from __future__ import annotations import asyncio import logging import httpx from app.binance import fetch_klines, load_tick_sizes, to_binance_symbol from app.config import Settings from app.heryon import heryon_nonce, to_tv_perp_ticker from app.indicators.fvg import FvgParams, evaluate_fvg, last_fvg_signal from app.indicators.ltf import LtfParams, attach_forming_6h, evaluate_ltf, last_ltf_signal from app.pipeline import deliver_generated from app.state import ScannerStore from app.watchlist import Watchlist, load_watchlist logger = logging.getLogger(__name__) LTF_15M_BARS = 600 LTF_6H_BARS = 250 FVG_6H_BARS = 400 async def run_scanner(settings: Settings) -> None: store = ScannerStore(settings.scanner_state_path) try: await load_tick_sizes() while True: try: watchlist = load_watchlist(settings.watchlist_path) await scan_once(settings, watchlist, store) except asyncio.CancelledError: raise except Exception: logger.exception("Scanner cycle failed") await asyncio.sleep(settings.scanner_poll_seconds) finally: store.close() async def scan_once( settings: Settings, watchlist: Watchlist, store: ScannerStore, ) -> None: ltf_cfg = watchlist.strategies.get("ltf") fvg_cfg = watchlist.strategies.get("fvg") async with httpx.AsyncClient(timeout=20.0) as client: for symbol in watchlist.symbols: pair = to_binance_symbol(symbol) if ltf_cfg is not None and ltf_cfg.enabled: try: await _scan_ltf(settings, store, client, pair, ltf_cfg.telegram_thread_id) except Exception: logger.exception("LTF scan failed for %s", pair) if fvg_cfg is not None and fvg_cfg.enabled: try: await _scan_fvg(settings, store, client, pair, fvg_cfg.telegram_thread_id) except Exception: logger.exception("FVG scan failed for %s", pair) def _thread(settings: Settings, override: int | None) -> int: if override is not None and override >= 1: return override return settings.telegram_message_thread_id async def _scan_ltf( settings: Settings, store: ScannerStore, client: httpx.AsyncClient, symbol: str, thread_id: int | None, ) -> None: df_15m = await fetch_klines( symbol, "15m", limit=LTF_15M_BARS, closed_only=True, client=client ) if df_15m.empty: return bar_ts = int(df_15m.index[-1].timestamp()) last = store.get_last_bar("ltf", symbol) if last is None: store.set_last_bar("ltf", symbol, bar_ts) logger.info("LTF primed %s at %s (skip history)", symbol, bar_ts) return if bar_ts <= last: return df_6h_closed = await fetch_klines( symbol, "6h", limit=LTF_6H_BARS, closed_only=True, client=client ) if df_6h_closed.empty: logger.warning("LTF %s: empty 6h klines", symbol) return df_6h = attach_forming_6h(df_6h_closed, df_15m) params = LtfParams(risk_usd=settings.risk_usd) analyzed = await asyncio.to_thread(evaluate_ltf, df_15m, df_6h, params) signal = last_ltf_signal(analyzed, params) await _emit_if_new(settings, store, "ltf", symbol, bar_ts, signal, thread_id) async def _scan_fvg( settings: Settings, store: ScannerStore, client: httpx.AsyncClient, symbol: str, thread_id: int | None, ) -> None: df_6h = await fetch_klines( symbol, "6h", limit=FVG_6H_BARS, closed_only=True, client=client ) if df_6h.empty: return bar_ts = int(df_6h.index[-1].timestamp()) last = store.get_last_bar("fvg", symbol) if last is None: store.set_last_bar("fvg", symbol, bar_ts) logger.info("FVG primed %s at %s (skip history)", symbol, bar_ts) return if bar_ts <= last: return params = FvgParams(risk_usd=settings.risk_usd) analyzed = await asyncio.to_thread(evaluate_fvg, df_6h, params) signal = last_fvg_signal(analyzed, params) await _emit_if_new(settings, store, "fvg", symbol, bar_ts, signal, thread_id) async def _emit_if_new( settings: Settings, store: ScannerStore, strategy: str, symbol: str, bar_ts: int, signal, thread_id: int | None, ) -> None: if signal is None: store.set_last_bar(strategy, symbol, bar_ts) logger.debug("%s %s new bar %s — no signal", strategy, symbol, bar_ts) return ticker = to_tv_perp_ticker(symbol) nonce = heryon_nonce(ticker, signal.side, signal.bar_open_ts) if store.nonce_sent(nonce): store.set_last_bar(strategy, symbol, bar_ts) logger.info("Skip duplicate nonce %s", nonce) return logger.info( "Signal %s %s %s bar=%s sl=%s tp1=%s", strategy, symbol, signal.side, bar_ts, signal.sl, signal.tp1, ) await deliver_generated( settings, signal, symbol, _thread(settings, thread_id), store=store ) store.mark_nonce(nonce) store.set_last_bar(strategy, symbol, bar_ts)