You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
842 lines
33 KiB
Python
842 lines
33 KiB
Python
"""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"
|