signal-platform/scripts/run_backtest_cadence_comparison.py

"""Run the four production cadence/lockdown arms on one offline snapshot.

The command executes the complete backtest once weekly and once daily. Each
backtest contains two otherwise identical live-policy portfolio arms: no
post-stop lockdown and the production five-session lockdown. It writes both
full reports plus one compact four-arm comparison report.
"""

from __future__ import annotations

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

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 _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "snapshot",
        help="SQLite snapshot created by scripts/create_backtest_snapshot.py.",
    )
    parser.add_argument(
        "--out-dir",
        default="reports",
        help="Directory for the weekly, daily, and comparison JSON reports.",
    )
    parser.add_argument(
        "--prefix",
        default=None,
        help="Output prefix. Defaults to backtest-cadence-<timestamp>.",
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=None,
        help="Override worker count; on a powerful offline PC use CPU count minus one.",
    )
    parser.add_argument(
        "--allow-spawn",
        action="store_true",
        help="Enable multiprocessing spawn for the offline Windows run.",
    )
    parser.add_argument("--quiet", action="store_true", help="Hide ticker progress.")
    return parser.parse_args()


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


def _write_json(path: Path, payload: dict) -> None:
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")


def _comparison_arms(report: dict) -> list[dict]:
    comparison = report.get("production_cadence_comparison") or {}
    arms = list(comparison.get("arms") or [])
    if len(arms) != 2:
        cadence = (report.get("params") or {}).get("entry_cadence", "unknown")
        raise RuntimeError(
            f"Expected two live comparison arms for {cadence}; found {len(arms)}"
        )
    return arms


def _print_arm(row: dict) -> None:
    print(
        f"  {row['arm']}: Sharpe {row.get('sharpe')}, "
        f"CAGR {row.get('cagr_pct')}%, DD {row.get('max_drawdown_pct')}%, "
        f"trades {row.get('trades')}, skipped cooldown {row.get('skipped_cooldown', 0)}"
    )


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"

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

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

    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    prefix = args.prefix or f"backtest-cadence-{datetime.now():%Y%m%d-%H%M%S}"

    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    reports: dict[str, dict] = {}
    try:
        async with Session() as db:
            for cadence in ("weekly", "daily"):
                last_progress: tuple[int, int] | None = None

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

                reports[cadence] = await run_backtest(
                    db,
                    progress_cb=progress,
                    target_model="production_gtl",
                    cadence=cadence,
                )
                if not args.quiet:
                    print("")
                _write_json(out_dir / f"{prefix}-{cadence}.json", reports[cadence])
    finally:
        await engine.dispose()

    arms = [
        *_comparison_arms(reports["weekly"]),
        *_comparison_arms(reports["daily"]),
    ]
    expected = {
        "prod_live_setup_weekly",
        "prod_live_setup_daily",
        "cooldown_5_weekly",
        "cooldown_5_daily",
    }
    if {row.get("arm") for row in arms} != expected:
        raise RuntimeError("The generated cadence report does not contain all four arms")
    arm_order = {
        "prod_live_setup_weekly": 0,
        "prod_live_setup_daily": 1,
        "cooldown_5_weekly": 2,
        "cooldown_5_daily": 3,
    }
    arms.sort(key=lambda row: arm_order[str(row["arm"])])

    comparison = {
        "generated_at": datetime.now().astimezone().isoformat(),
        "snapshot": str(snapshot.resolve()),
        "target_model": "production_gtl",
        "arms": arms,
        "full_reports": {
            cadence: str((out_dir / f"{prefix}-{cadence}.json").resolve())
            for cadence in ("weekly", "daily")
        },
        "note": (
            "All four arms use the same snapshot, activation settings, target model, "
            "live Admin exit policy, fees, sizing, and portfolio constraints. Within "
            "each cadence pair, only the five-session post-stop lockdown differs."
        ),
    }
    comparison_path = out_dir / f"{prefix}-comparison.json"
    _write_json(comparison_path, comparison)

    print(f"Comparison written: {comparison_path}")
    for row in arms:
        _print_arm(row)


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