signal-platform/scripts/run_backtest_snapshot.py

"""Run the existing backtest service against a local SQLite snapshot.

The runner is offline/read-only: it does not refresh benchmark prices and does
not cache the report back to any database. It writes a local JSON report.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def _sqlite_url(path: Path) -> str:
    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("snapshot", help="SQLite snapshot created by create_backtest_snapshot.py.")
    parser.add_argument(
        "--out",
        default=None,
        help="JSON report path. Defaults to reports/backtest-<timestamp>.json.",
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=None,
        help="Override backtest worker count. On an 8-thread machine, 6 is a good starting point.",
    )
    parser.add_argument(
        "--allow-spawn",
        action="store_true",
        help="Allow spawn multiprocessing for offline CLI runs, useful on Windows.",
    )
    parser.add_argument("--quiet", action="store_true", help="Hide progress output.")
    parser.add_argument(
        "--sr-variant",
        choices=(
            "production_control", "rr_aligned_control", "rewrite",
            "soft_zones", "confirmed_rounds", "gate_v2",
            "rewrite_legacy_primary", "soft_zones_legacy_primary",
            "confirmed_rounds_legacy_primary", "gate_v2_legacy_primary",
            "legacy_geometry_neutral", "legacy_pivots_only",
            "legacy_traffic_grid_only", "legacy_range_grid_touch",
            "legacy_range_grid_neutral",
        ),
        default=None,
        help="Research-only S/R detector/gate arm.",
    )
    parser.add_argument(
        "--entry-start",
        default=None,
        help="Include entries on/after YYYY-MM-DD.",
    )
    parser.add_argument(
        "--entry-end",
        default=None,
        help="Include entries on/before YYYY-MM-DD.",
    )
    parser.add_argument(
        "--sr-audit",
        action="store_true",
        help="Include candidate-level S/R audit rows for paired comparison.",
    )
    return parser.parse_args()


def _default_output_path() -> Path:
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    return Path("reports") / f"backtest-{stamp}.json"


def _pct(value: Any) -> str:
    return "-" if value is None else f"{float(value):+.1f}%"


def _drawdown_pct(value: Any) -> str:
    return "-" if value is None else f"-{abs(float(value)):.1f}%"


def _r(value: Any) -> str:
    return "-" if value is None else f"{float(value):+.2f}R"


def _print_summary(report: dict) -> None:
    qualified = report.get("overall_qualified") or {}
    all_setups = report.get("overall_all") or {}
    time_exit = {row.get("hold_days"): row for row in report.get("time_exit_sweep") or []}
    hold_30 = time_exit.get(30) or {}
    policies = {
        row.get("policy"): row
        for row in ((report.get("portfolio_sim") or {}).get("policies") or [])
    }
    hold_policy = policies.get("hold") or {}

    print("")
    print("Backtest summary")
    print(f"  candidates: {report.get('candidates')}")
    print(f"  qualified: {report.get('qualified')}")
    print(f"  all setups net avg R: {_r(all_setups.get('net_avg_r'))}")
    print(f"  qualified net avg R: {_r(qualified.get('net_avg_r'))}")
    print(f"  qualified total R: {_r(qualified.get('total_r'))}")
    print(f"  30d hold net avg R: {_r(hold_30.get('net_avg_r'))}")
    print(f"  30d hold total R: {_r(hold_30.get('total_r'))}")
    if hold_policy:
        print(f"  hold CAGR: {_pct(hold_policy.get('cagr_pct'))}")
        print(f"  hold max drawdown: {_drawdown_pct(hold_policy.get('max_drawdown_pct'))}")
        print(f"  hold Sharpe: {hold_policy.get('sharpe')}")
        print(f"  hold trades: {hold_policy.get('trades')}")

    variants = list((report.get("strategy_variants") or {}).get("variants") or [])
    ranked = sorted(
        (v for v in variants if v.get("sharpe") is not None),
        key=lambda v: v.get("sharpe"),
        reverse=True,
    )
    if ranked:
        print("  top strategy variants by Sharpe:")
        for row in ranked[:5]:
            print(
                "    "
                f"{row.get('variant')}: "
                f"Sharpe {row.get('sharpe')}, "
                f"CAGR {_pct(row.get('cagr_pct'))}, "
                f"DD {_drawdown_pct(row.get('max_drawdown_pct'))}, "
                f"trades {row.get('trades')}"
            )

    exits = list((report.get("exit_policy_variants") or {}).get("variants") or [])
    ranked_exits = sorted(
        (v for v in exits if v.get("sharpe") is not None),
        key=lambda v: v.get("sharpe"),
        reverse=True,
    )
    if ranked_exits:
        print("  exit policies for 80/20 entry by Sharpe:")
        for row in ranked_exits[:5]:
            print(
                "    "
                f"{row.get('exit_policy')}: "
                f"Sharpe {row.get('sharpe')}, "
                f"CAGR {_pct(row.get('cagr_pct'))}, "
                f"DD {_drawdown_pct(row.get('max_drawdown_pct'))}, "
                f"trades {row.get('trades')}"
            )


async def _main() -> None:
    args = _parse_args()
    snapshot = Path(args.snapshot)
    if not snapshot.exists():
        raise SystemExit(f"Snapshot not found: {snapshot}")

    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
    if args.allow_spawn:
        os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
    if args.sr_variant:
        os.environ["BACKTEST_SR_VARIANT"] = args.sr_variant
    if args.entry_start:
        os.environ["BACKTEST_ENTRY_START"] = args.entry_start
    if args.entry_end:
        os.environ["BACKTEST_ENTRY_END"] = args.entry_end
    if args.sr_audit:
        os.environ["BACKTEST_SR_AUDIT"] = "1"

    from app.config import settings
    from app.services.backtest_service import run_backtest

    if args.workers is not None:
        settings.backtest_workers = args.workers

    output = Path(args.out) if args.out else _default_output_path()
    output.parent.mkdir(parents=True, exist_ok=True)

    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    last_progress: tuple[int, int] | None = None

    def progress(done: int, total: int, symbol: str) -> None:
        nonlocal last_progress
        if args.quiet:
            return
        marker = (done, total)
        if marker == last_progress:
            return
        last_progress = marker
        label = f" {symbol}" if symbol else ""
        print(f"progress: {done}/{total}{label}", end="\r")

    try:
        async with Session() as db:
            report = await run_backtest(db, progress_cb=progress)
    finally:
        await engine.dispose()

    if not args.quiet:
        print("")
    with output.open("w", encoding="utf-8") as fh:
        json.dump(report, fh, indent=2)
        fh.write("\n")

    print(f"Report written: {output}")
    _print_summary(report)


if __name__ == "__main__":
    asyncio.run(_main())