fix(b3a-deploy): dedicated external-443 egress netpol for the ingest Job (Tiingo reachable past default-deny)

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

docs/superpowers/specs/2026-06-17-equity-backtest-design.md

@@ -0,0 +1,50 @@
+# B3 — Survivorship-free price-only backtest of the equity-factor sleeves
+
+**Status:** design approved 2026-06-17.
+**Why:** pre-screen the equity-factor constructions against history (DSR/OOS verdict) *before* committing weeks of blind forward. Fundamentals are DOW-30-only on our Tiingo plan (see [[project_fxhnt_b1_b2_paper_tracks_live]]), so this is **price-only** (momentum + low-vol) — which is exactly what the live sleeves now run, and prices are broad + survivorship-free.
+
+## Scope
+Backtest the 3 price-only constructions (`long`/`ls`/`tilt`, momentum+low-vol) over the full available price history, **survivorship-free**, → a gauntlet verdict per construction. The walk-forward engine also generalizes to the single-market price "kinds" (which already fit `domain/backtest.py run_backtest`), but the equity-factor sleeves are the target.
+
+## Reuse (do not rebuild)
+- `domain/strategies/equity_factor.py` — `momentum_score`, `lowvol_score`, `composite_price`, `construction_weights`, `book_return`.
+- `domain/backtest.py compute_stats` — Sharpe/CAGR/maxDD on any return series.
+- The gauntlet (`GauntletSettings`: dsr_min 0.95, oos_min_sharpe, max_is_oos_decay 0.50, oos_fraction 0.40) + its evaluation path (as used in `application/research.py`).
+- The DuckDB warehouse (`adapters/warehouse/duckdb_feature_store.py`, `WarehouseIngest`) — the SSOT cache.
+- `adapters/data/tiingo_daily.py` + `tiingo_universe.py` (supported-tickers with start/end dates).
+
+## Architecture
+### B3a — Point-in-time data ingest → warehouse (the heavy piece; build FIRST)
+- A `HistoricalEodIngest` service: for each ticker in the survivorship-free US universe (`supported_tickers`: assetType=Stock, USD, NYSE/NASDAQ/AMEX, **including delisted** via end-date), fetch full Tiingo EOD (date, adjClose, close, volume) and bulk-write to the warehouse (long-format features per `DuckDbFeatureStore`). Also persist each ticker's `start_date`/`end_date` (a `ticker_membership` table/feature) for PIT membership.
+- **Resumable + incremental:** track which tickers/date-ranges are already ingested (skip/append only new). Safe to re-run.
+- **Cost discipline (HARD):** the full universe is ~22k tickers / ~20GB — about half our 40GB/mo Tiingo bandwidth. So: (1) build + validate the ingest on a **small batch (~100 tickers)** first; (2) the **full pull is a separate, deliberate, checkpointed run** (log running ticker count + bytes; never silent). Do NOT run the full pull as part of the build task.
+
+### B3b — PIT reconstruction + walk-forward + verdict (after B3a data exists)
+- `universe_asof(D, n)` (pure over warehouse reads): tickers with `start ≤ D < end` (active at D), ranked by trailing dollar-volume (close×volume from warehouse over a window ending at D) → top-N.
+- `prices_asof` / returns from the warehouse.
+- Walk-forward (monthly): PIT universe → `compute_factor_scores` price-only (momentum+lowvol) **reading the warehouse** (a warehouse-backed `DailyBarClient`, not live Tiingo) → `construction_weights` → hold → `book_return` with cost → daily return series per construction.
+- Verdict: `compute_stats` + IS/OOS split (40% holdout) + DSR + decay → GO/NO_GO per construction → JSON report under `/data/surfer/backtest/` (+ optional cockpit `backtest_summary` field).
+
+## Defaults
+window = full available · rebalance monthly · cost **15 bps/turnover** (flat; conservative since it understates 1990s–2000s spreads) · OOS holdout 40% · n=150 · DSR trial-count = the 3 constructions (+ any factor variants) tested.
+
+## Testing
+- B3a: ingest service with a fake Tiingo client → warehouse round-trip (rows written, membership stored, resumable skip on re-run, delisted ticker retained). No live network in tests.
+- B3b: PIT reconstruction on a synthetic warehouse (delisted ticker excluded after end-date; dollar-volume rank as-of D; insufficient-history name dropped). Walk-forward determinism. Verdict on a known return series. No live network.
+- strict mypy, two-stage subagent review.
+
+## Out of scope
+- Fundamentals factors (need the Tiingo add-on — separate; value/quality kept in the domain for then).
+- Nightly automation of the backtest (on-demand/occasional; the data ingest can be nightly-incremental, the backtest is run deliberately).
+- Live execution.
+
+## Honest caveats
+- One-time data pull is real (~22k calls / ~20GB / ~1 day; bandwidth-tight) — hence the small-batch-first + checkpoint discipline.
+- Cost model is flat-approximate, not era-exact.
+- Tiingo adjClose handles splits+dividends (total return); delisted final returns rely on Tiingo's end-date data (verify a sample).
+
+## Build order
+1. **B3a** — warehouse schema + `HistoricalEodIngest` (resumable) + tests, validated on ~100 tickers. **[checkpoint before the full pull]**
+2. Full PIT ingest run (deliberate, monitored).
+3. **B3b** — `universe_asof` + warehouse-backed bars + walk-forward + verdict + report + tests.
+4. Wire a verdict surface (report / optional cockpit field).

infra/k8s/jobs/fxhnt-backtest-ingest-job.yaml

@@ -0,0 +1,63 @@
+# B3a-deploy — survivorship-free historical-EOD ingest as an isolated, RESUMABLE one-shot Job.
+#
+# Runs `fxhnt ingest-historical` (full survivorship-free US-common-stock daily history → the DEDICATED
+# backtest warehouse on its OWN PVC, never the live cockpit `fxhnt-surfer-data`). The ingest is resumable
+# (already-processed tickers are skipped via the price catalog + membership markers), so this Job is SAFE
+# to re-run — re-submitting it picks up where a prior run left off and the paid Tiingo fetches are not
+# repeated. Delete + re-apply to resume after a failure or to top up incrementally.
+#
+# SMALL-BATCH VALIDATION: before the full pull, validate against a few tickers by changing the args to
+#   args: ["ingest-historical", "--limit", "100"]
+# (ingests only the first 100 survivorship-free tickers into the same dedicated warehouse).
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: fxhnt-backtest-ingest
+  namespace: foxhunt
+  labels: { app.kubernetes.io/name: fxhnt-backtest-ingest, app.kubernetes.io/part-of: foxhunt }
+spec:
+  backoffLimit: 2
+  ttlSecondsAfterFinished: 604800   # 7d — keep the finished Job around for log inspection before GC
+  template:
+    metadata:
+      labels: { app.kubernetes.io/part-of: foxhunt, app.kubernetes.io/name: fxhnt-backtest-ingest }   # external-443 egress via the dedicated NetworkPolicy below (part-of/base egress only reaches internal CIDRs)
+    spec:
+      restartPolicy: Never
+      nodeSelector:
+        k8s.scaleway.com/pool-name: platform
+      imagePullSecrets:
+        - name: gitlab-registry
+      containers:
+        - name: ingest
+          image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/fxhnt-cockpit:latest
+          # Full pull (no --limit). For a small-batch validation run, use:
+          #   args: ["ingest-historical", "--limit", "100"]
+          command: ["fxhnt"]
+          args: ["ingest-historical"]
+          env:
+            - name: FXHNT_BACKTEST_WAREHOUSE_PATH
+              value: /backtest-data/warehouse.duckdb
+            - name: TIINGO_API_KEY
+              valueFrom: { secretKeyRef: { name: tiingo-api, key: key } }
+          volumeMounts:
+            - { name: backtest-data, mountPath: /backtest-data }
+      volumes:
+        - name: backtest-data
+          persistentVolumeClaim: { claimName: fxhnt-backtest-data }
+---
+# Egress for the ingest Job: DNS + external 443 (Tiingo). foxhunt has default-deny-all egress, and the
+# shared `argo-base-egress` (part-of:foxhunt) only permits 443 to INTERNAL CIDRs — Tiingo is external, so
+# the ingest needs its own external-443 rule (mirrors the `dagster` netpol's external egress).
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: fxhnt-backtest-ingest
+  namespace: foxhunt
+  labels: { app.kubernetes.io/part-of: foxhunt }
+spec:
+  podSelector: { matchLabels: { app.kubernetes.io/name: fxhnt-backtest-ingest } }
+  policyTypes: [Egress]
+  egress:
+    - ports: [{ port: 53, protocol: UDP }, { port: 53, protocol: TCP }]
+    - ports: [{ port: 443, protocol: TCP }]
+      to: [{ ipBlock: { cidr: 0.0.0.0/0, except: [10.32.0.0/16, 172.16.0.0/16] } }]

infra/k8s/storage/fxhnt-backtest-pvc.yaml

@@ -0,0 +1,17 @@
+# B3a-deploy — DEDICATED backtest warehouse PVC.
+#
+# The survivorship-free historical-EOD ingest (full US-common-stock daily history, incl. delisted names)
+# is large and long-running, so it gets its OWN RWO block volume — NEVER the live 5 GiB cockpit
+# `fxhnt-surfer-data` PVC. The ingest Job mounts this at /backtest-data and writes warehouse.duckdb here.
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: fxhnt-backtest-data
+  namespace: foxhunt
+  labels: { app.kubernetes.io/name: fxhnt-backtest, app.kubernetes.io/part-of: foxhunt }
+spec:
+  accessModes: ["ReadWriteOnce"]
+  storageClassName: sbs-default
+  resources:
+    requests:
+      storage: 20Gi

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

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

src/fxhnt/adapters/orchestration/assets.py

@@ -210,51 +210,58 @@ def poc_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
 
 
 @asset
-def eqfactor_long_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
-    """Paper-forward long-only equity-factor sleeve (Tiingo universe + fundamentals + daily bars)."""
+def eqfactor_scores(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Shared PRICE-ONLY factor cross-section fetched ONCE (Tiingo universe + daily bars, batched-concurrent).
+
+    The sleeve is price-only (momentum + low-vol): one adj_closes fetch per ticker yields both signals, so
+    prices alone — which Tiingo serves broadly — score the full universe. Fundamentals are NOT fetched (the
+    Tiingo plan covers DOW-30 only, which silently dropped ~270/300 names). The three equity-factor sleeves
+    consume this SAME precomputed scores dict. A transient network/API error logs a warning and yields an
+    EMPTY cross-section (sleeves then book nothing) rather than crashing the nightly run."""
     from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient
-    from fxhnt.adapters.data.tiingo_fundamentals import TiingoFundamentalsClient
     from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource
+    from fxhnt.application.equity_factor_strategy import compute_factor_scores
+
+    try:
+        s = compute_factor_scores(TiingoUniverseSource(), TiingoDailyClient(), n=150)
+    except (OSError, TimeoutError, ConnectionError) as e:   # transient: don't crash the run
+        context.log.warning(f"eqfactor_scores fetch failed: {e}")
+        s = {"date": "", "syms": [], "scores": [], "prices": {}}
+    context.log.info(f"eqfactor_scores: {len(s['syms'])} names scored")
+    return s
+
+
+@asset
+def eqfactor_long_nav(context: AssetExecutionContext, eqfactor_scores: dict) -> dict:  # type: ignore[type-arg]
+    """Paper-forward long-only equity-factor sleeve over the shared precomputed scores (no I/O of its own)."""
     from fxhnt.application.equity_factor_strategy import EquityFactorLong
+    from fxhnt.application.forward_tracker import ForwardTracker
 
-    return _run_paper_tracker(
-        context,
-        "eqfactor_long_nav",
-        lambda: EquityFactorLong(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient()),
-        "eqfactor_long_state",
-    )
+    st = ForwardTracker(EquityFactorLong(eqfactor_scores), f"{_data_dir()}/eqfactor_long_state.json").step()
+    context.log.info(f"eqfactor_long_nav: {st.forward_days}d through {st.last_date}")
+    return {"forward_days": st.forward_days, "last_date": st.last_date}
 
 
 @asset
-def eqfactor_ls_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
-    """Paper-forward market-neutral long-short equity-factor sleeve (Tiingo universe + fundamentals + daily bars)."""
-    from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient
-    from fxhnt.adapters.data.tiingo_fundamentals import TiingoFundamentalsClient
-    from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource
+def eqfactor_ls_nav(context: AssetExecutionContext, eqfactor_scores: dict) -> dict:  # type: ignore[type-arg]
+    """Paper-forward market-neutral long-short equity-factor sleeve over the shared precomputed scores."""
     from fxhnt.application.equity_factor_strategy import EquityFactorLS
+    from fxhnt.application.forward_tracker import ForwardTracker
 
-    return _run_paper_tracker(
-        context,
-        "eqfactor_ls_nav",
-        lambda: EquityFactorLS(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient()),
-        "eqfactor_ls_state",
-    )
+    st = ForwardTracker(EquityFactorLS(eqfactor_scores), f"{_data_dir()}/eqfactor_ls_state.json").step()
+    context.log.info(f"eqfactor_ls_nav: {st.forward_days}d through {st.last_date}")
+    return {"forward_days": st.forward_days, "last_date": st.last_date}
 
 
 @asset
-def eqfactor_tilt_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
-    """Paper-forward long-only rank-weighted (tilt) equity-factor sleeve (Tiingo universe + fundamentals + daily bars)."""
-    from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient
-    from fxhnt.adapters.data.tiingo_fundamentals import TiingoFundamentalsClient
-    from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource
+def eqfactor_tilt_nav(context: AssetExecutionContext, eqfactor_scores: dict) -> dict:  # type: ignore[type-arg]
+    """Paper-forward long-only rank-weighted (tilt) equity-factor sleeve over the shared precomputed scores."""
     from fxhnt.application.equity_factor_strategy import EquityFactorTilt
+    from fxhnt.application.forward_tracker import ForwardTracker
 
-    return _run_paper_tracker(
-        context,
-        "eqfactor_tilt_nav",
-        lambda: EquityFactorTilt(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient()),
-        "eqfactor_tilt_state",
-    )
+    st = ForwardTracker(EquityFactorTilt(eqfactor_scores), f"{_data_dir()}/eqfactor_tilt_state.json").step()
+    context.log.info(f"eqfactor_tilt_nav: {st.forward_days}d through {st.last_date}")
+    return {"forward_days": st.forward_days, "last_date": st.last_date}
 
 
 @asset(deps=[combined_forward_nav, sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav,

src/fxhnt/adapters/orchestration/definitions.py

@@ -11,6 +11,7 @@ from fxhnt.adapters.orchestration.assets import (
     crypto_bars,
     eqfactor_long_nav,
     eqfactor_ls_nav,
+    eqfactor_scores,
     eqfactor_tilt_nav,
     funding_nav,
     futures_bars,
@@ -25,7 +26,7 @@ combined_book_job = define_asset_job(
     selection=[
         crypto_bars, futures_bars, combined_forward_nav,
         sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav,
-        eqfactor_long_nav, eqfactor_ls_nav, eqfactor_tilt_nav,
+        eqfactor_scores, eqfactor_long_nav, eqfactor_ls_nav, eqfactor_tilt_nav,
         cockpit_forward,
     ],
 )
@@ -42,7 +43,7 @@ defs = Definitions(
     assets=[
         crypto_bars, futures_bars, combined_forward_nav,
         sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav,
-        eqfactor_long_nav, eqfactor_ls_nav, eqfactor_tilt_nav,
+        eqfactor_scores, eqfactor_long_nav, eqfactor_ls_nav, eqfactor_tilt_nav,
         cockpit_forward,
     ],
     jobs=[combined_book_job],

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

src/fxhnt/application/equity_factor_strategy.py

@@ -1,22 +1,39 @@
-"""Equity-factor target builder + the three live-booking ForwardStrategy services (long / ls / tilt).
+"""Shared batched-concurrent PRICE-ONLY factor-score builder + the three live-booking ForwardStrategy services.
 
-Composes the universe + fundamentals + daily-bar ports and the pure `equity_factor` domain into a set of
-target weights, then books a daily realized return on the held book and rebalances on the monthly boundary
-— the live-booking pattern of FundingCarryStrategy/CrossVenueStrategy. On the FIRST run the prior book is
-empty so the booked return is 0 and the tracker freezes (books nothing); it just seeds the initial target
-positions + their marking prices + the rebalance month into `extra`."""
+The universe + per-ticker daily bars are fetched ONCE (concurrently, not per-sleeve) into a single
+precomputed `scores` dict via `compute_factor_scores`; the three sleeves (long / ls / tilt) each consume
+that SAME shared dict and do NO I/O of their own.
+
+The sleeve is PRICE-ONLY (momentum + low-vol): a single `adj_closes` fetch per ticker yields BOTH the
+12-1 momentum and a trailing realized vol, so the whole cross-section needs nothing but prices — which
+Tiingo serves broadly. Fundamentals are intentionally NOT fetched: the current Tiingo plan exposes
+fundamentals for DOW-30 only, so a fundamentals fetch silently dropped ~270/300 names. The
+FundamentalsClient port + TiingoFundamentalsClient adapter are retained (unused here) for a future
+fundamentals upgrade.
+
+Each service marks the prior target book to today's precomputed prices, books the realized daily return
+(net of short-borrow for the long-short construction), and rebalances to a fresh factor book on the monthly
+boundary. On the FIRST run the prior book is empty so the booked return is 0 and the tracker freezes; it
+just seeds the initial target positions + their marking prices + the rebalance month into `extra`."""
 from __future__ import annotations
 
+import concurrent.futures
+import logging
+import math
 from collections.abc import Callable
+from statistics import fmean
 from typing import Any
 
 from fxhnt.application.forward_tracker import _today_iso
 from fxhnt.domain.strategies import equity_factor as ef
-from fxhnt.ports.market_data import DailyBarClient, FundamentalsClient, UniverseSource
+from fxhnt.ports.market_data import DailyBarClient, UniverseSource
+
+_logger = logging.getLogger(__name__)
 
 # Trading-day offsets for the 12-1 momentum window: skip the most recent ~21d, look back ~252d (1y).
 _LOOKBACK = 252
 _SKIP = 21
+_VOL_WINDOW = 90
 
 
 def _momentum_12_1(closes: dict[str, float]) -> float | None:
@@ -35,104 +52,124 @@ def _momentum_12_1(closes: dict[str, float]) -> float | None:
     return p_recent / p_then - 1.0
 
 
-def _factor_targets(universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient,
-                    construction: str, n: int, quantile: float = 0.2) -> dict[str, float]:
-    """Build the target weight book for one construction by composing the domain over the live ports.
+def _realized_vol(closes: dict[str, float], window: int = _VOL_WINDOW) -> float | None:
+    """Trailing realized vol = sample stdev of the last `window` daily log returns.
 
-    Pulls the top-n universe, then per symbol the value metrics (pe/pb), statement metrics
-    (piotroski/roe/debtEquity/grossMargin) and 12-1 momentum; runs them through the pure
-    value/quality/momentum/composite pipeline; maps the composite to construction weights; and returns
-    {sym: weight} dropping zero-weight names."""
+    Sorts the dates, takes the most-recent `window+1` closes (-> `window` returns), and returns their
+    population/sample stdev. Returns None when there is not enough history for a full window or when any
+    close in the window is non-positive (log return undefined)."""
+    dates = sorted(closes)
+    if len(dates) < window + 1:
+        return None
+    tail = [closes[d] for d in dates[-(window + 1):]]
+    rets: list[float] = []
+    for p0, p1 in zip(tail, tail[1:]):
+        if p0 is None or p1 is None or p0 <= 0 or p1 <= 0:
+            return None
+        rets.append(math.log(p1 / p0))
+    if len(rets) < 2:
+        return None
+    mu = fmean(rets)
+    return math.sqrt(fmean([(r - mu) ** 2 for r in rets]))
+
+
+def _gather_ticker(sym: str, bars: DailyBarClient) -> dict[str, Any] | None:
+    """Fetch one ticker's adjusted-close history ONCE and derive its 12-1 momentum, trailing realized vol
+    and latest close. Per-ticker resilient: ANY exception (the Tiingo daily endpoint 400/404s on a
+    no-coverage name, ...) returns None. A name with insufficient history for momentum is also dropped (a
+    name we can't score must not be in the book) rather than aborting the whole sleeve. Pure read — safe to
+    run concurrently across a thread pool."""
+    try:
+        closes = bars.adj_closes(sym)
+        mom = _momentum_12_1(closes)
+        if mom is None:
+            return None
+        price = closes[max(closes)] if closes else None
+        return {
+            "sym": sym,
+            "mom": mom,
+            "vol": _realized_vol(closes),
+            "price": price,
+        }
+    except Exception:
+        return None
+
+
+def _gather_universe(universe: UniverseSource, bars: DailyBarClient,
+                     n: int, max_workers: int) -> list[dict[str, Any]]:
+    """Pull the top-n universe and gather each ticker's prices CONCURRENTLY over a thread pool, dropping
+    (and counting) any ticker whose gather failed. Returns the survivors' per-ticker records in the
+    deterministic top-n order (concurrency fans out the I/O; results are re-ordered to the universe order
+    so the cross-section + downstream weights are reproducible regardless of completion order)."""
     syms = universe.top_n(n)
-    pe: list[float | None] = []
-    pb: list[float | None] = []
-    pio: list[float | None] = []
-    roe: list[float | None] = []
-    de: list[float | None] = []
-    gm: list[float | None] = []
-    mom: list[float | None] = []
-    for sym in syms:
-        m = fund.metrics(sym)
-        s = fund.statement_metrics(sym)
-        pe.append(m.get("peRatio"))
-        pb.append(m.get("pbRatio"))
-        pio.append(s.get("piotroskiFScore"))
-        roe.append(s.get("roe"))
-        de.append(s.get("debtEquity"))
-        gm.append(s.get("grossMargin"))
-        mom.append(_momentum_12_1(bars.adj_closes(sym)))
+    records: dict[str, dict[str, Any]] = {}
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
+        futures = {pool.submit(_gather_ticker, sym, bars): sym for sym in syms}
+        for fut in concurrent.futures.as_completed(futures):
+            rec = fut.result()
+            if rec is not None:
+                records[rec["sym"]] = rec
+    survivors = [records[sym] for sym in syms if sym in records]
+    _logger.info("equity-factor: gathered %d/%d, dropped %d (no data)",
+                 len(survivors), len(syms), len(syms) - len(survivors))
+    return survivors
 
-    v = ef.value_score(pe, pb)
-    q = ef.quality_score(pio, roe, de, gm)
-    mm = ef.momentum_score(mom)
-    c = ef.composite(v, q, mm)
-    w = ef.construction_weights(c, construction, quantile)
-    return {sym: weight for sym, weight in zip(syms, w) if weight != 0}
+
+def compute_factor_scores(universe: UniverseSource, bars: DailyBarClient,
+                          n: int = 150, max_workers: int = 12) -> dict[str, Any]:
+    """Fetch the universe + prices ONCE (batched-concurrent) and compute the SHARED PRICE-ONLY factor
+    cross-section the three sleeves consume. Returns {'date', 'syms', 'scores', 'prices'} over survivors
+    only — gather each ticker's adj_closes concurrently (deriving momentum AND trailing realized vol from
+    that single fetch), run the survivors through the pure momentum + low-vol -> composite_price pipeline,
+    and emit the aligned syms/scores plus a {sym: latest_close} price map."""
+    survivors = _gather_universe(universe, bars, n, max_workers)
+    syms = [r["sym"] for r in survivors]
+    mm = ef.momentum_score([r["mom"] for r in survivors])
+    lv = ef.lowvol_score([r["vol"] for r in survivors])
+    c = ef.composite_price(mm, lv)
+    prices = {r["sym"]: r["price"] for r in survivors if r["price"] is not None}
+    return {"date": _today_iso(), "syms": syms, "scores": c, "prices": prices}
 
 
 class _EquityFactorStrategy:
-    """Live-booking equity-factor ForwardStrategy. Marks the prior target book to today's adjusted closes,
-    books the realized return (net of short-borrow for the long-short construction), and rebalances to a
-    fresh factor book on the monthly boundary. Carry state (`positions`, `prices`, `last_rebal`) lives in
-    the tracker's opaque `extra`."""
+    """Live-booking equity-factor ForwardStrategy over a SHARED precomputed `scores` dict (no I/O). Marks
+    the prior target book to today's precomputed closes, books the realized return (net of short-borrow for
+    the long-short construction), and rebalances to a fresh factor book on the monthly boundary. Carry state
+    (`positions`, `prices`, `last_rebal`) lives in the tracker's opaque `extra`."""
 
-    def __init__(self, construction: str, universe: UniverseSource, fund: FundamentalsClient,
-                 bars: DailyBarClient, n: int = 300, borrow_annual: float = 0.01,
+    def __init__(self, construction: str, scores: dict[str, Any], borrow_annual: float = 0.01,
                  clock: Callable[[], str] = _today_iso) -> None:
         self._construction = construction
-        self._universe = universe
-        self._fund = fund
-        self._bars = bars
-        self._n = n
+        self._scores = scores
         self._borrow_annual = borrow_annual
         self._clock = clock
 
-    def _latest_close(self, sym: str) -> float | None:
-        series = self._bars.adj_closes(sym)
-        if not series:
-            return None
-        return series[max(series)]
-
     def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
         prev: dict[str, float] = extra.get("positions", {})
         prev_prices: dict[str, float] = extra.get("prices", {})
         last_rebal: str | None = extra.get("last_rebal")
 
         today = self._clock()
-        rebalancing = last_rebal is None or today[:7] != last_rebal[:7]
-
-        # Mark the prior book to today's latest adjusted closes (current prices of the held names).
-        cur_prices: dict[str, float] = {}
-        for sym in prev:
-            px = self._latest_close(sym)
-            if px is not None:
-                cur_prices[sym] = px
-
+        cur: dict[str, float] = self._scores["prices"]
         borrow_daily = (self._borrow_annual / 252.0) if self._construction == "ls" else 0.0
-        r = ef.book_return(prev, prev_prices, cur_prices, borrow_daily) if prev else 0.0
+        r = ef.book_return(prev, prev_prices, cur, borrow_daily) if prev else 0.0
 
-        if rebalancing:
-            new_positions = _factor_targets(self._universe, self._fund, self._bars, self._construction, self._n)
+        if last_rebal is None or today[:7] != last_rebal[:7]:
+            w = ef.construction_weights(self._scores["scores"], self._construction)
+            new = {s: wt for s, wt in zip(self._scores["syms"], w) if wt != 0}
             last_rebal = today
         else:
-            new_positions = dict(prev)
+            new = dict(prev)
 
-        # Marking prices for the next step: latest close of the new held set.
-        new_prices: dict[str, float] = {}
-        for sym in new_positions:
-            px = self._latest_close(sym)
-            if px is not None:
-                new_prices[sym] = px
-
-        return [(today, r)], {"positions": new_positions, "prices": new_prices, "last_rebal": last_rebal}
+        new_prices = {s: cur[s] for s in new if s in cur}
+        return [(today, r)], {"positions": new, "prices": new_prices, "last_rebal": last_rebal}
 
 
 class EquityFactorLong:
-    """Long-only top-quintile equity-factor sleeve."""
+    """Long-only top-quintile equity-factor sleeve over a shared precomputed `scores` dict."""
 
-    def __init__(self, universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient,
-                 n: int = 300, clock: Callable[[], str] = _today_iso) -> None:
-        self._impl = _EquityFactorStrategy("long", universe, fund, bars, n=n, clock=clock)
+    def __init__(self, scores: dict[str, Any], clock: Callable[[], str] = _today_iso) -> None:
+        self._impl = _EquityFactorStrategy("long", scores, clock=clock)
 
     def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
         return self._impl.advance(last_date, extra)
@@ -141,20 +178,19 @@ class EquityFactorLong:
 class EquityFactorLS:
     """Market-neutral long-short equity-factor sleeve (top quintile long, bottom quintile short, borrow cost)."""
 
-    def __init__(self, universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient,
-                 n: int = 300, borrow_annual: float = 0.01, clock: Callable[[], str] = _today_iso) -> None:
-        self._impl = _EquityFactorStrategy("ls", universe, fund, bars, n=n, borrow_annual=borrow_annual, clock=clock)
+    def __init__(self, scores: dict[str, Any], borrow_annual: float = 0.01,
+                 clock: Callable[[], str] = _today_iso) -> None:
+        self._impl = _EquityFactorStrategy("ls", scores, borrow_annual=borrow_annual, clock=clock)
 
     def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
         return self._impl.advance(last_date, extra)
 
 
 class EquityFactorTilt:
-    """Long-only rank-weighted (tilt) equity-factor sleeve."""
+    """Long-only rank-weighted (tilt) equity-factor sleeve over a shared precomputed `scores` dict."""
 
-    def __init__(self, universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient,
-                 n: int = 300, clock: Callable[[], str] = _today_iso) -> None:
-        self._impl = _EquityFactorStrategy("tilt", universe, fund, bars, n=n, clock=clock)
+    def __init__(self, scores: dict[str, Any], clock: Callable[[], str] = _today_iso) -> None:
+        self._impl = _EquityFactorStrategy("tilt", scores, clock=clock)
 
     def advance(self, last_date: str | None, extra: dict[str, Any]) -> tuple[list[tuple[str, float]], dict[str, Any]]:
         return self._impl.advance(last_date, extra)

src/fxhnt/application/historical_eod_ingest.py

@@ -0,0 +1,125 @@
+"""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 is SKIPPED (its paid bar fetch is not repeated) if it is EITHER already present
+in the warehouse price `catalog()` OR already has a `ticker_membership` row from a prior completed pass.
+Membership is upserted PER-TICKER immediately after each fetch attempt (empty or not), so "has a
+membership row" == "already attempted" — which is what lets a delisted name whose `bars()` returns `[]`
+(no price rows ever written) still be recorded as processed and NOT refetched on every incremental run.
+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, 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
+        refetched on the next incremental run)."""
+        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)
+
+        # Already-processed = present in the price catalog OR already has a membership row from a prior
+        # completed pass. Read both once up-front; membership is the "fetched-but-empty" marker.
+        present = {row[0] for row in self._store.catalog()}  # symbols already in the warehouse
+        attempted = {row[0] for row in self._store.read_membership()}  # symbols already fetched
+
+        ingested = 0
+        skipped = 0
+        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,
+                )
+
+        log.info(
+            "HistoricalEodIngest: done — %d ingested, %d skipped (already processed), %d empty, "
+            "%d rows written, %d membership rows upserted.",
+            ingested, skipped, empty, rows_written, members_written,
+        )
+        return {
+            "tickers": ingested,
+            "rows": rows_written,
+            "skipped": skipped,
+            "members": members_written,
+            "empty": empty,
+        }

src/fxhnt/cli.py

@@ -470,6 +470,44 @@ def warehouse_ingest(
         store.close()
 
 
+@app.command("ingest-historical")
+def ingest_historical(
+    limit: int = typer.Option(
+        0, "--limit",
+        help="ingest only the first N survivorship-free tickers (0 = full set; use a small N to validate)",
+    ),
+) -> None:
+    """B3a: resumable, survivorship-free historical daily-EOD ingest into the DEDICATED backtest warehouse
+    (settings.backtest_warehouse_path — its OWN PVC, never the live cockpit warehouse). Re-running is safe
+    and cheap (already-processed tickers are skipped). Pass --limit N for a small-batch validation run."""
+    from pathlib import Path
+
+    from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient
+    from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource
+    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
+    from fxhnt.application.historical_eod_ingest import HistoricalEodIngest
+
+    settings = get_settings()
+    Path(settings.backtest_warehouse_path).parent.mkdir(parents=True, exist_ok=True)
+
+    tickers: list[str] | None = None
+    if limit > 0:
+        tickers = [t[0] for t in TiingoUniverseSource().survivorship_free()[:limit]]
+        typer.echo(f"ingest-historical: small-batch validation — first {len(tickers)} tickers")
+
+    store = DuckDbFeatureStore(settings.backtest_warehouse_path)
+    try:
+        ingest = HistoricalEodIngest(TiingoDailyClient(), store, TiingoUniverseSource())
+        counts = ingest.ingest(tickers=tickers)
+    finally:
+        store.close()
+    typer.echo(
+        f"ingest-historical -> {settings.backtest_warehouse_path}: "
+        f"{counts['tickers']} ingested, {counts['skipped']} skipped, {counts['empty']} empty, "
+        f"{counts['rows']} rows, {counts['members']} membership rows"
+    )
+
+
 @app.command("warehouse-catalog")
 def warehouse_catalog() -> None:
     """Show the warehouse SSOT inventory: per-symbol bar count + date range (freshness)."""

src/fxhnt/config.py

@@ -106,6 +106,10 @@ class Settings(BaseSettings):
     analytical_path: str = Field(default=str(_DATA_DIR / "analytical.duckdb"))
     # Warehouse (point-in-time SSOT: bars, funding, microstructure features) — DuckDB embedded file.
     warehouse_path: str = Field(default=str(_DATA_DIR / "warehouse.duckdb"))
+    # Backtest warehouse: a DEDICATED, isolated warehouse for the survivorship-free historical-EOD
+    # ingest (B3a). NEVER the live cockpit warehouse — the historical pull is large + long-running, so
+    # it gets its own DuckDB file on its own PVC. The cluster Job sets this to /backtest-data/warehouse.duckdb.
+    backtest_warehouse_path: str = Field(default="/backtest-data/warehouse.duckdb")
     # Combined-book data path: "raw" (crypto_pit/.dbn files) or "warehouse" (SSOT). Default raw for safety;
     # the B0 Dagster deploy sets FXHNT_COMBINED_BOOK_DATA_SOURCE=warehouse.
     combined_book_data_source: str = Field(default="raw")

src/fxhnt/domain/strategies/equity_factor.py

@@ -1,5 +1,13 @@
-"""Pure cross-sectional factor math: winsorized z-scores, value/quality/momentum families, composite,
-and the three portfolio constructions (long / long-short / tilt) + realized book return. No I/O."""
+"""Pure cross-sectional factor math: winsorized z-scores, price/fundamentals factor families,
+composites, and the three portfolio constructions (long / long-short / tilt) + realized book return.
+No I/O.
+
+The ACTIVE sleeve is PRICE-ONLY (momentum + low-vol) via `composite_price`, because prices are
+available broadly across the universe (including delisted names). The fundamentals factors
+(`value_score`, `quality_score`) and the 3-factor `composite` are RETAINED for a future Tiingo
+fundamentals upgrade — the current Tiingo plan exposes fundamentals for DOW-30 only, so they cannot
+score a broad universe yet. They are intentionally not used by `composite_price`.
+"""
 from __future__ import annotations
 
 import math
@@ -52,10 +60,22 @@ def momentum_score(mom_12_1: list[float | None]) -> list[float]:
     return robust_z(mom_12_1)
 
 
+def lowvol_score(vols: list[float | None]) -> list[float]:
+    """Low-volatility anomaly: LOWER trailing realized vol -> HIGHER score (None stays neutral 0.0)."""
+    return robust_z([None if v is None else -v for v in vols])
+
+
 def composite(value_z: list[float], quality_z: list[float], momentum_z: list[float]) -> list[float]:
+    """3-factor (value+quality+momentum) composite. RETAINED for a future Tiingo fundamentals
+    upgrade (fundamentals currently DOW-30-only); NOT used by the active price-only sleeve."""
     return _mean_z([value_z, quality_z, momentum_z])
 
 
+def composite_price(momentum_z: list[float], lowvol_z: list[float]) -> list[float]:
+    """ACTIVE price-only composite for the pivoted sleeve: equal-weight mean of momentum + low-vol z."""
+    return _mean_z([momentum_z, lowvol_z])
+
+
 def _quintile_cut(scores: list[float], quantile: float) -> tuple[float, float]:
     s = sorted(scores)
     n = len(s)

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

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

tests/integration/test_equity_factor.py

@@ -22,6 +22,22 @@ def test_composite_is_equal_weight_of_families() -> None:
     assert c == pytest.approx([1.0, -1.0])
 
 
+def test_lowvol_score_low_vol_ranks_high() -> None:
+    # low-volatility anomaly: lowest trailing vol -> highest score, highest vol -> lowest
+    z = ef.lowvol_score([0.10, 0.20, 0.30, 0.40, 0.50, None])
+    assert z[0] == max(z[:5])                               # lowest vol -> highest score
+    assert z[4] == min(z[:5])                               # highest vol -> lowest score
+    assert z[5] == 0.0                                      # None -> neutral
+
+
+def test_composite_price_equal_weight() -> None:
+    c = ef.composite_price(momentum_z=[1.0, -1.0], lowvol_z=[1.0, -1.0])
+    assert c == pytest.approx([1.0, -1.0])
+    # momentum and lowvol disagree -> equal-weight mean
+    c2 = ef.composite_price(momentum_z=[2.0, 0.0], lowvol_z=[0.0, 2.0])
+    assert c2 == pytest.approx([1.0, 1.0])
+
+
 def test_long_only_weights_top_quintile_sum_to_one() -> None:
     scores = [float(i) for i in range(10)]                 # 0..9; top quintile = top 2 (8,9)
     w = ef.construction_weights(scores, "long", quantile=0.2)

tests/integration/test_equity_factor_strategy.py

@@ -1,13 +1,15 @@
-"""Equity-factor target builder + the three live-booking ForwardStrategy services (long / ls / tilt).
+"""Shared batched-concurrent factor-score builder + the three live-booking ForwardStrategy services.
 
-Deterministic fakes only (no network): a fixed >=6-name cross-section with known pe/pb/piotroski/roe/
-debtEquity/grossMargin and enough adj-close history for the 12-1 momentum. Each construction is driven
-through TWO ForwardTracker steps (first freezes inception + seeds positions; second books one row after a
-month-boundary rebalance), round-tripped through ForwardStateReader, and the target builder is asserted
-directly against the pure domain's expected top/bottom names."""
+Deterministic fakes only (no network): a fixed >=6-name cross-section with enough adj-close history for
+both the 12-1 momentum (~252d) and the trailing realized vol (~90d). The sleeve is PRICE-ONLY
+(momentum + low-vol); fundamentals are NOT fetched. `compute_factor_scores` is asserted against the pure
+domain pipeline directly (incl. dropping a raising ticker). Each construction is then driven through TWO
+ForwardTracker steps from a SHARED fixed `scores` dict (no clients): first freezes inception + seeds
+positions; second books one row after a month-boundary rebalance, round-tripped through ForwardStateReader."""
 from __future__ import annotations
 
 import json
+import urllib.error
 from pathlib import Path
 
 import pytest
@@ -17,8 +19,9 @@ from fxhnt.application.equity_factor_strategy import (
     EquityFactorLong,
     EquityFactorLS,
     EquityFactorTilt,
-    _factor_targets,
     _momentum_12_1,
+    _realized_vol,
+    compute_factor_scores,
 )
 from fxhnt.application.forward_tracker import ForwardTracker
 from fxhnt.domain.strategies import equity_factor as ef
@@ -33,52 +36,32 @@ class FakeUniverse:
         return list(self._tickers[:n])
 
 
-class FakeFundamentals:
-    def __init__(self, metrics_by_sym: dict[str, dict[str, float]],
-                 stmts_by_sym: dict[str, dict[str, float]]) -> None:
-        self._metrics = metrics_by_sym
-        self._stmts = stmts_by_sym
-
-    def metrics(self, symbol: str) -> dict[str, float]:
-        return dict(self._metrics.get(symbol, {}))
-
-    def statement_metrics(self, symbol: str) -> dict[str, float]:
-        return dict(self._stmts.get(symbol, {}))
-
-
 class FakeDailyBars:
-    def __init__(self, closes_by_sym: dict[str, dict[str, float]]) -> None:
+    def __init__(self, closes_by_sym: dict[str, dict[str, float]],
+                 raise_for: str | None = None, exc: Exception | None = None) -> None:
         self._data = closes_by_sym
+        # When set, adj_closes() raises for this one ticker — models a name the Tiingo daily endpoint
+        # 400/404s on (no coverage / odd symbol) so the name is dropped from the cross-section.
+        self._raise_for = raise_for
+        self._exc = exc
 
     def adj_closes(self, symbol: str) -> dict[str, float]:
+        if self._raise_for is not None and symbol == self._raise_for:
+            raise self._exc if self._exc is not None else RuntimeError(f"no coverage for {symbol}")
         return dict(self._data.get(symbol, {}))
 
 
 # --------------------------------------------------------------------------- cross-section fixture
-# Six names A..F, monotone best->worst on EVERY family so the composite ranking is unambiguous:
-#   A best (cheap: low pe/pb; high quality: high piotroski/roe/gross_margin, low debt; high momentum),
-#   F worst. With n=6, quantile=0.2 the top/bottom quintile each select the single extreme name.
+# Six names A..F, monotone best->worst on the price-only families (momentum + low-vol) so the composite
+# ranking is unambiguous: A best (highest 12-1 momentum, lowest realized vol), F worst. With n=6,
+# quantile=0.2 the top/bottom quintile each select the single extreme name.
 _SYMS = ["A", "B", "C", "D", "E", "F"]
-_METRICS = {  # value family: low pe/pb is GOOD (domain negates them)
-    "A": {"peRatio": 8.0, "pbRatio": 0.8},
-    "B": {"peRatio": 12.0, "pbRatio": 1.2},
-    "C": {"peRatio": 16.0, "pbRatio": 1.6},
-    "D": {"peRatio": 20.0, "pbRatio": 2.0},
-    "E": {"peRatio": 26.0, "pbRatio": 2.6},
-    "F": {"peRatio": 34.0, "pbRatio": 3.4},
-}
-_STMTS = {  # quality family: high piotroski/roe/gross_margin GOOD, high debt BAD
-    "A": {"piotroskiFScore": 9.0, "roe": 0.30, "debtEquity": 0.1, "grossMargin": 0.60},
-    "B": {"piotroskiFScore": 8.0, "roe": 0.25, "debtEquity": 0.3, "grossMargin": 0.52},
-    "C": {"piotroskiFScore": 7.0, "roe": 0.20, "debtEquity": 0.5, "grossMargin": 0.44},
-    "D": {"piotroskiFScore": 5.0, "roe": 0.14, "debtEquity": 0.8, "grossMargin": 0.36},
-    "E": {"piotroskiFScore": 3.0, "roe": 0.08, "debtEquity": 1.2, "grossMargin": 0.28},
-    "F": {"piotroskiFScore": 2.0, "roe": 0.02, "debtEquity": 1.8, "grossMargin": 0.20},
-}
 # 12-1 momentum strength per name (the total cumulative drift baked into the price path).
 _MOM = {"A": 0.40, "B": 0.30, "C": 0.20, "D": 0.10, "E": 0.00, "F": -0.10}
+# trailing realized-vol amplitude per name (zig-zag size on top of the drift): A calmest, F most volatile.
+_VOL = {"A": 0.00, "B": 0.01, "C": 0.02, "D": 0.03, "E": 0.04, "F": 0.05}
 
-_N_DAYS = 300  # >= 252 so 12-1 momentum is computable
+_N_DAYS = 300  # >= 252 so 12-1 momentum is computable; > 90 so the realized vol window is full
 
 
 def _iso(i: int) -> str:
@@ -86,27 +69,43 @@ def _iso(i: int) -> str:
     return (datetime.date(2024, 1, 1) + datetime.timedelta(days=i)).isoformat()
 
 
-def _price_path(total_drift: float) -> dict[str, float]:
-    """Monotone geometric path over _N_DAYS dates whose 12-1 window ratio encodes `total_drift`.
-
-    The 12-1 momentum reads closes[d_-21]/closes[d_-252]-1 over a strictly increasing path, so a constant
-    daily growth makes that ratio deterministic and rank-preserving in `total_drift`."""
+def _price_path(total_drift: float, vol_amp: float = 0.0) -> dict[str, float]:
+    """Monotone-drift geometric path over _N_DAYS dates whose 12-1 window ratio encodes `total_drift`,
+    with a deterministic alternating ±`vol_amp` wiggle so the trailing realized vol is rank-ordered by
+    `vol_amp`. The 12-1 momentum reads closes[d_-21]/closes[d_-252]-1; the ±vol_amp wiggle nets out across
+    the 21d/252d offsets (it lands on the same parity) so momentum stays deterministic and rank-preserving
+    in `total_drift` regardless of the vol amplitude."""
     g = (1.0 + total_drift) ** (1.0 / _N_DAYS)
-    return {_iso(i): 100.0 * (g ** i) for i in range(_N_DAYS)}
+    return {_iso(i): 100.0 * (g ** i) * (1.0 + (vol_amp if i % 2 == 0 else -vol_amp)) for i in range(_N_DAYS)}
 
 
-def _make_bars() -> FakeDailyBars:
-    return FakeDailyBars({s: _price_path(_MOM[s]) for s in _SYMS})
-
-
-def _make_fund() -> FakeFundamentals:
-    return FakeFundamentals(_METRICS, _STMTS)
+def _make_bars(**kwargs) -> FakeDailyBars:
+    return FakeDailyBars({s: _price_path(_MOM[s], _VOL[s]) for s in _SYMS}, **kwargs)
 
 
 def _make_universe() -> FakeUniverse:
     return FakeUniverse(_SYMS)
 
 
+def _domain_scores(syms: list[str]) -> list[float]:
+    """The pure-domain PRICE-ONLY composite for the given names, fed the same aligned inputs the builder
+    would (momentum + low-vol) — asserted against directly (not circularly via the builder)."""
+    moms: list[float | None] = [_momentum_12_1(_price_path(_MOM[s], _VOL[s])) for s in syms]
+    vols: list[float | None] = [_realized_vol(_price_path(_MOM[s], _VOL[s])) for s in syms]
+    return ef.composite_price(ef.momentum_score(moms), ef.lowvol_score(vols))
+
+
+def _fixed_scores(syms: list[str]) -> dict:
+    """A precomputed `scores` dict (the shared product of compute_factor_scores) for the given names, with
+    the latest close of each name's price path as its marking price — the form the services consume."""
+    return {
+        "date": "2026-01-15",
+        "syms": list(syms),
+        "scores": _domain_scores(syms),
+        "prices": {s: _price_path(_MOM[s], _VOL[s])[_iso(_N_DAYS - 1)] for s in syms},
+    }
+
+
 # --------------------------------------------------------------------------- _momentum_12_1
 def test_momentum_12_1_uses_252_21_window() -> None:
     closes = _price_path(0.40)
@@ -120,61 +119,71 @@ def test_momentum_12_1_none_when_too_short() -> None:
     assert _momentum_12_1(closes) is None
 
 
-# --------------------------------------------------------------------------- _factor_targets composition
-def test_factor_targets_long_selects_top_name() -> None:
-    targets = _factor_targets(_make_universe(), _make_fund(), _make_bars(), "long", n=6)
-    # long-only top quintile (single best name A) holds full weight; nothing else.
-    assert set(targets) == {"A"}
-    assert targets["A"] == pytest.approx(1.0)
-    assert all(w >= 0.0 for w in targets.values())
+# --------------------------------------------------------------------------- _realized_vol
+def test_realized_vol_orders_by_amplitude() -> None:
+    """A calmer price path (smaller wiggle) has strictly lower trailing realized vol."""
+    calm = _realized_vol(_price_path(0.10, 0.01))
+    choppy = _realized_vol(_price_path(0.10, 0.05))
+    assert calm is not None and choppy is not None
+    assert calm < choppy
 
 
-def test_factor_targets_ls_longs_best_shorts_worst() -> None:
-    targets = _factor_targets(_make_universe(), _make_fund(), _make_bars(), "ls", n=6)
-    assert targets["A"] == pytest.approx(1.0)    # best name long
-    assert targets["F"] == pytest.approx(-1.0)   # worst name short
-    # market-neutral, gross ~2
-    assert sum(targets.values()) == pytest.approx(0.0)
-    assert sum(abs(w) for w in targets.values()) == pytest.approx(2.0)
+def test_realized_vol_none_when_too_short() -> None:
+    closes = {_iso(i): 100.0 + i for i in range(30)}  # < 90 returns available
+    assert _realized_vol(closes, window=90) is None
 
 
-def test_factor_targets_tilt_is_long_only_overweights_best() -> None:
-    targets = _factor_targets(_make_universe(), _make_fund(), _make_bars(), "tilt", n=6)
-    assert all(w >= 0.0 for w in targets.values())
-    assert sum(targets.values()) == pytest.approx(1.0)
-    # best name A overweighted vs the median-and-below names (which are zeroed by the tilt cut).
-    assert targets["A"] == max(targets.values())
+# --------------------------------------------------------------------------- compute_factor_scores
+def test_compute_factor_scores_matches_domain_and_emits_prices() -> None:
+    """The shared builder must fetch each ticker's adj_closes ONCE and emit aligned syms/scores equal to
+    the pure-domain PRICE-ONLY composite (asserted directly, not circularly) plus a {sym: latest_close}."""
+    s = compute_factor_scores(_make_universe(), _make_bars(), n=6)
+    assert s["syms"] == _SYMS
+    assert s["scores"] == pytest.approx(_domain_scores(_SYMS))
+    # latest adjusted close of each name's price path is its marking price
+    for sym in _SYMS:
+        assert s["prices"][sym] == pytest.approx(_price_path(_MOM[sym], _VOL[sym])[_iso(_N_DAYS - 1)])
+    assert "date" in s
 
 
-def test_factor_targets_matches_domain_pipeline() -> None:
-    """_factor_targets must equal feeding the same aligned inputs through the pure domain directly."""
-    syms = _make_universe().top_n(6)
-    pe: list[float | None] = [_METRICS[s]["peRatio"] for s in syms]
-    pb: list[float | None] = [_METRICS[s]["pbRatio"] for s in syms]
-    pio: list[float | None] = [_STMTS[s]["piotroskiFScore"] for s in syms]
-    roe: list[float | None] = [_STMTS[s]["roe"] for s in syms]
-    de: list[float | None] = [_STMTS[s]["debtEquity"] for s in syms]
-    gm: list[float | None] = [_STMTS[s]["grossMargin"] for s in syms]
-    mom = [_momentum_12_1(_price_path(_MOM[s])) for s in syms]
-    c = ef.composite(ef.value_score(pe, pb), ef.quality_score(pio, roe, de, gm), ef.momentum_score(mom))
-    w = ef.construction_weights(c, "ls", 0.2)
-    expected = {s: wt for s, wt in zip(syms, w) if wt != 0}
-    assert _factor_targets(_make_universe(), _make_fund(), _make_bars(), "ls", n=6) == pytest.approx(expected)
+def test_compute_factor_scores_drops_raising_ticker() -> None:
+    """A ticker whose daily endpoint raises (Tiingo 400/404) is dropped from the cross-section without
+    aborting the build; survivors-only, and their scores equal the domain over just the survivors."""
+    bars = _make_bars(
+        raise_for="C",
+        exc=urllib.error.HTTPError("http://tiingo/daily/C", 404, "Not Found", {}, None),  # type: ignore[arg-type]
+    )
+    s = compute_factor_scores(_make_universe(), bars, n=6)
+
+    survivors = ["A", "B", "D", "E", "F"]
+    assert s["syms"] == survivors                 # C dropped, order preserved
+    assert "C" not in s["prices"]
+    assert s["scores"] == pytest.approx(_domain_scores(survivors))  # scored over survivors only
+
+
+def test_compute_factor_scores_drops_ticker_with_insufficient_history() -> None:
+    """A ticker without enough price history for momentum/vol is dropped (no momentum signal -> not scored)."""
+    data = {s: _price_path(_MOM[s], _VOL[s]) for s in _SYMS}
+    data["C"] = {_iso(i): 100.0 + i for i in range(50)}  # < 252: momentum None -> dropped
+    bars = FakeDailyBars(data)
+    s = compute_factor_scores(_make_universe(), bars, n=6)
+    assert s["syms"] == ["A", "B", "D", "E", "F"]
+    assert "C" not in s["prices"]
 
 
 # --------------------------------------------------------------------------- live-booking services
 def _drive_two_steps(strategy_factory, sid: str, tmp_path, *, ls: bool):
-    """First step freezes inception + seeds target positions; advance prices + cross a month boundary via a
-    clock; second step books exactly one row. Returns (st1, loaded0, loaded1, summary, rows, prev_targets)."""
-    bars = _make_bars()
-    fund = _make_fund()
-    universe = _make_universe()
+    """First step freezes inception + seeds target positions; cross a month boundary via a clock so the
+    second step rebalances and books exactly one row off the SHARED precomputed scores. The second step's
+    scores carry BUMPED prices so the prior book marks to a non-trivial realized return.
+    Returns (st1, loaded0, loaded1, summary, rows, prev_targets)."""
     p = str(tmp_path / f"{sid}_state.json")
 
     # clock: inception in month M, second run in month M+1 (forces a rebalance + a strictly-later date).
     day = ["2026-01-15"]
 
-    st0 = ForwardTracker(strategy_factory(universe, fund, bars, clock=lambda: day[0]), p).step()
+    scores0 = _fixed_scores(_SYMS)
+    st0 = ForwardTracker(strategy_factory(scores0, clock=lambda: day[0]), p).step()
     assert st0.forward_days == 0
     loaded0 = json.loads(Path(p).read_text())
     prev_targets = dict(loaded0["extra"]["positions"])
@@ -182,18 +191,14 @@ def _drive_two_steps(strategy_factory, sid: str, tmp_path, *, ls: bool):
     assert prev_targets                                  # seeded a real target book
     assert loaded0["extra"]["last_rebal"] == "2026-01-15"
 
-    # Advance every held name's latest price by appending a new dated close, then cross the month boundary.
-    new_idx = _N_DAYS
+    # Second run: bump every name's marking price (a new precomputed `scores` snapshot) and cross the month.
     bumps = {"A": 1.10, "B": 1.05, "C": 1.00, "D": 0.98, "E": 0.95, "F": 0.90}
-    cur_prices = {}
-    for s in _SYMS:
-        series = bars._data[s]
-        last = series[_iso(_N_DAYS - 1)]
-        series[_iso(new_idx)] = last * bumps[s]
-        cur_prices[s] = series[_iso(new_idx)]
+    scores1 = _fixed_scores(_SYMS)
+    cur_prices = {s: scores1["prices"][s] * bumps[s] for s in _SYMS}
+    scores1["prices"] = dict(cur_prices)
     day[0] = "2026-02-15"  # next month → month-changed rebalance, strictly later date → booked
 
-    st1 = ForwardTracker(strategy_factory(universe, fund, bars, clock=lambda: day[0]), p).step()
+    st1 = ForwardTracker(strategy_factory(scores1, clock=lambda: day[0]), p).step()
     assert st1.forward_days == 1
     loaded1 = json.loads(Path(p).read_text())
     summary, rows = ForwardStateReader().read(p, sid)
@@ -237,22 +242,19 @@ def test_equity_factor_tilt_service_round_trips(tmp_path) -> None:
 
 def test_no_rebalance_within_same_month(tmp_path) -> None:
     """Second step in the SAME month must NOT rebalance: positions/last_rebal unchanged, still books a row."""
-    bars = _make_bars()
-    fund = _make_fund()
-    universe = _make_universe()
     p = str(tmp_path / "eqfactor_long_state.json")
 
     day = ["2026-01-10"]
-    ForwardTracker(EquityFactorLong(universe, fund, bars, clock=lambda: day[0]), p).step()
+    scores0 = _fixed_scores(_SYMS)
+    ForwardTracker(EquityFactorLong(scores0, clock=lambda: day[0]), p).step()
     loaded0 = json.loads(Path(p).read_text())
     pos0 = loaded0["extra"]["positions"]
 
-    # advance prices, same month, strictly-later date
-    for s in _SYMS:
-        series = bars._data[s]
-        series[_iso(_N_DAYS)] = series[_iso(_N_DAYS - 1)] * 1.01
+    # new price snapshot, same month, strictly-later date
+    scores1 = _fixed_scores(_SYMS)
+    scores1["prices"] = {s: scores1["prices"][s] * 1.01 for s in _SYMS}
     day[0] = "2026-01-20"
-    st1 = ForwardTracker(EquityFactorLong(universe, fund, bars, clock=lambda: day[0]), p).step()
+    st1 = ForwardTracker(EquityFactorLong(scores1, clock=lambda: day[0]), p).step()
     assert st1.forward_days == 1
     loaded1 = json.loads(Path(p).read_text())
     assert loaded1["extra"]["positions"] == pos0                 # positions held (no rebalance)

tests/integration/test_historical_eod_ingest.py

@@ -0,0 +1,189 @@
+"""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_empty_bars_ticker_not_refetched(tmp_path) -> None:
+    """A name whose bars() returns [] (delisted, no price history) is recorded as PROCESSED via its
+    membership row on the first run, so the next incremental run does NOT re-fetch it (no wasted call)."""
+    store = DuckDbFeatureStore(str(tmp_path / "wh.duckdb"))
+    bar_client = FakeBarClient()
+
+    class EmptyBarUniverse:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        def survivorship_free(self) -> list[tuple[str, str, str, str]]:
+            self.calls += 1
+            # GONE has membership metadata but FakeBarClient has no bars for it → bars() returns []
+            return [
+                ("AAA", "NYSE", "2000-01-01", "2026-06-01"),
+                ("GONE", "AMEX", "1990-01-01", "1995-12-31"),
+            ]
+
+    svc = HistoricalEodIngest(bar_client, store, EmptyBarUniverse())
+
+    result = svc.ingest()
+    # GONE was fetched once (its empty result recorded), AAA fetched + written
+    assert sorted(bar_client.fetched) == ["AAA", "GONE"]
+    assert result["empty"] == 1  # GONE
+    assert result["tickers"] == 2  # both attempted this run
+    # GONE wrote no price rows but DID get a membership row (the "processed" marker)
+    assert "GONE" not in {c[0] for c in store.catalog()}
+    assert "GONE" in {r[0] for r in store.read_membership()}
+
+    # re-run: GONE is now marked processed (membership) → NOT refetched; AAA present in catalog → skipped
+    bar_client.fetched.clear()
+    result2 = svc.ingest()
+    assert bar_client.fetched == []  # zero calls, GONE included
+    assert result2["skipped"] == 2
+    assert result2["tickers"] == 0
+    assert result2["empty"] == 0
+    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()

tests/integration/test_orchestration_definitions.py

@@ -17,6 +17,9 @@ def test_definitions_load_with_assets_and_schedule() -> None:
     # B2: the three equity-factor assets are wired into the graph alongside the B1 six
     eqfactor = {"eqfactor_long_nav", "eqfactor_ls_nav", "eqfactor_tilt_nav"}
     assert eqfactor <= asset_keys, f"missing equity-factor assets: {eqfactor - asset_keys}"
+    # B2.1: the shared scores asset is wired in (fetched once, consumed by the three sleeves) — 14 total
+    assert "eqfactor_scores" in asset_keys, f"missing eqfactor_scores asset: {asset_keys}"
+    assert len(asset_keys) == 14, f"expected 14 assets, got {len(asset_keys)}: {asset_keys}"
 
     # --- schedule present with the right cron ---
     # defs.schedules is a list[ScheduleDefinition] (or None when empty)
@@ -36,3 +39,9 @@ def test_definitions_load_with_assets_and_schedule() -> None:
     upstream = {k.to_user_string() for k in deps}
     expected_upstream = {"combined_forward_nav"} | paper | eqfactor
     assert expected_upstream <= upstream, f"cockpit_forward missing upstream: {expected_upstream - upstream}"
+
+    # --- each equity-factor nav sleeve depends on the shared eqfactor_scores asset (fetched once) ---
+    for nav in ("eqfactor_long_nav", "eqfactor_ls_nav", "eqfactor_tilt_nav"):
+        nav_def = repo.assets_defs_by_key[AssetKey(nav)]
+        nav_up = {k.to_user_string() for k in nav_def.asset_deps[AssetKey(nav)]}
+        assert "eqfactor_scores" in nav_up, f"{nav} missing eqfactor_scores upstream dep: {nav_up}"

tests/unit/test_ingest_historical_cli.py

@@ -0,0 +1,87 @@
+"""B3a-deploy: the `ingest-historical` CLI entrypoint wires the historical-EOD ingest against the
+DEDICATED backtest warehouse path (never the live cockpit warehouse) and honors --limit. No network:
+the Tiingo clients are monkeypatched with fakes."""
+from __future__ import annotations
+
+from typer.testing import CliRunner
+
+from fxhnt.config import Settings
+
+
+def test_settings_has_backtest_warehouse_path_default() -> None:
+    s = Settings()
+    assert s.backtest_warehouse_path == "/backtest-data/warehouse.duckdb"
+
+
+def test_settings_backtest_path_env_override(monkeypatch) -> None:
+    monkeypatch.setenv("FXHNT_BACKTEST_WAREHOUSE_PATH", "/tmp/bt.duckdb")
+    s = Settings()
+    assert s.backtest_warehouse_path == "/tmp/bt.duckdb"
+
+
+class _FakeUniverse:
+    """Survivorship-free set: three members; .survivorship_free() returns (sym, exch, start, end)."""
+
+    def survivorship_free(self):
+        return [
+            ("AAA", "NYSE", "1990-01-02", "2026-06-01"),
+            ("BBB", "NASDAQ", "1995-03-01", "2010-12-31"),
+            ("CCC", "AMEX", "2000-06-15", "2026-06-01"),
+        ]
+
+
+class _FakeBars:
+    def bars(self, ticker, start_date=None):
+        return [{"date": "2020-01-02", "adjClose": 1.0, "close": 1.0, "volume": 100.0}]
+
+
+def test_ingest_historical_builds_store_at_backtest_path_and_honors_limit(tmp_path, monkeypatch) -> None:
+    """--limit N restricts to the first N tickers AND the ingest writes to the configured backtest path."""
+    import fxhnt.cli as cli
+    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
+
+    bt_path = tmp_path / "backtest" / "warehouse.duckdb"
+    monkeypatch.setenv("FXHNT_BACKTEST_WAREHOUSE_PATH", str(bt_path))
+    cli.get_settings.cache_clear()  # the settings are lru_cached — pick up the env override
+
+    # Patch the concrete adapters the command imports (function-local imports -> patch at source module).
+    monkeypatch.setattr("fxhnt.adapters.data.tiingo_daily.TiingoDailyClient", lambda *a, **k: _FakeBars())
+    monkeypatch.setattr("fxhnt.adapters.data.tiingo_universe.TiingoUniverseSource", lambda *a, **k: _FakeUniverse())
+
+    result = CliRunner().invoke(cli.app, ["ingest-historical", "--limit", "2"])
+    cli.get_settings.cache_clear()
+
+    assert result.exit_code == 0, result.output
+    assert bt_path.exists(), "ingest must create the warehouse at the configured backtest path"
+    assert "2 ingested" in result.output  # only the first 2 of 3 tickers (the --limit)
+
+    # The two limited tickers are in the dedicated warehouse; the third (CCC) is not.
+    store = DuckDbFeatureStore(str(bt_path))
+    try:
+        symbols = {row[0] for row in store.catalog()}
+    finally:
+        store.close()
+    assert symbols == {"AAA", "BBB"}
+
+
+def test_ingest_historical_full_set_when_no_limit(tmp_path, monkeypatch) -> None:
+    """No --limit (0) ingests the FULL survivorship-free set into the backtest warehouse."""
+    import fxhnt.cli as cli
+    from fxhnt.adapters.warehouse.duckdb_feature_store import DuckDbFeatureStore
+
+    bt_path = tmp_path / "warehouse.duckdb"
+    monkeypatch.setenv("FXHNT_BACKTEST_WAREHOUSE_PATH", str(bt_path))
+    cli.get_settings.cache_clear()
+    monkeypatch.setattr("fxhnt.adapters.data.tiingo_daily.TiingoDailyClient", lambda *a, **k: _FakeBars())
+    monkeypatch.setattr("fxhnt.adapters.data.tiingo_universe.TiingoUniverseSource", lambda *a, **k: _FakeUniverse())
+
+    result = CliRunner().invoke(cli.app, ["ingest-historical"])
+    cli.get_settings.cache_clear()
+
+    assert result.exit_code == 0, result.output
+    store = DuckDbFeatureStore(str(bt_path))
+    try:
+        symbols = {row[0] for row in store.catalog()}
+    finally:
+        store.close()
+    assert symbols == {"AAA", "BBB", "CCC"}