diff --git a/src/fxhnt/adapters/data/tiingo_daily.py b/src/fxhnt/adapters/data/tiingo_daily.py index d985054..2777493 100644 --- a/src/fxhnt/adapters/data/tiingo_daily.py +++ b/src/fxhnt/adapters/data/tiingo_daily.py @@ -82,6 +82,30 @@ class TiingoDailyClient: return self._crypto_closes(symbol, frm, to) return self._equity_closes(symbol, frm, to) + def bars(self, symbol: str, start_date: str | None = None) -> list[dict[str, Any]]: + """Full-history daily EOD bars for an equity symbol: per-date + {"date", "adjClose", "close", "volume"} ascending, over ALL available history (default + start_date = 1990-01-01, far before any US equity Tiingo carries). This is the + survivorship-free total-return + raw-close + volume read a walk-forward backtest needs. + + Reuses `_get` (key only in the Authorization header). Equity endpoint only — the daily + rows carry close/volume/adjClose; crypto routing is out of scope here (B3a is US equities).""" + frm = start_date or "1990-01-01" + to = dt.date.today().isoformat() + sym = urllib.parse.quote(symbol) + rows = self._get(_DAILY.format(base=self._base, sym=sym, frm=frm, to=to)) + out: list[dict[str, Any]] = [] + for r in rows or []: + out.append( + { + "date": str(r["date"])[:10], + "adjClose": float(r["adjClose"]), + "close": float(r["close"]), + "volume": float(r["volume"]), + } + ) + return out + def _equity_closes(self, symbol: str, frm: str, to: str) -> dict[str, float]: sym = urllib.parse.quote(symbol) rows = self._get(_DAILY.format(base=self._base, sym=sym, frm=frm, to=to)) diff --git a/src/fxhnt/adapters/data/tiingo_universe.py b/src/fxhnt/adapters/data/tiingo_universe.py index a751dbd..0b7b3ee 100644 --- a/src/fxhnt/adapters/data/tiingo_universe.py +++ b/src/fxhnt/adapters/data/tiingo_universe.py @@ -112,6 +112,41 @@ class TiingoUniverseSource: # --- filtering + ranking ----------------------------------------------------------------- + @staticmethod + def _is_us_common_stock(r: dict[str, str]) -> bool: + """The survivorship-free filter: US common stock in USD on a kept exchange (NO active gate — + delisted names pass). Listing-status (active vs delisted) is a SEPARATE concern applied by + callers that want only live names; the row itself is in-scope for the universe either way.""" + if not (r.get("ticker") or "").strip(): + return False + if (r.get("assetType") or "") != "Stock": + return False + if (r.get("priceCurrency") or "") != "USD": + return False + return (r.get("exchange") or "") in _US_EXCHANGES + + def survivorship_free(self) -> list[tuple[str, str, str, str]]: + """The FULL survivorship-free candidate set WITH dates: (ticker, exchange, startDate, endDate) + for every US common stock (Stock/USD/NYSE+NASDAQ+AMEX+…) the file carries, INCLUDING DELISTED + names. No active gate, no dollar-volume rank — this is the point-in-time-universe backbone a + walk-forward backtest reconstructs membership from. Reuses the supported-tickers + download/parse (`_fetch_candidates`); no zip logic duplicated.""" + out: list[tuple[str, str, str, str]] = [] + for r in self._fetch_candidates(): + if not self._is_us_common_stock(r): + continue + out.append( + ( + (r.get("ticker") or "").strip(), + (r.get("exchange") or ""), + str(r.get("startDate") or "")[:10], + str(r.get("endDate") or "")[:10], + ) + ) + log.info("TiingoUniverseSource: %d survivorship-free US common-stock candidates (incl. delisted).", + len(out)) + return out + def _candidates(self) -> list[str]: """Active US common-stock tickers from the supported-tickers file, in file order.""" rows = self._fetch_candidates() @@ -131,19 +166,12 @@ class TiingoUniverseSource: out: list[str] = [] for r in rows: - ticker = (r.get("ticker") or "").strip() - if not ticker: - continue - if (r.get("assetType") or "") != "Stock": - continue - if (r.get("priceCurrency") or "") != "USD": - continue - if (r.get("exchange") or "") not in _US_EXCHANGES: + if not self._is_us_common_stock(r): continue end = str(r.get("endDate") or "") if cutoff and end[:10] < cutoff: continue # delisted - out.append(ticker) + out.append((r.get("ticker") or "").strip()) return out @staticmethod diff --git a/src/fxhnt/adapters/warehouse/duckdb_feature_store.py b/src/fxhnt/adapters/warehouse/duckdb_feature_store.py index a274076..fe1c046 100644 --- a/src/fxhnt/adapters/warehouse/duckdb_feature_store.py +++ b/src/fxhnt/adapters/warehouse/duckdb_feature_store.py @@ -8,12 +8,21 @@ import duckdb from fxhnt.ports.warehouse import FeatureRow _SCHEMA = "CREATE TABLE IF NOT EXISTS features (symbol VARCHAR, ts BIGINT, feature VARCHAR, value DOUBLE)" +# ticker_membership: the survivorship-free listing index — each ticker's exchange + listing +# start/end dates, so a backtest reconstructs the point-in-time universe (incl. delisted names) +# straight from the warehouse, no live re-fetch. start_date/end_date kept as ISO TEXT (the +# supported-tickers file's native form; no parse/precision loss). +_MEMBERSHIP_SCHEMA = ( + "CREATE TABLE IF NOT EXISTS ticker_membership " + "(symbol TEXT PRIMARY KEY, exchange TEXT, start_date TEXT, end_date TEXT)" +) class DuckDbFeatureStore: def __init__(self, path: str) -> None: self._con = duckdb.connect(path) self._con.execute(_SCHEMA) + self._con.execute(_MEMBERSHIP_SCHEMA) def write_features(self, symbol: str, rows: list[FeatureRow]) -> None: flat = [(symbol, ts, name, float(value)) for ts, feats in rows for name, value in feats.items()] @@ -101,5 +110,26 @@ class DuckDbFeatureStore: out[symbol][int(ts) // 86_400] = float(value) return out + # --- ticker membership (survivorship-free listing index) --------------------------------- + + def upsert_membership(self, rows: list[tuple[str, str, str, str]]) -> None: + """Upsert (symbol, exchange, start_date, end_date) rows — idempotent on symbol (PK). + Re-running overwrites a symbol's row in place; no duplicates.""" + if not rows: + return + self._con.executemany( + "INSERT INTO ticker_membership VALUES (?, ?, ?, ?) " + "ON CONFLICT (symbol) DO UPDATE SET " + "exchange = excluded.exchange, start_date = excluded.start_date, " + "end_date = excluded.end_date", + [(s, e, sd, ed) for s, e, sd, ed in rows], + ) + + def read_membership(self) -> list[tuple[str, str, str, str]]: + """All membership rows: (symbol, exchange, start_date, end_date), symbol-ordered.""" + return self._con.execute( + "SELECT symbol, exchange, start_date, end_date FROM ticker_membership ORDER BY symbol" + ).fetchall() + def close(self) -> None: self._con.close() diff --git a/src/fxhnt/application/historical_eod_ingest.py b/src/fxhnt/application/historical_eod_ingest.py new file mode 100644 index 0000000..9455fcf --- /dev/null +++ b/src/fxhnt/application/historical_eod_ingest.py @@ -0,0 +1,109 @@ +"""HistoricalEodIngest — resumable, survivorship-free historical daily-EOD ingest into the +gold-layer warehouse, plus each ticker's listing membership. + +For the FULL survivorship-free US common-stock set (active AND delisted), per ticker: + 1. fetch full-history bars (adjClose + close + volume) from the (paid) Tiingo daily endpoint, + 2. bulk-write them as point-in-time features (`adjclose`/`close`/`volume`) keyed by epoch-second ts, + 3. record the ticker's (exchange, start_date, end_date) in `ticker_membership`. + +A later walk-forward backtest then reconstructs point-in-time universes (which names were listed on a +given date, incl. ones that have since died) AND their returns straight from the warehouse — no live +re-fetch, no survivorship bias. + +Resumability: a ticker whose bars are already present in the warehouse `catalog()` is SKIPPED — the +(paid) bar fetch is not repeated. Re-running is therefore safe AND cheap. The warehouse write is +idempotent per (symbol, ts, feature) anyway, so even a forced re-write produces no duplicate rows. + +Cost discipline: the service NEVER silently caps the set. It logs (INFO) the running ticker count + +an estimate of rows written every `batch_log_every`, and accepts a `tickers` subset so a small batch +can be validated before a full run. Dependencies (bar client / store / universe source) are injected, +so tests drive it with fakes — no network. +""" +from __future__ import annotations + +import datetime as dt +import logging + +from fxhnt.ports.market_data import HistoricalBarClient, SurvivorshipFreeUniverseSource +from fxhnt.ports.warehouse import FeatureRow, FeatureStore + +log = logging.getLogger(__name__) + +_EPOCH = dt.date(1970, 1, 1) + + +def _to_ts(iso_date: str) -> int: + """ISO date 'YYYY-MM-DD' → epoch-second at 00:00:00 UTC (the warehouse bar ts convention).""" + return (dt.date.fromisoformat(iso_date[:10]) - _EPOCH).days * 86_400 + + +class HistoricalEodIngest: + def __init__( + self, + bar_client: HistoricalBarClient, + store: FeatureStore, + universe: SurvivorshipFreeUniverseSource, + ) -> None: + self._bars = bar_client + self._store = store + self._universe = universe + + def ingest(self, tickers: list[str] | None = None, batch_log_every: int = 50) -> dict[str, int]: + """Ingest EOD bars + membership for `tickers` (default: the FULL survivorship-free set). + + Returns counters: {tickers, rows, skipped, members} — tickers actually ingested this run, + feature-bar rows written, tickers skipped as already-present, membership rows upserted.""" + members = self._universe.survivorship_free() + by_symbol = {sym: (sym, exch, sd, ed) for sym, exch, sd, ed in members} + + if tickers is None: + selected = [m[0] for m in members] + else: + # honor the explicit subset; keep only names we have membership metadata for + selected = [t for t in tickers if t in by_symbol] + missing = [t for t in tickers if t not in by_symbol] + if missing: + log.warning("HistoricalEodIngest: %d requested tickers not in universe (skipped): %s", + len(missing), missing) + + present = {row[0] for row in self._store.catalog()} # symbols already in the warehouse + + membership_rows = [by_symbol[t] for t in selected] + self._store.upsert_membership(membership_rows) + + ingested = 0 + skipped = 0 + rows_written = 0 + for i, ticker in enumerate(selected, start=1): + if ticker in present: + skipped += 1 + continue + raw = self._bars.bars(ticker) + feature_rows: list[FeatureRow] = [ + ( + _to_ts(str(b["date"])), + {"adjclose": float(b["adjClose"]), "close": float(b["close"]), "volume": float(b["volume"])}, + ) + for b in raw + ] + if feature_rows: + self._store.write_features_bulk([(ticker, feature_rows)]) + rows_written += len(feature_rows) + ingested += 1 + if i % batch_log_every == 0: + log.info( + "HistoricalEodIngest: %d/%d processed (%d ingested, %d skipped, ~%d rows written).", + i, len(selected), ingested, skipped, rows_written, + ) + + log.info( + "HistoricalEodIngest: done — %d ingested, %d skipped (already present), %d rows written, " + "%d membership rows upserted.", + ingested, skipped, rows_written, len(membership_rows), + ) + return { + "tickers": ingested, + "rows": rows_written, + "skipped": skipped, + "members": len(membership_rows), + } diff --git a/src/fxhnt/ports/market_data.py b/src/fxhnt/ports/market_data.py index 1dbf30b..68d644d 100644 --- a/src/fxhnt/ports/market_data.py +++ b/src/fxhnt/ports/market_data.py @@ -54,3 +54,16 @@ class UniverseSource(Protocol): def top_n(self, n: int) -> list[str]: """The n most-liquid US common-stock tickers by trailing dollar-volume (descending).""" ... + + +class HistoricalBarClient(Protocol): + def bars(self, symbol: str, start_date: str | None = None) -> list[dict[str, float | str]]: + """Full-history daily EOD bars: per-date {'date','adjClose','close','volume'} ascending.""" + ... + + +class SurvivorshipFreeUniverseSource(Protocol): + def survivorship_free(self) -> list[tuple[str, str, str, str]]: + """The FULL US common-stock candidate set WITH dates: (ticker, exchange, startDate, endDate), + INCLUDING delisted names — the point-in-time-universe backbone for a walk-forward backtest.""" + ... diff --git a/src/fxhnt/ports/warehouse.py b/src/fxhnt/ports/warehouse.py index 013f187..71f0aa7 100644 --- a/src/fxhnt/ports/warehouse.py +++ b/src/fxhnt/ports/warehouse.py @@ -37,3 +37,11 @@ class FeatureStore(Protocol): def read_panel(self, symbols: list[str], feature: str) -> dict[str, dict[int, float]]: """{symbol: {epoch_day: value}} for one feature across symbols.""" ... + + def upsert_membership(self, rows: list[tuple[str, str, str, str]]) -> None: + """Upsert (symbol, exchange, start_date, end_date) listing rows; idempotent on symbol.""" + ... + + def read_membership(self) -> list[tuple[str, str, str, str]]: + """All membership rows: (symbol, exchange, start_date, end_date).""" + ... diff --git a/tests/integration/test_historical_eod_ingest.py b/tests/integration/test_historical_eod_ingest.py new file mode 100644 index 0000000..34d5b7d --- /dev/null +++ b/tests/integration/test_historical_eod_ingest.py @@ -0,0 +1,150 @@ +"""HistoricalEodIngest — resumable survivorship-free historical EOD ingest into the warehouse. +NO live network: the Tiingo bar client + universe source are FAKES; the DuckDB store is real on a +tmp path. + +Invariants proven here: + (1) adjclose/close/volume are written as point-in-time features and round-trip via read_features + AND read_panel (the cross-sectional path a walk-forward backtest reads). + (2) ticker membership (symbol, exchange, start_date, end_date) is stored for EVERY name in the + survivorship-free set INCLUDING delisted ones (a delisted ticker is RETAINED, not dropped). + (3) re-running ingest is idempotent + cheap: no duplicate rows, and an already-present ticker is + SKIPPED (its bar client is not called again). + (4) a subset ingest(["AAA"]) touches ONLY AAA — no other ticker is fetched or written. +""" +from __future__ import annotations + +import datetime as dt + +from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore +from fxhnt.application.historical_eod_ingest import HistoricalEodIngest + +_EPOCH = dt.date(1970, 1, 1) + + +def _ts(iso: str) -> int: + """ISO date → epoch-second at 00:00:00 UTC (the warehouse bar ts convention).""" + return (dt.date.fromisoformat(iso) - _EPOCH).days * 86_400 + + +# --- fakes ------------------------------------------------------------------------------------- + +# Three names: two active, one DELISTED (old endDate). The delisted one must survive into membership. +_MEMBERS = [ + ("AAA", "NYSE", "2000-01-01", "2026-06-01"), + ("BBB", "NASDAQ", "2010-01-01", "2026-06-01"), + ("DEAD", "AMEX", "1998-01-01", "2005-03-15"), # delisted (survivorship-free retention target) +] + +_BARS = { + "AAA": [ + {"date": "2026-01-02", "adjClose": 90.0, "close": 100.0, "volume": 1_000.0}, + {"date": "2026-01-05", "adjClose": 91.5, "close": 101.0, "volume": 1_200.0}, + ], + "BBB": [ + {"date": "2026-01-02", "adjClose": 40.0, "close": 50.0, "volume": 2_000.0}, + ], + "DEAD": [ + {"date": "2005-03-14", "adjClose": 5.0, "close": 6.0, "volume": 300.0}, + {"date": "2005-03-15", "adjClose": 4.5, "close": 5.5, "volume": 280.0}, + ], +} + + +class FakeUniverseSource: + """Enumerates the survivorship-free candidate set WITH dates (incl. delisted).""" + + def __init__(self) -> None: + self.calls = 0 + + def survivorship_free(self) -> list[tuple[str, str, str, str]]: + self.calls += 1 + return list(_MEMBERS) + + +class FakeBarClient: + """Returns fixed full-history bars per ticker; records which tickers were fetched (cost proxy).""" + + def __init__(self) -> None: + self.fetched: list[str] = [] + + def bars(self, symbol: str, start_date: str | None = None) -> list[dict]: + self.fetched.append(symbol) + return list(_BARS.get(symbol, [])) + + +# --- tests ------------------------------------------------------------------------------------- + + +def test_features_written_and_round_trip(tmp_path) -> None: + store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) + svc = HistoricalEodIngest(FakeBarClient(), store, FakeUniverseSource()) + + result = svc.ingest() + + assert result["tickers"] == 3 + assert result["rows"] == 5 # AAA 2 + BBB 1 + DEAD 2 + + # round-trip via read_features: all three features present at the right ts + aaa = store.read_features("AAA") + assert aaa[0] == (_ts("2026-01-02"), {"adjclose": 90.0, "close": 100.0, "volume": 1_000.0}) + assert aaa[1] == (_ts("2026-01-05"), {"adjclose": 91.5, "close": 101.0, "volume": 1_200.0}) + + # round-trip via read_panel (the backtest cross-sectional read), per feature + adj = store.read_panel(["AAA", "BBB", "DEAD"], "adjclose") + assert adj["AAA"][_ts("2026-01-02") // 86_400] == 90.0 + assert adj["BBB"][_ts("2026-01-02") // 86_400] == 40.0 + assert adj["DEAD"][_ts("2005-03-15") // 86_400] == 4.5 + vol = store.read_panel(["DEAD"], "volume") + assert vol["DEAD"][_ts("2005-03-14") // 86_400] == 300.0 + store.close() + + +def test_delisted_membership_retained(tmp_path) -> None: + store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) + HistoricalEodIngest(FakeBarClient(), store, FakeUniverseSource()).ingest() + + members = {r[0]: r for r in store.read_membership()} + assert set(members) == {"AAA", "BBB", "DEAD"} + # the delisted name is RETAINED with its real (old) end_date — not dropped, not "active-ified" + assert members["DEAD"] == ("DEAD", "AMEX", "1998-01-01", "2005-03-15") + assert members["AAA"] == ("AAA", "NYSE", "2000-01-01", "2026-06-01") + store.close() + + +def test_rerun_is_idempotent_and_skips_present(tmp_path) -> None: + store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) + bar_client = FakeBarClient() + svc = HistoricalEodIngest(bar_client, store, FakeUniverseSource()) + + svc.ingest() + assert sorted(bar_client.fetched) == ["AAA", "BBB", "DEAD"] + rows_after_first = len(store.read_features("AAA")) + + # re-run: already-present tickers are SKIPPED (bar client not called again) + bar_client.fetched.clear() + result2 = svc.ingest() + assert bar_client.fetched == [] # nothing refetched + assert result2["skipped"] == 3 + + # no duplicate rows + assert len(store.read_features("AAA")) == rows_after_first + panel = store.read_panel(["AAA"], "close") + assert len(panel["AAA"]) == 2 # still exactly the two original days + store.close() + + +def test_subset_ingest_touches_only_named_ticker(tmp_path) -> None: + store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) + bar_client = FakeBarClient() + svc = HistoricalEodIngest(bar_client, store, FakeUniverseSource()) + + result = svc.ingest(tickers=["AAA"]) + + assert bar_client.fetched == ["AAA"] + assert result["tickers"] == 1 + # only AAA in the warehouse + assert [c[0] for c in store.catalog()] == ["AAA"] + # membership still recorded for AAA (from the universe metadata), none for the others + members = {r[0] for r in store.read_membership()} + assert members == {"AAA"} + store.close()