diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index 11f4fa8..e0a9fa5 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -94,6 +94,12 @@ HORIZON = 30 # trading days to resolve an outcome (matches the evaluat ATR_MULTIPLIER = 1.5 RANGE_FACTOR_LOOKBACK = 504 RANGE_FACTOR_MIN_LOG = 1.0 # approximately a 2.7x high/low span +STRUCTURAL_OVERLAY_VARIANT = "production_structural_overlay" +STRUCTURAL_OVERLAY_SOURCE_VARIANT = ( + "rewrite_range504_structural_legacy_primary" +) +STRUCTURAL_OVERLAY_WEIGHT = 0.05 +STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score" RANGE_RESIDUAL_VARIANTS = { "rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_primary2", @@ -157,6 +163,7 @@ SR_RESEARCH_VARIANTS = { "legacy_traffic_grid_only", "legacy_range_grid_touch", "legacy_range_grid_neutral", + STRUCTURAL_OVERLAY_VARIANT, *RANGE_FACTOR_VARIANTS, } @@ -172,7 +179,7 @@ def _sr_research_variant() -> str: def _sr_detector_variant(sr_variant: str) -> str: """Map factor-gated research arms to the detector they hold fixed.""" - if sr_variant == "production_range504": + if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}: return "production_control" if ( sr_variant == "rewrite_range504_legacy_primary" @@ -206,7 +213,11 @@ def _range_factor_allows(sr_variant: str, range_504_log: float) -> bool: def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float: """Preserve the deployed selector except in explicit primary-2 research.""" if ( - sr_variant in {"production_control", "production_range504"} + sr_variant in { + "production_control", + "production_range504", + STRUCTURAL_OVERLAY_VARIANT, + } or sr_variant.endswith("_legacy_primary") or sr_variant.startswith("legacy_") ): @@ -317,6 +328,8 @@ def _window_setups( window_records: list, config: dict, activation: dict, + *, + sr_variant: str | 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.""" @@ -343,7 +356,7 @@ def _window_setups( if atr <= 0: return [] - sr_variant = _sr_research_variant() + sr_variant = sr_variant or _sr_research_variant() detector_variant = _sr_detector_variant(sr_variant) range_504_log = _range_504_log(highs, lows) if sr_variant == "legacy_geometry_neutral": @@ -508,6 +521,55 @@ def _window_setups( return out +def _structural_overlay_window_setups( + window_records: list, + config: dict, + activation: dict, +) -> list[dict]: + """Production setups tagged by the clean structural/range candidate. + + The returned population and setup geometry remain production-identical. + The clean detector is only a point-in-time feature, so this arm can test a + ranking overlay without silently changing admission breadth or targets. + """ + production = _window_setups( + window_records, + config, + activation, + sr_variant="production_control", + ) + if not production: + return [] + structural = _window_setups( + window_records, + config, + activation, + sr_variant=STRUCTURAL_OVERLAY_SOURCE_VARIANT, + ) + structural_by_direction = {row["direction"]: row for row in structural} + tagged: list[dict] = [] + for production_row in production: + row = dict(production_row) + structural_row = structural_by_direction.get(row["direction"]) + row["sr_variant"] = STRUCTURAL_OVERLAY_VARIANT + row["structural_overlay_pass"] = bool( + structural_row and structural_row.get("meets_core") + ) + row["structural_overlay_rr"] = ( + float(structural_row["rr"]) if structural_row is not None else None + ) + row["structural_overlay_sources"] = ( + list(structural_row.get("primary_sources") or []) + if structural_row is not None else [] + ) + row["structural_overlay_gate_level_count"] = ( + int(structural_row.get("gate_level_count", 0) or 0) + if structural_row is not None else 0 + ) + tagged.append(row) + 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, @@ -595,6 +657,7 @@ def _replay_ticker( return candidates entry_start, entry_end = _backtest_entry_bounds() + sr_variant = _sr_research_variant() for i in range(MIN_LOOKBACK - 1, n - HORIZON, STEP_DAYS): as_of = records[i].date if entry_start is not None and as_of < entry_start: @@ -611,7 +674,17 @@ def _replay_ticker( ) vol_6m = _realized_vol_6m(closes, len(window) - 1) - for s in _window_setups(window, config, activation): + setups = ( + _structural_overlay_window_setups(window, config, activation) + if sr_variant == STRUCTURAL_OVERLAY_VARIANT + else _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 ) @@ -670,6 +743,12 @@ def _replay_ticker( "range_504_log": s["range_504_log"], "range_504_ratio": s["range_504_ratio"], "range_factor_pass": s["range_factor_pass"], + "structural_overlay_pass": s.get("structural_overlay_pass"), + "structural_overlay_rr": s.get("structural_overlay_rr"), + "structural_overlay_sources": s.get("structural_overlay_sources"), + "structural_overlay_gate_level_count": s.get( + "structural_overlay_gate_level_count" + ), "outcome": outcome, "target_hit": target_hit, "realized_r": realized_r, @@ -740,6 +819,8 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: raw_counts: list[int] = [] gate_counts: list[int] = [] range_logs: list[float] = [] + overlay_rows = 0 + overlay_pass = 0 for cand in candidates: sources = list(cand.get("primary_sources") or []) for source in sources: @@ -752,6 +833,9 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: raw_counts.append(int(cand.get("raw_level_count", 0) or 0)) gate_counts.append(int(cand.get("gate_level_count", 0) or 0)) range_logs.append(float(cand.get("range_504_log", 0.0) or 0.0)) + if cand.get("structural_overlay_pass") is not None: + overlay_rows += 1 + overlay_pass += int(bool(cand["structural_overlay_pass"])) def avg(values: list[float] | list[int]) -> float | None: return round(sum(values) / len(values), 3) if values else None @@ -771,6 +855,11 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: "range_factor_pass": sum( 1 for value in range_logs if value >= RANGE_FACTOR_MIN_LOG ), + "structural_overlay_rows": overlay_rows, + "structural_overlay_pass": overlay_pass, + "structural_overlay_weight": ( + STRUCTURAL_OVERLAY_WEIGHT if overlay_rows else None + ), } @@ -798,6 +887,25 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d "meets_core": bool(cand.get("meets_core")), "momentum_percentile": round(float(percentile), 6), "strategy_rank": round(float(cand.get(RESIDUAL_HIGH_VOL_BLEND_KEY, 0.0) or 0.0), 6), + "production_rank": round( + float(cand.get(RESIDUAL_HIGH_VOL_BLEND_80_20_KEY, 0.0) or 0.0), + 6, + ), + "structural_overlay_pass": cand.get("structural_overlay_pass"), + "structural_overlay_score": ( + round(float(cand[STRUCTURAL_OVERLAY_SCORE_KEY]), 6) + if cand.get(STRUCTURAL_OVERLAY_SCORE_KEY) is not None else None + ), + "structural_overlay_rr": ( + round(float(cand["structural_overlay_rr"]), 6) + if cand.get("structural_overlay_rr") is not None else None + ), + "structural_overlay_sources": list( + cand.get("structural_overlay_sources") or [] + ), + "structural_overlay_gate_level_count": int( + cand.get("structural_overlay_gate_level_count", 0) or 0 + ), "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 []), @@ -1333,6 +1441,23 @@ def _assign_residual_high_vol_blend(candidates: list[dict]) -> None: ) +def _assign_structural_overlay_score(candidates: list[dict]) -> None: + """Conservative rank nudge for production setups confirmed by clean S/R.""" + for cand in candidates: + if cand.get("structural_overlay_pass") is None: + cand[STRUCTURAL_OVERLAY_SCORE_KEY] = None + continue + production_rank = cand.get(RESIDUAL_HIGH_VOL_BLEND_80_20_KEY) + if production_rank is None: + cand[STRUCTURAL_OVERLAY_SCORE_KEY] = None + continue + structural_score = 100.0 if cand["structural_overlay_pass"] else 0.0 + cand[STRUCTURAL_OVERLAY_SCORE_KEY] = ( + float(production_rank) * (1.0 - STRUCTURAL_OVERLAY_WEIGHT) + + structural_score * STRUCTURAL_OVERLAY_WEIGHT + ) + + def _momentum_qualifies(cand: dict, threshold: float) -> bool: """Whether a candidate clears the floors (meets_core) and the momentum gate. Threshold 0 disables the momentum gate (floors only). The gate is long-only: @@ -2185,6 +2310,7 @@ PORTFOLIO_MONITOR_LOOKBACKS: tuple[dict, ...] = ( ) PRODUCTION_PORTFOLIO_STRATEGY = "residual80_highvol80_20_atr3" +STRUCTURAL_OVERLAY_PORTFOLIO_STRATEGY = "production_structural_overlay5_atr3" PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = ( { "strategy": "legacy_residual80_hold", @@ -2218,6 +2344,25 @@ PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = ( ) +def _portfolio_monitor_strategies() -> tuple[dict, ...]: + """Add the frozen overlay only inside its explicit research arm.""" + if _sr_research_variant() != STRUCTURAL_OVERLAY_VARIANT: + return PORTFOLIO_MONITOR_STRATEGIES + return PORTFOLIO_MONITOR_STRATEGIES + ({ + "strategy": STRUCTURAL_OVERLAY_PORTFOLIO_STRATEGY, + "label": "Research: production gate + 5% clean-structure rank overlay", + "description": ( + "Production-qualified universe and live exit, ranked by 95% current " + "80/20 strategy rank plus 5% clean structural/range confirmation." + ), + "entry_variant": "residual80_highvol_blend80_20_fixed10", + "exit_policy": "atr_trail3", + "ranking_key": STRUCTURAL_OVERLAY_SCORE_KEY, + "use_live_config": True, + "is_production": False, + },) + + def _entry_variant_config(variant: str) -> dict | None: return next((cfg for cfg in STRATEGY_VARIANTS if cfg["variant"] == variant), None) @@ -2504,16 +2649,19 @@ def _portfolio_monitor( ) -> dict: latest_ord = max((max(cols[0]) for cols in prices.values() if cols[0]), default=None) rows: list[dict] = [] - for strategy in PORTFOLIO_MONITOR_STRATEGIES: + strategies = _portfolio_monitor_strategies() + for strategy in strategies: entry_cfg = _entry_variant_config(str(strategy["entry_variant"])) if entry_cfg is None: continue - ranking_key = str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]) - # The production row must replay the LIVE configuration: the runtime - # qualification flag (Admin activation settings) instead of the frozen - # research-variant gate, and the Admin exit policy instead of the - # hardcoded 3x-trail/30d defaults. Research rows stay frozen so they - # remain comparable across runs. + ranking_key = str( + strategy.get("ranking_key") + or entry_cfg.get("ranking_key") + or entry_cfg["percentile_key"] + ) + # Live-config rows replay the runtime qualification flag and Admin exit + # policy. The overlay opts into this deliberately so only ordering + # changes relative to the production row. use_live = bool(strategy.get("use_live_config")) exit_policy = str(strategy["exit_policy"]) row_hold_days = hold_days @@ -2554,6 +2702,7 @@ def _portfolio_monitor( "description": strategy["description"], "is_production": bool(strategy.get("is_production")), "entry_variant": strategy["entry_variant"], + "ranking_key": ranking_key, "exit_policy": exit_policy, "live_exit_mode": live_exit_mode, "lookback": lookback["lookback"], @@ -2569,7 +2718,7 @@ def _portfolio_monitor( "description": s["description"], "is_production": bool(s.get("is_production")), } - for s in PORTFOLIO_MONITOR_STRATEGIES + for s in strategies ], "lookbacks": [ {"lookback": lb["lookback"], "label": lb["label"]} @@ -2578,7 +2727,9 @@ def _portfolio_monitor( "runs": rows, "note": ( "Portfolio monitor runs supported named strategies across cached lookbacks. " - "Local snapshot backtests remain the research surface for broad variant sweeps." + "The structural overlay appears only in its explicit research arm and changes " + "ordering, not production qualification. Local snapshot backtests remain the " + "research surface for broad variant sweeps." ), } @@ -3061,6 +3212,7 @@ async def run_backtest( _assign_activation_momentum_percentiles(candidates) _assign_residual_low_vol_blend(candidates) _assign_residual_high_vol_blend(candidates) + _assign_structural_overlay_score(candidates) current_min_pct = float(activation.get("min_momentum_percentile", 80.0)) for c in candidates: c["qualified"] = _momentum_qualifies(c, current_min_pct) @@ -3163,6 +3315,14 @@ async def run_backtest( "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), + "structural_overlay_weight": ( + STRUCTURAL_OVERLAY_WEIGHT + if _sr_research_variant() == STRUCTURAL_OVERLAY_VARIANT else None + ), + "structural_overlay_source_variant": ( + STRUCTURAL_OVERLAY_SOURCE_VARIANT + if _sr_research_variant() == STRUCTURAL_OVERLAY_VARIANT else None + ), "entry_start": ( _backtest_entry_bounds()[0].isoformat() if _backtest_entry_bounds()[0] is not None else None diff --git a/docs/research/sr-levels-and-exits.md b/docs/research/sr-levels-and-exits.md index 8f060a1..5c6139e 100644 --- a/docs/research/sr-levels-and-exits.md +++ b/docs/research/sr-levels-and-exits.md @@ -642,6 +642,39 @@ This is an apples-to-apples full-history diagnostic against the current live production path. It is not a new untouched holdout because the post-2024 data was already inspected while isolating the range factor. +Full-period results reject the clean range-gated candidate as a replacement: +it improves robust setup expectancy and drawdown, but cuts the qualified set +from 1,086 to 290 and the live-path book from 321 to 170 trades. Its 73 unique +qualified symbols are also concentrated (36.2% of setups in the top ten names), +so overlapping setup expectancy does not translate into independent portfolio +opportunity. + +### Structural confirmation as a ranking overlay + +The next experiment preserves the production detector, setup geometry, +qualified universe, activation gate, and live exit. For each production setup, +the clean detector is evaluated point-in-time only to attach a binary feature: +whether `rewrite_range504_structural_legacy_primary` also clears its core gate. +That confirmation receives a single pre-registered 5% weight: + +```text +overlay_rank = 95% * production_80_20_rank + 5% * structural_confirmation +``` + +There is deliberately no weight sweep and no union with clean-only setups. The +report must first reproduce the production qualified count and production book; +otherwise the comparison is invalid. Run the one-arm full-period diagnostic: + +```bash +.venv/bin/python scripts/run_sr_v2_matrix.py overlay --workers 14 +``` + +Output: `reports/backtest-sr-overlay-full.json`. The candidate advances only if +the overlay improves full-period Sharpe, does not worsen drawdown, and retains +at least 90% of production CAGR. The 1-year and 6-month rows must not both +deteriorate. This remains contaminated full-history research, not promotion +validation. + 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 3ea2bc5..c2e0d1d 100644 --- a/reports/README.md +++ b/reports/README.md @@ -41,3 +41,8 @@ The full-period production comparison writes: Those outputs should be retained as the final decision point for this research branch. + +The follow-up breadth-preserving ranking experiment writes +`backtest-sr-overlay-full.json`. Retain it only as the decision point for the +pre-registered 5% clean-structure overlay; exploratory weight sweeps should not +be committed. diff --git a/scripts/run_backtest_snapshot.py b/scripts/run_backtest_snapshot.py index 9f0c9da..5cf0f01 100644 --- a/scripts/run_backtest_snapshot.py +++ b/scripts/run_backtest_snapshot.py @@ -59,6 +59,7 @@ def _parse_args() -> argparse.Namespace: "production_range504", "rewrite_range504_legacy_primary", "rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_primary2", + "production_structural_overlay", ), default=None, help="Research-only S/R detector/gate arm.", @@ -160,6 +161,27 @@ def _print_summary(report: dict) -> None: f"trades {row.get('trades')}" ) + monitor_rows = [ + row + for row in ((report.get("portfolio_monitor") or {}).get("runs") or []) + if row.get("lookback") == "all" + and ( + row.get("is_production") + or row.get("strategy") == "production_structural_overlay5_atr3" + ) + ] + if monitor_rows: + print(" live-path full-period comparison:") + for row in monitor_rows: + print( + " " + f"{row.get('strategy')}: " + 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() diff --git a/scripts/run_sr_v2_matrix.py b/scripts/run_sr_v2_matrix.py index 315f219..6f64236 100644 --- a/scripts/run_sr_v2_matrix.py +++ b/scripts/run_sr_v2_matrix.py @@ -100,6 +100,11 @@ def _args() -> argparse.Namespace: help="Run current production and the frozen candidate over the full snapshot.", ) _add_common(full) + overlay = commands.add_parser( + "overlay", + help="Test the frozen 5% clean-structure rank overlay on production breadth.", + ) + _add_common(overlay) validate = commands.add_parser( "validate", help="Run production control and one locked arm from 2024-07-01.", @@ -210,6 +215,16 @@ def _full(args: argparse.Namespace) -> None: print("Full-period production comparison complete.") +def _overlay(args: argparse.Namespace) -> None: + _run_arm( + "production_structural_overlay", + args.snapshot, + args.workers, + output=ROOT / "reports" / "backtest-sr-overlay-full.json", + ) + print("Full-period structural ranking overlay complete.") + + def main() -> None: args = _args() if args.command == "train": @@ -225,6 +240,8 @@ def main() -> None: _train(args, arms, "backtest-sr-range-residual-train") elif args.command == "full": _full(args) + elif args.command == "overlay": + _overlay(args) else: _validate(args) diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index 6729c04..5603f9d 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -169,6 +169,40 @@ def test_residual_high_vol_blend_is_research_only_rank(): assert cands[0][bt.RESIDUAL_HIGH_VOL_BLEND_60_40_KEY] == 72.0 +def test_structural_overlay_is_a_frozen_five_percent_rank_nudge(): + cands = [ + { + bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: 80.0, + "structural_overlay_pass": True, + }, + { + bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: 80.0, + "structural_overlay_pass": False, + }, + {bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: 80.0}, + ] + + bt._assign_structural_overlay_score(cands) + + assert bt.STRUCTURAL_OVERLAY_WEIGHT == 0.05 + assert cands[0][bt.STRUCTURAL_OVERLAY_SCORE_KEY] == pytest.approx(81.0) + assert cands[1][bt.STRUCTURAL_OVERLAY_SCORE_KEY] == pytest.approx(76.0) + assert cands[2][bt.STRUCTURAL_OVERLAY_SCORE_KEY] is None + + +def test_structural_overlay_monitor_strategy_is_opt_in(monkeypatch): + monkeypatch.setenv("BACKTEST_SR_VARIANT", "production_control") + assert bt._portfolio_monitor_strategies() == bt.PORTFOLIO_MONITOR_STRATEGIES + + monkeypatch.setenv("BACKTEST_SR_VARIANT", bt.STRUCTURAL_OVERLAY_VARIANT) + strategies = bt._portfolio_monitor_strategies() + overlay = strategies[-1] + assert overlay["strategy"] == bt.STRUCTURAL_OVERLAY_PORTFOLIO_STRATEGY + assert overlay["ranking_key"] == bt.STRUCTURAL_OVERLAY_SCORE_KEY + assert overlay["use_live_config"] is True + assert overlay["is_production"] is False + + def test_strategy_variants_keep_only_current_research_candidates(): variants = {cfg["variant"]: cfg for cfg in bt.STRATEGY_VARIANTS} @@ -767,6 +801,7 @@ def test_sr_research_variant_is_explicit_and_validated(monkeypatch): "rewrite_range504_legacy_primary", "rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_primary2", + "production_structural_overlay", ): monkeypatch.setenv("BACKTEST_SR_VARIANT", variant) assert bt._sr_research_variant() == variant @@ -782,6 +817,7 @@ def test_range_factor_detector_mapping_is_explicit(): "rewrite_range504_structural_legacy_primary" ) == "rewrite" assert bt._sr_detector_variant("rewrite_range504_structural_primary2") == "rewrite" + assert bt._sr_detector_variant("production_structural_overlay") == "production_control" assert bt._sr_detector_variant("soft_zones_legacy_primary") == "soft_zones" @@ -816,6 +852,47 @@ def test_residual_arms_change_only_the_primary_rr_floor(): "rewrite_range504_structural_primary2", activation, ) == 2.0 + assert bt._primary_min_rr_for_variant( + "production_structural_overlay", + activation, + ) == 1.5 + + +def test_structural_overlay_tags_production_geometry_without_replacing_it(monkeypatch): + production = { + "direction": "long", + "meets_core": True, + "rr": 2.2, + "target": 111.0, + "primary_sources": ["volume_profile"], + "gate_level_count": 53, + } + structural = { + "direction": "long", + "meets_core": True, + "rr": 2.8, + "target": 114.0, + "primary_sources": ["pivot_point"], + "gate_level_count": 14, + } + + def fake_window_setups(*args, sr_variant=None, **kwargs): + if sr_variant == "production_control": + return [production] + assert sr_variant == bt.STRUCTURAL_OVERLAY_SOURCE_VARIANT + return [structural] + + monkeypatch.setattr(bt, "_window_setups", fake_window_setups) + rows = bt._structural_overlay_window_setups([], {}, {}) + + assert len(rows) == 1 + assert rows[0]["target"] == production["target"] + assert rows[0]["rr"] == production["rr"] + assert rows[0]["structural_overlay_pass"] is True + assert rows[0]["structural_overlay_rr"] == structural["rr"] + assert rows[0]["structural_overlay_sources"] == ["pivot_point"] + assert rows[0]["structural_overlay_gate_level_count"] == 14 + assert production.get("structural_overlay_pass") is None @pytest.mark.parametrize(