219 lines
7.1 KiB
Python
219 lines
7.1 KiB
Python
"""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",
|
|
),
|
|
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())
|