From 5f1e789984a375b8699ded8aae5dc1d2c2d7bbe0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 13 Sep 2026 18:55:09 +0800 Subject: [PATCH] fix: skip invalid inverse-vol candidates Co-Authored-By: Codex --- src/portfolio.py | 9 +++++++-- tests/test_backtest_accounting.py | 12 ++++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/portfolio.py b/src/portfolio.py index c79c1dd..ba4554a 100644 --- a/src/portfolio.py +++ b/src/portfolio.py @@ -2,6 +2,7 @@ from typing import Iterable +import numpy as np import pandas as pd @@ -16,11 +17,16 @@ def select_portfolio( if eligible.empty: return eligible + weighting = weighting.lower() + if weighting in {"inverse_vol", "inverse-vol", "inv_vol"} and "vol20" in eligible.columns: + # Missing risk estimates cannot produce a trusted inverse-vol weight. + # Exclude those candidates before top-N selection so one bad row does + # not turn every selected target weight into NaN. + eligible = eligible.loc[np.isfinite(eligible["vol20"])] selected = eligible.sort_values(score_column, ascending=False).head(top_n).copy() if selected.empty: return selected - weighting = weighting.lower() if weighting in {"inverse_vol", "inverse-vol", "inv_vol"} and "vol20" in selected.columns: inverse_vol = 1.0 / selected["vol20"].clip(lower=0.05) selected["target_weight"] = inverse_vol / inverse_vol.sum() @@ -43,4 +49,3 @@ def calculate_turnover(previous_weights: pd.Series, next_weights: pd.Series) -> previous_weights = previous_weights.reindex(next_weights.index).fillna(0.0) next_weights = next_weights.reindex(previous_weights.index).fillna(0.0) return float(0.5 * (next_weights - previous_weights).abs().sum()) - diff --git a/tests/test_backtest_accounting.py b/tests/test_backtest_accounting.py index a1b8cb2..a2c2791 100644 --- a/tests/test_backtest_accounting.py +++ b/tests/test_backtest_accounting.py @@ -109,6 +109,18 @@ def test_valid_frozen_empty_selection_produces_cash(self) -> None: self.assertEqual(len(result.returns), len(self.dates)) self.assertTrue(result.returns.eq(0).all()) + def test_inverse_vol_skips_top_score_with_missing_volatility(self) -> None: + panel = self.panel(("A", "B")) + panel["vol20"] = 0.2 + panel.loc[(slice(None), "A"), "final_score"] = 2.0 + panel.loc[(slice(None), "A"), "vol20"] = np.nan + panel.loc[(slice(None), "B"), "final_score"] = 1.0 + config = {"strategy": {**self.config["strategy"], "weighting": "inverse_vol"}} + + result = run_single_backtest(panel, "final_score", config) + + self.assertEqual(result.trades["symbol"].tolist(), ["B"]) + def test_nonfinite_eligible_scores_are_not_cash(self) -> None: for bad_score in (np.inf, -np.inf): with self.subTest(bad_score=bad_score):