diff --git a/README.md b/README.md index 804ee9c..5994b03 100644 --- a/README.md +++ b/README.md @@ -486,6 +486,20 @@ the run fails, they remain available for diagnosis; pass `--keep-arm-reports` to retain them after success too. No arm changes live scanner defaults or deploys anything. +The replacement matrix found no winning single constant. Its evidence-selected +follow-up keeps control geometry and isolates retained versus added cohorts in +one 13-arm confirmation/union run: + +```bash +# macOS/Linux +.venv/bin/python scripts/run_gtl_confirmation_matrix.py \ + backtest_snapshots/prod.sqlite --workers 12 +``` + +It first verifies exact control parity with the completed tuning matrix, then +checkpoints consolidated JSON and Markdown reports under +`reports/backtest-YYYYMMDD-gtl-confirmation-matrix.*`. + ### Reading a local backtest report The deployed **Signals → Track Record** page is deliberately trimmed to validation diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index dfecd1f..c718cee 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -111,6 +111,7 @@ STRUCTURAL_OVERLAY_WEIGHT = 0.05 STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score" EXPLICIT_TARGET_LADDER_VARIANT = "explicit_target_ladder" GTL_TUNING_VARIANT = "gtl_tuning" +GTL_CONFIRMATION_VARIANT = "gtl_confirmation" RANGE_RESIDUAL_VARIANTS = { "rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_primary2", @@ -167,6 +168,23 @@ class GTLResearchConfig: strength_scale=self.strength_scale, ) + +@dataclass(frozen=True) +class GTLConfirmationConfig: + """Research-only composition of the frozen GTL and tuned variants.""" + + name: str = "control" + mode: str = "intersection" + confirmations: tuple[GTLResearchConfig, ...] = () + + def __post_init__(self) -> None: + if not self.name.strip(): + raise ValueError("GTL confirmation config name must not be empty") + if self.mode not in {"intersection", "union"}: + raise ValueError("GTL confirmation mode must be intersection or union") + if self.mode == "union" and len(self.confirmations) != 1: + raise ValueError("GTL union mode requires exactly one tuned variant") + # Cross-sectional signal evaluation (factor IC). Each candidate signal is a # point-in-time number computed from closes alone (sentiment/fundamentals have no # history here), sampled one as-of per ISO week, and graded by how its rank @@ -222,6 +240,7 @@ SR_RESEARCH_VARIANTS = { "legacy_range_grid_neutral", EXPLICIT_TARGET_LADDER_VARIANT, GTL_TUNING_VARIANT, + GTL_CONFIRMATION_VARIANT, STRUCTURAL_OVERLAY_VARIANT, *RANGE_FACTOR_VARIANTS, } @@ -261,6 +280,47 @@ def _gtl_research_config() -> GTLResearchConfig: return _parse_gtl_research_config(os.getenv("BACKTEST_GTL_CONFIG", "")) +@lru_cache(maxsize=32) +def _parse_gtl_confirmation_config(raw: str) -> GTLConfirmationConfig: + """Parse one composition arm for the offline confirmation matrix.""" + if not raw.strip(): + return GTLConfirmationConfig() + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + raise ValueError("BACKTEST_GTL_CONFIRM_CONFIG must be valid JSON") from exc + if not isinstance(payload, dict): + raise ValueError("BACKTEST_GTL_CONFIRM_CONFIG must be a JSON object") + allowed = {"name", "mode", "confirmations"} + unknown = sorted(set(payload) - allowed) + if unknown: + raise ValueError( + f"Unknown GTL confirmation config fields: {', '.join(unknown)}" + ) + raw_confirmations = payload.get("confirmations", []) + if not isinstance(raw_confirmations, list): + raise ValueError("GTL confirmations must be a JSON array") + confirmations: list[GTLResearchConfig] = [] + for item in raw_confirmations: + if not isinstance(item, dict): + raise ValueError("Each GTL confirmation must be a JSON object") + confirmations.append(_parse_gtl_research_config(json.dumps(item))) + try: + return GTLConfirmationConfig( + name=payload.get("name", "control"), + mode=payload.get("mode", "intersection"), + confirmations=tuple(confirmations), + ) + except (AttributeError, TypeError, ValueError) as exc: + raise ValueError(f"Invalid GTL confirmation config: {exc}") from exc + + +def _gtl_confirmation_config() -> GTLConfirmationConfig: + return _parse_gtl_confirmation_config( + os.getenv("BACKTEST_GTL_CONFIRM_CONFIG", "") + ) + + def _sr_detector_variant(sr_variant: str) -> str: """Map factor-gated research arms to the detector they hold fixed.""" if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}: @@ -302,6 +362,7 @@ def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float: "production_range504", EXPLICIT_TARGET_LADDER_VARIANT, GTL_TUNING_VARIANT, + GTL_CONFIRMATION_VARIANT, STRUCTURAL_OVERLAY_VARIANT, } or sr_variant.endswith("_legacy_primary") @@ -416,6 +477,7 @@ def _window_setups( activation: dict, *, sr_variant: str | None = None, + gtl_research_config: GTLResearchConfig | None = None, ) -> list[dict]: """Rebuild the setup(s) at the last bar of ``window_records`` (the as-of date), using only those bars. Returns one dict per tradeable direction.""" @@ -444,7 +506,11 @@ def _window_setups( sr_variant = sr_variant or _sr_research_variant() detector_variant = _sr_detector_variant(sr_variant) - gtl_config = _gtl_research_config() if sr_variant == GTL_TUNING_VARIANT else None + gtl_config = ( + gtl_research_config or _gtl_research_config() + if sr_variant == GTL_TUNING_VARIANT + else None + ) range_504_log = _range_504_log(highs, lows) if sr_variant == "legacy_geometry_neutral": detected_levels = detect_sr_levels_legacy( @@ -683,6 +749,87 @@ def _structural_overlay_window_setups( return tagged +def _gtl_confirmation_window_setups( + window_records: list, + config: dict, + activation: dict, + *, + confirmation_config: GTLConfirmationConfig | None = None, +) -> list[dict]: + """Compose tuned GTLs around the frozen ladder without changing it silently. + + ``intersection`` retains the frozen setup geometry and requires every tuned + variant to clear the core gate in the same direction. ``union`` preserves a + frozen setup whenever it already clears the core gate, and otherwise admits + the one tuned variant's setup. This makes retained, removed, and added + cohorts explicit instead of conflating them in a replacement arm. + """ + research = confirmation_config or _gtl_confirmation_config() + production = _window_setups( + window_records, + config, + activation, + sr_variant=EXPLICIT_TARGET_LADDER_VARIANT, + ) + tuned_sets = [ + _window_setups( + window_records, + config, + activation, + sr_variant=GTL_TUNING_VARIANT, + gtl_research_config=tuned_config, + ) + for tuned_config in research.confirmations + ] + tuned_by_direction = [ + {row["direction"]: row for row in rows} + for rows in tuned_sets + ] + + def annotate(row: dict, passes: list[bool], source: str) -> dict: + tagged = dict(row) + tagged["sr_variant"] = GTL_CONFIRMATION_VARIANT + tagged["gtl_confirmation_name"] = research.name + tagged["gtl_confirmation_mode"] = research.mode + tagged["gtl_confirmation_source"] = source + tagged["gtl_confirmation_passes"] = passes + tagged["gtl_confirmation_all_pass"] = all(passes) + return tagged + + if research.mode == "intersection": + tagged: list[dict] = [] + for production_row in production: + direction = production_row["direction"] + passes = [ + bool(rows.get(direction) and rows[direction].get("meets_core")) + for rows in tuned_by_direction + ] + row = annotate(production_row, passes, "control") + row["meets_core"] = bool(production_row.get("meets_core")) and all( + passes + ) + tagged.append(row) + return tagged + + # Union mode is validated to contain exactly one tuned variant. Keep one + # setup per direction: frozen geometry wins whenever it already qualifies; + # tuned geometry is used only for a genuinely added core-qualified setup. + production_by_direction = {row["direction"]: row for row in production} + tuned_by_dir = tuned_by_direction[0] + tagged = [] + for direction in sorted(set(production_by_direction) | set(tuned_by_dir)): + production_row = production_by_direction.get(direction) + tuned_row = tuned_by_dir.get(direction) + tuned_pass = bool(tuned_row and tuned_row.get("meets_core")) + if production_row is not None and production_row.get("meets_core"): + tagged.append(annotate(production_row, [tuned_pass], "control")) + elif tuned_pass and tuned_row is not None: + tagged.append(annotate(tuned_row, [True], "tuned_addition")) + elif production_row is not None: + tagged.append(annotate(production_row, [tuned_pass], "control")) + return tagged + + def _stop_fill_r(direction: str, entry: float, stop: float, bar) -> float: """Realized R when the stop is hit on ``bar``: filled at the stop, or at the bar's open when price gapped through it — so a gap can lose more than −1R, @@ -787,16 +934,17 @@ def _replay_ticker( ) vol_6m = _realized_vol_6m(closes, len(window) - 1) - setups = ( - _structural_overlay_window_setups(window, config, activation) - if sr_variant == STRUCTURAL_OVERLAY_VARIANT - else _window_setups( + if sr_variant == STRUCTURAL_OVERLAY_VARIANT: + setups = _structural_overlay_window_setups(window, config, activation) + elif sr_variant == GTL_CONFIRMATION_VARIANT: + setups = _gtl_confirmation_window_setups(window, config, activation) + else: + setups = _window_setups( window, config, activation, sr_variant=sr_variant, ) - ) for s in setups: outcome, outcome_date = evaluate_setup_against_bars( s["direction"], s["stop"], s["target"], forward_bars, HORIZON @@ -862,6 +1010,13 @@ def _replay_ticker( "structural_overlay_gate_level_count": s.get( "structural_overlay_gate_level_count" ), + "gtl_confirmation_name": s.get("gtl_confirmation_name"), + "gtl_confirmation_mode": s.get("gtl_confirmation_mode"), + "gtl_confirmation_source": s.get("gtl_confirmation_source"), + "gtl_confirmation_passes": s.get("gtl_confirmation_passes"), + "gtl_confirmation_all_pass": s.get( + "gtl_confirmation_all_pass" + ), "outcome": outcome, "target_hit": target_hit, "realized_r": realized_r, @@ -934,6 +1089,9 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: range_logs: list[float] = [] overlay_rows = 0 overlay_pass = 0 + confirmation_rows = 0 + confirmation_pass = 0 + confirmation_tuned_additions = 0 for cand in candidates: sources = list(cand.get("primary_sources") or []) for source in sources: @@ -949,6 +1107,12 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: if cand.get("structural_overlay_pass") is not None: overlay_rows += 1 overlay_pass += int(bool(cand["structural_overlay_pass"])) + if cand.get("gtl_confirmation_all_pass") is not None: + confirmation_rows += 1 + confirmation_pass += int(bool(cand["gtl_confirmation_all_pass"])) + confirmation_tuned_additions += int( + cand.get("gtl_confirmation_source") == "tuned_addition" + ) def avg(values: list[float] | list[int]) -> float | None: return round(sum(values) / len(values), 3) if values else None @@ -973,6 +1137,9 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: "structural_overlay_weight": ( STRUCTURAL_OVERLAY_WEIGHT if overlay_rows else None ), + "gtl_confirmation_rows": confirmation_rows, + "gtl_confirmation_pass": confirmation_pass, + "gtl_confirmation_tuned_additions": confirmation_tuned_additions, } @@ -1019,6 +1186,15 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d "structural_overlay_gate_level_count": int( cand.get("structural_overlay_gate_level_count", 0) or 0 ), + "gtl_confirmation_name": cand.get("gtl_confirmation_name"), + "gtl_confirmation_mode": cand.get("gtl_confirmation_mode"), + "gtl_confirmation_source": cand.get("gtl_confirmation_source"), + "gtl_confirmation_passes": list( + cand.get("gtl_confirmation_passes") or [] + ), + "gtl_confirmation_all_pass": cand.get( + "gtl_confirmation_all_pass" + ), "rr": round(float(cand.get("rr", 0.0)), 6), "primary_prob": round(float(cand.get("primary_prob", 0.0)), 6), "primary_sources": list(cand.get("primary_sources") or []), @@ -3430,6 +3606,11 @@ async def run_backtest( if _sr_research_variant() == GTL_TUNING_VARIANT else None ), + "gtl_confirmation_config": ( + asdict(_gtl_confirmation_config()) + if _sr_research_variant() == GTL_CONFIRMATION_VARIANT + else None + ), "range_factor_lookback": RANGE_FACTOR_LOOKBACK, "range_factor_min_log": RANGE_FACTOR_MIN_LOG, "range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4), diff --git a/docs/research/sr-levels-and-exits.md b/docs/research/sr-levels-and-exits.md index 32a5e48..ef6ab17 100644 --- a/docs/research/sr-levels-and-exits.md +++ b/docs/research/sr-levels-and-exits.md @@ -845,6 +845,40 @@ the top 5%. Passing identifies a candidate for forward paper validation, not an automatic deployment. The post-2024 interval has already influenced this research, so the split is a robustness check rather than a pristine holdout. +Result on the 2026-07-13 snapshot: **20/20 arms completed; no replacement arm +passed all six checks.** The frozen control remained best on full-period Sharpe +(2.03), CAGR (50.0%), and post-2024 Sharpe (2.78), with 321 production trades +and 21.4% drawdown. The closest replacement, 0.25% touch padding, still fell to +Sharpe 1.94 / CAGR 47.9%. This rejects direct constant replacement; the inherited +behavior is not explained by one obvious GTL knob. + +The paired cohorts do expose a narrower mechanism worth testing: + +| Variant | Retained control setups | Added by variant | Removed from control | +|---|---|---|---| +| 0.25% touch | 1,049 at +0.214R (+0.058 ex-top-5%) | 18 at -0.235R | 37 at -0.056R | +| Strength 1000 | 1,037 at +0.225R (+0.069) | 122 at +0.301R (+0.152) | 49 at +0.142R (-0.044) | +| 0.25% merge | 791 at +0.238R (+0.078) | 365 at +0.161R (+0.023) | 295 at +0.129R (-0.008) | +| Grid without pivots | 428 at +0.232R (+0.100) | 392 at +0.176R (+0.046) | 658 at +0.208R (+0.041) | + +Replacement mixes the retained and added cohorts and also discards the removed +cohort, so its portfolio result cannot say which part helped. The follow-up +`gtl_confirmation` matrix therefore preserves frozen control geometry and +decomposes each selected variant into: + +- **intersection** — only control setups also core-qualified by the variant; +- **union** — all core-qualified control setups plus genuinely added variant + setups, using tuned geometry only for those additions. + +It also tests pre-registered intersections among the three high-breadth +confirmers. Its control path must exactly reproduce the completed tuning +matrix before any research arm is accepted. + +```bash +.venv/bin/python scripts/run_gtl_confirmation_matrix.py \ + backtest_snapshots/prod.sqlite --workers 12 +``` + The post-2024 window has been opened and is now analysis data, not a valid final promotion holdout. These arms can isolate mechanism, but neither may ship without new future data or a separately pre-registered walk-forward protocol. diff --git a/reports/README.md b/reports/README.md index 0fecf75..f06b495 100644 --- a/reports/README.md +++ b/reports/README.md @@ -69,3 +69,13 @@ The next decision point is generated by the one-command GTL parameter run: These two consolidated files are the reports worth retaining. The runner uses a hidden temporary directory for its 20 full per-arm reports and removes it after successful consolidation unless `--keep-arm-reports` is supplied. + +The completed 2026-07-13 matrix found no single-parameter replacement that +passed all robustness checks. Retain its JSON and Markdown as that decision +point. The evidence-selected retained-versus-added decomposition writes: + +- `backtest-YYYYMMDD-gtl-confirmation-matrix.json` +- `backtest-YYYYMMDD-gtl-confirmation-matrix.md` + +Those become the next decision point; detailed per-arm reports remain temporary +unless explicitly retained. diff --git a/scripts/run_backtest_snapshot.py b/scripts/run_backtest_snapshot.py index 21330e3..77f3487 100644 --- a/scripts/run_backtest_snapshot.py +++ b/scripts/run_backtest_snapshot.py @@ -62,6 +62,7 @@ def _parse_args() -> argparse.Namespace: "production_structural_overlay", "explicit_target_ladder", "gtl_tuning", + "gtl_confirmation", ), default=None, help="Research-only S/R detector/gate arm.", @@ -71,6 +72,14 @@ def _parse_args() -> argparse.Namespace: default=None, help="Research-only GTL configuration as a JSON object (requires --sr-variant gtl_tuning).", ) + parser.add_argument( + "--gtl-confirm-config", + default=None, + help=( + "Research-only GTL intersection/union configuration as JSON " + "(requires --sr-variant gtl_confirmation)." + ), + ) parser.add_argument( "--entry-start", default=None, @@ -210,6 +219,12 @@ async def _main() -> None: if args.sr_variant != "gtl_tuning": raise SystemExit("--gtl-config requires --sr-variant gtl_tuning") os.environ["BACKTEST_GTL_CONFIG"] = args.gtl_config + if args.gtl_confirm_config: + if args.sr_variant != "gtl_confirmation": + raise SystemExit( + "--gtl-confirm-config requires --sr-variant gtl_confirmation" + ) + os.environ["BACKTEST_GTL_CONFIRM_CONFIG"] = args.gtl_confirm_config if args.entry_start: os.environ["BACKTEST_ENTRY_START"] = args.entry_start if args.entry_end: diff --git a/scripts/run_gtl_confirmation_matrix.py b/scripts/run_gtl_confirmation_matrix.py new file mode 100644 index 0000000..cc8ee2f --- /dev/null +++ b/scripts/run_gtl_confirmation_matrix.py @@ -0,0 +1,360 @@ +"""Run the evidence-selected GTL confirmation/union matrix with one command. + +The first GTL tuning matrix tested replacements. This follow-up decomposes the +four informative variants into retained-only intersections and control-plus- +addition unions while preserving frozen control geometry wherever possible. +""" + +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] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts import run_gtl_tuning_matrix as common # noqa: E402 + +RUNNER = ROOT / "scripts" / "run_backtest_snapshot.py" +REFERENCE_MATRIX = ROOT / "reports" / "backtest-20260713-gtl-tuning-matrix.json" + +TOUCH = {"name": "touch_0_25pct", "touch_tolerance": 0.0025} +STRENGTH = {"name": "strength_1000", "strength_scale": 1000.0} +MERGE = {"name": "merge_0_25pct", "merge_tolerance": 0.0025} +GRID_ONLY = {"name": "pivots_none", "include_pivots": False} + + +def _arm( + name: str, + description: str, + mode: str, + *confirmations: dict[str, Any], +) -> dict[str, Any]: + return { + "name": name, + "description": description, + "mode": mode, + "confirmations": list(confirmations), + } + + +# Pre-registered from the replacement matrix's paired cohorts. Replacement +# arms that plainly removed strong control setups or added weak cohorts are not +# repeated here. +GTL_CONFIRMATION_ARMS: tuple[dict[str, Any], ...] = ( + _arm("control", "Frozen explicit GTL; composition-path parity control.", "intersection"), + _arm( + "touch_intersection", + "Retain control setups also qualified with 0.25% touch padding.", + "intersection", + TOUCH, + ), + _arm( + "touch_union", + "Keep control and admit additions from 0.25% touch padding.", + "union", + TOUCH, + ), + _arm( + "strength_intersection", + "Retain control setups also qualified at strength scale 1000.", + "intersection", + STRENGTH, + ), + _arm( + "strength_union", + "Keep control and admit additions from strength scale 1000.", + "union", + STRENGTH, + ), + _arm( + "merge_intersection", + "Retain control setups also qualified with 0.25% proposal merging.", + "intersection", + MERGE, + ), + _arm( + "merge_union", + "Keep control and admit additions from 0.25% proposal merging.", + "union", + MERGE, + ), + _arm( + "grid_intersection", + "Retain control setups also qualified by the range grid without pivots.", + "intersection", + GRID_ONLY, + ), + _arm( + "grid_union", + "Keep control and admit additions from the range grid without pivots.", + "union", + GRID_ONLY, + ), + _arm( + "touch_strength_intersection", + "Require both tighter-touch and faster-strength confirmation.", + "intersection", + TOUCH, + STRENGTH, + ), + _arm( + "touch_merge_intersection", + "Require both tighter-touch and tighter-merge confirmation.", + "intersection", + TOUCH, + MERGE, + ), + _arm( + "strength_merge_intersection", + "Require both faster-strength and tighter-merge confirmation.", + "intersection", + STRENGTH, + MERGE, + ), + _arm( + "touch_strength_merge_intersection", + "Require all three high-breadth confirmation variants.", + "intersection", + TOUCH, + STRENGTH, + MERGE, + ), +) + + +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-confirmation-matrix.json." + ), + ) + parser.add_argument("--keep-arm-reports", action="store_true") + return parser.parse_args() + + +def _config(arm: dict[str, Any]) -> dict[str, Any]: + return { + "name": arm["name"], + "mode": arm["mode"], + "confirmations": arm["confirmations"], + } + + +def _signature(arm: dict) -> dict: + return { + "candidates": arm.get("candidates"), + "qualified": arm.get("qualified"), + "qualified_net_avg_r": arm.get("qualified_net_avg_r"), + "qualified_net_avg_r_ex_top5": arm.get("qualified_net_avg_r_ex_top5"), + "full_book": arm.get("full_book"), + "holdout": arm.get("holdout"), + } + + +def _reference_control() -> dict | None: + if not REFERENCE_MATRIX.exists(): + return None + with REFERENCE_MATRIX.open(encoding="utf-8") as handle: + payload = json.load(handle) + return next( + (arm for arm in payload.get("arms") or [] if arm.get("name") == "control"), + None, + ) + + +def _write_markdown(path: Path, payload: dict) -> None: + rows = [ + "# GTL confirmation/union matrix", + "", + f"Status: **{payload['status']}** ", + f"Holdout split: `{payload['holdout_split']}` ", + f"Completed arms: {len(payload['arms'])}/{payload['arm_count']}", + "", + "| Arm | Mode | 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"], + arm["config"]["mode"], + 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(( + "", + "## Interpretation guardrail", + "", + "Intersections test the retained control cohort; unions test control plus genuinely added setups. The post-2024 interval is a robustness check, not a pristine holdout. Passing does not authorize deployment.", + "", + )) + 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: Path, + workers: int, + holdout_split: str, + output: Path, +) -> None: + command = [ + sys.executable, + str(RUNNER), + str(snapshot), + "--workers", str(workers), + "--allow-spawn", + "--sr-variant", "gtl_confirmation", + "--gtl-confirm-config", json.dumps(_config(arm), 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-confirmation-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-confirmation-work-{stamp}" + work_dir.mkdir(parents=True, exist_ok=False) + + reference = _reference_control() + 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_CONFIRMATION_ARMS), + "reference_matrix": str(REFERENCE_MATRIX) if reference else None, + "arms": [], + } + 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_CONFIRMATION_ARMS, start=1): + name = arm["name"] + output = work_dir / f"{index:02d}-{name}.json" + arm_outputs.append(output) + print(f"\n[{index}/{len(GTL_CONFIRMATION_ARMS)}] {name}", flush=True) + print(f" {arm['description']}", flush=True) + _run_arm(arm, snapshot, args.workers, args.holdout_split, output) + report = common._load_report(output) + compact = common._compact_arm(report, _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 reference is not None and _signature(compact) != _signature(reference): + raise RuntimeError( + "Confirmation-path control does not reproduce the frozen GTL matrix control" + ) + payload["control_parity"] = "pass" if reference is not None else "not_checked" + else: + compact["screen"] = common._screen_arm(compact, payload["arms"][0]) + 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["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 confirmation matrix complete.") + print(f" JSON: {out_path}") + print(f" Markdown: {markdown_path}") + if payload["advancing_arms"]: + print(" Passing arms: " + ", ".join(payload["advancing_arms"])) + else: + print(" No arm passed every pre-registered screen.") + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index e35172b..18df51f 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -805,6 +805,7 @@ def test_sr_research_variant_is_explicit_and_validated(monkeypatch): "production_structural_overlay", "explicit_target_ladder", "gtl_tuning", + "gtl_confirmation", ): monkeypatch.setenv("BACKTEST_SR_VARIANT", variant) assert bt._sr_research_variant() == variant @@ -864,6 +865,7 @@ def test_residual_arms_change_only_the_primary_rr_floor(): activation, ) == 1.5 assert bt._primary_min_rr_for_variant("gtl_tuning", activation) == 1.5 + assert bt._primary_min_rr_for_variant("gtl_confirmation", activation) == 1.5 def test_gtl_research_config_parses_and_rejects_unknown_fields(): @@ -883,6 +885,29 @@ def test_gtl_research_config_parses_and_rejects_unknown_fields(): bt._parse_gtl_research_config("{") +def test_gtl_confirmation_config_parses_and_validates_composition(): + config = bt._parse_gtl_confirmation_config(json.dumps({ + "name": "touch_strength_intersection", + "mode": "intersection", + "confirmations": [ + {"name": "touch", "touch_tolerance": 0.0025}, + {"name": "strength", "strength_scale": 1000.0}, + ], + })) + assert config.name == "touch_strength_intersection" + assert config.mode == "intersection" + assert len(config.confirmations) == 2 + assert config.confirmations[0].touch_tolerance == 0.0025 + assert config.confirmations[1].strength_scale == 1000.0 + + with pytest.raises(ValueError, match="exactly one tuned variant"): + bt._parse_gtl_confirmation_config(json.dumps({ + "name": "invalid_union", + "mode": "union", + "confirmations": [], + })) + + def test_structural_overlay_tags_production_geometry_without_replacing_it(monkeypatch): production = { "direction": "long", @@ -1031,6 +1056,77 @@ def test_window_setups_routes_gtl_tuning_config(monkeypatch): assert captured["config"].grid_bins == 12 +def test_gtl_confirmation_intersection_keeps_control_geometry(monkeypatch): + production = [{ + "direction": "long", + "target": 111.0, + "rr": 2.2, + "meets_core": True, + "sr_variant": bt.EXPLICIT_TARGET_LADDER_VARIANT, + }] + + def fake_window_setups(*args, sr_variant=None, gtl_research_config=None, **kwargs): + if sr_variant == bt.EXPLICIT_TARGET_LADDER_VARIANT: + return production + assert sr_variant == bt.GTL_TUNING_VARIANT + return [{ + "direction": "long", + "target": 115.0, + "rr": 3.0, + "meets_core": gtl_research_config.name == "pass", + }] + + monkeypatch.setattr(bt, "_window_setups", fake_window_setups) + research = bt.GTLConfirmationConfig( + name="two_filters", + mode="intersection", + confirmations=( + bt.GTLResearchConfig(name="pass"), + bt.GTLResearchConfig(name="fail"), + ), + ) + rows = bt._gtl_confirmation_window_setups( + [], {}, {}, confirmation_config=research + ) + + assert len(rows) == 1 + assert rows[0]["target"] == 111.0 + assert rows[0]["rr"] == 2.2 + assert rows[0]["meets_core"] is False + assert rows[0]["gtl_confirmation_passes"] == [True, False] + assert production[0]["meets_core"] is True + + +def test_gtl_confirmation_union_uses_tuned_geometry_only_for_addition(monkeypatch): + production = [ + {"direction": "long", "target": 111.0, "meets_core": False}, + {"direction": "short", "target": 90.0, "meets_core": True}, + ] + tuned = [ + {"direction": "long", "target": 115.0, "meets_core": True}, + {"direction": "short", "target": 85.0, "meets_core": False}, + ] + + def fake_window_setups(*args, sr_variant=None, **kwargs): + return production if sr_variant == bt.EXPLICIT_TARGET_LADDER_VARIANT else tuned + + monkeypatch.setattr(bt, "_window_setups", fake_window_setups) + research = bt.GTLConfirmationConfig( + name="strength_union", + mode="union", + confirmations=(bt.GTLResearchConfig(name="strength"),), + ) + rows = bt._gtl_confirmation_window_setups( + [], {}, {}, confirmation_config=research + ) + by_direction = {row["direction"]: row for row in rows} + + assert by_direction["long"]["target"] == 115.0 + assert by_direction["long"]["gtl_confirmation_source"] == "tuned_addition" + assert by_direction["short"]["target"] == 90.0 + assert by_direction["short"]["gtl_confirmation_source"] == "control" + + @pytest.mark.parametrize( "variant", ["legacy_geometry_neutral", "legacy_range_grid_neutral"], diff --git a/tests/unit/test_gtl_confirmation_matrix.py b/tests/unit/test_gtl_confirmation_matrix.py new file mode 100644 index 0000000..d6d763a --- /dev/null +++ b/tests/unit/test_gtl_confirmation_matrix.py @@ -0,0 +1,39 @@ +"""Tests for the evidence-selected GTL composition matrix.""" + +from __future__ import annotations + +from scripts import run_gtl_confirmation_matrix as matrix + + +def test_confirmation_matrix_is_pre_registered_and_well_formed(): + assert len(matrix.GTL_CONFIRMATION_ARMS) == 13 + assert matrix.GTL_CONFIRMATION_ARMS[0]["name"] == "control" + names = [arm["name"] for arm in matrix.GTL_CONFIRMATION_ARMS] + assert len(names) == len(set(names)) + + for arm in matrix.GTL_CONFIRMATION_ARMS: + config = matrix._config(arm) + assert config["mode"] in {"intersection", "union"} + if config["mode"] == "union": + assert len(config["confirmations"]) == 1 + for confirmation in config["confirmations"]: + assert confirmation["name"] in { + "touch_0_25pct", + "strength_1000", + "merge_0_25pct", + "pivots_none", + } + + +def test_signature_uses_only_control_parity_fields(): + arm = { + "candidates": 100, + "qualified": 10, + "qualified_net_avg_r": 0.2, + "qualified_net_avg_r_ex_top5": 0.05, + "full_book": {"sharpe": 2.0}, + "holdout": {"train": {"sharpe": 1.0}}, + "unrelated": "ignored", + } + assert "unrelated" not in matrix._signature(arm) + assert matrix._signature(arm)["full_book"] == {"sharpe": 2.0}