diff --git a/src/fxhnt/application/equity_backtest_runner.py b/src/fxhnt/application/equity_backtest_runner.py index 9e06c3b..c084a96 100644 --- a/src/fxhnt/application/equity_backtest_runner.py +++ b/src/fxhnt/application/equity_backtest_runner.py @@ -45,6 +45,61 @@ def _iso(epoch_day: int) -> str: return (dt.date(1970, 1, 1) + dt.timedelta(days=epoch_day)).isoformat() +def select_backtest_universe( + store: DuckDbFeatureStore, + *, + min_history: int = 300, + top_k: int = 2500, +) -> dict[str, float]: + """Bounded, survivorship-safe candidate universe for the backtest. + + Returns {symbol: peak_yearly_avg_dollar_volume} for the `top_k` most-liquid + symbols that have at least `min_history` daily bars. Liquidity is ranked by + PEAK calendar-year average dollar-volume (close*volume), NOT lifetime average, + so names that were highly liquid only while alive (later delisted) or only + recently (recent IPOs) are still included — avoiding survivorship bias in the + candidate set. This caps memory: the runner then loads only these symbols. + """ + # ONE DuckDB query, all aggregation pushed into the engine (no rows pulled into + # Python). Self-join `features` on (symbol, ts): one side close, other volume → + # per-bar dollar-volume. Bucket by approximate calendar year, take the per-year + # mean dollar-volume, then the PEAK year per symbol. Bar count comes from the + # 'close' feature. Filter bars >= min_history, rank by peak yearly dv desc. + sql = """ + WITH dv AS ( + SELECT c.symbol AS symbol, + c.ts AS ts, + c.value * v.value AS dollar_vol + FROM features c + JOIN features v + ON c.symbol = v.symbol AND c.ts = v.ts + WHERE c.feature = 'close' AND v.feature = 'volume' + ), + yearly AS ( + SELECT symbol, + CAST(ts / (86400 * 365.25) AS INTEGER) AS yr, + avg(dollar_vol) AS yr_avg_dv, + count(*) AS bars + FROM dv + GROUP BY symbol, yr + ), + agg AS ( + SELECT symbol, + max(yr_avg_dv) AS peak_yearly_dv, + sum(bars) AS total_bars + FROM yearly + GROUP BY symbol + ) + SELECT symbol, peak_yearly_dv + FROM agg + WHERE total_bars >= ? AND peak_yearly_dv > 0.0 + ORDER BY peak_yearly_dv DESC + LIMIT ? + """ + rows = store._con.execute(sql, [min_history, top_k]).fetchall() + return {str(sym): float(dv) for sym, dv in rows} + + class EquityBacktestRunner: def __init__( self, @@ -54,19 +109,42 @@ class EquityBacktestRunner: cost_bps_per_turnover: float = 15.0, borrow_annual: float = 0.0, momentum_min_history: int = 252, + candidate_top_k: int | None = None, + candidate_min_history: int = 300, ) -> None: self._store = store self._n = n self._cost = cost_bps_per_turnover / 1e4 self._borrow_daily = borrow_annual / _TRADING_DAYS_YEAR self._mom_min = momentum_min_history + self._candidate_top_k = candidate_top_k + self._candidate_min_history = candidate_min_history def run(self) -> BacktestRunResult: members = self._store.read_membership() - syms = [m[0] for m in members] - adj = self._store.read_panel(syms, "adjclose") - close = self._store.read_panel(syms, "close") - vol = self._store.read_panel(syms, "volume") + + # Low-memory candidate path: pre-filter to a bounded, survivorship-safe set + # of liquid names, load ONLY the adjclose panel for them (not close/volume), + # and rank rebalances by a STATIC peak-yearly-liquidity dict. The alive-gate + # stays point-in-time (membership tuples), only the liquidity ranking is + # static. None-path = unchanged full-panel trailing-dollar-vol behavior. + liq: dict[str, float] | None = None + if self._candidate_top_k is not None: + liq = select_backtest_universe( + self._store, + min_history=self._candidate_min_history, + top_k=self._candidate_top_k, + ) + members = [m for m in members if m[0] in liq] + adj = self._store.read_panel(list(liq), "adjclose") + close: dict[str, dict[int, float]] = {} + vol: dict[str, dict[int, float]] = {} + syms = list(liq) + else: + syms = [m[0] for m in members] + adj = self._store.read_panel(syms, "adjclose") + close = self._store.read_panel(syms, "close") + vol = self._store.read_panel(syms, "volume") all_days = sorted({d for s in adj.values() for d in s}) all_iso = [_iso(d) for d in all_days] @@ -80,13 +158,18 @@ class EquityBacktestRunner: for k, rebal in enumerate(rebals): rebal_day = epoch_day(rebal) - lo = rebal_day - _DOLLAR_VOL_WINDOW_DAYS - dollar_vol: dict[str, float] = {} - for s in syms: - cl, vl = close.get(s, {}), vol.get(s, {}) - acc = [cl[d] * vl[d] for d in cl if lo <= d <= rebal_day and d in vl] - if acc: - dollar_vol[s] = float(np.mean(acc)) + dollar_vol: dict[str, float] + if liq is not None: + # static peak-yearly-liquidity ranking; alive-gate is still PIT + dollar_vol = liq + else: + lo = rebal_day - _DOLLAR_VOL_WINDOW_DAYS + dollar_vol = {} + for s in syms: + cl, vl = close.get(s, {}), vol.get(s, {}) + acc = [cl[d] * vl[d] for d in cl if lo <= d <= rebal_day and d in vl] + if acc: + dollar_vol[s] = float(np.mean(acc)) uni = universe_asof(members, dollar_vol, rebal, self._n) mom: list[float | None] = [] diff --git a/tests/integration/test_equity_backtest_runner.py b/tests/integration/test_equity_backtest_runner.py index 79989fe..db0d56d 100644 --- a/tests/integration/test_equity_backtest_runner.py +++ b/tests/integration/test_equity_backtest_runner.py @@ -8,6 +8,7 @@ from fxhnt.application.equity_backtest_runner import ( BacktestRunResult, EquityBacktestRunner, evaluate_constructions, + select_backtest_universe, ) from fxhnt.domain.equity_backtest import month_end_rebalance_dates @@ -140,6 +141,86 @@ def test_evaluate_constructions_short_series_emits_clean_insufficient_data_verdi json.dumps(report) # must not raise (no NaN) +def _build_liquidity_warehouse(path: str) -> DuckDbFeatureStore: + """5 symbols with deliberately differing history-length and dollar-volume so + the peak-yearly-liquidity pre-filter can be exercised: + + - LIVELONG : long history, HIGH dollar-volume throughout (top liquidity) + - DELISTED : SHORT lifetime (well above min_history though) but VERY HIGH + dollar-volume while alive — must be INCLUDED (survivorship-safe; + peak-yearly ranking sees its liquid alive years) + - ILLIQUID : long history but tiny dollar-volume throughout — ranks low, + dropped by a small top_k + - RECENTIPO : short-but-sufficient history, moderate dollar-volume + - TOOSHORT : fewer than min_history bars — must be EXCLUDED + """ + store = DuckDbFeatureStore(path) + + def write(sym: str, start_iso: str, n_bars: int, close: float, vol: float) -> None: + start = dt.date.fromisoformat(start_iso) + rows = [] + for i in range(n_bars): + day = start + dt.timedelta(days=i) + ts = (day - dt.date(1970, 1, 1)).days * _SPD + rows.append((ts, {"adjclose": close, "close": close, "volume": vol})) + store.write_features(sym, rows) + store.upsert_membership([(sym, "NYSE", start_iso, + (start + dt.timedelta(days=n_bars)).isoformat())]) + + write("LIVELONG", "2010-01-01", 3000, close=100.0, vol=5_000_000.0) # dv = 5e8 + write("DELISTED", "2012-01-01", 600, close=200.0, vol=8_000_000.0) # dv = 1.6e9 (highest) + write("ILLIQUID", "2010-01-01", 3000, close=10.0, vol=1_000.0) # dv = 1e4 (lowest) + write("RECENTIPO", "2023-01-01", 500, close=50.0, vol=2_000_000.0) # dv = 1e8 + write("TOOSHORT", "2024-01-01", 50, close=100.0, vol=9_000_000.0) # < min_history + return store + + +def test_select_backtest_universe_filters_and_ranks(tmp_path): + store = _build_liquidity_warehouse(str(tmp_path / "liq.duckdb")) + liq = select_backtest_universe(store, min_history=300, top_k=2500) + store.close() + + # dict[str, float] with positive values + assert isinstance(liq, dict) + assert all(isinstance(s, str) for s in liq) + assert all(isinstance(v, float) and v > 0.0 for v in liq.values()) + + # TOOSHORT (< min_history bars) is excluded + assert "TOOSHORT" not in liq + # DELISTED short-but-liquid name is INCLUDED (survivorship-safe) + assert "DELISTED" in liq + # long-history liquid name present + assert "LIVELONG" in liq + + # ranking by peak-yearly dollar-volume: DELISTED (1.6e9) > LIVELONG (5e8) > ILLIQUID (1e4) + assert liq["DELISTED"] > liq["LIVELONG"] > liq["ILLIQUID"] + + +def test_select_backtest_universe_top_k_drops_illiquid_tail(tmp_path): + store = _build_liquidity_warehouse(str(tmp_path / "liq.duckdb")) + # top_k=2 keeps only the two most-liquid (DELISTED, LIVELONG); ILLIQUID dropped + liq = select_backtest_universe(store, min_history=300, top_k=2) + store.close() + assert len(liq) == 2 + assert set(liq) == {"DELISTED", "LIVELONG"} + assert "ILLIQUID" not in liq + + +def test_runner_candidate_path_matches_membership_gating(tmp_path): + store = _build_warehouse(str(tmp_path / "wh.duckdb")) + runner = EquityBacktestRunner( + store, n=10, candidate_top_k=100, candidate_min_history=252, + ) + result = runner.run() + store.close() + # bounded path works end-to-end: 3 construction series, finite, non-trivial + assert set(result.returns_by_construction) == {"long", "ls", "tilt"} + for series in result.returns_by_construction.values(): + assert len(series) > 50 + assert np.isfinite(series).all() + assert result.n_names_avg > 0.0 + + def test_evaluate_constructions_sr_variance_is_cross_construction_computed_once(monkeypatch): # FIX #1: sr_variance must be the variance of the per-construction IS Sharpes # ACROSS the constructions, computed ONCE and shared by every evaluate() call —