perf(b3a): parallelize ingest fetch (ThreadPool); DuckDB writes stay main-thread (single-writer safe)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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, "

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