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.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,
) -> 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_store,
        _build_merged_snapshot_store,
        load_manifest,
        resolve_config,
    )

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

    # 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
    store = store.slice_by_date_range(start_date, end_date)

    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

                    trades.append({
                        "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", "")),
                    })
    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

    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,
        },
    }


def _snapshot_needs_refresh(
    snapshot_id: str,
    end_date: dt.date,
    snapshot_dir: str = "data/datasets/snapshots",
) -> bool:
    """Check if the Parquet snapshot is stale (doesn't cover end_date)."""
    import json

    manifest_path = Path(snapshot_dir) / snapshot_id / "manifest.json"
    if not manifest_path.exists():
        return True

    try:
        manifest = json.loads(manifest_path.read_text())
        created = manifest.get("created_at_utc", "")[:10]
        if created and dt.date.fromisoformat(created) < end_date - dt.timedelta(days=7):
            return True
    except Exception:
        return True

    # Check if the latest event_date in the data covers end_date
    train_path = Path(snapshot_dir) / snapshot_id / "train.parquet"
    test_path = Path(snapshot_dir) / snapshot_id / "test.parquet"
    latest_path = test_path if test_path.exists() else train_path
    if not latest_path.exists():
        return True

    try:
        import pyarrow.parquet as pq
        table = pq.read_table(str(latest_path), columns=["event_date"])
        dates = table.column("event_date").to_pylist()
        max_date = max(dates) if dates else ""
        if isinstance(max_date, str):
            max_date = dt.date.fromisoformat(max_date[:10])
        # Stale if snapshot's latest event is more than 14 days before end_date
        return max_date < end_date - dt.timedelta(days=14)
    except Exception:
        return True


async def _refresh_snapshot(
    snapshot_id: str,
    universe_profile: str | None,
    console=None,
) -> None:
    """Re-run pipeline steps and re-export the snapshot."""
    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: Re-export snapshot
    if console:
        console.print("  [dim]4/4 Exporting snapshot...[/]")
    try:
        from libs.db.session import get_session
        from libs.export.snapshot_export import export_dataset_snapshot

        async with get_session() as session:
            await export_dataset_snapshot(
                session=session,
                snapshot_id=snapshot_id,
                split_policy="temporal_70_15_15",
                output_dir="data/datasets/snapshots",
                feature_versions=["market_v1", "event_v1"],
                universe_profile=universe_profile,
            )
        if console:
            console.print("  [green]Snapshot refreshed.[/]")
    except Exception as exc:
        if console:
            console.print(f"  [red]Snapshot export failed: {exc}[/]")
        raise


async def run_backtest(
    configs: list[str],
    capital: float,
    start_date: dt.date,
    end_date: dt.date,
    db_dsn: str,
    oracle_url: str,
    console=None,
) -> 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).
    """
    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 for each config
    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 = config.dataset_snapshot_id

        if _snapshot_needs_refresh(snapshot_id, end_date):
            # Determine universe_profile from snapshot export config
            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:
                await _refresh_snapshot(snapshot_id, universe_profile, console=console)
            except Exception as exc:
                if console:
                    console.print(f"[yellow]Snapshot refresh failed, using existing data: {exc}[/]")

    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,
        )
        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