diff --git a/deploy/docker/truenas-compose.yml b/deploy/docker/truenas-compose.yml index 2cbb8af..c3a7b86 100644 --- a/deploy/docker/truenas-compose.yml +++ b/deploy/docker/truenas-compose.yml @@ -197,6 +197,8 @@ services: while true; do if [ "$$(date +%H:%M)" = "20:00" ] && [ "$$(date +%u)" -le 5 ]; then python -m research.premarket_eval || echo "[scheduler] eval FAILED" + # SPY/QQQ ingest -> forward source for the labeler's market-context features (plan 009) + python -m research.ingestion.market_context || echo "[scheduler] market_context ingest FAILED" python -m research.eod_reconcile || echo "[scheduler] reconcile FAILED" # Flywheel: grow the whole-market dataset first, then fit the tuner on # it. 20:00 ET is after close, so replay of "today" is complete data — diff --git a/docker-compose.yml b/docker-compose.yml index a54e5c2..5969c3e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,10 @@ x-app-env: &app-env # Self-contained default: the in-stack postgres. Overrides .env. DATABASE_URL: postgresql://${POSTGRES_USER:-momentum}:${POSTGRES_PASSWORD:-momentum}@postgres:5432/${POSTGRES_DB:-momentum} PG_SKIP_SCHEMA: "0" + # catalyst enrichment ON as a labeler FEATURE source (plan 009) — loop + scheduler + # (labeler builds) both inherit this anchor. Needs a reachable Ollama (see .env); + # enabled-but-down never impedes trading (fully gated). + NEWS_ENRICH_ENABLED: "1" services: postgres: @@ -141,6 +145,8 @@ services: if [ "$$(date +%H:%M)" = "20:00" ] && [ "$$(date +%u)" -le 5 ]; then echo "[scheduler] premarket_eval $$(date)" python -m research.premarket_eval || echo "[scheduler] eval FAILED" + # SPY/QQQ ingest -> forward source for the labeler's market-context features (plan 009) + python -m research.ingestion.market_context || echo "[scheduler] market_context ingest FAILED" python -m research.eod_reconcile || echo "[scheduler] reconcile FAILED" # Flywheel: grow the whole-market dataset first, then fit the tuner on # it. 20:00 ET is after close, so replay of "today" is complete data — diff --git a/research/ingestion/market_context.py b/research/ingestion/market_context.py new file mode 100644 index 0000000..4b732ec --- /dev/null +++ b/research/ingestion/market_context.py @@ -0,0 +1,76 @@ +"""Nightly SPY/QQQ ingest — the forward source for the labeler's market-context +features (tape_regime + spy_intraday_return, plan 009). + +Index ETFs never enter the gapper discovery universe, so the live loop never +ingests them; without this job market context can only be backfilled over the +~30 sessions of SPY data that happen to exist. This pulls the full session's +minute bars for SPY/QQQ (SIP feed — the job runs at 20:00 ET, past the 15-min +free-key embargo) plus daily bars, through the same upsert helpers the rest of +ingestion uses (idempotent ON CONFLICT). Mirrors research.sip_backfill's shape. + + python -m research.ingestion.market_context # today + python -m research.ingestion.market_context --date D # one session (YYYY-MM-DD) +""" + +from __future__ import annotations + +import argparse +import os +from datetime import date, datetime, timedelta + +from dotenv import load_dotenv + +load_dotenv(os.path.join(os.path.dirname(__file__), "..", "..", ".env")) + +from research.ingestion.market_data import ( # noqa: E402 + upsert_daily_bars, upsert_minute_bars) +from research.multi_schema import open_research_db # noqa: E402 + +INDEX_SYMBOLS = ("SPY", "QQQ") + + +def ingest_index_session(con, client, d, symbols=INDEX_SYMBOLS) -> tuple[int, int]: + """Ingest one session's SPY/QQQ minute bars (04:00-20:00 ET, SIP) + recent daily + bars. Returns (minute_rows, daily_rows). Unlike sip_backfill.refetch_minute_session + this fetches a FIXED symbol list, so it seeds SPY/QQQ even when they have no prior + rows that day.""" + d = d if isinstance(d, date) else date.fromisoformat(str(d)) + syms = list(symbols) + start = f"{d.isoformat()}T08:00:00Z" # 04:00 ET + end = f"{(d + timedelta(days=1)).isoformat()}T00:00:00Z" # 20:00 ET + # respect the free-key SIP embargo when the session is TODAY (end in the past) + embargo = (datetime.utcnow() - timedelta(minutes=16)).strftime("%Y-%m-%dT%H:%M:%SZ") + if d == date.today() and embargo < end: + end = embargo + minute_rows = 0 + bars_by_symbol = client.get_minute_bars(syms, start_iso=start, end_iso=end, feed="sip") + for sym, bars in bars_by_symbol.items(): + if bars: + minute_rows += upsert_minute_bars(con, sym, bars, source_provider="alpaca_sip") + daily_rows = 0 + start_daily = (datetime.utcnow() - timedelta(days=60)).strftime("%Y-%m-%dT00:00:00Z") + try: + daily_by_symbol = client.get_daily_bars(syms, start_iso=start_daily, feed="sip") + for sym, bars in daily_by_symbol.items(): + if bars: + daily_rows += upsert_daily_bars(con, sym, bars) + except Exception as exc: # noqa: BLE001 — minute bars are what market context needs + print(f"index daily ingest skipped: {exc}") + return minute_rows, daily_rows + + +def main() -> None: + ap = argparse.ArgumentParser(description="Nightly SPY/QQQ ingest (market-context source)") + ap.add_argument("--date", help="one session date (YYYY-MM-DD); default = today") + args = ap.parse_args() + from alpaca_paper.client import AlpacaPaperClient + from alpaca_paper.settings import AlpacaPaperSettings + con = open_research_db("market") + client = AlpacaPaperClient(AlpacaPaperSettings.from_env()) + d = date.fromisoformat(args.date) if args.date else date.today() + m, dr = ingest_index_session(con, client, d) + print(f"market-context ingest {d}: {m} SPY/QQQ minute rows, {dr} daily rows") + + +if __name__ == "__main__": + main() diff --git a/tests/unit/research/test_market_context_ingest.py b/tests/unit/research/test_market_context_ingest.py new file mode 100644 index 0000000..87d9edb --- /dev/null +++ b/tests/unit/research/test_market_context_ingest.py @@ -0,0 +1,40 @@ +"""SPY/QQQ market-context ingest (plan 009): fake-client bars land as minute_bars +and round-trip through the labeler's _preload_spy. Isolated :memory: schema.""" + +from datetime import date + +from research.ingestion.market_context import ingest_index_session +from research.labeler import _preload_spy +from storage.db import get_connection + +D = date(2026, 6, 2) + + +class _FakeClient: + """Minimal AlpacaPaperClient stand-in for SPY/QQQ minute + daily bars.""" + + def get_minute_bars(self, symbols, start_iso, end_iso, feed=None): + # two RTH minute bars (13:30/13:31Z = 09:30/09:31 ET) per requested symbol + out = {} + for s in symbols: + out[s] = [ + {"t": f"{D.isoformat()}T13:30:00Z", "o": 400, "h": 401, "l": 399, "c": 400.5, "v": 1000, "vw": 400.3}, + {"t": f"{D.isoformat()}T13:31:00Z", "o": 400.5, "h": 402, "l": 400, "c": 401.5, "v": 1200, "vw": 401.0}, + ] + return out + + def get_daily_bars(self, symbols, start_iso, feed=None): + return {s: [{"t": f"{D.isoformat()}T00:00:00Z", "o": 399, "h": 403, "l": 398, "c": 401.5, "v": 5_000_000, "vw": 400.8}] + for s in symbols} + + +def test_index_ingest_lands_and_preloads(): + con = get_connection(":memory:") + try: + m, dr = ingest_index_session(con, _FakeClient(), D) + assert m == 4 and dr == 2 # 2 symbols x 2 minute bars, x1 daily each + spy = _preload_spy(con) # the labeler's market-context source + assert D in spy and len(spy[D]) == 2 + assert set(spy[D].columns) >= {"timestamp", "open", "high", "low", "close"} + finally: + con.close()