""" Gerchik Scanner — Instrument filter + level scan loop. Scans top futures pairs, finds levels on 1H, detects patterns on 15m. """ import asyncio import logging from dataclasses import dataclass from typing import Callable, Optional from binance.client import Client from gerchik_config import ( GERCHIK_MIN_VOLUME_24H, GERCHIK_MIN_ATR_PCT, GERCHIK_TOP_N_PAIRS, GERCHIK_LEVEL_CANDLES, GERCHIK_SL_BUFFER_PCT, GERCHIK_TP1_RR, GERCHIK_TP2_RR, GERCHIK_TP3_RR, GERCHIK_MAX_SL_PCT, ) from gerchik_levels import find_levels from gerchik_models import detect_all_models, GerchikSignal logger = logging.getLogger(__name__) # Reuse from config.py from config import SKIP_TICKERS # Extra blacklist for Gerchik only (too volatile for level trading) GERCHIK_EXTRA_SKIP = {"NOM", "1000NOM"} def get_top_pairs(client: Client, top_n: int = GERCHIK_TOP_N_PAIRS) -> list[str]: """ Get top N USDT perpetual futures pairs by 24h volume. Excludes SKIP_TICKERS. """ try: tickers = client.futures_ticker() usdt_pairs = [] for t in tickers: symbol = t["symbol"] if not symbol.endswith("USDT"): continue # Extract base ticker base = symbol.replace("USDT", "") if base in SKIP_TICKERS or base in GERCHIK_EXTRA_SKIP: continue volume_usdt = float(t.get("quoteVolume", 0)) if volume_usdt < GERCHIK_MIN_VOLUME_24H: continue usdt_pairs.append({ "symbol": symbol, "volume": volume_usdt, "price": float(t.get("lastPrice", 0)), }) # Sort by volume descending usdt_pairs.sort(key=lambda x: x["volume"], reverse=True) result = [p["symbol"] for p in usdt_pairs[:top_n]] return result except Exception as e: logger.error(f"Failed to get top pairs: {e}") return [] def fetch_klines(client: Client, symbol: str, interval: str, limit: int) -> dict: """ Fetch OHLCV klines from Binance. Returns dict with lists: opens, highs, lows, closes, volumes. Uses completed candles only (drops last incomplete candle). """ try: raw = client.futures_klines(symbol=symbol, interval=interval, limit=limit + 1) # Drop last candle (incomplete) if len(raw) > 1: raw = raw[:-1] opens = [float(k[1]) for k in raw] highs = [float(k[2]) for k in raw] lows = [float(k[3]) for k in raw] closes = [float(k[4]) for k in raw] volumes = [float(k[5]) for k in raw] return { "opens": opens, "highs": highs, "lows": lows, "closes": closes, "volumes": volumes, } except Exception as e: logger.error(f"Failed to fetch klines {symbol} {interval}: {e}") return {"opens": [], "highs": [], "lows": [], "closes": [], "volumes": []} def calc_atr(highs: list[float], lows: list[float], closes: list[float], period: int = 14) -> float: """Calculate ATR as percentage of current price.""" if len(closes) < period + 1: return 0 trs = [] for i in range(1, len(closes)): tr = max( highs[i] - lows[i], abs(highs[i] - closes[i - 1]), abs(lows[i] - closes[i - 1]), ) trs.append(tr) if len(trs) < period: return 0 atr = sum(trs[-period:]) / period current_price = closes[-1] return (atr / current_price * 100) if current_price > 0 else 0 @dataclass class NearbyLevel: """A level that price is approaching — for watchlist/alerts.""" symbol: str level_price: float level_type: str level_strength: float level_touches: int distance_pct: float # How far price is from level (%) current_price: float side: str # Expected trade direction if bounce # Threshold: price within this % of a level = "nearby" NEARBY_THRESHOLD_PCT = 0.5 async def scan_for_gerchik_signals( client: Client, on_signal: Callable, skip_symbols: set, ) -> dict: """ Full scan: get pairs → fetch candles → find levels → detect patterns. Args: client: Binance client on_signal: async callback for each signal found skip_symbols: symbols to skip (already have positions) Returns: dict with scan results: signals_found, checked, nearby_levels, levels_total """ pairs = get_top_pairs(client) if not pairs: logger.warning("No pairs found for Gerchik scan") return {"signals_found": 0, "checked": 0, "nearby": [], "levels_total": 0} signals_found = 0 checked = 0 nearby_levels: list[NearbyLevel] = [] total_levels = 0 for symbol in pairs: if symbol in skip_symbols: continue try: # Fetch 1H candles for levels klines_1h = fetch_klines(client, symbol, "1h", GERCHIK_LEVEL_CANDLES) if len(klines_1h["closes"]) < 100: continue # Check ATR filter atr_pct = calc_atr(klines_1h["highs"], klines_1h["lows"], klines_1h["closes"]) if atr_pct < GERCHIK_MIN_ATR_PCT: continue checked += 1 current_price = klines_1h["closes"][-1] # Find levels on 1H levels = find_levels( klines_1h["opens"], klines_1h["highs"], klines_1h["lows"], klines_1h["closes"], current_price, ) if not levels: continue total_levels += len(levels) # Track nearby levels (price approaching strong level) for level in levels[:5]: dist_pct = abs(level.price - current_price) / current_price * 100 if dist_pct <= NEARBY_THRESHOLD_PCT and level.strength >= 40: side = "BUY" if level.price < current_price else "SELL" nearby_levels.append(NearbyLevel( symbol=symbol, level_price=level.price, level_type=level.level_type, level_strength=level.strength, level_touches=level.touches, distance_pct=round(dist_pct, 3), current_price=current_price, side=side, )) # Fetch 15m candles for pattern detection klines_15m = fetch_klines(client, symbol, "15m", 100) if len(klines_15m["closes"]) < 20: continue # Try each level for pattern match for level in levels[:5]: # Only check nearest 5 levels signal = detect_all_models( symbol=symbol, level=level, opens_15m=klines_15m["opens"], highs_15m=klines_15m["highs"], lows_15m=klines_15m["lows"], closes_15m=klines_15m["closes"], volumes_15m=klines_15m["volumes"], closes_1h=klines_1h["closes"], current_price=current_price, sl_buffer_pct=GERCHIK_SL_BUFFER_PCT, tp1_rr=GERCHIK_TP1_RR, tp2_rr=GERCHIK_TP2_RR, tp3_rr=GERCHIK_TP3_RR, max_sl_pct=GERCHIK_MAX_SL_PCT, ) if signal: await on_signal(signal) signals_found += 1 break # One signal per symbol max # Rate limit: don't hammer Binance await asyncio.sleep(0.1) except Exception as e: logger.error(f"Error scanning {symbol}: {e}", exc_info=True) continue # Sort nearby by strength (strongest first) nearby_levels.sort(key=lambda n: n.level_strength, reverse=True) logger.info( f"[gerchik_scanner] Scan: {checked} checked, " f"{total_levels} levels, {len(nearby_levels)} nearby, " f"{signals_found} signals" ) return { "signals_found": signals_found, "checked": checked, "nearby": nearby_levels[:10], # Top 10 nearest "levels_total": total_levels, } def scan_levels_for_symbol(client: Client, symbol: str) -> list: """ Find levels for a single symbol (for /levels command). Returns list of Level objects. """ klines = fetch_klines(client, symbol, "1h", GERCHIK_LEVEL_CANDLES) if len(klines["closes"]) < 100: return [] current_price = klines["closes"][-1] return find_levels( klines["opens"], klines["highs"], klines["lows"], klines["closes"], current_price, )