288 lines
10 KiB
Python
288 lines
10 KiB
Python
"""Run the pre-registered GTL strength-confirmation sensitivity band.
|
|
|
|
The strength-1000 intersection was the only composition arm to improve
|
|
full/train/test Sharpe, but it missed the unchanged drawdown guardrail. This
|
|
matrix checks whether that result is a stable one-dimensional plateau rather
|
|
than tuning the guardrail or launching another broad parameter search.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import sys
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from scripts import run_gtl_confirmation_matrix as composition # noqa: E402
|
|
from scripts import run_gtl_tuning_matrix as common # noqa: E402
|
|
|
|
REFERENCE_TUNING = ROOT / "reports" / "backtest-20260713-gtl-tuning-matrix.json"
|
|
REFERENCE_COMPOSITION = (
|
|
ROOT / "reports" / "backtest-20260713-gtl-confirmation-matrix.json"
|
|
)
|
|
STRENGTH_SCALES = (625.0, 750.0, 875.0, 1000.0, 1125.0, 1250.0, 1500.0, 2000.0)
|
|
|
|
GTL_STRENGTH_ARMS: tuple[dict[str, Any], ...] = (
|
|
composition._arm(
|
|
"control",
|
|
"Frozen GTL composition-path parity control.",
|
|
"intersection",
|
|
),
|
|
*(
|
|
composition._arm(
|
|
f"strength_{int(scale)}_intersection",
|
|
f"Require control confirmation at traffic-strength scale {scale:g}.",
|
|
"intersection",
|
|
{"name": f"strength_{int(scale)}", "strength_scale": scale},
|
|
)
|
|
for scale in STRENGTH_SCALES
|
|
),
|
|
)
|
|
|
|
|
|
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")
|
|
parser.add_argument(
|
|
"--out",
|
|
default=None,
|
|
help=(
|
|
"Consolidated JSON path. Defaults to "
|
|
"reports/backtest-YYYYMMDD-gtl-strength-sensitivity.json."
|
|
),
|
|
)
|
|
parser.add_argument("--keep-arm-reports", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def _reference_arm(path: Path, name: str) -> dict | None:
|
|
if not path.exists():
|
|
return None
|
|
with path.open(encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
return next(
|
|
(arm for arm in payload.get("arms") or [] if arm.get("name") == name),
|
|
None,
|
|
)
|
|
|
|
|
|
def _write_markdown(path: Path, payload: dict) -> None:
|
|
rows = [
|
|
"# GTL strength-confirmation sensitivity",
|
|
"",
|
|
f"Status: **{payload['status']}** ",
|
|
f"Holdout split: `{payload['holdout_split']}` ",
|
|
f"Completed arms: {len(payload['arms'])}/{payload['arm_count']}",
|
|
"",
|
|
"| Arm | Qualified | 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 {}
|
|
screen = arm.get("screen") or {}
|
|
rows.append(
|
|
"| "
|
|
+ " | ".join((
|
|
arm["name"],
|
|
str(arm.get("qualified") or "-"),
|
|
common._fmt(full.get("sharpe")),
|
|
common._fmt(full.get("cagr_pct"), 1),
|
|
common._fmt(full.get("max_drawdown_pct"), 1),
|
|
str(full.get("trades") or "-"),
|
|
common._fmt((holdout.get("train") or {}).get("sharpe")),
|
|
common._fmt((holdout.get("test") or {}).get("sharpe")),
|
|
common._fmt(arm.get("qualified_net_avg_r_ex_top5"), 3),
|
|
f"{screen.get('passed', '-')}/{screen.get('total', '-')}",
|
|
))
|
|
+ " |"
|
|
)
|
|
rows.extend((
|
|
"",
|
|
"## Pre-registered interpretation",
|
|
"",
|
|
"The six original guardrails remain unchanged. A stable candidate requires at least two adjacent non-control scales to pass all six; an isolated passing scale is rejected as sensitivity, not promoted.",
|
|
"",
|
|
))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("\n".join(rows), encoding="utf-8")
|
|
|
|
|
|
def _stable_plateau_pairs(arms: list[dict]) -> list[list[str]]:
|
|
sensitivity = [arm for arm in arms if arm.get("name") != "control"]
|
|
return [
|
|
[left["name"], right["name"]]
|
|
for left, right in zip(sensitivity, sensitivity[1:], strict=False)
|
|
if (left.get("screen") or {}).get("advances")
|
|
and (right.get("screen") or {}).get("advances")
|
|
]
|
|
|
|
|
|
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
|
|
|
|
reference_control = _reference_arm(REFERENCE_TUNING, "control")
|
|
reference_1000 = _reference_arm(
|
|
REFERENCE_COMPOSITION,
|
|
"strength_intersection",
|
|
)
|
|
if reference_control is None or reference_1000 is None:
|
|
raise SystemExit(
|
|
"The completed GTL tuning and confirmation matrices are required "
|
|
"for control/replication checks"
|
|
)
|
|
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
|
default_out = (
|
|
ROOT / "reports" / f"backtest-{stamp[:8]}-gtl-strength-sensitivity.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-strength-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),
|
|
"workers": args.workers,
|
|
"holdout_split": args.holdout_split,
|
|
"arm_count": len(GTL_STRENGTH_ARMS),
|
|
"strength_scales": list(STRENGTH_SCALES),
|
|
"arms": [],
|
|
"promotion_rule": (
|
|
"At least two adjacent non-control scales must pass all six original "
|
|
"guardrails."
|
|
),
|
|
}
|
|
common._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_STRENGTH_ARMS, start=1):
|
|
name = arm["name"]
|
|
output = work_dir / f"{index:02d}-{name}.json"
|
|
arm_outputs.append(output)
|
|
print(f"\n[{index}/{len(GTL_STRENGTH_ARMS)}] {name}", flush=True)
|
|
print(f" {arm['description']}", flush=True)
|
|
composition._run_arm(
|
|
arm,
|
|
snapshot,
|
|
args.workers,
|
|
args.holdout_split,
|
|
output,
|
|
)
|
|
report = common._load_report(output)
|
|
compact = common._compact_arm(
|
|
report,
|
|
composition._config(arm),
|
|
control_report,
|
|
)
|
|
compact["description"] = arm["description"]
|
|
if control_report is None:
|
|
control_report = report
|
|
compact["screen"] = {
|
|
"checks": {}, "passed": 0, "total": 0, "advances": False,
|
|
}
|
|
if composition._signature(compact) != composition._signature(
|
|
reference_control
|
|
):
|
|
raise RuntimeError(
|
|
"Strength sensitivity control does not reproduce frozen GTL control"
|
|
)
|
|
payload["control_parity"] = "pass"
|
|
else:
|
|
compact["screen"] = common._screen_arm(
|
|
compact,
|
|
payload["arms"][0],
|
|
)
|
|
if name == "strength_1000_intersection":
|
|
if composition._signature(compact) != composition._signature(
|
|
reference_1000
|
|
):
|
|
raise RuntimeError(
|
|
"Strength-1000 arm does not reproduce the completed composition result"
|
|
)
|
|
payload["strength_1000_replication"] = "pass"
|
|
payload["arms"].append(compact)
|
|
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
|
|
common._write_json(out_path, payload)
|
|
_write_markdown(markdown_path, payload)
|
|
except BaseException 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)
|
|
common._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["stable_plateau_pairs"] = _stable_plateau_pairs(payload["arms"])
|
|
payload["stable_candidate"] = bool(payload["stable_plateau_pairs"])
|
|
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)
|
|
common._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 strength sensitivity complete.")
|
|
print(f" JSON: {out_path}")
|
|
print(f" Markdown: {markdown_path}")
|
|
if payload["stable_candidate"]:
|
|
pairs = [" + ".join(pair) for pair in payload["stable_plateau_pairs"]]
|
|
print(" Stable passing plateau: " + "; ".join(pairs))
|
|
else:
|
|
print(" No stable adjacent passing plateau.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|