diff --git a/src/fxhnt/application/historical_eod_ingest.py b/src/fxhnt/application/historical_eod_ingest.py index 86ab1ed..79a7557 100644 --- a/src/fxhnt/application/historical_eod_ingest.py +++ b/src/fxhnt/application/historical_eod_ingest.py @@ -25,6 +25,7 @@ so tests drive it with fakes — no network. """ from __future__ import annotations +import concurrent.futures import datetime as dt import logging @@ -52,9 +53,24 @@ class HistoricalEodIngest: self._store = store self._universe = universe - def ingest(self, tickers: list[str] | None = None, batch_log_every: int = 50) -> dict[str, int]: + def ingest( + self, + tickers: list[str] | None = None, + batch_log_every: int = 50, + max_workers: int = 12, + ) -> dict[str, int]: """Ingest EOD bars + membership for `tickers` (default: the FULL survivorship-free set). + Concurrency: the (I/O-bound, urllib) bar FETCH is parallelized across a bounded + `ThreadPoolExecutor(max_workers)` — at ~22k tickers this is the ~35h→~3-4h win. The DuckDB + store is SINGLE-WRITER (concurrent writes to one connection/file corrupt it), so EVERY write + (`write_features_bulk` + `upsert_membership`) is performed on THIS (main) thread as futures + complete via `as_completed`. Worker threads ONLY call `bars()` and return data; they never + touch the store. `max_workers` only affects speed, never the result (writes are per-symbol so + completion order is irrelevant). Per-ticker resilience is preserved: a fetch that raises → + that ticker is skipped+counted but STILL gets a membership "attempted" marker so it is not + refetched next run, exactly as the serial path did. + Returns counters: {tickers, rows, skipped, members, empty} — tickers actually ingested this run, feature-bar rows written, tickers skipped as already-processed, membership rows upserted, and tickers whose bars came back empty (fetched-but-no-price-history, recorded so they are not @@ -82,34 +98,56 @@ class HistoricalEodIngest: rows_written = 0 empty = 0 members_written = 0 - for i, ticker in enumerate(selected, start=1): - if ticker in present or ticker in attempted: - 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) - else: - empty += 1 - # Upsert membership PER-TICKER right after the fetch attempt (empty or not): this marks the - # ticker as processed so a name with no price history is not refetched on the next run. - self._store.upsert_membership([by_symbol[ticker]]) - members_written += 1 - ingested += 1 - if i % batch_log_every == 0: - log.info( - "HistoricalEodIngest: %d/%d processed (%d ingested, %d skipped, %d empty, " - "~%d rows written).", - i, len(selected), ingested, skipped, empty, rows_written, - ) + + # To-do = selection minus already-processed (skip logic stays on the main thread, unchanged). + todo = [t for t in selected if t not in present and t not in attempted] + skipped = len(selected) - len(todo) + + completed = 0 + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool: + # Submit one fetch future per to-do ticker. Worker threads ONLY call bars() — never the + # store — so the DuckDB single-writer invariant is upheld by construction. + future_to_ticker = {pool.submit(self._bars.bars, ticker): ticker for ticker in todo} + for future in concurrent.futures.as_completed(future_to_ticker): + ticker = future_to_ticker[future] + completed += 1 + # Per-ticker resilience: a fetch that raised is skipped+counted (treated as empty for + # price rows) but STILL gets a membership "attempted" marker below so it is not + # refetched next run — exactly as the serial path behaved on a per-ticker exception. + try: + raw = future.result() + except Exception as exc: # noqa: BLE001 — per-ticker resilience, never abort the run + log.warning("HistoricalEodIngest: fetch failed for %s (skipped): %s", ticker, exc) + raw = [] + + 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 + ] + # SERIAL WRITES on the main thread only (DuckDB single-writer). + if feature_rows: + self._store.write_features_bulk([(ticker, feature_rows)]) + rows_written += len(feature_rows) + else: + empty += 1 + # Upsert membership right after the fetch attempt (empty/raised or not): this marks the + # ticker as processed so a name with no price history is not refetched on the next run. + self._store.upsert_membership([by_symbol[ticker]]) + members_written += 1 + ingested += 1 + if completed % batch_log_every == 0: + log.info( + "HistoricalEodIngest: %d/%d processed (%d ingested, %d skipped, %d empty, " + "~%d rows written).", + skipped + completed, len(selected), ingested, skipped, empty, rows_written, + ) log.info( "HistoricalEodIngest: done — %d ingested, %d skipped (already processed), %d empty, " diff --git a/tests/integration/test_historical_eod_ingest.py b/tests/integration/test_historical_eod_ingest.py index 5efbbfb..10e18e7 100644 --- a/tests/integration/test_historical_eod_ingest.py +++ b/tests/integration/test_historical_eod_ingest.py @@ -14,6 +14,7 @@ Invariants proven here: from __future__ import annotations import datetime as dt +import threading from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore from fxhnt.application.historical_eod_ingest import HistoricalEodIngest @@ -172,6 +173,120 @@ def test_empty_bars_ticker_not_refetched(tmp_path) -> None: store.close() +def test_concurrent_ingest_writes_all_and_is_thread_safe(tmp_path) -> None: + """With max_workers>1 the FETCH runs on a thread pool but every DuckDB WRITE must happen on the + MAIN thread (DuckDB is single-writer; concurrent writes to one connection/file corrupt it). Drive + 30 tickers (1 empty + 1 raising) and assert: all non-empty rows written, empty marked, raiser + skipped, counts exact, AND the store was only ever touched from the calling (main) thread.""" + main_ident = threading.get_ident() + n = 30 + raiser = "T07" # one ticker whose fetch raises + empty_ticker = "T11" # one ticker whose bars come back empty + syms = [f"T{i:02d}" for i in range(n)] + members = [(s, "NYSE", "2000-01-01", "2026-06-01") for s in syms] + + bars_by_sym: dict[str, list[dict]] = {} + for i, s in enumerate(syms): + if s in (raiser, empty_ticker): + continue + # i+1 bars per ticker so we can verify exact row totals + bars_by_sym[s] = [ + {"date": f"2026-01-{d + 1:02d}", "adjClose": 1.0 + d, "close": 2.0 + d, "volume": 100.0 + d} + for d in range(i + 1) + ] + expected_rows = sum(len(b) for b in bars_by_sym.values()) + nonempty_syms = set(bars_by_sym) + + class ManyUniverse: + def survivorship_free(self) -> list[tuple[str, str, str, str]]: + return list(members) + + class ConcurrentBarClient: + """Fetched on worker threads. Returns fixed bars; one ticker raises.""" + + def __init__(self) -> None: + self.fetched: list[str] = [] + self._lock = threading.Lock() + + def bars(self, symbol: str, start_date: str | None = None) -> list[dict]: + with self._lock: + self.fetched.append(symbol) + if symbol == raiser: + raise RuntimeError("boom") + return list(bars_by_sym.get(symbol, [])) + + class ThreadAssertingStore: + """Wraps the real DuckDB store; records the calling thread of every mutating call and asserts + it is the MAIN thread (worker threads must NEVER touch the store).""" + + def __init__(self, inner: DuckDbFeatureStore) -> None: + self._inner = inner + self.write_threads: set[int] = set() + + def catalog(self): + return self._inner.catalog() + + def read_membership(self): + return self._inner.read_membership() + + def write_features_bulk(self, items): + self.write_threads.add(threading.get_ident()) + return self._inner.write_features_bulk(items) + + def upsert_membership(self, rows): + self.write_threads.add(threading.get_ident()) + return self._inner.upsert_membership(rows) + + def read_features(self, symbol): + return self._inner.read_features(symbol) + + def read_panel(self, symbols, feature): + return self._inner.read_panel(symbols, feature) + + def close(self): + return self._inner.close() + + inner = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) + store = ThreadAssertingStore(inner) + bar_client = ConcurrentBarClient() + svc = HistoricalEodIngest(bar_client, store, ManyUniverse()) + + result = svc.ingest(max_workers=8) + + # every ticker was attempted exactly once (raiser included, empty included) + assert sorted(bar_client.fetched) == sorted(syms) + # counts exact under concurrency + assert result["tickers"] == n # all attempted (incl. raiser + empty) + assert result["rows"] == expected_rows + # both the empty-bars ticker and the raiser produced no price rows → counted as "empty" + # (no-rows-written), and both still get a membership "attempted" marker. + assert result["empty"] == 2 # empty_ticker + raiser (both produced zero price rows) + assert result["members"] == n # every attempted ticker gets a membership "processed" marker + assert result["skipped"] == 0 + + # THREAD SAFETY: every store write happened on the MAIN thread only + assert store.write_threads == {main_ident} + + # all non-empty tickers' rows are present and correct + for s in nonempty_syms: + feats = store.read_features(s) + assert len(feats) == len(bars_by_sym[s]) + # empty ticker wrote no price rows but IS in membership (processed marker) + assert empty_ticker not in {c[0] for c in store.catalog()} + assert empty_ticker in {r[0] for r in store.read_membership()} + # raiser wrote no price rows but IS recorded as attempted (membership) so it is not refetched + assert raiser not in {c[0] for c in store.catalog()} + assert raiser in {r[0] for r in store.read_membership()} + + # re-run: nothing refetched (all now present/attempted) + bar_client.fetched.clear() + result2 = svc.ingest(max_workers=8) + assert bar_client.fetched == [] + assert result2["skipped"] == n + assert result2["tickers"] == 0 + store.close() + + def test_subset_ingest_touches_only_named_ticker(tmp_path) -> None: store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb")) bar_client = FakeBarClient()