"""Unit tests for libs/backtest/snapshot_store.py."""
from __future__ import annotations

import datetime as dt
import asyncio
import json
from pathlib import Path
from types import SimpleNamespace

import pyarrow as pa
import pyarrow.parquet as pq
import pytest


def _build_store_from_fixture(tmp_path: Path) -> object:
    """Build a SnapshotStore directly from test data (no DB/HTTP)."""
    from libs.backtest.snapshot_store import SnapshotStore

    candidates = {
        dt.date(2026, 1, 6): [
            {
                "event_id": "EVT::TEST::001",
                "symbol": "AAPL",
                "execution_date": dt.date(2026, 1, 6),
                "entry_date": "2026-01-06",
                "entry_price": 150.0,
                "score": 0.8,
                "sector": "Technology",
                "event_type": "earnings",
                "event_timestamp": "2026-01-05T21:00:00+00:00",
                "filing_time_bucket": "post_market",
                "reaction_date": "2026-01-05",
                "avg_dollar_volume": 5_000_000.0,
                "atr_14": 3.0,
            }
        ],
        dt.date(2026, 1, 7): [
            {
                "event_id": "EVT::TEST::002",
                "symbol": "MSFT",
                "execution_date": dt.date(2026, 1, 7),
                "entry_date": "2026-01-07",
                "entry_price": 300.0,
                "score": 0.6,
                "sector": "Technology",
                "event_type": "guidance",
                "event_timestamp": "2026-01-06T20:00:00+00:00",
                "filing_time_bucket": "post_market",
                "reaction_date": "2026-01-06",
                "avg_dollar_volume": 10_000_000.0,
                "atr_14": 5.0,
            }
        ],
    }

    bars = {
        "AAPL": {
            dt.date(2026, 1, 6): {"date": dt.date(2026, 1, 6), "open": 150.0, "high": 155.0, "low": 148.0, "close": 152.0, "volume": 1_000_000},
            dt.date(2026, 1, 7): {"date": dt.date(2026, 1, 7), "open": 152.0, "high": 162.0, "low": 150.0, "close": 159.0, "volume": 900_000},
        },
        "MSFT": {
            dt.date(2026, 1, 7): {"date": dt.date(2026, 1, 7), "open": 300.0, "high": 310.0, "low": 295.0, "close": 305.0, "volume": 500_000},
        },
    }

    return SnapshotStore(
        candidates_by_exec_date=candidates,
        bars_by_symbol_date=bars,
    )


class TestSnapshotStoreQuery:
    def test_get_candidates_for_date(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        rows = store.get_candidates_for_date(dt.date(2026, 1, 6))
        assert len(rows) == 1
        assert rows[0]["symbol"] == "AAPL"

    def test_get_candidates_empty_date(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        rows = store.get_candidates_for_date(dt.date(2026, 1, 1))
        assert rows == []

    def test_get_candidates_for_reaction_date(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        rows = store.get_candidates_for_reaction_date(dt.date(2026, 1, 5))
        assert len(rows) == 1
        assert rows[0]["symbol"] == "AAPL"

    def test_no_lookahead(self, tmp_path):
        """Candidates for Jan 7 should NOT appear when querying Jan 6."""
        store = _build_store_from_fixture(tmp_path)
        rows = store.get_candidates_for_date(dt.date(2026, 1, 6))
        symbols = [r["symbol"] for r in rows]
        assert "MSFT" not in symbols

    def test_get_bar_exists(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        bar = store.get_bar("AAPL", dt.date(2026, 1, 6))
        assert bar is not None
        assert bar["open"] == 150.0
        assert bar["close"] == 152.0

    def test_get_bar_missing_returns_none(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        bar = store.get_bar("AAPL", dt.date(2025, 12, 31))
        assert bar is None

    def test_get_bar_unknown_symbol_returns_none(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        assert store.get_bar("UNKNOWN", dt.date(2026, 1, 6)) is None

    def test_all_execution_dates_sorted(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        dates = store.all_execution_dates()
        assert dates == sorted(dates)
        assert dt.date(2026, 1, 6) in dates
        assert dt.date(2026, 1, 7) in dates

    def test_all_reaction_dates_sorted(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        dates = store.all_reaction_dates()
        assert dates == [dt.date(2026, 1, 5), dt.date(2026, 1, 6)]

    def test_macro_default_empty(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        macro = store.get_macro_for_date(dt.date(2026, 1, 6))
        assert macro == {}

    def test_get_market_features_from_stored_bars(self, tmp_path):
        from libs.backtest.snapshot_store import SnapshotStore

        store = SnapshotStore(
            candidates_by_exec_date={},
            bars_by_symbol_date={
                "QQQ": {
                    dt.date(2026, 1, 5): {"date": dt.date(2026, 1, 5), "open": 100.0, "high": 102.0, "low": 99.0, "close": 101.0, "volume": 100},
                    dt.date(2026, 1, 6): {"date": dt.date(2026, 1, 6), "open": 103.0, "high": 108.0, "low": 102.0, "close": 107.0, "volume": 500},
                }
            },
        )

        features = store.get_market_features("QQQ", dt.date(2026, 1, 6))
        assert features["event_close"] == pytest.approx(107.0)
        assert features["reaction_day_return"] == pytest.approx((107.0 - 101.0) / 101.0)
        assert features["gap_size"] == pytest.approx((103.0 - 101.0) / 101.0)
        assert features["close_location"] == pytest.approx((107.0 - 102.0) / (108.0 - 102.0))

    def test_candidates_copy_returned(self, tmp_path):
        """Modifying returned list should not affect internal state."""
        store = _build_store_from_fixture(tmp_path)
        rows1 = store.get_candidates_for_date(dt.date(2026, 1, 6))
        rows1.append({"extra": "data"})
        rows2 = store.get_candidates_for_date(dt.date(2026, 1, 6))
        assert len(rows2) == 1  # unchanged

    def test_slice_by_date_range_filters_execution_and_reaction_dates(self, tmp_path):
        store = _build_store_from_fixture(tmp_path)
        sliced = store.slice_by_date_range(dt.date(2026, 1, 7), dt.date(2026, 1, 7))
        assert sliced.all_execution_dates() == [dt.date(2026, 1, 7)]
        assert sliced.get_candidates_for_date(dt.date(2026, 1, 7))[0]["symbol"] == "MSFT"
        assert sliced.get_candidates_for_reaction_date(dt.date(2026, 1, 5)) == []
        assert sliced.get_candidates_for_reaction_date(dt.date(2026, 1, 6)) == []

    def test_slice_by_date_range_trims_macro(self, tmp_path):
        from libs.backtest.snapshot_store import SnapshotStore

        store = SnapshotStore(
            candidates_by_exec_date={
                dt.date(2026, 1, 6): [{"event_id": "E1", "symbol": "AAPL", "reaction_date": "2026-01-05"}],
                dt.date(2026, 1, 7): [{"event_id": "E2", "symbol": "MSFT", "reaction_date": "2026-01-06"}],
            },
            bars_by_symbol_date={},
            macro_by_date={
                dt.date(2026, 1, 6): {"spy_close": 100.0},
                dt.date(2026, 1, 7): {"spy_close": 101.0},
            },
        )

        sliced = store.slice_by_date_range(dt.date(2026, 1, 7), dt.date(2026, 1, 7))
        assert sliced.get_macro_for_date(dt.date(2026, 1, 6)) == {}
        assert sliced.get_macro_for_date(dt.date(2026, 1, 7)) == {"spy_close": 101.0}


class TestSnapshotStoreLoadGuard:
    def test_raises_in_running_event_loop(self, tmp_path):
        """load() should raise RuntimeError if called from a running event loop."""
        import asyncio

        from libs.backtest.snapshot_store import SnapshotStore

        async def _test():
            with pytest.raises(RuntimeError, match="running event loop"):
                SnapshotStore.load(tmp_path, "train", "http://localhost", "postgres://")

        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, 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

        parquet_path = tmp_path / "train.parquet"
        table = pa.table({
            "event_id": ["EVT::ROW::001"],
            "ticker": ["AAPL"],
            "event_date": ["2026-01-05"],
            "event_type": ["earnings_release"],
            "reaction_date": ["2026-01-05"],
            "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):
            return (
                {
                    "AAPL": {
                        dt.date(2026, 1, 5): {
                            "date": dt.date(2026, 1, 5),
                            "open": 149.0,
                            "high": 153.0,
                            "low": 148.0,
                            "close": 150.0,
                            "volume": 1_000_000,
                        },
                        dt.date(2026, 1, 6): {
                            "date": dt.date(2026, 1, 6),
                            "open": 151.0,
                            "high": 156.0,
                            "low": 150.0,
                            "close": 155.0,
                            "volume": 900_000,
                        },
                    }
                },
                {"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
        assert rows[0]["symbol"] == "AAPL"
        assert rows[0]["event_type"] == "earnings_release"
        assert rows[0]["event_date"] == dt.date(2026, 1, 5)
        assert rows[0]["score"] == pytest.approx(0.77)
        assert rows[0]["event_timestamp"] is not None

    def test_async_load_keeps_snapshot_metadata_when_db_has_drifted(self, tmp_path, monkeypatch):
        from libs.backtest.snapshot_store import SnapshotStore

        parquet_path = tmp_path / "train.parquet"
        table = pa.table({
            "event_id": ["EVT::ROW::DRIFTED"],
            "ticker": ["AAPL"],
            "event_date": ["2026-01-05"],
            "event_type": ["earnings_release"],
            "reaction_date": ["2026-01-05"],
            "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 {
                "EVT::ROW::DRIFTED": {
                    "issuer_id": "ISSUER::001",
                    "event_date": dt.date(2026, 1, 4),
                    "event_type": "shareholder_vote",
                    "event_timestamp": dt.datetime(2026, 1, 4, 21, 0, tzinfo=dt.UTC),
                    "ticker": "MSFT",
                }
            }

        async def _fake_price_data(symbols, *args, **kwargs):
            return (
                {
                    sym: {
                        dt.date(2026, 1, 5): {
                            "date": dt.date(2026, 1, 5),
                            "open": 149.0,
                            "high": 153.0,
                            "low": 148.0,
                            "close": 150.0,
                            "volume": 1_000_000,
                        },
                        dt.date(2026, 1, 6): {
                            "date": dt.date(2026, 1, 6),
                            "open": 151.0,
                            "high": 156.0,
                            "low": 150.0,
                            "close": 155.0,
                            "volume": 900_000,
                        },
                    }
                    for sym in symbols
                },
                {sym: 125_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(
                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
        assert rows[0]["symbol"] == "AAPL"
        assert rows[0]["event_type"] == "earnings_release"
        assert rows[0]["event_date"] == dt.date(2026, 1, 5)
        assert rows[0]["issuer_id"] == "ISSUER::001"

    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"]

    def test_attach_recent_sector_cluster_features_is_pit_safe(self, tmp_path):
        from libs.backtest.snapshot_store import SnapshotStore

        candidates_by_exec_date = {
            dt.date(2026, 1, 3): [
                {
                    "event_id": "EVT::1",
                    "event_type": "earnings_release",
                    "sector": "Technology",
                    "event_date": dt.date(2026, 1, 2),
                    "event_timestamp": dt.datetime(2026, 1, 2, 21, 0, tzinfo=dt.UTC),
                    "reaction_day_return": 0.15,
                    "volume_ratio": 3.0,
                    "close_location": 0.85,
                    "market_cap_proxy": 50_000_000_000.0,
                }
            ],
            dt.date(2026, 1, 6): [
                {
                    "event_id": "EVT::2",
                    "event_type": "earnings_release",
                    "sector": "Technology",
                    "event_date": dt.date(2026, 1, 5),
                    "event_timestamp": dt.datetime(2026, 1, 5, 21, 0, tzinfo=dt.UTC),
                    "reaction_day_return": 0.04,
                    "volume_ratio": 1.4,
                    "close_location": 0.58,
                    "market_cap_proxy": 20_000_000_000.0,
                }
            ],
            dt.date(2026, 1, 8): [
                {
                    "event_id": "EVT::3",
                    "event_type": "earnings_release",
                    "sector": "Technology",
                    "event_date": dt.date(2026, 1, 7),
                    "event_timestamp": dt.datetime(2026, 1, 7, 21, 0, tzinfo=dt.UTC),
                    "reaction_day_return": 0.03,
                    "volume_ratio": 1.3,
                    "close_location": 0.55,
                    "market_cap_proxy": 18_000_000_000.0,
                }
            ],
        }

        SnapshotStore._attach_recent_sector_cluster_features(candidates_by_exec_date)

        leader = candidates_by_exec_date[dt.date(2026, 1, 3)][0]
        follower_1 = candidates_by_exec_date[dt.date(2026, 1, 6)][0]
        follower_2 = candidates_by_exec_date[dt.date(2026, 1, 8)][0]

        assert leader["sector_recent_event_count_3d"] == 0.0
        assert leader["sector_recent_leader_count_3d"] == 0.0
        assert leader["sector_recent_leader_reaction_max_3d"] is None

        assert follower_1["sector_recent_event_count_3d"] == 1.0
        assert follower_1["sector_recent_leader_count_3d"] == 1.0
        assert follower_1["sector_recent_leader_reaction_max_3d"] == pytest.approx(0.15)

        assert follower_2["sector_recent_event_count_3d"] == 1.0
        assert follower_2["sector_recent_leader_count_3d"] == 0.0
        assert follower_2["sector_recent_leader_reaction_max_3d"] is None


class TestSnapshotStoreSectorFetch:
    def test_fetch_sectors_falls_back_from_placeholder_oracle(self, tmp_path, monkeypatch):
        import libs.oracle_client as oracle_mod
        from libs.backtest.snapshot_store import SnapshotStore

        class FakeOracleClient:
            def __init__(self, base_url: str) -> None:
                self.base_url = base_url

            async def __aenter__(self):
                return self

            async def __aexit__(self, *args):
                return None

        class FakeCompanyService:
            def __init__(self, client) -> None:
                self.client = client

            async def get_company(self, symbol: str):
                if symbol == "BAX":
                    return SimpleNamespace(
                        sector="Technology",
                        industry="Software",
                        exchange=None,
                        market_cap=None,
                    )
                return SimpleNamespace(
                    sector="Utilities",
                    industry="Utilities - Regulated Water",
                    exchange="NYSE",
                    market_cap=123.0,
                )

        monkeypatch.setattr(oracle_mod, "OracleClient", FakeOracleClient)
        monkeypatch.setattr(oracle_mod, "CompanyService", FakeCompanyService)
        monkeypatch.setattr(
            SnapshotStore,
            "_sector_cache_path",
            staticmethod(lambda: tmp_path / "sector_cache.json"),
        )
        monkeypatch.setattr(
            SnapshotStore,
            "_fetch_sector_from_yfinance",
            staticmethod(lambda symbol: "Healthcare" if symbol == "BAX" else "UNKNOWN"),
        )

        result = asyncio.run(
            SnapshotStore._fetch_sectors(["BAX", "AWK"], oracle_url="http://unused")
        )

        assert result == {"BAX": "Healthcare", "AWK": "Utilities"}
        cache = json.loads((tmp_path / "sector_cache.json").read_text())
        assert cache["BAX"] == "Healthcare"
        assert cache["AWK"] == "Utilities"