run_live.py

import json
import sys
import time
import threading
from pathlib import Path
import pandas as pd

sys.path.append(str(Path(__file__).resolve().parents[1]))

from src.config.loader import load_config
from src.core.logger import setup_logging
from src.core.timeutils import now_utc
from src.infra.exchange_binance import BinanceClient
from src.infra.ws_binance import BinanceWsClient
from src.infra.telegram_client import TelegramClient
from src.infra.storage import write_healthcheck_db, init_live_db
from src.ml.model_xgb import load_model
from src.ml.features import load_feature_list
from src.live.scanner import scan_once, scan_once_cached, _load_symbols, _prepare_1h
from src.live.state_machine import TradeStateMachine
from src.live.monitor import monitor_trades, monitor_trades_ws
from src.core.timeutils import sleep_until_next_close, timeframe_to_seconds
from src.structure.pivots import find_swings, detect_structure


def main():
    logger = setup_logging("live")
    cfg = load_config()
    exchange = BinanceClient(
        cfg.exchange.base_url,
        cfg.exchange.timeout_sec,
        cfg.exchange.max_retries,
        cfg.exchange.backoff_sec,
        cfg.exchange.min_interval_ms,
    )
    ws_kline_client = BinanceWsClient(cfg.exchange.ws_url, on_log=logger.info)
    ws_btc_client = BinanceWsClient(cfg.exchange.ws_url, on_log=logger.info)
    ws_book_client = BinanceWsClient(cfg.exchange.ws_url, on_log=logger.info)
    telegram = TelegramClient(
        cfg.telegram.enabled,
        cfg.telegram.bot_token,
        cfg.telegram.chat_id,
        cfg.telegram.timeout_sec,
        cfg.telegram.max_retries,
        cfg.telegram.backoff_sec,
    )
    events_db = cfg.database.name
    init_live_db(events_db, prefix="live")
    state_machine = TradeStateMachine(
        telegram=telegram,
        events_db=events_db,
        entry_expiry_hours=cfg.filters.entry_expiry_hours,
        db_prefix="live",
    )
    model_entry_path = Path("data/models/model_entry.joblib")
    entry_model_path = model_entry_path if model_entry_path.exists() else Path("data/models/model.joblib")
    entry_model = load_model(str(entry_model_path))
    logger.info("entry model: %s", entry_model_path)
    strategy_models = {}
    strategy_thresholds = {}

    strategy_cfg_map = {
        "reclaim": cfg.strategies.reclaim,
        "breakout": cfg.strategies.breakout,
        "trend_pullback": cfg.strategies.trend_pullback,
        "mean_reversion": cfg.strategies.mean_reversion,
        "liquidity_sweep": cfg.strategies.liquidity_sweep,
        "volatility_expansion": cfg.strategies.volatility_expansion,
    }
    for strategy_name, strategy_cfg in strategy_cfg_map.items():
        threshold_cfg = float(getattr(strategy_cfg, "threshold", cfg.ml.threshold))
        model_path = Path(getattr(strategy_cfg, "model_path", "") or "")
        meta_path = model_path.parent / "meta_entry.json"
        if meta_path.exists():
            try:
                meta_data = json.loads(meta_path.read_text(encoding="utf-8"))
                threshold_cfg = float(meta_data.get("threshold", threshold_cfg))
            except Exception as exc:
                logger.warning("failed read strategy meta (%s): %s", strategy_name, exc)
        strategy_thresholds[strategy_name] = threshold_cfg
        if model_path.exists():
            try:
                strategy_models[strategy_name] = load_model(str(model_path))
                logger.info("strategy model (%s): %s", strategy_name, model_path)
            except Exception as exc:
                logger.warning("failed load strategy model (%s): %s", strategy_name, exc)
                strategy_models[strategy_name] = entry_model
        else:
            logger.warning("strategy model missing (%s), fallback to default model: %s", strategy_name, model_path)
            strategy_models[strategy_name] = entry_model
    tp_model = None
    tp_model_path = Path("data/models/model_tp.joblib")
    if tp_model_path.exists():
        tp_model = load_model(str(tp_model_path))
        logger.info("tp model: %s", tp_model_path)
    meta_path = Path("data/models/meta_entry.json")
    legacy_meta_path = Path("data/models/meta.json")
    if meta_path.exists():
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
    elif legacy_meta_path.exists():
        meta = json.loads(legacy_meta_path.read_text(encoding="utf-8"))
    else:
        meta = {}
    entry_threshold = cfg.ml.threshold
    entry_feature_order = meta.get("feature_order") or load_feature_list()
    logger.info(
        "live config: entry_tf=%s trend_tf=%s symbols=%d threshold=%.4f features=%d",
        cfg.timeframes.entry_tf,
        cfg.timeframes.trend_tf,
        len(_load_symbols(cfg.universe.symbols_file)),
        float(entry_threshold),
        len(entry_feature_order),
    )
    tp_meta_path = Path("data/models/meta_tp.json")
    if tp_meta_path.exists():
        tp_meta = json.loads(tp_meta_path.read_text(encoding="utf-8"))
        tp_feature_order = tp_meta.get("feature_order") or entry_feature_order
    else:
        tp_feature_order = entry_feature_order

    symbols = _load_symbols(cfg.universe.symbols_file)
    kline_symbols = [s for s in symbols if s != "BTCUSDT"]
    cache_15m = {}
    cache_1h = {}
    btc_15m_cache = []
    btc_1h_cache = []
    live_cache_path = Path("data/processed/live_cache.json")

    seed_entry_limit = int(cfg.live.entry_limit)
    seed_trend_limit = int(cfg.live.trend_limit)
    cache_entry_limit = int(cfg.live.entry_limit)
    cache_trend_limit = int(cfg.live.trend_limit)

    def _seed_cache():
        for symbol in symbols:
            df_15m = exchange.fetch_ohlcv(symbol, cfg.timeframes.entry_tf, limit=seed_entry_limit)
            df_1h = exchange.fetch_ohlcv(symbol, cfg.timeframes.trend_tf, limit=seed_trend_limit)
            cache_15m[symbol] = df_15m.to_dict("records")
            cache_1h[symbol] = df_1h.to_dict("records")
        btc = exchange.fetch_btc_ohlcv(cfg.timeframes.entry_tf, limit=seed_entry_limit)
        btc_15m_cache.extend(btc.to_dict("records"))
        btc_trend = exchange.fetch_btc_ohlcv(cfg.timeframes.trend_tf, limit=seed_trend_limit)
        btc_1h_cache.extend(btc_trend.to_dict("records"))
        logger.info(
            "seed cache: symbols=%d entry_limit=%d trend_limit=%d btc_entry=%d btc_trend=%d",
            len(symbols),
            seed_entry_limit,
            seed_trend_limit,
            len(btc_15m_cache),
            len(btc_1h_cache),
        )

    def _write_realtime_cache():
        payload = {"updated_ts": int(now_utc().timestamp() * 1000), "symbols": {}}
        entry_tf = cfg.timeframes.entry_tf
        trend_tf = cfg.timeframes.trend_tf
        for symbol in symbols:
            if symbol == "BTCUSDT":
                payload["symbols"][symbol] = {
                    entry_tf: list(btc_15m_cache),
                    trend_tf: list(btc_1h_cache),
                }
            else:
                payload["symbols"][symbol] = {
                    entry_tf: list(cache_15m.get(symbol, [])),
                    trend_tf: list(cache_1h.get(symbol, [])),
                }
        live_cache_path.parent.mkdir(parents=True, exist_ok=True)
        live_cache_path.write_text(json.dumps(payload, ensure_ascii=True), encoding="utf-8")

    def _log_gap(symbol: str, interval: str, last_ts: int | None, new_ts: int, expected_ms: int):
        if last_ts is None:
            return
        gap_ms = new_ts - last_ts
        if gap_ms > int(expected_ms * 1.5):
            logger.warning(
                "gap detected: symbol=%s interval=%s last_ts=%d new_ts=%d gap_ms=%d expected_ms=%d",
                symbol,
                interval,
                last_ts,
                new_ts,
                gap_ms,
                expected_ms,
            )

    entry_ms = timeframe_to_seconds(cfg.timeframes.entry_tf) * 1000
    trend_ms = timeframe_to_seconds(cfg.timeframes.trend_tf) * 1000

    def _update_cache(cache: dict, symbol: str, kline: dict, interval: str, expected_ms: int):
        entry = {
            "ts": int(kline.get("t")),
            "open": float(kline.get("o")),
            "high": float(kline.get("h")),
            "low": float(kline.get("l")),
            "close": float(kline.get("c")),
            "volume": float(kline.get("v")),
        }
        series = cache.setdefault(symbol, [])
        last_ts = series[-1]["ts"] if series else None
        _log_gap(symbol, interval, last_ts, entry["ts"], expected_ms)
        if not series or series[-1]["ts"] != entry["ts"]:
            series.append(entry)
        else:
            series[-1] = entry
        if len(series) > cache_entry_limit:
            del series[0]

    def _update_cache_list(series: list, symbol: str, kline: dict, interval: str, expected_ms: int):
        entry = {
            "ts": int(kline.get("t")),
            "open": float(kline.get("o")),
            "high": float(kline.get("h")),
            "low": float(kline.get("l")),
            "close": float(kline.get("c")),
            "volume": float(kline.get("v")),
        }
        last_ts = series[-1]["ts"] if series else None
        _log_gap(symbol, interval, last_ts, entry["ts"], expected_ms)
        if not series or series[-1]["ts"] != entry["ts"]:
            series.append(entry)
        else:
            series[-1] = entry
        if len(series) > cache_entry_limit:
            del series[0]

    _seed_cache()
    _write_realtime_cache()

    cooldowns = {}
    last_scan_ts = 0
    monitor_thread = None

    def _start_monitor():
        nonlocal monitor_thread
        if monitor_thread and monitor_thread.is_alive():
            return
        monitor_thread = threading.Thread(
            target=monitor_trades_ws,
            args=(state_machine, ws_book_client),
            kwargs={"max_minutes": None},
            daemon=True,
        )
        monitor_thread.start()

    def _on_kline(symbol: str, interval: str, k: dict):
        nonlocal last_scan_ts
        is_closed = bool(k.get("x"))
        if interval == cfg.timeframes.entry_tf:
            if symbol == "BTCUSDT":
                _update_cache_list(btc_15m_cache, symbol, k, interval, entry_ms)
            else:
                _update_cache(cache_15m, symbol, k, interval, entry_ms)
            if is_closed:
                close_ts = int(k.get("T", 0))
                if close_ts and close_ts != last_scan_ts:
                    last_scan_ts = close_ts
                    open_symbols = {
                        info["signal"]["candidate"].symbol
                        for info in state_machine.trades.values()
                        if info["state"].value not in ["closed", "expired"]
                    }
                    signals = scan_once_cached(
                        cfg,
                        symbols,
                        cache_15m,
                        cache_1h,
                        btc_15m_cache,
                        btc_1h_cache,
                        entry_model,
                        tp_model,
                        entry_feature_order,
                        tp_feature_order,
                        entry_threshold,
                        telegram,
                        cooldowns,
                        open_symbols,
                        strategy_models,
                        strategy_thresholds,
                    )
                    for signal in signals:
                        symbol_sig = signal["candidate"].symbol
                        direction_sig = signal["candidate"].direction.value
                        if state_machine.has_open_trade(symbol_sig):
                            if state_machine.cancel_new_if_opposite(symbol_sig, direction_sig):
                                state_machine.on_new_signal(signal["trade_id"], signal)
                            continue
                        state_machine.on_new_signal(signal["trade_id"], signal)
                    logger.info("scan complete: close_ts=%d signals=%d", close_ts, len(signals))
                    if signals:
                        _start_monitor()
                    _write_realtime_cache()
        if interval == cfg.timeframes.trend_tf:
            if symbol == "BTCUSDT":
                _update_cache_list(btc_1h_cache, symbol, k, interval, trend_ms)
                if len(btc_1h_cache) > cache_trend_limit:
                    del btc_1h_cache[0]
                return
            _update_cache(cache_1h, symbol, k, interval, trend_ms)
            if len(cache_1h.get(symbol, [])) > cache_trend_limit:
                del cache_1h[symbol][0]
            if is_closed:
                df_1h = pd.DataFrame(cache_1h.get(symbol, []))
                if not df_1h.empty:
                    df_1h = _prepare_1h(df_1h, cfg)
                    latest = df_1h.iloc[-1]
                    trend_row = {
                        "ema50_1h": float(latest["ema50_1h"]),
                        "ema200_1h": float(latest["ema200_1h"]),
                        "adx_1h": float(latest["adx_1h"]),
                    }
                    swings = find_swings(df_1h)
                    struct = detect_structure(swings) if swings else None
                    state_machine.on_trend_update(symbol, trend_row, struct)

    while True:
        try:
            ws_kline_client.start_klines(kline_symbols, [cfg.timeframes.entry_tf, cfg.timeframes.trend_tf], _on_kline)
            ws_btc_client.start_klines(["BTCUSDT"], [cfg.timeframes.entry_tf, cfg.timeframes.trend_tf], _on_kline)
            while True:
                open_trades = sum(
                    1
                    for info in state_machine.trades.values()
                    if getattr(info.get("state"), "value", info.get("state")) not in ["closed", "expired"]
                )
                write_healthcheck_db(
                    events_db,
                    {"ts": int(now_utc().timestamp() * 1000), "open_trades": open_trades},
                    prefix="live",
                )
                time.sleep(5)
        except Exception as exc:
            logger.exception("live loop error: %s", exc)
            ws_kline_client.stop()
            ws_btc_client.stop()
            ws_book_client.stop()
            monitor_trades(state_machine, exchange, poll_seconds=30, max_minutes=None)
        sleep_until_next_close(cfg.timeframes.entry_tf)


if __name__ == "__main__":
    main()