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):