fithia2/apps/paper_trader/backtest_sim.py

"""Paper-trading backtest simulator.

Refactored to use the SAME BacktestRunner + SnapshotStore as
apps/backtester/run.py. This guarantees identical scoring, engine
matching, and position sizing between `fithia2 paper backtest` and
the research backtester.

Previous implementation used PaperTradingEngine + EventDetector +
MockBroker which had different scoring functions, feature computation,
and data sources — causing divergent results.
"""
from __future__ import annotations

import datetime as dt
import math
import statistics
import tempfile
from pathlib import Path
from typing import Any

from libs.backtest.snapshots import (
    resolve_snapshot,
    resolve_snapshot_path as _resolve_registry_snapshot_path,
)
from libs.common.config import get_settings
from libs.common.logging import get_logger

logger = get_logger(__name__)


def run_backtest_session_sync(
    session_name: str,
    config_path: str,
    initial_equity: float,
    start_date: dt.date,
    end_date: dt.date,
    parking_preset: str | None = None,
    idle_alpha_preset: str | None = None,
    form4_sleeve_preset: str | None = None,
    ownership_sleeve_preset: str | None = None,
    risk_off_alpha_sleeve_preset: str | None = None,
    non_core_allocator_v2_mode: str | None = None,
    snapshot_id_override: str | None = None,
    fixed_capital_sizing: bool = False,
    daily_budget_reset: bool = False,
) -> dict[str, Any]:
    """Run a single strategy using BacktestRunner (same as research backtester).

    Uses the existing Parquet snapshot + BacktestRunner pipeline so results
    match `python -m apps.backtester.run --manifest <config>` exactly.
    """
    from apps.backtester.run import (
        BacktestRunner,
        _build_merged_snapshot_store,
        _compute_max_effective_mhd,
        _extend_store_to_requested_window,
        load_manifest,
        resolve_config,
    )

    manifest = load_manifest(config_path)
    config = resolve_config(manifest, snapshot_id_override=snapshot_id_override)

    # Auto-refresh snapshot if stale (covers web direct_runner path).
    # Uses _last_market_closed_date() as the reference so the snapshot updates
    # once market closes today (same criteria as price bar extension).
    import asyncio as _asyncio
    from apps.backtester.run import _last_market_closed_date
    _snap_id = config.canonical_snapshot_id or config.dataset_snapshot_id
    if _snapshot_needs_refresh(_snap_id, _last_market_closed_date()):
        _asyncio.run(_refresh_snapshot(_snap_id, universe_profile=None))

    # Apply parking preset override (CLI --parking option)
    if parking_preset:
        config.risk.cash_parking_preset = parking_preset
        config.risk.apply_parking_preset()
    if idle_alpha_preset:
        config.idle_alpha_sleeve_preset = idle_alpha_preset
        config.apply_idle_alpha_sleeve_preset()
    if form4_sleeve_preset:
        config.form4_capture_sleeve_preset = form4_sleeve_preset
        config.apply_form4_capture_sleeve_preset()
    if ownership_sleeve_preset:
        config.ownership_capture_sleeve_preset = ownership_sleeve_preset
        config.apply_ownership_capture_sleeve_preset()
    if risk_off_alpha_sleeve_preset:
        config.risk_off_alpha_sleeve_preset = risk_off_alpha_sleeve_preset
        config.apply_risk_off_alpha_sleeve_preset()
    if non_core_allocator_v2_mode:
        config.non_core_allocator_v2.enabled = True
        config.non_core_allocator_v2.mode = non_core_allocator_v2_mode
    config.risk.fixed_capital_sizing = fixed_capital_sizing
    config.risk.daily_budget_reset = daily_budget_reset

    # Use merged store (train+valid+test) to cover the full date range.
    store = _build_merged_snapshot_store(manifest, config, snapshot_dir_override=None)

    # Slice to requested date range.
    # When lookback_entry_enabled, extend slice start backward so pre-start
    # events survive and BacktestRunner._collect_lookback_candidates() can find them.
    if config.execution.lookback_entry_enabled:
        max_mhd = _compute_max_effective_mhd(config)
        import datetime as _dt
        lookback_start = start_date - _dt.timedelta(days=max_mhd * 2)
        store = store.slice_by_date_range(lookback_start, end_date)
    else:
        store = store.slice_by_date_range(start_date, end_date)
    store = _extend_store_to_requested_window(
        store=store,
        config=config,
        start_date=start_date,
        end_date=end_date,
        snapshot_dir_override=None,
    )

    runner = BacktestRunner(
        manifest=manifest,
        config=config,
        store=store,
        initial_equity=initial_equity,
        split_name="paper_backtest",
    )

    # Run with temporary output directory
    tmp_dir = tempfile.mkdtemp(prefix="paper_bt_")
    try:
        result = runner.run(output_root=tmp_dir)
    except Exception:
        result = runner.run(output_root=None)

    # Convert to paper backtest format
    return _convert_from_runner(session_name, config_path, initial_equity, result, tmp_dir)


def _convert_from_runner(
    session_name: str,
    config_path: str,
    initial_equity: float,
    result: Any,
    tmp_dir: str,
) -> dict[str, Any]:
    """Convert BacktestRunner result to paper backtest output format."""
    equity_curve: list[dict] = []
    trades: list[dict] = []

    # Load from Parquet artifacts
    try:
        import pyarrow.parquet as pq
        import glob
        import os

        run_dirs = sorted(glob.glob(os.path.join(tmp_dir, "bt_*")))
        if run_dirs:
            run_dir = Path(run_dirs[0])

            # Equity curve
            eq_path = run_dir / "artifacts" / "daily_equity_curve.parquet"
            if eq_path.exists():
                eq_df = pq.read_table(str(eq_path)).to_pandas()
                for _, row in eq_df.iterrows():
                    d = row.get("date")
                    if isinstance(d, str):
                        d = dt.date.fromisoformat(d[:10])
                    equity_curve.append({
                        "date": d,
                        "equity": float(row.get("equity", initial_equity)),
                    })

            # Trade blotter
            bl_path = run_dir / "artifacts" / "trade_blotter.parquet"
            if bl_path.exists():
                bl_df = pq.read_table(str(bl_path)).to_pandas()
                for _, row in bl_df.iterrows():
                    entry_px = row.get("entry_price")
                    exit_px = row.get("exit_price")
                    shares = int(row.get("shares", 0))
                    pnl_pct = float(row.get("pnl_pct", 0.0))
                    pnl_dollar = pnl_pct * float(entry_px or 0) * shares if entry_px else 0.0

                    trade = {
                        "symbol": str(row.get("symbol", "")),
                        "entry_date": str(row.get("entry_date", "-")),
                        "exit_date": str(row.get("exit_date", "-")),
                        "entry_price": float(entry_px) if entry_px is not None else None,
                        "exit_price": float(exit_px) if exit_px is not None else None,
                        "shares": shares,
                        "pnl": pnl_dollar,
                        "reason": str(row.get("exit_reason", "-")),
                        "event_type": str(row.get("event_type", "-")),
                        "score": float(row.get("score", 0.0)),
                        "engine_id": str(row.get("engine_id", "")),
                        "trade_sleeve": str(row.get("trade_sleeve", "") or ""),
                    }
                    # Skip same-day KILL_SWITCH — backtest period end artifact
                    if trade["entry_date"] == trade["exit_date"] and trade["reason"] == "KILL_SWITCH":
                        continue
                    trades.append(trade)
    except Exception as exc:
        logger.warning("backtest_sim_artifact_load_failed", error=str(exc))

    # Compute summary stats
    final_equity = equity_curve[-1]["equity"] if equity_curve else initial_equity
    total_return_pct = (final_equity - initial_equity) / initial_equity * 100

    pnls = [t["pnl"] for t in trades]
    wins = [p for p in pnls if p > 0]
    win_rate = len(wins) / len(pnls) * 100 if pnls else 0.0

    equities = [r["equity"] for r in equity_curve]
    daily_returns = [
        (equities[i] - equities[i - 1]) / equities[i - 1]
        for i in range(1, len(equities))
        if equities[i - 1] > 0
    ]
    if len(daily_returns) >= 2:
        mean_r = statistics.mean(daily_returns)
        std_r = statistics.stdev(daily_returns)
        sharpe = (mean_r / std_r) * math.sqrt(252) if std_r > 0 else 0.0
    else:
        sharpe = 0.0

    peak = initial_equity
    max_dd_pct = 0.0
    for eq in equities:
        if eq > peak:
            peak = eq
        dd = (peak - eq) / peak * 100 if peak > 0 else 0.0
        if dd > max_dd_pct:
            max_dd_pct = dd

    # Extract full MetricsBundle if available
    metrics_bundle: dict = {}
    try:
        if hasattr(result, "metrics") and result.metrics is not None:
            metrics_bundle = result.metrics.model_dump()
    except Exception:
        pass

    # Snapshot coverage info from the canonical snapshot manifest
    snapshot_coverage_end_date: str | None = None
    snapshot_last_refresh_utc: str | None = None
    try:
        import json as _json
        from pathlib import Path as _Path
        _snap_manifest = _Path("data/parquet") / (config.canonical_snapshot_id or config.dataset_snapshot_id) / "manifest.json"
        if _snap_manifest.exists():
            _snap_meta = _json.loads(_snap_manifest.read_text())
            snapshot_coverage_end_date = _snap_meta.get("coverage_end_date")
            snapshot_last_refresh_utc = _snap_meta.get("last_refresh_utc")
    except Exception:
        pass

    return {
        "session_name": session_name,
        "config_path": config_path,
        "initial_equity": initial_equity,
        "equity_curve": equity_curve,
        "trades": trades,
        "all_entries": [],
        "all_exits": [],
        "summary": {
            "return_pct": total_return_pct,
            "final_equity": final_equity,
            "max_dd_pct": max_dd_pct,
            "trade_count": len(trades),
            "win_rate": win_rate,
            "sharpe": sharpe,
        },
        "metrics_bundle": metrics_bundle,
        "snapshot_coverage_end_date": snapshot_coverage_end_date,
        "snapshot_last_refresh_utc": snapshot_last_refresh_utc,
    }


def _snapshot_needs_refresh(
    snapshot_id: str,
    end_date: dt.date,
    snapshot_dir: str | None = None,
) -> bool:
    """Refresh only when no existing snapshot covers the requested end date.

    Skips refresh if already refreshed today (marker file).
    """
    resolution = resolve_snapshot(snapshot_id, snapshot_dir=snapshot_dir)
    if resolution.refresh_policy == "manual_only":
        return False
    if not resolution.is_registry_managed and snapshot_dir is None:
        return False
    if _snapshot_has_required_coverage(snapshot_id=snapshot_id, end_date=end_date, snapshot_dir=snapshot_dir):
        return False
    # Check if we already attempted refresh today (avoid repeated pipeline runs)
    snapshot_path = _resolve_snapshot_path(snapshot_id, snapshot_dir=snapshot_dir)
    if snapshot_path:
        marker = snapshot_path / ".last_refresh"
        if marker.exists():
            try:
                last_refresh = dt.date.fromisoformat(marker.read_text().strip()[:10])
                if last_refresh >= dt.date.today():
                    return False  # already refreshed today
            except Exception:
                pass
    return True


def _snapshot_has_required_coverage(
    snapshot_id: str,
    end_date: dt.date,
    snapshot_dir: str | None = None,
    grace_days: int = 1,
) -> bool:
    """Return True when an existing snapshot already covers the requested date.

    grace_days: max allowed gap between snapshot's latest event and end_date.
    Default 1 (covers weekends/holidays with 1-day lag). Paired with the
    .last_refresh marker this ensures at most one refresh per calendar day.
    """
    snapshot_path = _resolve_snapshot_path(snapshot_id, snapshot_dir=snapshot_dir)
    if snapshot_path is None:
        return False

    train_path = snapshot_path / "train.parquet"
    valid_path = snapshot_path / "valid.parquet"
    test_path = snapshot_path / "test.parquet"
    parquet_paths = [path for path in (test_path, valid_path, train_path) if path.exists()]
    if not parquet_paths:
        return False

    try:
        import pyarrow.parquet as pq

        max_date: dt.date | None = None
        for parquet_path in parquet_paths:
            table = pq.read_table(str(parquet_path), columns=["event_date"])
            dates = table.column("event_date").to_pylist()
            if not dates:
                continue
            candidate = max(dates)
            if isinstance(candidate, str):
                candidate = dt.date.fromisoformat(candidate[:10])
            if isinstance(candidate, dt.datetime):
                candidate = candidate.date()
            if isinstance(candidate, dt.date) and (max_date is None or candidate > max_date):
                max_date = candidate
        if max_date is None:
            return False
        return max_date >= end_date - dt.timedelta(days=grace_days)
    except Exception:
        return False


def _resolve_snapshot_path(
    snapshot_id: str,
    snapshot_dir: str | None = None,
) -> Path | None:
    """Resolve the on-disk snapshot directory using the same fallback order as the runner."""
    return _resolve_registry_snapshot_path(snapshot_id, snapshot_dir=snapshot_dir)


async def _refresh_snapshot(
    snapshot_id: str,
    universe_profile: str | None,
    console=None,
    *,
    manual: bool = False,
) -> None:
    """Re-run pipeline steps and re-export the snapshot."""
    resolution = resolve_snapshot(snapshot_id)
    if not resolution.is_registry_managed:
        raise RuntimeError(
            f"Snapshot '{snapshot_id}' is not registry-managed; phase-1 refresh only supports canonical snapshots."
        )
    if resolution.refresh_policy == "manual_only" and not manual:
        if console:
            console.print(
                f"\n[bold yellow]Snapshot '{snapshot_id}' resolves to frozen canonical "
                f"'{resolution.canonical_snapshot_id}' — skipping auto-refresh.[/]"
            )
        return
    if console and resolution.requested_snapshot_id != resolution.canonical_snapshot_id:
        console.print(
            f"\n[bold cyan]Resolved snapshot:[/] {resolution.requested_snapshot_id} "
            f"→ {resolution.canonical_snapshot_id}"
        )
    if console:
        console.print("\n[bold yellow]Snapshot stale — refreshing pipeline...[/]")

    # Step 1: Run pending pipeline steps
    if console:
        console.print("  [dim]1/4 Polling new filings...[/]")
    try:
        from apps.pipeline.filing_poller.main import poll_filings
        from libs.common.ids import new_job_run_id
        await poll_filings(new_job_run_id())
    except Exception as exc:
        if console:
            console.print(f"  [yellow]Filing poller skipped: {exc}[/]")

    if console:
        console.print("  [dim]2/4 Fetching exhibits...[/]")
    try:
        from apps.pipeline.filing_fetcher.main import fetch_exhibits
        from libs.common.ids import new_job_run_id
        await fetch_exhibits(new_job_run_id())
    except Exception as exc:
        if console:
            console.print(f"  [yellow]Fetcher skipped: {exc}[/]")

    if console:
        console.print("  [dim]3/4 Parsing events & building features...[/]")
    try:
        from apps.pipeline.event_parser.main import run_event_parser
        from libs.common.ids import new_job_run_id
        await run_event_parser(new_job_run_id())
    except Exception as exc:
        if console:
            console.print(f"  [yellow]Parser skipped: {exc}[/]")

    try:
        from apps.pipeline.feature_builder.main import run_feature_builder
        from libs.common.ids import new_job_run_id
        await run_feature_builder(new_job_run_id())
    except Exception as exc:
        if console:
            console.print(f"  [yellow]Feature builder skipped: {exc}[/]")

    try:
        from apps.pipeline.label_generator.main import run_label_generator
        from libs.common.ids import new_job_run_id
        await run_label_generator(new_job_run_id())
    except Exception as exc:
        if console:
            console.print(f"  [yellow]Label generator skipped: {exc}[/]")

    # Step 2: Update canonical snapshot (incremental first, full rebuild as fallback)
    if console:
        console.print("  [dim]4/4 Exporting snapshot...[/]")
    try:
        from libs.export.canonical_snapshots import (
            build_canonical_snapshot,
            incremental_update_canonical_snapshot,
        )

        snapshot_path = _resolve_snapshot_path(resolution.canonical_snapshot_id)
        use_incremental = snapshot_path is not None and snapshot_path.exists()

        if use_incremental:
            if console:
                console.print("  [dim]Incremental update (new events only)...[/]")
            try:
                await incremental_update_canonical_snapshot(snapshot_id)
            except Exception as inc_exc:
                if console:
                    console.print(f"  [yellow]Incremental failed ({inc_exc}), falling back to full rebuild...[/]")
                await build_canonical_snapshot(snapshot_id, manual=manual)
        else:
            if console:
                console.print("  [dim]Full rebuild (no existing snapshot)...[/]")
            await build_canonical_snapshot(snapshot_id, manual=manual)

        if console:
            console.print("  [green]Snapshot refreshed.[/]")
        # Write marker to avoid re-refreshing today
        snapshot_path = _resolve_snapshot_path(resolution.canonical_snapshot_id)
        if snapshot_path:
            (snapshot_path / ".last_refresh").write_text(dt.date.today().isoformat())
    except Exception as exc:
        if console:
            console.print(f"  [red]Snapshot export failed: {exc}[/]")
        raise


def run_backtest(
    configs: list[str],
    capital: float,
    start_date: dt.date,
    end_date: dt.date,
    db_dsn: str,
    oracle_url: str,
    console=None,
    parking_preset: str | None = None,
    idle_alpha_preset: str | None = None,
    form4_sleeve_preset: str | None = None,
    ownership_sleeve_preset: str | None = None,
    risk_off_alpha_sleeve_preset: str | None = None,
    non_core_allocator_v2_mode: str | None = None,
    snapshot_id_override: str | None = None,
    auto_refresh: bool = True,
) -> list[dict[str, Any]]:
    """Run multiple strategies sequentially using BacktestRunner.

    Automatically refreshes the Parquet snapshot if it doesn't cover
    the requested end_date (runs pipeline + re-export).

    This is a SYNC function — runs async pipeline steps via asyncio.run()
    before the sync BacktestRunner, avoiding nested event loop issues.
    """
    import asyncio
    from libs.common.time_utils import is_trading_day
    from libs.common.logging import configure_logging

    all_days = [
        start_date + dt.timedelta(days=i)
        for i in range((end_date - start_date).days + 1)
    ]
    trading_days = [d for d in all_days if is_trading_day(d)]

    if not trading_days:
        raise ValueError(f"No trading days found between {start_date} and {end_date}")

    if console:
        console.print(f"[bold]Trading days:[/] {trading_days[0]} → {trading_days[-1]} ({len(trading_days)} days)")
        console.print("[bold]Engine:[/] BacktestRunner (identical to research backtester)")

    # Check if snapshots need refresh (async pipeline, run before sync backtest)
    for config_path in configs:
        from apps.backtester.run import load_manifest, resolve_config
        manifest = load_manifest(config_path)
        config = resolve_config(manifest, snapshot_id_override=snapshot_id_override)
        snapshot_id = config.requested_snapshot_id or config.dataset_snapshot_id

        if auto_refresh and _snapshot_needs_refresh(snapshot_id, end_date):
            universe_profile = None
            if "midlarge" in snapshot_id:
                universe_profile = "midlarge-liquid-long-v1"
            elif "midwide" in snapshot_id:
                universe_profile = "midwide-liquid-long-v1"
            elif "smallcap" in snapshot_id:
                universe_profile = "smallcap-liquid-long-v1"

            if console:
                console.print(f"\n[bold yellow]Snapshot '{snapshot_id}' is stale — refreshing...[/]")
            try:
                asyncio.run(_refresh_snapshot(snapshot_id, universe_profile, console=console, manual=False))
            except Exception:
                if _snapshot_has_required_coverage(snapshot_id, end_date):
                    if console:
                        console.print("  [yellow]Refresh failed, but existing snapshot still covers the requested period. Using current snapshot.[/]")
                else:
                    raise

    configure_logging("WARNING")

    results = []
    for config_path in configs:
        session_name = Path(config_path).stem
        if console:
            console.print(f"\n[bold cyan]Running:[/] {session_name}")
        result = run_backtest_session_sync(
            session_name=session_name,
            config_path=config_path,
            initial_equity=capital,
            start_date=start_date,
            end_date=end_date,
            parking_preset=parking_preset,
            idle_alpha_preset=idle_alpha_preset,
            form4_sleeve_preset=form4_sleeve_preset,
            ownership_sleeve_preset=ownership_sleeve_preset,
            risk_off_alpha_sleeve_preset=risk_off_alpha_sleeve_preset,
            non_core_allocator_v2_mode=non_core_allocator_v2_mode,
            snapshot_id_override=snapshot_id_override,
        )

        results.append(result)

        if console and result["summary"]["trade_count"] > 0:
            s = result["summary"]
            console.print(
                f"  Trades: {s['trade_count']}, "
                f"Return: {s['return_pct']:+.2f}%, "
                f"MaxDD: {s['max_dd_pct']:.2f}%, "
                f"WR: {s['win_rate']:.0f}%"
            )

    return results


def load_snapshot_store_for_session(
    config: Any,
    oracle_url: str,
    db_dsn: str,
    *,
    auto_refresh: bool = True,
) -> Any:
    """Load SnapshotStore for a live paper trading session.

    Mirrors the snapshot loading done in run_backtest(), but for live use.
    Returns None if snapshot_id is not configured or loading fails (caller
    should fall back to EventDetector).

    This is a SYNC function — safe to call from _make_engine() / make_engine()
    before an event loop is started.
    """
    import asyncio
    from libs.backtest.snapshot_store import SnapshotStore

    snapshot_id = getattr(config, "dataset_snapshot_id", None)
    if not snapshot_id:
        logger.warning("snapshot_store_no_snapshot_id", config=str(config))
        return None

    try:
        if auto_refresh and _snapshot_needs_refresh(snapshot_id, dt.date.today()):
            universe_profile = None
            if "midlarge" in snapshot_id:
                universe_profile = "midlarge-liquid-long-v1"
            elif "midwide" in snapshot_id:
                universe_profile = "midwide-liquid-long-v1"
            elif "smallcap" in snapshot_id:
                universe_profile = "smallcap-liquid-long-v1"
            try:
                asyncio.run(_refresh_snapshot(snapshot_id, universe_profile, manual=False))
            except Exception as exc:
                logger.warning("snapshot_store_refresh_failed", error=str(exc))
                # Write the marker so we don't retry the (failing) refresh on
                # every engine init throughout the day.
                _snap_path_for_marker = _resolve_snapshot_path(snapshot_id)
                if _snap_path_for_marker:
                    try:
                        (_snap_path_for_marker / ".last_refresh").write_text(
                            dt.date.today().isoformat()
                        )
                    except Exception:
                        pass
                if not _snapshot_has_required_coverage(snapshot_id, dt.date.today()):
                    # Snapshot is stale (missing recent days) but may still contain
                    # lookback candidates from earlier dates (e.g. an event filed
                    # 1-2 weeks ago that we haven't entered yet).  Use it anyway —
                    # the engine will fall back to EventDetector for today's fresh
                    # events; the stale snapshot is better than nothing for lookback.
                    logger.warning(
                        "snapshot_store_stale_using_anyway",
                        snapshot_id=snapshot_id,
                    )

        snapshot_path = _resolve_snapshot_path(snapshot_id)
        if snapshot_path is None:
            logger.warning("snapshot_store_path_not_found", snapshot_id=snapshot_id)
            return None

        from apps.backtester.run import _resolve_scoring_fn
        scoring_fn = _resolve_scoring_fn(config)

        return SnapshotStore.load_merged(
            snapshot_dir=snapshot_path,
            split_names=["train", "valid", "test"],
            oracle_url=oracle_url,
            db_dsn=db_dsn,
            scoring_fn=scoring_fn,
        )
    except Exception as exc:
        logger.warning("snapshot_store_load_failed", error=str(exc))
        return None