trading/scanner_engine.py

from __future__ import annotations

import gc
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

from trading.market_registry import get_scan_results_path, to_provider_symbol
from trading.universe_provider import get_market_universe


def _default_backtest_end() -> str:
    """Return the most recent completed trading weekday as YYYY-MM-DD."""
    d = date.today() - timedelta(days=1)
    while d.weekday() >= 5:  # Saturday=5, Sunday=6
        d -= timedelta(days=1)
    return d.isoformat()


@dataclass(frozen=True)
class ScanConfig:
    market_id: str
    universe_name: str
    title: str
    symbols: Optional[List[str]] = None
    max_symbols: Optional[int] = None
    min_data_days: int = 252
    min_avg_volume: int = 500_000
    min_price: float = 1.0
    backtest_start: str = "2023-01-01"
    backtest_end: str = field(default_factory=_default_backtest_end)
    train_window: int = 252
    days_ahead: int = 7
    # --- Eligibility gates (tightened to resist overfit / selection bias) ---
    # Rationale: backtest dir_acc ~57% historically collapsed to 37% live, the
    # signature of an edge barely above coin-flip being amplified by picking the
    # best N of ~100 stocks. Stricter full-period gates + a recent out-of-sample
    # (OOS) holdout gate require the edge to PERSIST on unseen recent data.
    min_dir_acc: float = 0.55          # was 0.53 — must clear coin-flip noise band
    min_sharpe: float = 0.30           # was 0.0  — demand real risk-adjusted edge
    min_hit_rate: float = 45.0         # was 40.0
    min_trades: int = 8                # was 3   — kills 4-trade "100% hit" noise
    # OOS holdout gates (recent 30% of eval window the selection never optimised on)
    min_oos_dir_acc: float = 0.52      # recent edge must stay above coin-flip
    require_oos: bool = True           # if no holdout could be computed → exclude
    min_oos_return_pct: float = -2.0   # recent strategy return not deeply negative
    # Chunked / resumable scan support: if set, run_scan() processes at most
    # N additional Stage 2 stocks per call and then returns without marking
    # `completed=True`. Subsequent calls resume where it left off (work file
    # holds progress). Used by HF free-tier external cron pinger.
    max_stage2_per_run: Optional[int] = None

    @property
    def final_results_file(self) -> Path:
        return get_scan_results_path(self.market_id, completed=True)

    @property
    def work_results_file(self) -> Path:
        return get_scan_results_path(self.market_id, completed=False)

    @property
    def universe_key(self) -> str:
        parts = [self.market_id, self.universe_name]
        if self.symbols:
            parts.append("symbols=" + ",".join(self.symbols))
        if self.max_symbols:
            parts.append(f"max={self.max_symbols}")
        return "|".join(parts)


def _load_results(config: ScanConfig) -> Dict[str, Any]:
    src = None
    if config.work_results_file.exists():
        src = config.work_results_file
    elif config.final_results_file.exists():
        src = config.final_results_file

    if src is not None and src.exists():
        try:
            return json.loads(src.read_text())
        except Exception:
            pass
    return {
        "market_id": config.market_id,
        "scan_started": None,
        "universe": None,
        "universe_key": None,
        "stage1": {},
        "stage2": {},
        "completed": False,
    }


def _save_results(config: ScanConfig, data: Dict[str, Any]) -> None:
    config.work_results_file.parent.mkdir(parents=True, exist_ok=True)
    data["market_id"] = config.market_id
    data["updated_at"] = datetime.now(timezone.utc).isoformat()
    tmp = config.work_results_file.with_suffix(".tmp")
    tmp.write_text(json.dumps(data, indent=2, default=str))
    tmp.rename(config.work_results_file)


def _publish_final_if_complete(config: ScanConfig, data: Dict[str, Any]) -> None:
    if not data.get("completed", False):
        return
    config.final_results_file.parent.mkdir(parents=True, exist_ok=True)
    tmp = config.final_results_file.with_suffix(".tmp")
    tmp.write_text(json.dumps(data, indent=2, default=str))
    tmp.rename(config.final_results_file)


def stage1_prefilter(symbol: str, config: ScanConfig) -> Dict[str, Any]:
    import yfinance as yf

    ticker = to_provider_symbol(symbol, market_id=config.market_id)
    result: Dict[str, Any] = {
        "market_id": config.market_id,
        "symbol": symbol,
        "provider_symbol": ticker,
        "passed": False,
        "reason": "",
        "avg_volume": 0.0,
        "data_days": 0,
        "last_price": 0.0,
        "checked_at": datetime.now(timezone.utc).isoformat(),
    }

    try:
        df = yf.download(ticker, period="2y", progress=False, auto_adjust=True)
        if df is None or df.empty:
            result["reason"] = "no_data"
            return result

        data_days = len(df)
        result["data_days"] = data_days
        if data_days < config.min_data_days:
            result["reason"] = f"insufficient_data ({data_days} < {config.min_data_days} days)"
            return result

        avg_vol = 0.0
        if "Volume" in df.columns:
            avg_vol_raw = df["Volume"].mean()
            avg_vol = float(avg_vol_raw.iloc[0]) if hasattr(avg_vol_raw, "iloc") else float(avg_vol_raw)
        result["avg_volume"] = round(avg_vol, 0)
        if avg_vol < config.min_avg_volume:
            result["reason"] = f"low_volume ({avg_vol:,.0f} < {config.min_avg_volume:,.0f})"
            return result

        last_price = float(df["Close"].iloc[-1].iloc[0]) if hasattr(df["Close"].iloc[-1], "iloc") else float(df["Close"].iloc[-1])
        result["last_price"] = round(last_price, 2)
        if last_price < config.min_price:
            result["reason"] = f"below_min_price (price={last_price:.2f} < {config.min_price:.2f})"
            return result

        result["passed"] = True
        result["reason"] = "OK"
        return result
    except Exception as exc:
        result["reason"] = f"error: {exc}"
        return result


def stage2_backtest(symbol: str, config: ScanConfig, logger: logging.Logger) -> Dict[str, Any]:
    from analysis.walk_forward_backtest import walk_forward_backtest

    result: Dict[str, Any] = {
        "market_id": config.market_id,
        "symbol": symbol,
        "eligible": False,
        "reason": "",
        "evaluated_at": datetime.now(timezone.utc).isoformat(),
    }

    try:
        _, metrics = walk_forward_backtest(
            symbol=symbol,
            start_date=config.backtest_start,
            end_date=config.backtest_end,
            market_id=config.market_id,
            days_ahead=config.days_ahead,
            train_window=config.train_window,
            model_type="rf",
            use_technical_gate=True,
            initial_capital=100_000.0,
            commission_bps=10.0,
            slippage_bps=10.0,
            exit_rule="signal_or_fixed",
            max_hold_days=config.days_ahead,
            stop_loss_pct=0.05,
            take_profit_pct=0.10,
            trailing_stop_pct=0.07,
            max_position_pct=0.50,
            max_risk_per_trade_pct=0.02,
        )

        dir_acc = metrics["direction_accuracy"]
        sharpe = metrics["sharpe"]
        hit_rate = metrics.get("hit_rate_pct", 0.0)
        total_ret = metrics["total_return_pct"]
        trades = metrics["trades_count"]
        max_dd = metrics.get("max_drawdown_pct", 0.0)
        oos_dir_acc = metrics.get("oos_direction_accuracy")
        oos_samples = metrics.get("oos_samples", 0)
        oos_ret = metrics.get("oos_return_pct")

        result.update({
            "dir_acc": round(dir_acc, 4),
            "sharpe": round(sharpe, 3),
            "hit_rate": round(hit_rate, 1),
            "total_return_pct": round(total_ret, 1),
            "trades": trades,
            "max_drawdown_pct": round(max_dd, 1),
            "oos_dir_acc": round(oos_dir_acc, 4) if oos_dir_acc is not None else None,
            "oos_samples": oos_samples,
            "oos_return_pct": round(oos_ret, 1) if oos_ret is not None else None,
        })

        reasons: List[str] = []
        # --- Full-period gates ---
        if dir_acc < config.min_dir_acc:
            reasons.append(f"dir_acc={dir_acc:.1%}<{config.min_dir_acc:.0%}")
        if sharpe < config.min_sharpe:
            reasons.append(f"sharpe={sharpe:.3f}<{config.min_sharpe}")
        if hit_rate < config.min_hit_rate:
            reasons.append(f"hit_rate={hit_rate:.1f}%<{config.min_hit_rate}%")
        if trades < config.min_trades:
            reasons.append(f"trades={trades}<{config.min_trades}")
        # --- Out-of-sample holdout gates (the real overfit defense) ---
        if oos_dir_acc is None:
            if config.require_oos:
                reasons.append("no_oos_holdout(insufficient_recent_samples)")
        else:
            if oos_dir_acc < config.min_oos_dir_acc:
                reasons.append(f"oos_dir_acc={oos_dir_acc:.1%}<{config.min_oos_dir_acc:.0%}")
            if oos_ret is not None and oos_ret < config.min_oos_return_pct:
                reasons.append(f"oos_ret={oos_ret:.1f}%<{config.min_oos_return_pct}%")

        result["eligible"] = len(reasons) == 0
        result["reason"] = "; ".join(reasons) if reasons else "OK"
        return result
    except Exception as exc:
        result["reason"] = f"backtest_error: {exc}"
        logger.warning("Stage 2 failed for %s: %s", symbol, exc)
        return result


def get_universe(config: ScanConfig, logger: logging.Logger) -> List[str]:
    if config.symbols:
        symbols = list(dict.fromkeys(config.symbols))
        if config.max_symbols:
            symbols = symbols[: config.max_symbols]
        logger.info("Using explicit symbol subset for %s/%s: %s", config.market_id, config.universe_name, ", ".join(symbols))
        return symbols
    try:
        result = get_market_universe(config.market_id, config.universe_name)
        symbols = result.symbols
        if config.max_symbols:
            symbols = symbols[: config.max_symbols]
            logger.info(
                "Fetched %d stocks from %s/%s, limited to first %d symbols",
                len(result.symbols), result.market_id, result.name, len(symbols),
            )
        else:
            logger.info("Fetched %d stocks from %s/%s", len(result.symbols), result.market_id, result.name)
        return symbols
    except Exception as exc:
        logger.error("Failed to fetch universe %s/%s: %s", config.market_id, config.universe_name, exc)
        if config.market_id == "bist":
            return [
                "THYAO", "AKBNK", "GARAN", "EREGL", "SISE",
                "TUPRS", "KCHOL", "ASELS", "BIMAS", "SAHOL",
                "YKBNK", "HALKB", "VAKBN", "TCELL", "ARCLK",
                "PETKM", "TOASO", "KOZAA", "KOZAL", "SASA",
                "TAVHL", "TTKOM", "ENKAI", "FROTO", "EKGYO",
                "PGSUS", "SOKM", "DOHOL", "GUBRF", "ISCTR",
            ]
        if config.market_id == "us":
            return [
                "AAPL", "MSFT", "AMZN", "GOOGL", "META",
                "NVDA", "TSLA", "BRK-B", "JPM", "JNJ",
                "V", "UNH", "PG", "HD", "MA",
                "DIS", "BAC", "XOM", "PFE", "CSCO",
                "ABBV", "AVGO", "KO", "PEP", "TMO",
                "COST", "MRK", "WMT", "ABT", "CRM",
            ]
        return []


def run_scan(config: ScanConfig, logger: logging.Logger, *, force: bool = False, stage1_only: bool = False) -> Dict[str, Any]:
    symbols = get_universe(config, logger)
    if not symbols:
        logger.error("No symbols found for universe: %s", config.universe_name)
        return {}

    data = _load_results(config)
    existing_key = data.get("universe_key") or data.get("universe")
    is_same_universe = existing_key == config.universe_key
    if not is_same_universe or force:
        logger.info("Starting fresh scan for %s/%s (%d stocks)", config.market_id, config.universe_name, len(symbols))
        data = {
            "market_id": config.market_id,
            "scan_started": datetime.now(timezone.utc).isoformat(),
            "universe": config.universe_name,
            "universe_key": config.universe_key,
            "total_stocks": len(symbols),
            "stage1": {},
            "stage2": {},
            "completed": False,
        }
        _save_results(config, data)
    else:
        s1_done = len(data.get("stage1", {}))
        s2_done = len(data.get("stage2", {}))
        logger.info("Resuming scan: %d/%d Stage1, %d Stage2 done", s1_done, len(symbols), s2_done)

    if is_same_universe and not force and data.get("completed"):
        s1_done_map = data.get("stage1", {}) or {}
        s2_done_map = data.get("stage2", {}) or {}
        s1_passed_done = [s for s, v in s1_done_map.items() if isinstance(v, dict) and v.get("passed")]
        if s1_passed_done and len(s2_done_map) >= len(s1_passed_done):
            logger.info(
                "Scan already completed for %s/%s; returning existing results without updating timestamps",
                config.market_id,
                config.universe_name,
            )
            return data

    logger.info("=" * 60)
    logger.info("STAGE 1: Liquidity & Data Quality Pre-Filter")
    logger.info("=" * 60)

    s1_todo = [s for s in symbols if s not in data.get("stage1", {})]
    total_s1 = len(symbols)
    done_s1 = total_s1 - len(s1_todo)

    for i, sym in enumerate(s1_todo, start=done_s1 + 1):
        t0 = time.time()
        result = stage1_prefilter(sym, config)
        elapsed = time.time() - t0
        status = "PASS" if result["passed"] else f"FAIL ({result['reason']})"
        logger.info("[Stage1 %d/%d] %s: %s (%.1fs)", i, total_s1, sym, status, elapsed)
        data.setdefault("stage1", {})[sym] = result
        _save_results(config, data)
        gc.collect()

    s1_passed = [s for s, v in data["stage1"].items() if v.get("passed")]
    s1_failed = [s for s, v in data["stage1"].items() if not v.get("passed")]
    logger.info("")
    logger.info("Stage 1 Results: %d PASS / %d FAIL out of %d", len(s1_passed), len(s1_failed), total_s1)
    logger.info("Passed: %s", ", ".join(sorted(s1_passed)))
    logger.info("")

    if stage1_only:
        data["completed"] = True
        _save_results(config, data)
        _publish_final_if_complete(config, data)
        return data

    logger.info("=" * 60)
    logger.info("STAGE 2: ML Walk-Forward Backtest")
    logger.info("=" * 60)

    s2_todo = [s for s in s1_passed if s not in data.get("stage2", {})]
    total_s2 = len(s1_passed)
    done_s2 = total_s2 - len(s2_todo)
    times_s2: List[float] = []
    chunk_limit = config.max_stage2_per_run
    processed_this_run = 0

    for i, sym in enumerate(s2_todo, start=done_s2 + 1):
        t0 = time.time()
        result = stage2_backtest(sym, config, logger)
        elapsed = time.time() - t0
        times_s2.append(elapsed)

        if result["eligible"]:
            status = (
                f"ELIGIBLE (ret={result.get('total_return_pct', 0):.1f}%, "
                f"sharpe={result.get('sharpe', 0):.3f}, "
                f"hit={result.get('hit_rate', 0):.0f}%)"
            )
        else:
            status = f"EXCLUDED ({result['reason']})"

        avg_time = sum(times_s2) / len(times_s2)
        remaining = total_s2 - i
        eta_min = (remaining * avg_time) / 60
        logger.info("[Stage2 %d/%d] %s: %s (%.1fs, ETA: %.0f min)", i, total_s2, sym, status, elapsed, eta_min)
        data.setdefault("stage2", {})[sym] = result
        _save_results(config, data)
        gc.collect()

        processed_this_run += 1
        if chunk_limit and processed_this_run >= chunk_limit and i < total_s2:
            data["chunk_paused_at"] = datetime.now(timezone.utc).isoformat()
            data["chunk_last_index"] = i
            _save_results(config, data)
            logger.info(
                "Chunk limit reached: processed %d Stage 2 stocks this run "
                "(%d/%d total). Pausing — resume on next call.",
                processed_this_run, i, total_s2,
            )
            return data  # NOT completed=True; next call resumes

    data["completed"] = True
    data["scan_finished"] = datetime.now(timezone.utc).isoformat()
    _save_results(config, data)
    _publish_final_if_complete(config, data)
    print_report(config, logger, data)
    return data


def print_report(config: ScanConfig, logger: logging.Logger, data: Optional[Dict[str, Any]] = None) -> None:
    if data is None:
        data = _load_results(config)

    s2 = data.get("stage2", {})
    if not s2:
        logger.info("No Stage 2 results found. Run scan first.")
        return

    eligible = {s: v for s, v in s2.items() if v.get("eligible")}
    excluded = {s: v for s, v in s2.items() if not v.get("eligible")}

    print("\n" + "=" * 70)
    print(f"  {config.title} SCAN RESULTS — {data.get('universe', '?').upper()}")
    print("=" * 70)
    s1 = data.get("stage1", {})
    s1_pass = sum(1 for v in s1.values() if v.get("passed"))
    s1_fail = sum(1 for v in s1.values() if not v.get("passed"))
    print(f"\nStage 1 (Liquidity Filter):  {s1_pass} pass / {s1_fail} fail / {len(s1)} total")
    print(f"Stage 2 (ML Backtest):       {len(eligible)} eligible / {len(excluded)} excluded / {len(s2)} tested")

    if eligible:
        print(f"\n{'─' * 70}")
        print(f"  ELIGIBLE STOCKS ({len(eligible)})")
        print(f"{'─' * 70}")
        print(f"  {'Symbol':<10} {'Return%':>9} {'Sharpe':>8} {'DirAcc':>7} {'OOSAcc':>7} {'OOSRet%':>8} {'Trades':>7}")
        print(f"  {'─'*10} {'─'*9} {'─'*8} {'─'*7} {'─'*7} {'─'*8} {'─'*7}")
        sorted_eligible = sorted(eligible.items(), key=lambda kv: kv[1].get("sharpe", -999), reverse=True)
        for sym, metrics in sorted_eligible:
            _oa = metrics.get("oos_dir_acc")
            _or = metrics.get("oos_return_pct")
            print(
                f"  {sym:<10} {metrics.get('total_return_pct', 0):>+8.1f}% "
                f"{metrics.get('sharpe', 0):>8.3f} "
                f"{metrics.get('dir_acc', 0)*100:>6.1f}% "
                f"{(_oa*100 if _oa is not None else 0):>6.1f}% "
                f"{(_or if _or is not None else 0):>+7.1f}% "
                f"{metrics.get('trades', 0):>7d}"
            )

    if excluded:
        print(f"\n{'─' * 70}")
        print(f"  EXCLUDED STOCKS ({len(excluded)})")
        print(f"{'─' * 70}")
        for sym, metrics in sorted(excluded.items()):
            print(f"  {sym:<10} {metrics.get('reason', 'unknown')}")