diff --git a/src/us_equity_strategies/research/soxl_core_optimization.py b/src/us_equity_strategies/research/soxl_core_optimization.py index 5539cbb..59e5ace 100644 --- a/src/us_equity_strategies/research/soxl_core_optimization.py +++ b/src/us_equity_strategies/research/soxl_core_optimization.py @@ -24,6 +24,7 @@ VOLATILITY_SCALING_READBACK_SCHEMA = "qsl.research.soxl_volatility_scaling_readback.v1" RSI2_MEAN_REVERSION_SCHEMA = "qsl.research.soxl_rsi2_mean_reversion.v1" RSI2_MEAN_REVERSION_READBACK_SCHEMA = "qsl.research.soxl_rsi2_mean_reversion_readback.v1" +RSI2_ACCEPTANCE_CONTRACT = "SOXL_CONSTRAINED_COMPOUNDING_ACCEPTANCE_HARDENING_V1" CANDIDATE_WINDOWS = (140, 160, 180, 200) BASELINE_WINDOW_DAYS = 200 INITIAL_EQUITY = 100_000.0 @@ -915,33 +916,61 @@ def simulate_rsi2_mean_reversion_candidate(source: OfflineInput, candidate_id: s return tuple(points) -def _select_rsi2_mean_reversion_winner(validation: dict[str, Sequence[dict[str, float | int | None | str]]]) -> str: +def _rsi2_acceptance_window_metrics(points: Sequence[DailyPoint], soxx: Sequence[InputRow], raw_start: int, raw_end: int) -> dict[str, float | int | None | str | bool]: + metrics = _volatility_metrics_with_soxx(points, soxx, raw_start, raw_end) + selected_points = points[raw_start - BASELINE_WINDOW_DAYS:raw_end - BASELINE_WINDOW_DAYS + 1] + selected_soxx = soxx[raw_start - 1:raw_end + 1] + if len(selected_points) != raw_end - raw_start + 1 or len(selected_soxx) != len(selected_points) + 1: + _fail("WINDOW_BOUNDARY_INVALID") + observation_count = len(selected_points) + cumulative_return = selected_soxx[-1].close / selected_soxx[0].close - 1.0 + cagr = (1.0 + cumulative_return) ** (252.0 / observation_count) - 1.0 + if not math.isfinite(cumulative_return) or not math.isfinite(cagr): + _fail("RSI2_ACCEPTANCE_METRICS_INVALID") + metrics.update({ + "exposure_session_count": sum(point.quantity > 0.0 for point in selected_points), + "activity_observed": any(point.quantity > 0.0 or point.gross_traded_notional_at_open > 0.0 for point in selected_points), + "soxx_close_path_cumulative_return": cumulative_return, + "soxx_close_path_cagr": cagr, + }) + return metrics + + +def _select_rsi2_mean_reversion_winner(validation: dict[str, Sequence[dict[str, float | int | None | str | bool]]]) -> str | None: if tuple(validation) != RSI2_MEAN_REVERSION_CANDIDATES: _fail("VALIDATION_METRICS_INVALID") - ranked: list[tuple[tuple[float, float, float, float, float, int], str]] = [] + baseline = validation["UNSCALED_SMA200"] + if len(baseline) != 3 or not all(item.get("activity_observed") is True for item in baseline): + return None + ranked: list[tuple[tuple[float, float, float, float, int], str]] = [] for index, candidate_id in enumerate(RSI2_MEAN_REVERSION_CANDIDATES): metrics = validation[candidate_id] if len(metrics) != 3: _fail("VALIDATION_METRICS_INVALID") + if not all(item.get("activity_observed") is True for item in metrics): + continue + if candidate_id != "UNSCALED_SMA200" and not _all_strictly_better(metrics, baseline): + continue ranked.append(( ( + -_median(_metric_values(metrics, "cagr")), _median([abs(value) for value in _metric_values(metrics, "max_drawdown")]), -_median(_metric_values(metrics, "expected_shortfall_95")), - _median(_metric_values(metrics, "annualized_volatility")), - -_median(_metric_values(metrics, "cagr")), _median(_metric_values(metrics, "turnover")), index, ), candidate_id, )) - return min(ranked)[1] + return min(ranked)[1] if ranked else None def _invalid_rsi2_mean_reversion(code: str) -> dict[str, Any]: return { "schema": RSI2_MEAN_REVERSION_SCHEMA, "evidence_valid": False, "failure_codes": [code], "outcome": "NO_IMPROVEMENT", "research_recommendation": None, "research_only": True, "live_adoption_authorized": False, "size_zero_required": True, - "plugin_control": dict(RSI2_MEAN_REVERSION_PLUGIN_CONTROL), + "plugin_control": dict(RSI2_MEAN_REVERSION_PLUGIN_CONTROL), "acceptance_contract": RSI2_ACCEPTANCE_CONTRACT, + "acceptance_classes": {"activity": {"selection_windows_all_active": False, "wfa_qualifying_test_window_count": 0, "wfa_at_least_two_of_three": False}, "benchmark_relative_compounding": {"final_holdout_drawdown_no_worse_than_soxx": False, "final_holdout_cumulative_return_strictly_greater_than_soxx": False, "final_holdout_cagr_strictly_greater_than_soxx": False, "eligible": False}}, + "recommendation_eligible": False, "r4a_eligible": False, } @@ -965,14 +994,27 @@ def run_soxl_rsi2_mean_reversion(source: object, *, plugin_control: object = RSI if [(point.date, point.end_equity.hex(), point.cash.hex(), point.quantity.hex()) for point in zero] != [(point.date, point.equity.hex(), point.cash.hex(), point.soxl_quantity.hex()) for point in baseline.equity_curve]: _fail("SMA200_ZERO_PARITY_FAILED") validation = { - candidate_id: [_volatility_metrics_with_soxx(simulations[candidate_id]["C2_5"], soxx, *WINDOWS[name]) for name in ("F1_VALIDATION", "F2_VALIDATION", "F3_VALIDATION")] + candidate_id: [_rsi2_acceptance_window_metrics(simulations[candidate_id]["C2_5"], soxx, *WINDOWS[name]) for name in ("F1_VALIDATION", "F2_VALIDATION", "F3_VALIDATION")] for candidate_id in RSI2_MEAN_REVERSION_CANDIDATES } winner = _select_rsi2_mean_reversion_winner(validation) + if winner is None: + gates = {"selection_windows_all_active": False} + return { + "schema": RSI2_MEAN_REVERSION_SCHEMA, "evidence_valid": True, "failure_codes": ["NO_QUALIFYING_VALIDATION_CANDIDATE", "SELECTION_WINDOWS_NOT_ALL_ACTIVE"], "outcome": "NO_IMPROVEMENT", + "research_recommendation": None, "research_only": True, "live_adoption_authorized": False, "size_zero_required": True, + "plugin_control": dict(RSI2_MEAN_REVERSION_PLUGIN_CONTROL), "input_digest": source.input_digest, + "baseline": "UNSCALED_SMA200", "candidates": list(RSI2_MEAN_REVERSION_CANDIDATES), + "signal_rule": {"trend": "SOXX_CLOSE_STRICTLY_ABOVE_INCLUSIVE_SMA200", "rsi": "WILDER_RSI2", "execution": "NEXT_SOXL_OPEN", "exposure": "LONG_OR_CASH"}, + "validation_metrics_c2_5": validation, "locked_winner": None, "post_lock_metrics": {}, "evidence_gates": gates, + "acceptance_contract": RSI2_ACCEPTANCE_CONTRACT, + "acceptance_classes": {"activity": {"selection_windows_all_active": False, "wfa_qualifying_test_window_count": 0, "wfa_at_least_two_of_three": False}, "benchmark_relative_compounding": {"final_holdout_drawdown_no_worse_than_soxx": False, "final_holdout_cumulative_return_strictly_greater_than_soxx": False, "final_holdout_cagr_strictly_greater_than_soxx": False, "eligible": False}}, + "recommendation_eligible": False, "r4a_eligible": False, + } exposed = tuple(dict.fromkeys((winner, "UNSCALED_SMA200"))) post_lock = { candidate_id: { - name: {scenario.scenario_id: _volatility_metrics_with_soxx(simulations[candidate_id][scenario.scenario_id], soxx, *WINDOWS[name]) for scenario in SCENARIOS} + name: {scenario.scenario_id: _rsi2_acceptance_window_metrics(simulations[candidate_id][scenario.scenario_id], soxx, *WINDOWS[name]) for scenario in SCENARIOS} for name in ("F1_TEST", "F2_TEST", "F3_TEST", "FINAL_HOLDOUT") } for candidate_id in exposed @@ -986,11 +1028,32 @@ def run_soxl_rsi2_mean_reversion(source: object, *, plugin_control: object = RSI final_stress, baseline_final_stress = selected_post["FINAL_HOLDOUT"]["C5_10_STRESS"], baseline_post["FINAL_HOLDOUT"]["C5_10_STRESS"] final_returns = tuple(point.daily_return for point in simulations[winner]["C2_5"][WINDOWS["FINAL_HOLDOUT"][0] - BASELINE_WINDOW_DAYS:]) loss_probability = _terminal_loss_probability(final_returns) + wfa_qualifying = sum( + candidate["activity_observed"] is True + and float(candidate["cumulative_return"]) > 0.0 + and abs(float(candidate["max_drawdown"])) < abs(float(reference["max_drawdown"])) + and float(candidate["expected_shortfall_95"]) > float(reference["expected_shortfall_95"]) + for candidate, reference in zip(selected_tests, baseline_tests, strict=True) + ) + cagr_beats_soxx = float(final_c2["cagr"]) > float(final_c2["soxx_close_path_cagr"]) + return_beats_soxx = float(final_c2["cumulative_return"]) > float(final_c2["soxx_close_path_cumulative_return"]) + if cagr_beats_soxx != return_beats_soxx: + _fail("RSI2_ACCEPTANCE_COMPARISON_INVALID") + activity = {"selection_windows_all_active": all(item["activity_observed"] is True for item in selected_validation), "wfa_qualifying_test_window_count": wfa_qualifying, "wfa_at_least_two_of_three": wfa_qualifying >= 2} + benchmark_relative_compounding = { + "final_holdout_drawdown_no_worse_than_soxx": abs(float(final_c2["max_drawdown"])) <= abs(float(final_c2["soxx_close_path_max_drawdown"])), + "final_holdout_cumulative_return_strictly_greater_than_soxx": return_beats_soxx, + "final_holdout_cagr_strictly_greater_than_soxx": cagr_beats_soxx, + } + benchmark_relative_compounding["eligible"] = all(benchmark_relative_compounding.values()) gates = { - "validation_medians_strictly_improve_drawdown_and_es95": _all_strictly_better(selected_validation, baseline_validation), - "wfa_tests_at_least_two_strictly_improve_drawdown_and_es95": sum(abs(float(candidate["max_drawdown"])) < abs(float(reference["max_drawdown"])) and float(candidate["expected_shortfall_95"]) > float(reference["expected_shortfall_95"]) for candidate, reference in zip(selected_tests, baseline_tests, strict=True)) >= 2, + "selection_windows_all_active": activity["selection_windows_all_active"], + "validation_medians_strictly_improve_drawdown_and_es95": winner != "UNSCALED_SMA200" and _all_strictly_better(selected_validation, baseline_validation), + "wfa_same_window_conjunction_at_least_two_of_three": activity["wfa_at_least_two_of_three"], "final_c2_5_strictly_improves_drawdown_and_es95": abs(float(final_c2["max_drawdown"])) < abs(float(baseline_final_c2["max_drawdown"])) and float(final_c2["expected_shortfall_95"]) > float(baseline_final_c2["expected_shortfall_95"]), - "wfa_c2_5_returns_at_least_two_positive": sum(float(item["cumulative_return"]) > 0.0 for item in selected_tests) >= 2, + "final_holdout_drawdown_no_worse_than_soxx": benchmark_relative_compounding["final_holdout_drawdown_no_worse_than_soxx"], + "final_holdout_cumulative_return_strictly_greater_than_soxx": benchmark_relative_compounding["final_holdout_cumulative_return_strictly_greater_than_soxx"], + "final_holdout_cagr_strictly_greater_than_soxx": benchmark_relative_compounding["final_holdout_cagr_strictly_greater_than_soxx"], "final_c2_5_return_strictly_positive": float(final_c2["cumulative_return"]) > 0.0, "final_c5_10_stress_return_strictly_positive": float(final_stress["cumulative_return"]) > 0.0, "mc_c2_5_terminal_loss_probability_strictly_below_half": loss_probability < 0.5, @@ -1002,7 +1065,7 @@ def run_soxl_rsi2_mean_reversion(source: object, *, plugin_control: object = RSI holdout_soxx_drawdown = _soxx_drawdown(soxx, *WINDOWS["FINAL_HOLDOUT"]) return { "schema": RSI2_MEAN_REVERSION_SCHEMA, "evidence_valid": True, "failure_codes": failures, - "outcome": "CHARACTERIZATION_CANDIDATE_FOUND" if found else "NO_IMPROVEMENT", + "outcome": "COMPOUNDING_CANDIDATE_FOUND" if found else "NO_IMPROVEMENT", "research_recommendation": {"candidate_id": winner} if found else None, "research_only": True, "live_adoption_authorized": False, "size_zero_required": True, "plugin_control": dict(RSI2_MEAN_REVERSION_PLUGIN_CONTROL), "input_digest": source.input_digest, @@ -1011,6 +1074,9 @@ def run_soxl_rsi2_mean_reversion(source: object, *, plugin_control: object = RSI "validation_metrics_c2_5": validation, "locked_winner": winner, "post_lock_metrics": post_lock, "evidence_gates": gates, "soxx_drawdown_comparison": {"final_holdout_max_drawdown": holdout_soxx_drawdown, "matches_or_beats_soxx_drawdown": abs(float(final_c2["max_drawdown"])) <= abs(holdout_soxx_drawdown)}, + "acceptance_contract": RSI2_ACCEPTANCE_CONTRACT, + "acceptance_classes": {"activity": activity, "benchmark_relative_compounding": benchmark_relative_compounding}, + "recommendation_eligible": found, "r4a_eligible": found, } except OptimizationError as exc: return _invalid_rsi2_mean_reversion(str(exc)) diff --git a/tests/test_soxl_rsi2_mean_reversion.py b/tests/test_soxl_rsi2_mean_reversion.py index 9647257..c067b99 100644 --- a/tests/test_soxl_rsi2_mean_reversion.py +++ b/tests/test_soxl_rsi2_mean_reversion.py @@ -8,10 +8,12 @@ import pytest +import us_equity_strategies.research.soxl_core_optimization as optimization from us_equity_strategies.research.soxl_soxx_offline_input_contract import InputRow, OfflineInput from us_equity_strategies.research.soxl_soxx_typed_baseline_result import run_typed_baseline from us_equity_strategies.research.soxl_core_optimization import ( BASELINE_WINDOW_DAYS, + DailyPoint, RSI2_MEAN_REVERSION_CANDIDATES, RSI2_MEAN_REVERSION_PLUGIN_CONTROL, RSI2_MEAN_REVERSION_SCHEMA, @@ -112,7 +114,7 @@ def test_rsi_state_machine_costs_and_exposure_bounds() -> None: def test_validation_selection_tie_and_runner_isolation(monkeypatch: pytest.MonkeyPatch) -> None: - metric = {"max_drawdown": -0.1, "expected_shortfall_95": -0.02, "annualized_volatility": 0.2, "cagr": 0.1, "turnover": 1.0} + metric = {"max_drawdown": -0.1, "expected_shortfall_95": -0.02, "annualized_volatility": 0.2, "cagr": 0.1, "turnover": 1.0, "activity_observed": True} validation = {candidate: [metric] * 3 for candidate in RSI2_MEAN_REVERSION_CANDIDATES} assert _select_rsi2_mean_reversion_winner(validation) == "UNSCALED_SMA200" monkeypatch.setattr("us_equity_strategies.research.soxl_core_optimization._terminal_loss_probability", lambda _: 0.0) @@ -124,6 +126,89 @@ def test_validation_selection_tie_and_runner_isolation(monkeypatch: pytest.Monke assert run_soxl_rsi2_mean_reversion(None)["evidence_valid"] is False +def test_acceptance_hardening_metrics_and_validation_selection() -> None: + assert hasattr(optimization, "_rsi2_acceptance_window_metrics") + metric = { + "max_drawdown": -0.1, + "expected_shortfall_95": -0.02, + "annualized_volatility": 9.0, + "cagr": 0.1, + "turnover": 1.0, + "activity_observed": True, + } + cagr_first = dict(metric, cagr=0.2, max_drawdown=-0.09, expected_shortfall_95=-0.01, annualized_volatility=9.0) + lower_drawdown = dict(metric, cagr=0.11, max_drawdown=-0.05, expected_shortfall_95=-0.01, annualized_volatility=0.01) + inactive = dict(metric, cagr=0.9, max_drawdown=-0.01, expected_shortfall_95=-0.01, activity_observed=False) + validation = { + "UNSCALED_SMA200": [metric] * 3, + "RSI2_ENTRY_5_EXIT_70": [cagr_first] * 3, + "RSI2_ENTRY_10_EXIT_70": [lower_drawdown] * 3, + "RSI2_ENTRY_15_EXIT_70": [inactive] * 3, + } + assert _select_rsi2_mean_reversion_winner(validation) == "RSI2_ENTRY_5_EXIT_70" + validation["UNSCALED_SMA200"] = [dict(metric, activity_observed=False)] * 3 + assert _select_rsi2_mean_reversion_winner(validation) is None + + +def test_acceptance_window_metrics_preserve_carried_and_exit_activity() -> None: + points = ( + DailyPoint("2024-01-01", 100.0, 101.0, 0.0, 1.0, 0.01, False, 0.0, 0.0, 0.0), + DailyPoint("2024-01-02", 101.0, 100.0, 100.0, 0.0, -0.01, True, 0.0, 0.0, 10.0), + ) + source = _source() + metrics = optimization._rsi2_acceptance_window_metrics(points, tuple(row for row in source.rows if row.symbol == "SOXX"), 200, 201) + expected_return = 120.1 / 119.9 - 1.0 + assert metrics["exposure_session_count"] == 1 + assert metrics["activity_observed"] is True + assert metrics["soxx_close_path_cumulative_return"] == pytest.approx(expected_return) + assert metrics["soxx_close_path_cagr"] == pytest.approx((1.0 + expected_return) ** 126.0 - 1.0) + all_cash = tuple(DailyPoint(point.date, point.start_equity, point.end_equity, point.cash, 0.0, point.daily_return, False, 0.0, 0.0, 0.0) for point in points) + assert optimization._rsi2_acceptance_window_metrics(all_cash, tuple(row for row in source.rows if row.symbol == "SOXX"), 200, 201)["activity_observed"] is False + + +def test_acceptance_rejections_are_reported_in_evidence_gates_and_failure_codes(monkeypatch: pytest.MonkeyPatch) -> None: + selector = optimization._select_rsi2_mean_reversion_winner + monkeypatch.setattr("us_equity_strategies.research.soxl_core_optimization._select_rsi2_mean_reversion_winner", lambda _: None) + activity_rejected = run_soxl_rsi2_mean_reversion(_source()) + assert activity_rejected["evidence_gates"] == {"selection_windows_all_active": False} + assert activity_rejected["failure_codes"] == ["NO_QUALIFYING_VALIDATION_CANDIDATE", "SELECTION_WINDOWS_NOT_ALL_ACTIVE"] + + original = optimization._rsi2_acceptance_window_metrics + monkeypatch.setattr(optimization, "_select_rsi2_mean_reversion_winner", selector) + + def benchmark_rejected(*args: object) -> dict[str, float | int | None | str | bool]: + metrics = original(*args) # type: ignore[arg-type] + if args[3] == 627: + metrics["soxx_close_path_cumulative_return"] = 100.0 + metrics["soxx_close_path_cagr"] = 100.0 + return metrics + + monkeypatch.setattr(optimization, "_rsi2_acceptance_window_metrics", benchmark_rejected) + result = run_soxl_rsi2_mean_reversion(_source()) + for key in ( + "final_holdout_drawdown_no_worse_than_soxx", + "final_holdout_cumulative_return_strictly_greater_than_soxx", + "final_holdout_cagr_strictly_greater_than_soxx", + ): + assert key in result["evidence_gates"] + assert (key.upper() in result["failure_codes"]) is (result["evidence_gates"][key] is False) + + +def test_hardened_result_is_fail_closed_and_exposes_acceptance_contract(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("us_equity_strategies.research.soxl_core_optimization._terminal_loss_probability", lambda _: 0.0) + result = run_soxl_rsi2_mean_reversion(_source()) + assert result["acceptance_contract"] == "SOXL_CONSTRAINED_COMPOUNDING_ACCEPTANCE_HARDENING_V1" + assert set(result["acceptance_classes"]) == {"activity", "benchmark_relative_compounding"} + assert {"exposure_session_count", "activity_observed", "soxx_close_path_cumulative_return", "soxx_close_path_cagr"} <= set( + result["validation_metrics_c2_5"]["UNSCALED_SMA200"][0] + ) + assert result["recommendation_eligible"] is False + assert result["r4a_eligible"] is False + assert result["research_recommendation"] is None + assert result["live_adoption_authorized"] is False + assert result["size_zero_required"] is True + + def test_persistence_set_once_strict_readback_and_symlink_rejection(tmp_path: Path) -> None: repo, head, blobs = _provenance_repo(tmp_path) result = {"schema": RSI2_MEAN_REVERSION_SCHEMA, "value": 1.0}