feat(b3b): bounded survivorship-safe candidate pre-filter + adjclose-only low-memory run path

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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] = []

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 —