Fix PIT snapshot store regressions for backtests

libs/backtest/selector.py

@@ -87,9 +87,58 @@ def build_candidate(
     ):
         return None
 
+    source_symbol = str(row.get("source_symbol") or row.get("symbol", row.get("ticker", ""))).upper()
+    trade_symbol_mode = (
+        strategy_engine.trade_symbol_mode
+        if strategy_engine is not None
+        else "event"
+    )
+    trade_symbol = source_symbol
+    execution_entry_price = row.get("entry_price") or row.get("entry_price_est")
+    execution_event_close = row.get("event_close")
+    execution_avg_dollar_volume = row.get(
+        "avg_dollar_volume_20d",
+        row.get("avg_dollar_volume", 0.0),
+    )
+    execution_atr_14_raw = row.get("atr_14")
+    execution_features = {k: v for k, v in row.items() if k not in _RESERVED_KEYS}
+
+    if strategy_engine and trade_symbol_mode == "sector_etf":
+        proxy_symbol = str(row.get("sector_etf_proxy") or "").upper()
+        if not proxy_symbol:
+            logger.debug(
+                "skip_candidate_no_sector_etf_proxy",
+                event_id=event_id,
+                source_symbol=source_symbol,
+                engine_id=strategy_engine.engine_id,
+            )
+            return None
+        trade_symbol = proxy_symbol
+        execution_entry_price = row.get("sector_etf_entry_price") or execution_entry_price
+        execution_event_close = row.get("sector_etf_event_close") or execution_event_close
+        execution_avg_dollar_volume = (
+            row.get("sector_etf_avg_dollar_volume")
+            or execution_avg_dollar_volume
+        )
+        execution_atr_14_raw = row.get("sector_etf_atr_14")
+        execution_features.update(
+            {
+                "source_symbol": source_symbol,
+                "trade_symbol_mode": trade_symbol_mode,
+                "proxy_trade_symbol": trade_symbol,
+                "proxy_reference_sector": row.get("sector"),
+                "source_event_close": row.get("event_close"),
+                "source_reaction_day_low": row.get("reaction_day_low"),
+                "source_reaction_day_high": row.get("reaction_day_high"),
+                "event_close": row.get("sector_etf_event_close"),
+                "reaction_day_low": row.get("sector_etf_reaction_day_low"),
+                "reaction_day_high": row.get("sector_etf_reaction_day_high"),
+            }
+        )
+
     if strategy_engine and strategy_engine.entry_timing_policy == "reaction_close":
         execution_date = reaction_date
-        entry_price_est = row.get("event_close") or row.get("entry_price_est")
+        entry_price_est = execution_event_close or row.get("entry_price_est")
         if not entry_price_est:
             logger.debug(
                 "skip_candidate_no_event_close",
@@ -98,7 +147,7 @@ def build_candidate(
             )
             return None
     else:
-        entry_price_est = row.get("entry_price") or row.get("entry_price_est")
+        entry_price_est = execution_entry_price
 
     if not entry_price_est:
         logger.warning("skip_candidate_no_entry_price", event_id=event_id)
@@ -109,8 +158,8 @@ def build_candidate(
         return None
 
     score = float(row.get("score", 0.0))
-    avg_dollar_volume = float(row.get("avg_dollar_volume", 0.0))
-    atr_14_raw = row.get("atr_14")
+    avg_dollar_volume = float(execution_avg_dollar_volume or 0.0)
+    atr_14_raw = execution_atr_14_raw
     atr_14 = float(atr_14_raw) if atr_14_raw is not None else None
     event_direction = str(row.get("event_direction", "")).lower()
     guidance_status = str(row.get("guidance_status", "")).lower()
@@ -229,7 +278,8 @@ def build_candidate(
 
     return Candidate(
         event_id=event_id,
-        symbol=str(row.get("symbol", row.get("ticker", ""))),
+        symbol=trade_symbol,
+        source_symbol=source_symbol,
         issuer_id=row.get("issuer_id"),
         score=score,
         sector=str(row.get("sector") or "UNKNOWN"),
@@ -356,6 +406,7 @@ def build_candidate(
             if strategy_engine
             else None
         ),
+        trade_symbol_mode=trade_symbol_mode,
         trade_direction=trade_direction,
         parent_position_id=row.get("parent_position_id"),
         is_add_on=bool(row.get("is_add_on", False)),
@@ -364,7 +415,7 @@ def build_candidate(
             if row.get("forced_shares") not in (None, "")
             else None
         ),
-        features={k: v for k, v in row.items() if k not in _RESERVED_KEYS},
+        features=execution_features,
     )
 
 
@@ -1056,11 +1107,8 @@ def select_candidates(
     excluded_symbols = {symbol.upper() for symbol in (excluded_symbols or set())}
     for row in raw_rows:
         event_id = str(row.get("event_id", ""))
-        symbol = str(row.get("symbol", row.get("ticker", ""))).upper()
         if event_id and event_id in excluded_event_ids:
             continue
-        if symbol and symbol in excluded_symbols:
-            continue
         prepared_row = _prepare_row_for_strategy_engine(
             row,
             signal_config=signal_config,
@@ -1072,6 +1120,8 @@ def select_candidates(
             engine_lookup=engine_lookup,
         )
         if c is not None:
+            if c.symbol and c.symbol.upper() in excluded_symbols:
+                continue
             candidates.append(c)
 
     candidates = filter_by_universe(candidates, universe_config)
@@ -1086,6 +1136,8 @@ def select_candidates(
     if event_type_profiles:
         candidates = filter_by_event_type(candidates, event_type_profiles)
     candidates = rank_candidates(candidates, signal_config.ranking_fields)
+    if strategy_engine and strategy_engine.trade_symbol_mode != "event":
+        candidates = _dedupe_candidates_by_symbol(candidates)
     candidates = truncate_candidates(
         candidates,
         truncate_to if truncate_to is not None else signal_config.max_candidates_per_day,
@@ -1093,6 +1145,18 @@ def select_candidates(
     return candidates
 
 
+def _dedupe_candidates_by_symbol(candidates: list[Candidate]) -> list[Candidate]:
+    seen_symbols: set[str] = set()
+    deduped: list[Candidate] = []
+    for candidate in candidates:
+        symbol = candidate.symbol.upper()
+        if symbol in seen_symbols:
+            continue
+        seen_symbols.add(symbol)
+        deduped.append(candidate)
+    return deduped
+
+
 def _resolve_score_threshold(
     signal_config: SignalConfig,
     strategy_engine: StrategyEngineConfig | None,

tests/unit/backtest/test_selector.py

@@ -85,6 +85,17 @@ class TestBuildCandidate:
         assert c is not None
         assert c.sector == "UNKNOWN"
 
+    def test_prefers_avg_dollar_volume_20d_when_present(self):
+        from libs.backtest.selector import build_candidate
+
+        row = _make_raw_row(
+            avg_dollar_volume=999_000_000.0,
+            avg_dollar_volume_20d=12_345_678.0,
+        )
+        c = build_candidate(row)
+        assert c is not None
+        assert c.avg_dollar_volume == 12_345_678.0
+
     def test_score_bucket_classification(self):
         from libs.backtest.selector import build_candidate
 
@@ -119,6 +130,39 @@ class TestBuildCandidate:
         assert c.timing_class == "same_day"
         assert c.trade_direction == "short"
 
+    def test_sector_etf_proxy_uses_proxy_trade_fields(self):
+        from libs.backtest.selector import build_candidate
+
+        engine = StrategyEngineConfig(
+            engine_id="sector_etf_proxy",
+            event_types=["earnings"],
+            trade_symbol_mode="sector_etf",
+        )
+        row = _make_raw_row(
+            score=0.9,
+            event_close=150.0,
+            reaction_day_low=145.0,
+            reaction_day_high=153.0,
+            sector_etf_proxy="XLK",
+            sector_etf_event_close=210.0,
+            sector_etf_entry_price=211.5,
+            sector_etf_reaction_day_low=206.0,
+            sector_etf_reaction_day_high=212.0,
+            sector_etf_avg_dollar_volume=250_000_000.0,
+            sector_etf_atr_14=4.2,
+        )
+        c = build_candidate(row, strategy_engine=engine)
+        assert c is not None
+        assert c.symbol == "XLK"
+        assert c.source_symbol == "AAPL"
+        assert c.trade_symbol_mode == "sector_etf"
+        assert c.entry_price_est == 211.5
+        assert c.avg_dollar_volume == 250_000_000.0
+        assert c.atr_14 == 4.2
+        assert c.features["event_close"] == 210.0
+        assert c.features["reaction_day_low"] == 206.0
+        assert c.features["source_event_close"] == 150.0
+
     def test_engine_can_force_long_direction_for_negative_reaction(self):
         from libs.backtest.selector import build_candidate
 
@@ -1741,6 +1785,79 @@ class TestSelectCandidates:
         assert "C" not in symbols  # low ADV
         assert "D" not in symbols  # below min_price
 
+    def test_sector_etf_proxy_dedupes_by_trade_symbol(self):
+        from libs.backtest.selector import select_candidates
+
+        rows = [
+            _make_raw_row(
+                event_id="EVT::TEST::001",
+                symbol="AAPL",
+                score=0.9,
+                sector="Technology",
+                sector_etf_proxy="XLK",
+                sector_etf_event_close=210.0,
+                sector_etf_entry_price=211.0,
+                sector_etf_avg_dollar_volume=250_000_000.0,
+                sector_etf_atr_14=4.0,
+            ),
+            _make_raw_row(
+                event_id="EVT::TEST::002",
+                symbol="MSFT",
+                score=0.8,
+                sector="Technology",
+                sector_etf_proxy="XLK",
+                sector_etf_event_close=210.0,
+                sector_etf_entry_price=211.0,
+                sector_etf_avg_dollar_volume=250_000_000.0,
+                sector_etf_atr_14=4.0,
+            ),
+        ]
+        u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000)
+        s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5)
+        engine = StrategyEngineConfig(
+            engine_id="sector_etf_proxy",
+            event_types=["earnings"],
+            trade_symbol_mode="sector_etf",
+        )
+
+        result = select_candidates(rows, u, s, strategy_engine=engine)
+        assert len(result) == 1
+        assert result[0].symbol == "XLK"
+        assert result[0].source_symbol == "AAPL"
+
+    def test_sector_etf_proxy_respects_excluded_trade_symbol(self):
+        from libs.backtest.selector import select_candidates
+
+        rows = [
+            _make_raw_row(
+                event_id="EVT::TEST::001",
+                symbol="AAPL",
+                score=0.9,
+                sector="Technology",
+                sector_etf_proxy="XLK",
+                sector_etf_event_close=210.0,
+                sector_etf_entry_price=211.0,
+                sector_etf_avg_dollar_volume=250_000_000.0,
+                sector_etf_atr_14=4.0,
+            ),
+        ]
+        u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000)
+        s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5)
+        engine = StrategyEngineConfig(
+            engine_id="sector_etf_proxy",
+            event_types=["earnings"],
+            trade_symbol_mode="sector_etf",
+        )
+
+        result = select_candidates(
+            rows,
+            u,
+            s,
+            strategy_engine=engine,
+            excluded_symbols={"XLK"},
+        )
+        assert result == []
+
     def test_pipeline_with_event_type_profiles(self):
         from libs.backtest.selector import select_candidates
 

tests/unit/backtest/test_snapshot_store.py

@@ -177,24 +177,136 @@ class TestSnapshotStoreLoadGuard:
         asyncio.run(_test())
 
 
+class TestSnapshotStoreRuntimeCache:
+    def test_load_merged_reuses_runtime_cache(self, tmp_path, monkeypatch):
+        from libs.backtest.snapshot_store import SnapshotStore
+
+        (tmp_path / "manifest.json").write_text(json.dumps({"snapshot_id": "test"}))
+        for split_name in ("train", "valid", "test"):
+            pq.write_table(
+                pa.table({"event_id": [f"EVT::{split_name}"], "ticker": ["AAPL"]}),
+                tmp_path / f"{split_name}.parquet",
+            )
+
+        calls = {"count": 0}
+        data = {
+            "candidates_by_exec_date": {
+                dt.date(2026, 1, 6): [{"event_id": "EVT::001", "symbol": "AAPL"}],
+            },
+            "bars_by_symbol_date": {
+                "AAPL": {
+                    dt.date(2026, 1, 6): {"date": dt.date(2026, 1, 6), "close": 100.0},
+                }
+            },
+            "macro_by_date": {},
+        }
+
+        async def _fake_async_load_merged(*args, **kwargs):
+            calls["count"] += 1
+            return data
+
+        monkeypatch.setattr(SnapshotStore, "_async_load_merged", _fake_async_load_merged)
+
+        first = SnapshotStore.load_merged(
+            snapshot_dir=tmp_path,
+            split_names=["train", "valid", "test"],
+            oracle_url="http://localhost",
+            db_dsn="postgres://localhost/test",
+            scoring_fn=None,
+        )
+        assert calls["count"] == 1
+        assert first.get_candidates_for_date(dt.date(2026, 1, 6))[0]["symbol"] == "AAPL"
+
+        second = SnapshotStore.load_merged(
+            snapshot_dir=tmp_path,
+            split_names=["train", "valid", "test"],
+            oracle_url="http://localhost",
+            db_dsn="postgres://localhost/test",
+            scoring_fn=None,
+        )
+        assert calls["count"] == 1
+        assert second.get_candidates_for_date(dt.date(2026, 1, 6))[0]["symbol"] == "AAPL"
+
+    def test_load_merged_waits_for_inflight_runtime_cache(self, tmp_path, monkeypatch):
+        from libs.backtest.snapshot_store import SnapshotStore
+
+        (tmp_path / "manifest.json").write_text(json.dumps({"snapshot_id": "test"}))
+        for split_name in ("train", "valid", "test"):
+            pq.write_table(
+                pa.table({"event_id": [f"EVT::{split_name}"], "ticker": ["AAPL"]}),
+                tmp_path / f"{split_name}.parquet",
+            )
+
+        data = {
+            "candidates_by_exec_date": {
+                dt.date(2026, 1, 6): [{"event_id": "EVT::001", "symbol": "AAPL"}],
+            },
+            "bars_by_symbol_date": {
+                "AAPL": {
+                    dt.date(2026, 1, 6): {"date": dt.date(2026, 1, 6), "close": 100.0},
+                }
+            },
+            "macro_by_date": {},
+        }
+        calls = {"builder": 0}
+
+        async def _fake_async_load_merged(*args, **kwargs):
+            calls["builder"] += 1
+            return data
+
+        monkeypatch.setattr(SnapshotStore, "_async_load_merged", _fake_async_load_merged)
+        monkeypatch.setattr(SnapshotStore, "_try_load_runtime_cache", lambda *args, **kwargs: None)
+        monkeypatch.setattr(SnapshotStore, "_acquire_runtime_cache_lock", lambda *args, **kwargs: False)
+        monkeypatch.setattr(SnapshotStore, "_wait_for_runtime_cache", lambda *args, **kwargs: data)
+
+        result = SnapshotStore.load_merged(
+            snapshot_dir=tmp_path,
+            split_names=["train", "valid", "test"],
+            oracle_url="http://localhost",
+            db_dsn="postgres://localhost/test",
+            scoring_fn=None,
+        )
+        assert calls["builder"] == 0
+        assert result.get_candidates_for_date(dt.date(2026, 1, 6))[0]["symbol"] == "AAPL"
+
+
 class TestSnapshotStoreFromParquet:
     def test_compute_date_range(self, tmp_path):
         from libs.backtest.snapshot_store import SnapshotStore
 
         rows = [
+            {"event_date": "2026-01-03"},
             {"entry_date": "2026-01-05"},
             {"entry_date": "2026-01-10"},
             {"entry_date": "2026-01-07"},
             {"reaction_date": "2026-01-04"},
         ]
         result = SnapshotStore._compute_date_range(rows)
-        assert result == (dt.date(2026, 1, 4), dt.date(2026, 1, 10))
+        assert result == (dt.date(2026, 1, 3), dt.date(2026, 1, 10))
 
     def test_compute_date_range_empty(self, tmp_path):
         from libs.backtest.snapshot_store import SnapshotStore
 
         assert SnapshotStore._compute_date_range([]) is None
 
+    def test_backfill_avg_dollar_volume_prefers_row_level_20d_value(self, tmp_path):
+        from libs.backtest.snapshot_store import SnapshotStore
+
+        row = {
+            "avg_dollar_volume_20d": 12_345_678.0,
+            "avg_dollar_volume": None,
+        }
+        SnapshotStore._backfill_avg_dollar_volume_features(
+            row,
+            symbol="AAPL",
+            event_date=dt.date(2026, 1, 6),
+            bars_by_symbol={},
+            price_bar_cache={},
+            fallback_avg_dvol=999_000_000.0,
+        )
+        assert row["avg_dollar_volume_20d"] == 12_345_678.0
+        assert row["avg_dollar_volume"] == 12_345_678.0
+
     def test_async_load_falls_back_to_parquet_metadata_when_db_unavailable(self, tmp_path, monkeypatch):
         from libs.backtest.snapshot_store import SnapshotStore
 
@@ -271,6 +383,240 @@ class TestSnapshotStoreFromParquet:
         assert rows[0]["score"] == pytest.approx(0.77)
         assert rows[0]["event_timestamp"] is not None
 
+    def test_async_load_backfills_missing_price_derived_features(self, tmp_path, monkeypatch):
+        from libs.backtest.snapshot_store import SnapshotStore
+
+        parquet_path = tmp_path / "train.parquet"
+        event_date = dt.date(2026, 1, 5)
+        table = pa.table({
+            "event_id": ["EVT::ROW::002"],
+            "ticker": ["AAPL"],
+            "event_date": [event_date.isoformat()],
+            "event_type": ["other_material_event"],
+            "reaction_date": [event_date.isoformat()],
+            "entry_date": ["2026-01-06"],
+            "event_close": [150.0],
+            "entry_price": [151.0],
+            "atr_14": [3.0],
+        })
+        pq.write_table(table, parquet_path)
+
+        async def _fake_event_meta(*args, **kwargs):
+            return {}
+
+        async def _fake_price_data(*args, **kwargs):
+            bars = {}
+            start = event_date - dt.timedelta(days=120)
+            series = {}
+            for i in range(121):
+                d = start + dt.timedelta(days=i)
+                close = 100.0 + i * 0.5 + ((i % 5) - 2) * 0.1
+                series[d] = {
+                    "date": d,
+                    "open": close - 0.4,
+                    "high": close + 0.8,
+                    "low": close - 0.9,
+                    "close": close,
+                    "volume": 1_000_000 + i * 1000,
+                }
+            bars["AAPL"] = series
+            return bars, {"AAPL": 125_000_000.0}
+
+        async def _fake_sectors(*args, **kwargs):
+            return {"AAPL": "Technology"}
+
+        async def _fake_macro(*args, **kwargs):
+            return {}
+
+        monkeypatch.setattr(SnapshotStore, "_fetch_event_metadata", staticmethod(_fake_event_meta))
+        monkeypatch.setattr(SnapshotStore, "_fetch_price_data", staticmethod(_fake_price_data))
+        monkeypatch.setattr(SnapshotStore, "_fetch_sectors", staticmethod(_fake_sectors))
+        monkeypatch.setattr(SnapshotStore, "_fetch_macro", staticmethod(_fake_macro))
+        monkeypatch.setattr(SnapshotStore, "_fetch_spy_macro", staticmethod(_fake_macro))
+
+        data = asyncio.run(
+            SnapshotStore._async_load(
+                tmp_path,
+                "train",
+                oracle_url="http://localhost:18001",
+                db_dsn="postgresql+asyncpg://unused",
+                scoring_fn=lambda row: 0.77,
+            )
+        )
+        store = SnapshotStore(**data)
+        rows = store.get_candidates_for_date(dt.date(2026, 1, 6))
+
+        assert len(rows) == 1
+        row = rows[0]
+        assert row["pre_event_hurst_60d"] is not None
+        assert row["pre_event_entropy_60d"] is not None
+        assert row["pre_event_bb_position"] is not None
+        assert row["pre_event_gravitational_pull"] is not None
+        assert row["pre_event_market_temperature"] is not None
+
+    def test_async_load_merged_dedupes_rows_and_fetches_once(self, tmp_path, monkeypatch):
+        from libs.backtest.snapshot_store import SnapshotStore
+
+        event_date = dt.date(2026, 1, 5)
+        train_table = pa.table({
+            "event_id": ["EVT::ROW::003"],
+            "ticker": ["AAPL"],
+            "event_date": [event_date.isoformat()],
+            "event_type": ["earnings_release"],
+            "reaction_date": [event_date.isoformat()],
+            "entry_date": ["2026-01-06"],
+            "event_close": [150.0],
+            "entry_price": [151.0],
+            "atr_14": [3.0],
+        })
+        valid_table = pa.table({
+            "event_id": ["EVT::ROW::003", "EVT::ROW::004"],
+            "ticker": ["AAPL", "MSFT"],
+            "event_date": [event_date.isoformat(), "2026-01-07"],
+            "event_type": ["earnings_release", "guidance"],
+            "reaction_date": [event_date.isoformat(), "2026-01-07"],
+            "entry_date": ["2026-01-06", "2026-01-08"],
+            "event_close": [150.0, 300.0],
+            "entry_price": [151.0, 301.0],
+            "atr_14": [3.0, 5.0],
+        })
+        pq.write_table(train_table, tmp_path / "train.parquet")
+        pq.write_table(valid_table, tmp_path / "valid.parquet")
+
+        call_state: dict[str, object] = {"price_calls": 0, "date_range": None}
+
+        async def _fake_event_meta(*args, **kwargs):
+            return {}
+
+        async def _fake_price_data(symbols, date_range, *args, **kwargs):
+            call_state["price_calls"] = int(call_state["price_calls"]) + 1
+            call_state["date_range"] = date_range
+            bars = {}
+            for sym in symbols:
+                bars[sym] = {
+                    event_date: {
+                        "date": event_date,
+                        "open": 100.0,
+                        "high": 101.0,
+                        "low": 99.0,
+                        "close": 100.5,
+                        "volume": 1_000_000,
+                    },
+                    dt.date(2026, 1, 7): {
+                        "date": dt.date(2026, 1, 7),
+                        "open": 101.0,
+                        "high": 102.0,
+                        "low": 100.0,
+                        "close": 101.5,
+                        "volume": 1_000_000,
+                    },
+                    dt.date(2026, 1, 8): {
+                        "date": dt.date(2026, 1, 8),
+                        "open": 102.0,
+                        "high": 103.0,
+                        "low": 101.0,
+                        "close": 102.5,
+                        "volume": 1_000_000,
+                    },
+                }
+            return bars, {sym: 100_000_000.0 for sym in symbols}
+
+        async def _fake_sectors(symbols, *args, **kwargs):
+            return {sym: "Technology" for sym in symbols}
+
+        async def _fake_macro(*args, **kwargs):
+            return {}
+
+        monkeypatch.setattr(SnapshotStore, "_fetch_event_metadata", staticmethod(_fake_event_meta))
+        monkeypatch.setattr(SnapshotStore, "_fetch_price_data", staticmethod(_fake_price_data))
+        monkeypatch.setattr(SnapshotStore, "_fetch_sectors", staticmethod(_fake_sectors))
+        monkeypatch.setattr(SnapshotStore, "_fetch_macro", staticmethod(_fake_macro))
+        monkeypatch.setattr(SnapshotStore, "_fetch_spy_macro", staticmethod(_fake_macro))
+
+        data = asyncio.run(
+            SnapshotStore._async_load_merged(
+                tmp_path,
+                ["train", "valid", "test"],
+                oracle_url="http://localhost:18001",
+                db_dsn="postgresql+asyncpg://unused",
+                scoring_fn=lambda row: 0.55,
+            )
+        )
+        store = SnapshotStore(**data)
+
+        assert call_state["price_calls"] == 1
+        assert call_state["date_range"] == (dt.date(2026, 1, 5), dt.date(2026, 1, 8))
+        assert len(store.get_candidates_for_date(dt.date(2026, 1, 6))) == 1
+        assert len(store.get_candidates_for_date(dt.date(2026, 1, 8))) == 1
+
+    def test_materialize_snapshot_dir_persists_runtime_backfilled_columns(self, tmp_path, monkeypatch):
+        from libs.backtest.snapshot_store import SnapshotStore
+
+        event_date = dt.date(2026, 1, 5)
+        table = pa.table({
+            "event_id": ["EVT::ROW::005"],
+            "ticker": ["AAPL"],
+            "event_date": [event_date.isoformat()],
+            "reaction_date": [event_date.isoformat()],
+            "entry_date": ["2026-01-06"],
+            "event_close": [150.0],
+            "entry_price": [151.0],
+            "macro_vix": [25.0],
+            "macro_hy_spread": [4.5],
+        })
+        pq.write_table(table, tmp_path / "train.parquet")
+        (tmp_path / "manifest.json").write_text(json.dumps({"snapshot_id": "unit_test_snapshot"}))
+
+        async def _fake_price_data(symbols, date_range, *args, **kwargs):
+            start = event_date - dt.timedelta(days=120)
+            bars = {
+                "AAPL": {
+                    (start + dt.timedelta(days=i)): {
+                        "date": start + dt.timedelta(days=i),
+                        "open": 100.0 + i,
+                        "high": 100.5 + i,
+                        "low": 99.5 + i,
+                        "close": 100.0 + i,
+                        "volume": 1_000_000 + i,
+                    }
+                    for i in range(121)
+                }
+            }
+            return bars, {"AAPL": 100_000_000.0}
+
+        async def _fake_macro(*args, **kwargs):
+            return {
+                event_date - dt.timedelta(days=1): {"T10Y2Y": 0.55},
+                event_date: {"T10Y2Y": 0.60},
+            }
+
+        monkeypatch.setattr(SnapshotStore, "_fetch_price_data", staticmethod(_fake_price_data))
+        monkeypatch.setattr(SnapshotStore, "_fetch_macro", staticmethod(_fake_macro))
+
+        materialized = SnapshotStore.materialize_snapshot_dir(
+            tmp_path,
+            oracle_url="http://localhost:18001",
+            db_dsn="postgresql+asyncpg://unused",
+        )
+
+        updated = pq.read_table(tmp_path / "train.parquet")
+        row = updated.to_pylist()[0]
+
+        assert "pre_event_hurst_60d" in updated.column_names
+        assert "pre_event_market_temperature" in updated.column_names
+        assert "macro_t10y2y" in updated.column_names
+        assert row["pre_event_hurst_60d"] is not None
+        assert row["pre_event_entropy_60d"] is not None
+        assert row["pre_event_bb_position"] is not None
+        assert row["pre_event_market_temperature"] is not None
+        assert row["macro_t10y2y"] == pytest.approx(0.60)
+        assert "pre_event_hurst_60d" in materialized
+        assert "macro_t10y2y" in materialized
+
+        manifest = json.loads((tmp_path / "manifest.json").read_text())
+        assert "pre_event_hurst_60d" in manifest["materialized_feature_columns"]
+        assert "macro_t10y2y" in manifest["materialized_feature_columns"]
+
 
 class TestSnapshotStoreSectorFetch:
     def test_fetch_sectors_falls_back_from_placeholder_oracle(self, tmp_path, monkeypatch):