gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(b3a): resumable survivorship-free historical EOD ingest -> warehouse + ticker membership

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-17 00:09:49 +02:00
parent 8d363e14b5
commit 371a16f94c
7 changed files with 371 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/fxhnt/adapters/data/tiingo_daily.py
View File

@@ -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))

46
src/fxhnt/adapters/data/tiingo_universe.py
View File

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

30
src/fxhnt/adapters/warehouse/duckdb_feature_store.py
View File

@@ -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()

109
src/fxhnt/application/historical_eod_ingest.py Normal file
View File

@@ -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),
}

13
src/fxhnt/ports/market_data.py
View File

@@ -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."""
...

8
src/fxhnt/ports/warehouse.py
View File

@@ -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)."""
...

150
tests/integration/test_historical_eod_ingest.py Normal file
View File

@@ -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()