419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""Run the complete Gate Target Ladder tuning matrix with one command.
|
|
|
|
Every arm is a full production-parity backtest. Arms run sequentially so each
|
|
one can use the requested worker pool without competing with another arm. Large
|
|
per-arm reports live in a temporary run directory and are removed after a
|
|
successful consolidation unless ``--keep-arm-reports`` is supplied.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import subprocess
|
|
import sys
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER = ROOT / "scripts" / "run_backtest_snapshot.py"
|
|
|
|
BASE_CONFIG: dict[str, Any] = {
|
|
"lookback_bars": None,
|
|
"grid_bins": 20,
|
|
"include_pivots": True,
|
|
"pivot_window": 2,
|
|
"touch_tolerance": 0.005,
|
|
"merge_tolerance": 0.005,
|
|
"strength_scale": 500.0,
|
|
"zone_tolerance": 0.02,
|
|
"candidate_limit": 5,
|
|
"max_target_atr": None,
|
|
}
|
|
|
|
# Single-variable arms only. The control value is represented by BASE_CONFIG;
|
|
# there is deliberately no Cartesian product.
|
|
GTL_TUNING_ARMS: tuple[dict[str, Any], ...] = (
|
|
{"name": "control", "description": "Frozen explicit GTL defaults."},
|
|
{"name": "lookback_252", "description": "One-year GTL history.", "lookback_bars": 252},
|
|
{"name": "lookback_504", "description": "Two-year GTL history.", "lookback_bars": 504},
|
|
{"name": "lookback_756", "description": "Three-year GTL history.", "lookback_bars": 756},
|
|
{"name": "candidates_8", "description": "Retain up to eight candidates before probability.", "candidate_limit": 8},
|
|
{"name": "candidates_all", "description": "Score every eligible target before primary selection.", "candidate_limit": None},
|
|
{"name": "max_atr_5_5", "description": "Universal 5.5 ATR maximum target distance.", "max_target_atr": 5.5},
|
|
{"name": "max_atr_8", "description": "Universal 8 ATR maximum target distance.", "max_target_atr": 8.0},
|
|
{"name": "touch_0", "description": "Strict candle-range crossings with no touch padding.", "touch_tolerance": 0.0},
|
|
{"name": "touch_0_25pct", "description": "Use 0.25% padding when counting price traffic.", "touch_tolerance": 0.0025},
|
|
{"name": "merge_0_25pct", "description": "Merge GTL proposals within 0.25%.", "merge_tolerance": 0.0025},
|
|
{"name": "merge_1pct", "description": "Merge GTL proposals within 1%.", "merge_tolerance": 0.01},
|
|
{"name": "zones_1pct", "description": "Cluster target zones within 1%.", "zone_tolerance": 0.01},
|
|
{"name": "zones_3pct", "description": "Cluster target zones within 3%.", "zone_tolerance": 0.03},
|
|
{"name": "grid_12", "description": "Use 12 evenly spaced range centers.", "grid_bins": 12},
|
|
{"name": "grid_32", "description": "Use 32 evenly spaced range centers.", "grid_bins": 32},
|
|
{"name": "pivots_none", "description": "Range grid only; omit swing pivots.", "include_pivots": False},
|
|
{"name": "pivots_11bar", "description": "Use an 11-bar swing-pivot window.", "pivot_window": 5},
|
|
{"name": "strength_250", "description": "Slower traffic-strength saturation.", "strength_scale": 250.0},
|
|
{"name": "strength_1000", "description": "Faster traffic-strength saturation.", "strength_scale": 1000.0},
|
|
)
|
|
|
|
BOOK_FIELDS = (
|
|
"sharpe",
|
|
"cagr_pct",
|
|
"max_drawdown_pct",
|
|
"trades",
|
|
"win_rate",
|
|
"avg_hold_days",
|
|
"skipped_book_full",
|
|
)
|
|
|
|
|
|
def _args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"snapshot",
|
|
nargs="?",
|
|
default="backtest_snapshots/prod.sqlite",
|
|
help="Local SQLite snapshot path.",
|
|
)
|
|
parser.add_argument("--workers", type=int, default=7)
|
|
parser.add_argument(
|
|
"--holdout-split",
|
|
default="2024-07-01",
|
|
help="Disjoint train/test split included in every arm.",
|
|
)
|
|
parser.add_argument(
|
|
"--out",
|
|
default=None,
|
|
help="Consolidated JSON path. Defaults to reports/backtest-YYYYMMDD-gtl-tuning-matrix.json.",
|
|
)
|
|
parser.add_argument(
|
|
"--keep-arm-reports",
|
|
action="store_true",
|
|
help="Keep the large temporary per-arm JSON reports after consolidation.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def _arm_config(arm: dict[str, Any]) -> dict[str, Any]:
|
|
config = {**BASE_CONFIG, **{key: value for key, value in arm.items() if key != "description"}}
|
|
return config
|
|
|
|
|
|
def _load_report(path: Path) -> dict:
|
|
with path.open(encoding="utf-8") as handle:
|
|
report = json.load(handle)
|
|
if report.get("sr_candidate_audit") is None:
|
|
raise ValueError(f"Report lacks sr_candidate_audit: {path}")
|
|
return report
|
|
|
|
|
|
def _compact_book(row: dict | None) -> dict | None:
|
|
if row is None:
|
|
return None
|
|
return {key: row.get(key) for key in BOOK_FIELDS}
|
|
|
|
|
|
def _full_book(report: dict) -> dict | None:
|
|
runs = ((report.get("portfolio_monitor") or {}).get("runs") or [])
|
|
return _compact_book(next(
|
|
(
|
|
row
|
|
for row in runs
|
|
if row.get("is_production") and row.get("lookback") == "all"
|
|
),
|
|
None,
|
|
))
|
|
|
|
|
|
def _holdout_books(report: dict) -> dict[str, dict | None]:
|
|
rows = ((report.get("holdout") or {}).get("rows") or [])
|
|
return {
|
|
window: _compact_book(next((row for row in rows if row.get("window") == window), None))
|
|
for window in ("train", "test")
|
|
}
|
|
|
|
|
|
def _audit_key(row: dict) -> tuple[str, str, str]:
|
|
return row["symbol"], row["date"], row["direction"]
|
|
|
|
|
|
def _cohort_stats(rows: list[dict]) -> dict:
|
|
net = [float(row.get("net_r", 0.0)) for row in rows]
|
|
trimmed = sorted(net, reverse=True)[math.ceil(len(net) * 0.05):]
|
|
return {
|
|
"count": len(rows),
|
|
"net_avg_r": round(sum(net) / len(net), 4) if net else None,
|
|
"net_avg_r_ex_top5": round(sum(trimmed) / len(trimmed), 4) if trimmed else None,
|
|
}
|
|
|
|
|
|
def _cohort_comparison(control: dict, variant: dict) -> dict:
|
|
control_rows = {
|
|
_audit_key(row): row for row in control.get("sr_candidate_audit") or []
|
|
}
|
|
variant_rows = {
|
|
_audit_key(row): row for row in variant.get("sr_candidate_audit") or []
|
|
}
|
|
control_q = {key for key, row in control_rows.items() if row.get("qualified")}
|
|
variant_q = {key for key, row in variant_rows.items() if row.get("qualified")}
|
|
retained = control_q & variant_q
|
|
added = variant_q - control_q
|
|
removed = control_q - variant_q
|
|
return {
|
|
"retained": _cohort_stats([variant_rows[key] for key in retained]),
|
|
"added": _cohort_stats([variant_rows[key] for key in added]),
|
|
"removed": _cohort_stats([control_rows[key] for key in removed]),
|
|
}
|
|
|
|
|
|
def _compact_arm(report: dict, config: dict, control: dict | None) -> dict:
|
|
qualified = report.get("overall_qualified") or {}
|
|
result = {
|
|
"name": config["name"],
|
|
"config": config,
|
|
"candidates": report.get("candidates"),
|
|
"qualified": report.get("qualified"),
|
|
"qualified_net_avg_r": qualified.get("net_avg_r"),
|
|
"qualified_net_avg_r_ex_top5": qualified.get("net_avg_r_ex_top5"),
|
|
"full_book": _full_book(report),
|
|
"holdout": _holdout_books(report),
|
|
"gtl_diagnostics": report.get("sr_variant_diagnostics"),
|
|
"cohort_vs_control": (
|
|
_cohort_comparison(control, report) if control is not None else None
|
|
),
|
|
}
|
|
return result
|
|
|
|
|
|
def _screen_arm(arm: dict, control: dict) -> dict:
|
|
full = arm.get("full_book") or {}
|
|
base_full = control.get("full_book") or {}
|
|
train = (arm.get("holdout") or {}).get("train") or {}
|
|
base_train = (control.get("holdout") or {}).get("train") or {}
|
|
test = (arm.get("holdout") or {}).get("test") or {}
|
|
base_test = (control.get("holdout") or {}).get("test") or {}
|
|
|
|
def at_least(value: Any, baseline: Any) -> bool:
|
|
return value is not None and baseline is not None and float(value) >= float(baseline)
|
|
|
|
control_trades = float(base_full.get("trades") or 0.0)
|
|
arm_trades = float(full.get("trades") or 0.0)
|
|
checks = {
|
|
"full_sharpe_not_worse": at_least(full.get("sharpe"), base_full.get("sharpe")),
|
|
"train_sharpe_not_worse": at_least(train.get("sharpe"), base_train.get("sharpe")),
|
|
"test_sharpe_not_worse": at_least(test.get("sharpe"), base_test.get("sharpe")),
|
|
"drawdown_not_worse": (
|
|
full.get("max_drawdown_pct") is not None
|
|
and base_full.get("max_drawdown_pct") is not None
|
|
and abs(float(full["max_drawdown_pct"]))
|
|
<= abs(float(base_full["max_drawdown_pct"]))
|
|
),
|
|
"retains_80pct_trades": control_trades > 0 and arm_trades >= control_trades * 0.8,
|
|
"robust_expectancy_positive": (
|
|
arm.get("qualified_net_avg_r_ex_top5") is not None
|
|
and float(arm["qualified_net_avg_r_ex_top5"]) > 0
|
|
),
|
|
}
|
|
return {
|
|
"checks": checks,
|
|
"passed": sum(checks.values()),
|
|
"total": len(checks),
|
|
"advances": all(checks.values()),
|
|
}
|
|
|
|
|
|
def _write_json(path: Path, payload: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, indent=2)
|
|
handle.write("\n")
|
|
|
|
|
|
def _fmt(value: Any, digits: int = 2) -> str:
|
|
return "-" if value is None else f"{float(value):.{digits}f}"
|
|
|
|
|
|
def _write_markdown(path: Path, payload: dict) -> None:
|
|
rows = [
|
|
"# GTL tuning matrix",
|
|
"",
|
|
f"Status: **{payload['status']}** ",
|
|
f"Holdout split: `{payload['holdout_split']}` ",
|
|
f"Completed arms: {len(payload['arms'])}/{payload['arm_count']}",
|
|
"",
|
|
"| Arm | Full Sharpe | CAGR | Max DD | Trades | Train Sharpe | Test Sharpe | Ex-top-5% R | Screen |",
|
|
"|---|---:|---:|---:|---:|---:|---:|---:|---:|",
|
|
]
|
|
for arm in payload["arms"]:
|
|
full = arm.get("full_book") or {}
|
|
holdout = arm.get("holdout") or {}
|
|
train = holdout.get("train") or {}
|
|
test = holdout.get("test") or {}
|
|
screen = arm.get("screen") or {}
|
|
rows.append(
|
|
"| "
|
|
+ " | ".join((
|
|
arm["name"],
|
|
_fmt(full.get("sharpe")),
|
|
_fmt(full.get("cagr_pct"), 1),
|
|
_fmt(full.get("max_drawdown_pct"), 1),
|
|
str(full.get("trades") or "-"),
|
|
_fmt(train.get("sharpe")),
|
|
_fmt(test.get("sharpe")),
|
|
_fmt(arm.get("qualified_net_avg_r_ex_top5"), 3),
|
|
f"{screen.get('passed', '-')}/{screen.get('total', '-')}",
|
|
))
|
|
+ " |"
|
|
)
|
|
rows.extend((
|
|
"",
|
|
"## Interpretation guardrail",
|
|
"",
|
|
"The post-2024 interval has already informed prior research. The train/test columns are robustness checks, not a pristine holdout. A passing arm is a candidate for forward paper validation, not automatic production promotion.",
|
|
"",
|
|
))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("\n".join(rows), encoding="utf-8")
|
|
|
|
|
|
def _run_arm(
|
|
*,
|
|
arm: dict[str, Any],
|
|
snapshot: str,
|
|
workers: int,
|
|
holdout_split: str,
|
|
output: Path,
|
|
) -> None:
|
|
config = _arm_config(arm)
|
|
command = [
|
|
sys.executable,
|
|
str(RUNNER),
|
|
snapshot,
|
|
"--workers", str(workers),
|
|
"--allow-spawn",
|
|
"--sr-variant", "gtl_tuning",
|
|
"--gtl-config", json.dumps(config, separators=(",", ":")),
|
|
"--holdout-split", holdout_split,
|
|
"--sr-audit",
|
|
"--out", str(output),
|
|
]
|
|
subprocess.run(command, cwd=ROOT, check=True)
|
|
|
|
|
|
def main() -> None:
|
|
args = _args()
|
|
snapshot = Path(args.snapshot).resolve()
|
|
if not snapshot.exists():
|
|
raise SystemExit(f"Snapshot not found: {snapshot}")
|
|
if args.workers < 1:
|
|
raise SystemExit("--workers must be at least 1")
|
|
try:
|
|
date.fromisoformat(args.holdout_split)
|
|
except ValueError as exc:
|
|
raise SystemExit("--holdout-split must use YYYY-MM-DD") from exc
|
|
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
|
default_out = ROOT / "reports" / f"backtest-{stamp[:8]}-gtl-tuning-matrix.json"
|
|
out_path = Path(args.out) if args.out else default_out
|
|
if not out_path.is_absolute():
|
|
out_path = ROOT / out_path
|
|
markdown_path = out_path.with_suffix(".md")
|
|
work_dir = ROOT / "reports" / f".gtl-tuning-work-{stamp}"
|
|
work_dir.mkdir(parents=True, exist_ok=False)
|
|
|
|
payload: dict[str, Any] = {
|
|
"status": "running",
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"snapshot": str(snapshot.resolve()),
|
|
"workers": args.workers,
|
|
"holdout_split": args.holdout_split,
|
|
"arm_count": len(GTL_TUNING_ARMS),
|
|
"arms": [],
|
|
"caveat": (
|
|
"The split is a robustness check, not a pristine holdout; post-2024 "
|
|
"data has already informed earlier research."
|
|
),
|
|
}
|
|
_write_json(out_path, payload)
|
|
_write_markdown(markdown_path, payload)
|
|
|
|
control_report: dict | None = None
|
|
arm_outputs: list[Path] = []
|
|
try:
|
|
for index, arm in enumerate(GTL_TUNING_ARMS, start=1):
|
|
name = str(arm["name"])
|
|
output = work_dir / f"{index:02d}-{name}.json"
|
|
arm_outputs.append(output)
|
|
print(f"\n[{index}/{len(GTL_TUNING_ARMS)}] GTL arm: {name}", flush=True)
|
|
print(f" {arm['description']}", flush=True)
|
|
_run_arm(
|
|
arm=arm,
|
|
snapshot=str(snapshot),
|
|
workers=args.workers,
|
|
holdout_split=args.holdout_split,
|
|
output=output,
|
|
)
|
|
report = _load_report(output)
|
|
config = _arm_config(arm)
|
|
compact = _compact_arm(report, config, control_report)
|
|
compact["description"] = arm["description"]
|
|
if control_report is None:
|
|
control_report = report
|
|
compact["screen"] = {
|
|
"checks": {},
|
|
"passed": 0,
|
|
"total": 0,
|
|
"advances": False,
|
|
}
|
|
else:
|
|
compact["screen"] = _screen_arm(compact, payload["arms"][0])
|
|
payload["arms"].append(compact)
|
|
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
|
|
_write_json(out_path, payload)
|
|
_write_markdown(markdown_path, payload)
|
|
except Exception as exc:
|
|
payload["status"] = "failed"
|
|
payload["failed_at"] = datetime.now(timezone.utc).isoformat()
|
|
payload["error"] = f"{type(exc).__name__}: {exc}"
|
|
payload["work_dir"] = str(work_dir)
|
|
_write_json(out_path, payload)
|
|
_write_markdown(markdown_path, payload)
|
|
raise
|
|
|
|
payload["status"] = "complete"
|
|
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
|
|
payload["advancing_arms"] = [
|
|
arm["name"] for arm in payload["arms"] if (arm.get("screen") or {}).get("advances")
|
|
]
|
|
payload["ranking_by_full_sharpe"] = [
|
|
arm["name"]
|
|
for arm in sorted(
|
|
payload["arms"],
|
|
key=lambda row: float((row.get("full_book") or {}).get("sharpe") or -math.inf),
|
|
reverse=True,
|
|
)
|
|
]
|
|
if args.keep_arm_reports:
|
|
payload["arm_report_directory"] = str(work_dir)
|
|
_write_json(out_path, payload)
|
|
_write_markdown(markdown_path, payload)
|
|
|
|
if not args.keep_arm_reports:
|
|
for path in arm_outputs:
|
|
path.unlink(missing_ok=True)
|
|
work_dir.rmdir()
|
|
|
|
print("\nGTL tuning matrix complete.")
|
|
print(f" JSON: {out_path}")
|
|
print(f" Markdown: {markdown_path}")
|
|
if payload["advancing_arms"]:
|
|
print(f" Arms passing every pre-registered screen: {', '.join(payload['advancing_arms'])}")
|
|
else:
|
|
print(" No arm passed every pre-registered screen.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|