feat(b2): wire 3 equity-factor assets into the Dagster graph

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

docs/superpowers/plans/2026-06-16-equity-factor-sleeve.md

@@ -0,0 +1,324 @@
+# Equity-Factor Sleeve (B2) — Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** A cross-sectional multi-factor (value+quality+momentum) US-equity book on a dynamic top-N-by-dollar-volume universe, built as three construction variants (long-only / long-short / long-tilt), each a `ForwardStrategy` paper-forward track the cockpit ingests.
+
+**Architecture:** Pure factor math in `domain/strategies/equity_factor.py`; Tiingo fundamentals + universe behind new ports/adapters (reusing `TiingoDailyClient` for prices); three live-booking `ForwardStrategy` services sharing one factor-target builder; three Dagster assets feeding the existing `cockpit_forward`.
+
+**Tech Stack:** Python 3.13, numpy, urllib (Tiingo), Dagster 1.12.x, pytest, strict mypy.
+
+**Spec:** `docs/superpowers/specs/2026-06-16-equity-factor-sleeve-design.md`
+
+**Conventions (same as B1):** tests `cd ~/Work/fxhnt && .venv/bin/python -m pytest <path> -v`; type-check `.venv/bin/python -m mypy src/fxhnt`. Orchestration modules MUST NOT use `from __future__ import annotations`; every other new module SHOULD. Use `dict[str, Any]` to match the `ForwardStrategy` Protocol. Commit per task with trailer `Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>`. No live network in tests (fake the clients / monkeypatch `_get`). Tiingo key via `os.environ["TIINGO_API_KEY"]`, header `Authorization: Token <key>`, NEVER printed.
+
+---
+
+## File Structure
+| File | Responsibility |
+|---|---|
+| `src/fxhnt/ports/market_data.py` (modify) | add `FundamentalsClient` + `UniverseSource` Protocols |
+| `src/fxhnt/domain/strategies/equity_factor.py` (create) | pure: winsorized z-score, family scores, composite, 3 construction weightings, book return |
+| `src/fxhnt/adapters/data/tiingo_fundamentals.py` (create) | `TiingoFundamentalsClient` — daily metrics + statements |
+| `src/fxhnt/adapters/data/tiingo_universe.py` (create) | `TiingoUniverseSource` — supported-tickers → filter → dollar-volume rank → top-N |
+| `src/fxhnt/application/equity_factor_strategy.py` (create) | `_factor_targets` builder + 3 `ForwardStrategy` services |
+| `src/fxhnt/registry.py` (modify) | 3 entries: eqfactor_long / eqfactor_ls / eqfactor_tilt |
+| `src/fxhnt/adapters/orchestration/assets.py` (modify) | 3 assets + cockpit_forward deps |
+| `src/fxhnt/adapters/orchestration/definitions.py` (modify) | register 3 assets in job + Definitions |
+| `tests/integration/test_equity_factor.py` (create) | domain math tests |
+| `tests/integration/test_equity_factor_strategy.py` (create) | service + round-trip tests |
+| `tests/integration/test_tiingo_fundamentals.py`, `test_tiingo_universe.py` (create) | adapter tests (mocked HTTP) |
+
+---
+
+## Task 1: Ports — FundamentalsClient + UniverseSource
+
+**Files:** Modify `src/fxhnt/ports/market_data.py`.
+
+- [ ] **Step 1: Append the ports** (no test — exercised by later fakes):
+
+```python
+class FundamentalsClient(Protocol):
+    def metrics(self, symbol: str) -> dict[str, float]:
+        """Latest daily valuation metrics: {'peRatio','pbRatio','marketCap', ...} (most recent available)."""
+        ...
+    def statement_metrics(self, symbol: str) -> dict[str, float]:
+        """Latest reported statement metrics: {'piotroskiFScore','roe','debtEquity','grossMargin', ...}."""
+        ...
+
+
+class UniverseSource(Protocol):
+    def top_n(self, n: int) -> list[str]:
+        """The n most-liquid US common-stock tickers by trailing dollar-volume (descending)."""
+        ...
+```
+
+- [ ] **Step 2:** `.venv/bin/python -m mypy src/fxhnt/ports/market_data.py` → clean.
+- [ ] **Step 3:** commit `feat(b2): FundamentalsClient + UniverseSource ports`.
+
+---
+
+## Task 2: Domain — `equity_factor.py` (the correctness-critical heart)
+
+**Files:** Create `src/fxhnt/domain/strategies/equity_factor.py`; Test `tests/integration/test_equity_factor.py`.
+
+- [ ] **Step 1: Write the failing tests** (`tests/integration/test_equity_factor.py`):
+
+```python
+"""Pure equity-factor math: winsorized z-score, family/composite scores, 3 construction weightings, book return."""
+from __future__ import annotations
+
+import math
+
+import pytest
+
+from fxhnt.domain.strategies import equity_factor as ef
+
+
+def test_robust_z_winsorizes_and_standardizes() -> None:
+    z = ef.robust_z([1.0, 2.0, 3.0, 4.0, 100.0], k=2.0)   # 100 is an outlier
+    assert max(z) == pytest.approx(2.0)                    # winsorized at +k
+    assert abs(sum(z) ) < 1e-9 or True                     # not asserting mean-0 (winsor shifts it); just bounded
+    assert all(-2.0 - 1e-9 <= v <= 2.0 + 1e-9 for v in z)
+
+
+def test_robust_z_missing_is_neutral_zero() -> None:
+    z = ef.robust_z([1.0, None, 3.0])
+    assert z[1] == 0.0                                     # missing -> neutral
+
+
+def test_composite_is_equal_weight_of_families() -> None:
+    c = ef.composite(value_z=[1.0, -1.0], quality_z=[1.0, -1.0], momentum_z=[1.0, -1.0])
+    assert c == pytest.approx([1.0, -1.0])                 # mean of three equal series
+
+
+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)
+    assert sum(w) == pytest.approx(1.0)
+    assert w[9] == pytest.approx(0.5) and w[8] == pytest.approx(0.5)
+    assert all(w[i] == 0.0 for i in range(8))
+
+
+def test_long_short_is_market_neutral_gross_two() -> None:
+    scores = [float(i) for i in range(10)]
+    w = ef.construction_weights(scores, "ls", quantile=0.2)
+    assert sum(w) == pytest.approx(0.0)                    # market-neutral
+    assert sum(abs(x) for x in w) == pytest.approx(2.0)    # gross 2.0
+    assert w[9] == pytest.approx(0.5) and w[0] == pytest.approx(-0.5)
+
+
+def test_tilt_is_long_only_rank_weighted_sum_one() -> None:
+    scores = [float(i) for i in range(10)]
+    w = ef.construction_weights(scores, "tilt")
+    assert sum(w) == pytest.approx(1.0)
+    assert all(x >= 0.0 for x in w)                        # long-only
+    assert w[9] > w[5] >= 0.0                              # monotone in score, below-center -> 0
+
+
+def test_book_return_weighted_minus_short_borrow() -> None:
+    prev_w = {"A": 0.5, "B": -0.5}
+    prev_p = {"A": 100.0, "B": 100.0}
+    cur_p = {"A": 110.0, "B": 90.0}                        # A +10%, B -10% (short B gains)
+    r = ef.book_return(prev_w, prev_p, cur_p, borrow_daily=0.0)
+    assert r == pytest.approx(0.5 * 0.10 + (-0.5) * (-0.10))   # +0.10
+    r2 = ef.book_return(prev_w, prev_p, cur_p, borrow_daily=0.001)
+    assert r2 == pytest.approx(r - 0.001 * 0.5)            # borrow charged on short notional (|−0.5|)
+```
+
+- [ ] **Step 2: Run → FAIL** (`.venv/bin/python -m pytest tests/integration/test_equity_factor.py -v`).
+
+- [ ] **Step 3: Implement `src/fxhnt/domain/strategies/equity_factor.py`:**
+
+```python
+"""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."""
+from __future__ import annotations
+
+import math
+from statistics import fmean, pstdev
+
+
+def robust_z(values: list[float | None], k: float = 3.0) -> list[float]:
+    """Cross-sectional z-score with winsorization at ±k sigma. Missing/non-finite -> neutral 0.0."""
+    present = [v for v in values if v is not None and math.isfinite(v)]
+    if len(present) < 2:
+        return [0.0 for _ in values]
+    mu = fmean(present)
+    sd = pstdev(present)
+    if sd == 0.0:
+        return [0.0 for _ in values]
+    out: list[float] = []
+    for v in values:
+        if v is None or not math.isfinite(v):
+            out.append(0.0)
+        else:
+            out.append(max(-k, min(k, (v - mu) / sd)))
+    return out
+
+
+def _mean_z(cols: list[list[float]]) -> list[float]:
+    n = len(cols[0])
+    return [fmean([col[i] for col in cols]) for i in range(n)]
+
+
+def value_score(pe: list[float | None], pb: list[float | None]) -> list[float]:
+    # cheap = high score: negate pe/pb; non-positive pe (negative earnings) -> neutral via None
+    pe_clean = [None if (p is None or p <= 0) else -p for p in pe]
+    pb_clean = [None if (p is None or p <= 0) else -p for p in pb]
+    return _mean_z([robust_z(pe_clean), robust_z(pb_clean)])
+
+
+def quality_score(piotroski: list[float | None], roe: list[float | None],
+                  debt_equity: list[float | None], gross_margin: list[float | None]) -> list[float]:
+    neg_de = [None if d is None else -d for d in debt_equity]
+    return _mean_z([robust_z(piotroski), robust_z(roe), robust_z(neg_de), robust_z(gross_margin)])
+
+
+def momentum_score(mom_12_1: list[float | None]) -> list[float]:
+    return robust_z(mom_12_1)
+
+
+def composite(value_z: list[float], quality_z: list[float], momentum_z: list[float]) -> list[float]:
+    return _mean_z([value_z, quality_z, momentum_z])
+
+
+def _quintile_cut(scores: list[float], quantile: float) -> tuple[float, float]:
+    s = sorted(scores)
+    n = len(s)
+    lo = s[max(0, int(quantile * n) - 1)]
+    hi = s[min(n - 1, n - int(quantile * n))]
+    return lo, hi
+
+
+def construction_weights(scores: list[float], construction: str, quantile: float = 0.2) -> list[float]:
+    """Map composite scores -> portfolio weights for one of: 'long', 'ls', 'tilt'."""
+    n = len(scores)
+    if n == 0:
+        return []
+    if construction == "tilt":
+        med = sorted(scores)[n // 2]
+        pos = [max(0.0, s - med) for s in scores]
+        tot = sum(pos)
+        return [p / tot for p in pos] if tot > 0 else [1.0 / n] * n
+    lo, hi = _quintile_cut(scores, quantile)
+    top = [i for i, s in enumerate(scores) if s >= hi]
+    bot = [i for i, s in enumerate(scores) if s <= lo]
+    w = [0.0] * n
+    if top:
+        for i in top:
+            w[i] += 1.0 / len(top)              # long leg sums to +1
+    if construction == "ls" and bot:
+        for i in bot:
+            w[i] += -1.0 / len(bot)             # short leg sums to -1 (net 0, gross 2)
+    return w
+
+
+def book_return(prev_weights: dict[str, float], prev_prices: dict[str, float],
+                cur_prices: dict[str, float], borrow_daily: float = 0.0) -> float:
+    """Realized one-period book return of the held weights, minus per-day borrow cost on short notional."""
+    r = 0.0
+    short_notional = 0.0
+    for sym, wt in prev_weights.items():
+        p0, p1 = prev_prices.get(sym), cur_prices.get(sym)
+        if p0 and p1 and p0 > 0:
+            r += wt * (p1 / p0 - 1.0)
+        if wt < 0:
+            short_notional += -wt
+    return r - borrow_daily * short_notional
+```
+
+- [ ] **Step 4: Run → PASS.** mypy clean.
+- [ ] **Step 5: Commit** `feat(b2): equity-factor domain (z-score, composite, 3 constructions, book return)`.
+
+---
+
+## Task 3: TiingoFundamentalsClient
+
+**Files:** Create `src/fxhnt/adapters/data/tiingo_fundamentals.py`; Test `tests/integration/test_tiingo_fundamentals.py`.
+
+**Endpoints (verified live):** `GET {base}/tiingo/fundamentals/{ticker}/daily?startDate=...` → list of `{date, peRatio, pbRatio, marketCap, enterpriseVal, trailingPEG1Y}` (use the most recent). `GET {base}/tiingo/fundamentals/{ticker}/statements?startDate=...` → list of statement objects; the relevant metrics (`piotroskiFScore, roe, debtEquity, grossMargin, profitMargin, revenueQoQ`) appear in the statement data — the implementer must inspect one live response (header-auth, NEVER print the key) to confirm the exact nesting and map it; tests use fixtures of that shape.
+
+- [ ] **Step 1: Write failing tests** — a `FakeFundamentals`-free direct test: monkeypatch the client's `_get` to return fixed daily + statement fixtures; assert `metrics("AAPL")` returns the latest `{peRatio, pbRatio, marketCap}` and `statement_metrics("AAPL")` returns `{piotroskiFScore, roe, debtEquity, grossMargin}` from the latest statement. (Write the fixtures to match the live shape you confirm in Step 3.)
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement** `TiingoFundamentalsClient(api_key=None, base_url="https://api.tiingo.com")`: `_get` retry helper with `Authorization: Token <key>` header (mirror `tiingo_daily.py`), `metrics()` hits `/daily` (latest row), `statement_metrics()` hits `/statements` (latest, mapping the confirmed field locations). Key from `os.environ["TIINGO_API_KEY"]`, never logged. First run one live call to confirm the statements shape, then code to it.
+- [ ] **Step 4: Run → PASS.** mypy clean.
+- [ ] **Step 5: Commit** `feat(b2): TiingoFundamentalsClient (daily metrics + statements)`.
+
+---
+
+## Task 4: TiingoUniverseSource
+
+**Files:** Create `src/fxhnt/adapters/data/tiingo_universe.py`; Test `tests/integration/test_tiingo_universe.py`.
+
+**Approach:** Tiingo publishes `https://apimedia.tiingo.com/docs/tiingo/daily/supported_tickers.zip` (CSV: ticker, exchange, assetType, priceCurrency, startDate, endDate). Filter `assetType == "Stock"`, US exchanges (`NYSE,NASDAQ,NYSE ARCA,AMEX`), `priceCurrency == "USD"`, active (endDate recent). From that candidate set, rank by trailing ~30d dollar-volume (close×volume from `TiingoDailyClient`, summed) and return the top-N. To bound cost, cap the candidate set the volume-rank pulls EOD for (e.g. take a large pre-set of liquid names, or sample) — document whatever cap is chosen with a `log`/comment (no silent truncation).
+
+- [ ] **Step 1: Write failing test** — monkeypatch the supported-tickers fetch + the per-ticker volume lookup with fixtures; assert filtering (non-Stock/non-US/inactive dropped) and that `top_n(3)` returns the 3 highest dollar-volume tickers in order.
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement** `TiingoUniverseSource(daily_client, api_key=None)`: fetch+cache the supported-tickers CSV (parse from the zip), filter, compute dollar-volume per candidate over the last ~30 days via `daily_client.adj_closes` (or raw close×volume — note: `adj_closes` returns total-return closes; for dollar-volume use a raw close×volume call or approximate with adjClose×volume — acceptable for ranking), return top-N. Make the candidate cap explicit + logged.
+- [ ] **Step 4: Run → PASS.** mypy clean.
+- [ ] **Step 5: Commit** `feat(b2): TiingoUniverseSource (supported-tickers + dollar-volume top-N)`.
+
+---
+
+## Task 5: Factor-target builder + 3 ForwardStrategy services
+
+**Files:** Create `src/fxhnt/application/equity_factor_strategy.py`; Test `tests/integration/test_equity_factor_strategy.py`.
+
+The three services share `_factor_targets(universe, fundamentals, prices) -> dict[str,float]`: pull the universe, fetch each name's metrics+statements+12-1 momentum (from prices), compute family scores → composite → `construction_weights(scores, construction)` → `{sym: weight}`. Each service is a live-booking `ForwardStrategy` (like `FundingCarryStrategy`): monthly rebalance, daily mark.
+
+- [ ] **Step 1: Write failing test** — fakes for `UniverseSource`, `FundamentalsClient`, `DailyBarClient` returning a small deterministic cross-section (≥6 names with known pe/pb/piotroski/roe/momentum). For each construction (`long`/`ls`/`tilt`): drive the service through two `ForwardTracker` steps (first run freezes; advance the fake prices + cross a month boundary; second run books one row). Assert: the booked return equals the hand-computed `book_return` of the prior targets; `extra` carries `positions` + `last_rebal`; round-trip via `ForwardStateReader` (days-list). Assert `ls` nets ~0 gross 2 in its targets; `long`/`tilt` long-only.
+
+```python
+class FakeUniverse:
+    def __init__(self, tickers): self._t = tickers
+    def top_n(self, n): return self._t[:n]
+
+class FakeFundamentals:
+    def __init__(self, m, s): self._m, self._s = m, s
+    def metrics(self, sym): return self._m[sym]
+    def statement_metrics(self, sym): return self._s[sym]
+# FakeDailyBars from test_paper_strategies (adj_closes) — reuse the same shape.
+```
+
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement** `equity_factor_strategy.py`:
+  - `_momentum_12_1(closes: dict[str,float]) -> float | None`: 12-1 month return from a date-keyed close series (return over [t-252, t-21], skip last ~21d), None if insufficient history.
+  - `_factor_targets(universe, fund, bars, construction, n) -> dict[str,float]`: build the cross-section arrays (aligned ticker order), call domain `value_score/quality_score/momentum_score/composite/construction_weights`, return `{sym: w}` for nonzero weights.
+  - `class _EquityFactorStrategy` with `__init__(self, construction, universe, fund, bars, n=300, borrow_annual=0.01, clock=_today_iso)`; `advance(last_date, extra)`: read `positions`/`last_rebal` from extra; fetch current prices for held ∪ (rebalance ? new targets); compute today's `book_return(prev_positions, prev_prices, cur_prices, borrow_daily=borrow_annual/252)`; if first run or month changed → recompute targets via `_factor_targets`; return `([(self._clock(), r)], {"positions":..., "prices":..., "last_rebal":...})`.
+  - Three thin public classes: `EquityFactorLong`, `EquityFactorLS`, `EquityFactorTilt` wiring `construction="long"/"ls"/"tilt"` (and borrow only for ls).
+- [ ] **Step 4: Run → PASS** (+ full suite). mypy clean.
+- [ ] **Step 5: Commit** `feat(b2): equity-factor target builder + 3 ForwardStrategy services`.
+
+---
+
+## Task 6: Registry + orchestration
+
+**Files:** Modify `src/fxhnt/registry.py`, `adapters/orchestration/assets.py`, `definitions.py`; Test extend `tests/integration/test_orchestration_definitions.py`.
+
+- [ ] **Step 1: Write failing test** — assert the Definitions asset keys include `eqfactor_long_nav, eqfactor_ls_nav, eqfactor_tilt_nav` and `cockpit_forward` deps include them.
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement**
+  - `registry.py`: 3 entries with `state_file` `eqfactor_long_state` / `eqfactor_ls_state` / `eqfactor_tilt_state`, display names, sleeve `"equity-factor"`, venue `"us-equity"`, `gate_spec={"min_days":60,"min_total_return":0.0,"min_sharpe":0.3}`.
+  - `assets.py`: 3 `@asset`s (`eqfactor_long_nav` etc.) following the B1 paper-asset shape via `_run_paper_tracker`, building each service from `TiingoUniverseSource(TiingoDailyClient())`, `TiingoFundamentalsClient()`, `TiingoDailyClient()`. Add all 3 to `cockpit_forward` `deps`.
+  - `definitions.py`: add the 3 assets to `Definitions(assets=...)` + the job `selection`.
+- [ ] **Step 4: Run → PASS** (+ full suite + the definitions count is now 13). mypy clean.
+- [ ] **Step 5: Commit** `feat(b2): wire 3 equity-factor assets into the Dagster graph`.
+
+---
+
+## Task 7: Full gate + deploy + verify
+
+- [ ] **Step 1:** Full suite + mypy: `cd ~/Work/fxhnt && .venv/bin/python -m pytest -q && .venv/bin/python -m mypy src/fxhnt` (all green; no new mypy errors). Commit any fixes.
+- [ ] **Step 2:** Merge `feat/equity-factor-sleeve` → master + push.
+- [ ] **Step 3:** Build + deploy the cockpit image (Argo `fxhnt-cockpit` at master HEAD); confirm build+deploy Succeeded (no new image deps — torch already in; Tiingo key already wired).
+- [ ] **Step 4:** Launch one `combined_book_forward_job` materialization (run-launcher, not exec); poll to SUCCESS.
+- [ ] **Step 5:** Verify the cockpit: `SELECT strategy_id, days, gate_status FROM forward_summary` shows the 3 eqfactor strategies present (days=0/WAIT on first run — inception freeze, correct).
+- [ ] **Step 6:** Update memory (`project_lifecycle_b0_state` or a B2 note): 3 equity-factor variants live; recommend the 30yr-backtest layer next.
+
+---
+
+## Self-Review
+**Spec coverage:** universe top-N (Task 4) ✓; value+quality+momentum composite (Task 2) ✓; 3 constructions (Task 2 weights + Task 5 services) ✓; monthly rebalance + live-booking + borrow cost (Tasks 2,5) ✓; ports/adapters reusing TiingoDailyClient (Tasks 1,3,4) ✓; registry + 3 assets → cockpit_forward (Task 6) ✓; deterministic domain + fake-port tests + round-trip (Tasks 2,5) ✓; deploy/verify (Task 7) ✓. Backtest layer correctly OUT of scope.
+**Placeholder scan:** the two adapter tasks (3,4) instruct a one-time live shape-confirm for the statements endpoint + supported-tickers parse rather than guessing the exact nesting — concrete (endpoints given) not vague; everything correctness-critical (domain Task 2) is full literal code with exact expected test values.
+**Type consistency:** `construction_weights(scores, construction, quantile)`, `book_return(prev_weights, prev_prices, cur_prices, borrow_daily)`, `robust_z(values, k)`, `composite(value_z, quality_z, momentum_z)` used identically in domain (Task 2) and services (Task 5). State-file basenames in Task 6 match the registry entries. Services emit days-list (no state_reader change — confirmed against B1).

docs/superpowers/specs/2026-06-16-equity-factor-sleeve-design.md

@@ -0,0 +1,81 @@
+# Equity-Factor Sleeve — Design (B2)
+
+**Status:** design (approved in substance 2026-06-16)
+**Predecessor:** B1 (6 paper-track strategies + Tiingo equity data). Reuses the `ForwardStrategy`/`ForwardTracker` port, the cockpit `STRATEGY_REGISTRY` + `cockpit_forward` ingest, and `TiingoDailyClient`.
+
+## Goal
+A systematic cross-sectional **multi-factor equity book** on a broad liquid US universe, run as **three construction variants** — long-only, long-short market-neutral, and long-tilt — each a separate `ForwardStrategy` paper-forward track through the existing gate. The point is platform-native: build all three, let the forward-gate + DSR decide which construction holds up out-of-sample. Diversifies the (crypto-heavy) book into a new asset class using data we already pay for (Tiingo fundamentals + 30yr EOD).
+
+## What this is NOT (scope boundary)
+- **NOT a backtest** — this is the forward-track sleeve. A 30yr backtest/DSR pre-screen layer over the same factor logic is a **separate increment** (strongly recommended to pair, so candidates are screened before months of blind forward). Out of scope here.
+- **NOT live execution** — paper-forward only (no broker orders).
+- **NOT analyst/estimate data** — Benzinga (earnings estimates/ratings) is not entitled; we use Tiingo *actuals* + price.
+
+## Data (verified live on our Tiingo Power key)
+- **EOD** (`/tiingo/daily/{t}/prices`) — adjusted OHLCV, 30yr; via existing `TiingoDailyClient`.
+- **Fundamentals daily** (`/tiingo/fundamentals/{t}/daily`) — `peRatio`, `pbRatio`, `marketCap`, `enterpriseVal`, `trailingPEG1Y`.
+- **Fundamentals statements** (`/tiingo/fundamentals/{t}/statements`) — `piotroskiFScore`, `roe`, `roa`, `debtEquity`, `grossMargin`, `profitMargin`, `revenueQoQ`, … (85 fields via `/definitions`).
+- **Supported tickers** (Tiingo `supported_tickers.zip`) — the candidate list for the universe screener (assetType=Stock, US exchanges, active).
+
+## Architecture (hexagonal, fits the platform)
+```
+domain/strategies/equity_factor.py   pure: z-scores, composite, quantile/rank weights, book return (no I/O)
+ports/market_data.py (extend)        FundamentalsClient + UniverseSource Protocols
+adapters/data/tiingo_fundamentals.py TiingoFundamentalsClient (daily metrics + statements)
+adapters/data/tiingo_universe.py     TiingoUniverseSource (supported-tickers → filter → dollar-volume rank → top-N)
+application/equity_factor_strategy.py 3 ForwardStrategy services (long / ls / tilt) sharing the universe+factor compute
+registry.py (extend)                 3 entries: eqfactor_long, eqfactor_ls, eqfactor_tilt
+adapters/orchestration/assets.py     3 assets → cockpit_forward deps
+```
+
+### Universe screener (`TiingoUniverseSource`)
+`top_n(n=300) -> list[str]`: load Tiingo supported-tickers (cache the zip), filter to US common stock + active, rank by trailing ~30d **dollar-volume** (close×volume from EOD over the candidate set), return the top-N. Refreshed at each monthly rebalance. (Default N=300; configurable 100–500.) To bound API cost, the volume rank uses a coarse pre-filter (e.g. the exchange-listed common-stock subset) before pulling EOD.
+
+### Factor computation (`domain/strategies/equity_factor.py`, pure)
+Per rebalance, for the universe cross-section:
+- **Value** = z(−peRatio) + z(−pbRatio) (cheap = high score; guard non-positive/None).
+- **Quality** = z(piotroskiFScore) + z(roe) + z(−debtEquity) + z(grossMargin).
+- **Momentum** = z(12-1 month price return) (skip most recent month).
+- Each raw factor is **winsorized** (e.g. ±3σ) then cross-sectional **z-scored**; missing inputs → neutral 0 for that factor (a name needs ≥2 of 3 families present to be scored).
+- **Composite** = equal-weight mean of the three family z-scores.
+Pure functions: `zscore(values)`, `winsorize(values, k)`, `composite(value_z, quality_z, momentum_z)`, `quantile_weights(scores, construction)`.
+
+### Three construction variants (the `construction` param)
+- **`eqfactor_long`** — top-quintile, equal-weight long, weights sum to 1.0.
+- **`eqfactor_ls`** — long top-quintile / short bottom-quintile, equal-weight each leg → market-neutral (Σw=0, gross=2.0). Realized return nets a **borrow cost** on the short leg (default 1%/yr, applied per-day).
+- **`eqfactor_tilt`** — long-only budget, rank-weighted overweight/underweight (weights ∝ max(0, rank-centered score), normalized to sum 1.0). No shorts.
+
+### Live-booking ForwardStrategy services
+Three services (one per construction) share a common `_factor_targets(as_of)` that calls the universe + fundamentals + price clients and returns target weights via the domain. Each `advance(last_date, extra)` (the live-booking pattern, like funding/poc):
+- reads prior `positions` (+ `last_rebal`) from `extra`,
+- **marks** the held book to today's prices → daily book return (minus borrow cost on shorts for `ls`),
+- **rebalances monthly** (`today.month != last_rebal.month` or first run): recompute targets, update positions,
+- returns `([(today_iso, book_return)], updated_extra)`; the `ForwardTracker` books forward-only (inception freeze on first run), emits the days-list state the cockpit ingests.
+
+### Registry + orchestration
+Three `STRATEGY_REGISTRY` entries (state_file basenames `eqfactor_long_state` / `eqfactor_ls_state` / `eqfactor_tilt_state`; gate_spec e.g. `min_days=60, min_total_return=0, min_sharpe=0.3`). Three Dagster `@asset`s (no `from __future__ import annotations`) mirroring the B1 paper assets, added to `cockpit_forward` deps + Definitions/job. They fetch live HTTP (no DuckDB) → run in parallel.
+
+## Testing (deterministic; this is the "test correctly" answer)
+- **Domain (no network):** fixed factor-input cross-section → assert exact z-scores (winsorized), composite, quantile membership, and each construction's weights (long Σ=1; ls Σ=0 & gross=2; tilt Σ=1, monotone in score); book-return = Σ(weight×asset-return); borrow-cost applied to ls shorts.
+- **Universe screener:** fake supported-tickers + volumes → assert filter + top-N by dollar-volume.
+- **Services:** fake `FundamentalsClient`/`DailyBarClient`/`UniverseSource` (deterministic fixtures, no network) → drive each construction through `ForwardTracker` → round-trip via `ForwardStateReader` (days-list summary); assert monthly-rebalance triggers + `extra` positions; ls books a borrow cost.
+- **No lookahead:** forward-only (we only use data known as-of now) — inherent; documented.
+- strict mypy, full suite green, two-stage review per task (spec then code-quality), faithful to B1's discipline.
+
+## Discipline / caveats
+- Factor premia are partly **crowded/decayed** — no edge is claimed; the **forward-gate + DSR** decide. Realistic small weighting expected.
+- **Long-short realism:** borrow cost modeled (1%/yr default); assumes shortability of liquid large-caps (reasonable for top-N).
+- **Point-in-time:** forward-only, so a reporting-lag lookahead cannot occur live; the future backtest layer MUST handle fundamentals reporting lag (flagged for that increment).
+- Data cost: Tiingo Power limits (100k req/day, 108k symbols/mo) comfortably cover a 300-name monthly rebalance.
+
+## Defaults (adjust if desired)
+universe **top-300** by dollar-volume · **monthly** rebalance · short **borrow 1%/yr** · winsor ±3σ · quintile cut-offs.
+
+## Build order (subagent-driven, next via writing-plans)
+1. Ports: `FundamentalsClient` + `UniverseSource`.
+2. `TiingoFundamentalsClient` (daily + statements) + fake.
+3. `TiingoUniverseSource` (supported-tickers + dollar-volume rank) + fake.
+4. Domain `equity_factor.py` (z-score/composite/winsor/quantile + 3 constructions + book return) — heavy unit tests.
+5. Three `ForwardStrategy` services (long/ls/tilt) sharing factor compute.
+6. Registry (3 entries) + 3 Dagster assets + cockpit_forward deps + Definitions.
+7. Full gate + deploy + verify all 3 in cockpit.

src/fxhnt/adapters/data/tiingo_fundamentals.py

@@ -0,0 +1,117 @@
+"""urllib FundamentalsClient — Tiingo fundamentals for the equity-factor sleeve. Two reads:
+
+  metrics(symbol):            latest DAILY valuation metrics (marketCap, enterpriseVal, peRatio,
+                              pbRatio, trailingPEG1Y, ...). Endpoint serves a flat JSON list of
+                              dated rows; we read the LATEST (max date) row.
+                                GET {base}/tiingo/fundamentals/{ticker}/daily?startDate=YYYY-MM-DD
+                                → [{date, marketCap, enterpriseVal, peRatio, pbRatio, trailingPEG1Y}]
+
+  statement_metrics(symbol):  latest reported STATEMENT metrics (piotroskiFScore, roe, roa,
+                              debtEquity, grossMargin, profitMargin, revenueQoQ, bookVal, epsDil, ...).
+                              Endpoint serves a JSON list of period entries; we take the LATEST
+                              (max date) entry and flatten its statementData sections (overview +
+                              incomeStatement + balanceSheet + cashFlow) into {dataCode: float(value)}.
+                                GET {base}/tiingo/fundamentals/{ticker}/statements?startDate=YYYY-MM-DD
+                                → [{date, year, quarter, statementData: {overview: [{dataCode, value}],
+                                    incomeStatement: [...], balanceSheet: [...], cashFlow: [...]}}]
+
+Both default a startDate window of ~800 days (well inside entitlement; gives several recent quarters
+of statements and a long daily-metrics history) and read the most-recent available period.
+
+Auth: header `Authorization: Token <key>` with TIINGO_API_KEY (env). The key is used ONLY in the
+request header — never logged or interpolated into a URL.
+
+NOTE: shapes above are coded to the documented/verified Tiingo response shapes; this build env had
+no TIINGO_API_KEY, so the first live run will confirm. None-valued fields are skipped.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import os
+import time
+import urllib.parse
+import urllib.request
+from typing import Any
+
+_DAILY = "{base}/tiingo/fundamentals/{sym}/daily?startDate={frm}&format=json"
+_STATEMENTS = "{base}/tiingo/fundamentals/{sym}/statements?startDate={frm}&format=json"
+
+# Valuation fields read from a fundamentals/daily row (all but `date`).
+_DAILY_FIELDS = ("marketCap", "enterpriseVal", "peRatio", "pbRatio", "trailingPEG1Y")
+# statementData sections flattened by statement_metrics.
+_STATEMENT_SECTIONS = ("overview", "incomeStatement", "balanceSheet", "cashFlow")
+
+
+class TiingoFundamentalsClient:
+    def __init__(
+        self,
+        api_key: str | None = None,
+        lookback_days: int = 800,
+        base_url: str = "https://api.tiingo.com",
+    ) -> None:
+        self._key = api_key if api_key is not None else os.environ["TIINGO_API_KEY"]
+        self._lookback_days = lookback_days
+        self._base = base_url.rstrip("/")
+
+    def _get(self, url: str, tries: int = 4) -> Any:
+        for a in range(tries):
+            try:
+                # The key lives ONLY in this header — never in the URL or any log line.
+                req = urllib.request.Request(url, headers={"Authorization": f"Token {self._key}"})
+                return json.loads(urllib.request.urlopen(req, timeout=30).read())
+            except Exception:  # noqa: BLE001 — transient HTTP; retry then re-raise
+                if a == tries - 1:
+                    raise
+                time.sleep(2 * (a + 1))
+        raise RuntimeError("unreachable")
+
+    def _start(self) -> str:
+        return (dt.date.today() - dt.timedelta(days=self._lookback_days)).isoformat()
+
+    @staticmethod
+    def _to_float(value: Any) -> float | None:
+        if value is None:
+            return None
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _latest(rows: list[dict[str, Any]]) -> dict[str, Any]:
+        """The row/entry with the max `date` (lexical compare on ISO/`YYYY-MM-DD...` strings)."""
+        return max(rows, key=lambda r: str(r.get("date", "")))
+
+    def metrics(self, symbol: str) -> dict[str, float]:
+        sym = urllib.parse.quote(symbol)
+        rows: list[dict[str, Any]] = self._get(_DAILY.format(base=self._base, sym=sym, frm=self._start())) or []
+        if not rows:
+            return {}
+        latest = self._latest(rows)
+        out: dict[str, float] = {}
+        for field in _DAILY_FIELDS:
+            v = self._to_float(latest.get(field))
+            if v is not None:
+                out[field] = v
+        return out
+
+    def statement_metrics(self, symbol: str) -> dict[str, float]:
+        sym = urllib.parse.quote(symbol)
+        entries: list[dict[str, Any]] = (
+            self._get(_STATEMENTS.format(base=self._base, sym=sym, frm=self._start())) or []
+        )
+        if not entries:
+            return {}
+        latest = self._latest(entries)
+        data: dict[str, Any] = latest.get("statementData") or {}
+        out: dict[str, float] = {}
+        for section in _STATEMENT_SECTIONS:
+            for item in data.get(section) or []:
+                code = item.get("dataCode")
+                if code is None:
+                    continue
+                v = self._to_float(item.get("value"))
+                if v is not None:
+                    out[str(code)] = v
+        return out

src/fxhnt/adapters/data/tiingo_universe.py

@@ -0,0 +1,194 @@
+"""urllib UniverseSource — the N most-liquid US common stocks by trailing dollar-volume.
+
+Two cheap reads (one snapshot per run):
+
+  candidates:  Tiingo publishes a supported-tickers CSV (zipped) listing every symbol it carries.
+                 GET https://apimedia.tiingo.com/docs/tiingo/daily/supported_tickers.zip
+                 → one CSV, columns: ticker,exchange,assetType,priceCurrency,startDate,endDate
+               We download+parse it once (cached on the instance) and filter to US common stock:
+                 assetType == "Stock", priceCurrency == "USD",
+                 exchange ∈ {NYSE, NASDAQ, NYSE ARCA, AMEX, BATS},
+                 and ACTIVE — endDate within ~7d of the file's max endDate (drops delisted names).
+               This yields thousands of candidates. It is NOT auth'd (public docs asset).
+
+  volumes:     Tiingo's IEX batch endpoint prices many tickers in one call.
+                 GET {base}/iex/?tickers=t1,t2,...   (comma-separated, chunked ~90/call)
+                 → [{ticker, last/tngoLast, volume, ...}]
+               dollar-volume ≈ last × volume (today's IEX volume — a consistent cross-sectional
+               liquidity proxy, fine for RANKING). One snapshot, summed across chunks.
+
+The FULL filtered candidate set is fed to the (auth'd, paid) IEX volume rank — no alphabetical
+pre-slice — so "top-N by dollar-volume" ranks every liquid name, not just early-alphabet ones.
+`max_candidates` (default 30000) is a runaway safety guard ONLY: it comfortably covers the entire
+filtered US-common-stock set (~22k names), so it never triggers in practice. If the filtered set
+ever exceeds it we LOG a WARNING naming how many were dropped — never a silent truncation. We then
+return the top-`n` by dollar-volume, descending.
+
+Auth: header `Authorization: Token <key>` with TIINGO_API_KEY (env). The key is used ONLY in the
+IEX request header — never logged or interpolated into a URL. The supported-tickers zip is public
+and is fetched WITHOUT the key.
+
+NOTE: the supported-tickers CSV and IEX shapes above are coded to the documented Tiingo shapes;
+this build env had no TIINGO_API_KEY, so the first live run will confirm. Rows missing
+last/volume (or yielding zero dollar-volume) are skipped.
+"""
+from __future__ import annotations
+
+import csv
+import io
+import json
+import logging
+import os
+import time
+import urllib.parse
+import urllib.request
+import zipfile
+from typing import Any
+
+log = logging.getLogger(__name__)
+
+_SUPPORTED_TICKERS_URL = "https://apimedia.tiingo.com/docs/tiingo/daily/supported_tickers.zip"
+_IEX = "{base}/iex/?tickers={tickers}"
+
+# US common-stock exchanges we keep (everything else — OTC, foreign boards — is dropped).
+# Verified live: Stock rows only ever match NYSE / NASDAQ / AMEX; NYSE ARCA / BATS never appear
+# among Stock rows (harmless to keep). OTC strings (PINK, OTCGREY, OTCMKTS, …) + foreign (ASX)
+# are excluded by not being in this set.
+_US_EXCHANGES = {"NYSE", "NASDAQ", "NYSE ARCA", "AMEX", "BATS"}
+# A name is "active" if its endDate is within this many days of the file's max endDate.
+_ACTIVE_TOL_DAYS = 7
+# IEX batch size — comma-separated tickers per call.
+_CHUNK = 90
+
+
+class TiingoUniverseSource:
+    def __init__(
+        self,
+        api_key: str | None = None,
+        base_url: str = "https://api.tiingo.com",
+        max_candidates: int = 30000,
+    ) -> None:
+        self._key = api_key if api_key is not None else os.environ["TIINGO_API_KEY"]
+        self._base = base_url.rstrip("/")
+        self._max_candidates = max_candidates
+
+    # --- HTTP layer (monkeypatched in tests) -------------------------------------------------
+
+    def _fetch_candidates(self) -> list[dict[str, str]]:
+        """Download+unzip the public supported-tickers CSV → list of row dicts. NO api key."""
+        for a in range(4):
+            try:
+                raw = urllib.request.urlopen(_SUPPORTED_TICKERS_URL, timeout=60).read()
+                break
+            except Exception:  # noqa: BLE001 — transient HTTP; retry then re-raise
+                if a == 3:
+                    raise
+                time.sleep(2 * (a + 1))
+        with zipfile.ZipFile(io.BytesIO(raw)) as zf:
+            name = zf.namelist()[0]
+            text = zf.read(name).decode("utf-8", errors="replace")
+        return list(csv.DictReader(io.StringIO(text)))
+
+    def _fetch_volumes(self, tickers: list[str]) -> list[dict[str, Any]]:
+        """IEX batch snapshot for `tickers` (chunked). Returns concatenated row dicts."""
+        out: list[dict[str, Any]] = []
+        for i in range(0, len(tickers), _CHUNK):
+            chunk = tickers[i : i + _CHUNK]
+            url = _IEX.format(base=self._base, tickers=urllib.parse.quote(",".join(chunk)))
+            out.extend(self._get(url) or [])
+        return out
+
+    def _get(self, url: str, tries: int = 4) -> Any:
+        for a in range(tries):
+            try:
+                # The key lives ONLY in this header — never in the URL or any log line.
+                req = urllib.request.Request(url, headers={"Authorization": f"Token {self._key}"})
+                return json.loads(urllib.request.urlopen(req, timeout=30).read())
+            except Exception:  # noqa: BLE001 — transient HTTP; retry then re-raise
+                if a == tries - 1:
+                    raise
+                time.sleep(2 * (a + 1))
+        raise RuntimeError("unreachable")
+
+    # --- filtering + ranking -----------------------------------------------------------------
+
+    def _candidates(self) -> list[str]:
+        """Active US common-stock tickers from the supported-tickers file, in file order."""
+        rows = self._fetch_candidates()
+        end_dates = [str(r.get("endDate") or "") for r in rows if r.get("endDate")]
+        max_end = max(end_dates) if end_dates else ""
+        # "active" cutoff: max endDate minus tolerance (lexical ISO compare; cheap, no parse).
+        cutoff = ""
+        if max_end:
+            import datetime as dt
+
+            try:
+                cutoff = (
+                    dt.date.fromisoformat(max_end[:10]) - dt.timedelta(days=_ACTIVE_TOL_DAYS)
+                ).isoformat()
+            except ValueError:
+                cutoff = ""
+
+        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:
+                continue
+            end = str(r.get("endDate") or "")
+            if cutoff and end[:10] < cutoff:
+                continue  # delisted
+            out.append(ticker)
+        return out
+
+    @staticmethod
+    def _dollar_volume(row: dict[str, Any]) -> float:
+        last = row.get("last")
+        if last is None:
+            last = row.get("tngoLast")
+        vol = row.get("volume")
+        try:
+            return float(last) * float(vol)  # type: ignore[arg-type]
+        except (TypeError, ValueError):
+            return 0.0
+
+    def top_n(self, n: int) -> list[str]:
+        candidates = self._candidates()
+        n_filtered = len(candidates)
+        # Runaway guard only: the filtered US-common-stock set (~22k) sits well under the default
+        # 30000, so this never trips in practice. If it ever does, WARN (never silent) so the
+        # dropped names are visible — but DO NOT pre-slice by file order before this point.
+        if n_filtered > self._max_candidates:
+            dropped = n_filtered - self._max_candidates
+            log.warning(
+                "TiingoUniverseSource: %d filtered candidates exceed runaway guard "
+                "max_candidates=%d; dropping %d by file order (raise max_candidates to widen).",
+                n_filtered,
+                self._max_candidates,
+                dropped,
+            )
+            candidates = candidates[: self._max_candidates]
+
+        # Rank the FULL filtered set by dollar-volume — no alphabetical pre-slice bias.
+        rows = self._fetch_volumes(candidates)
+        ranked = [
+            (str(r.get("ticker") or ""), self._dollar_volume(r))
+            for r in rows
+            if r.get("ticker")
+        ]
+        ranked = [(t, dv) for t, dv in ranked if dv > 0.0]
+        ranked.sort(key=lambda tv: tv[1], reverse=True)
+        result = [t for t, _ in ranked[:n]]
+        log.info(
+            "TiingoUniverseSource: %d filtered candidates, %d priced with dollar-volume>0, "
+            "returning top %d.",
+            n_filtered,
+            len(ranked),
+            len(result),
+        )
+        return result

src/fxhnt/adapters/orchestration/assets.py

@@ -209,7 +209,56 @@ def poc_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
     return _run_paper_tracker(context, "poc_nav", lambda: MomentumVrpStrategy(), "poc_state")
 
 
-@asset(deps=[combined_forward_nav, sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav])
+@asset
+def eqfactor_long_nav(context: AssetExecutionContext) -> dict:  # type: ignore[type-arg]
+    """Paper-forward long-only 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
+    from fxhnt.application.equity_factor_strategy import EquityFactorLong
+
+    return _run_paper_tracker(
+        context,
+        "eqfactor_long_nav",
+        lambda: EquityFactorLong(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient()),
+        "eqfactor_long_state",
+    )
+
+
+@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
+    from fxhnt.application.equity_factor_strategy import EquityFactorLS
+
+    return _run_paper_tracker(
+        context,
+        "eqfactor_ls_nav",
+        lambda: EquityFactorLS(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient()),
+        "eqfactor_ls_state",
+    )
+
+
+@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
+    from fxhnt.application.equity_factor_strategy import EquityFactorTilt
+
+    return _run_paper_tracker(
+        context,
+        "eqfactor_tilt_nav",
+        lambda: EquityFactorTilt(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient()),
+        "eqfactor_tilt_state",
+    )
+
+
+@asset(deps=[combined_forward_nav, sixtyforty_nav, multistrat_nav, gd_nav, funding_nav, crossvenue_nav, poc_nav,
+             eqfactor_long_nav, eqfactor_ls_nav, eqfactor_tilt_nav])
 def cockpit_forward(context: AssetExecutionContext) -> None:
     """Normalize tracker state files and upsert rows + summary into the operational cockpit DB."""
     from fxhnt.application.forward_ingest import ingest_forward_state

src/fxhnt/adapters/orchestration/definitions.py

@@ -9,6 +9,9 @@ from fxhnt.adapters.orchestration.assets import (
     combined_forward_nav,
     crossvenue_nav,
     crypto_bars,
+    eqfactor_long_nav,
+    eqfactor_ls_nav,
+    eqfactor_tilt_nav,
     funding_nav,
     futures_bars,
     gd_nav,
@@ -22,6 +25,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,
         cockpit_forward,
     ],
 )
@@ -38,6 +42,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,
         cockpit_forward,
     ],
     jobs=[combined_book_job],

src/fxhnt/application/equity_factor_strategy.py

@@ -0,0 +1,160 @@
+"""Equity-factor target builder + the three live-booking ForwardStrategy services (long / ls / tilt).
+
+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`."""
+from __future__ import annotations
+
+from collections.abc import Callable
+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
+
+# Trading-day offsets for the 12-1 momentum window: skip the most recent ~21d, look back ~252d (1y).
+_LOOKBACK = 252
+_SKIP = 21
+
+
+def _momentum_12_1(closes: dict[str, float]) -> float | None:
+    """12-1 month price momentum from a date-keyed adjusted-close series.
+
+    Sorts the dates, requires >= ~252 points, and returns the price ratio over [~252d ago, ~21d ago]
+    (i.e. the trailing-year return that EXCLUDES the most recent ~month, the standard 12-1 specification).
+    Returns None when the history is too short."""
+    dates = sorted(closes)
+    if len(dates) < _LOOKBACK:
+        return None
+    p_then = closes[dates[-_LOOKBACK]]
+    p_recent = closes[dates[-_SKIP]]
+    if p_then is None or p_then <= 0:
+        return 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.
+
+    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."""
+    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)))
+
+    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}
+
+
+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`."""
+
+    def __init__(self, construction: str, universe: UniverseSource, fund: FundamentalsClient,
+                 bars: DailyBarClient, n: int = 300, 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._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
+
+        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
+
+        if rebalancing:
+            new_positions = _factor_targets(self._universe, self._fund, self._bars, self._construction, self._n)
+            last_rebal = today
+        else:
+            new_positions = 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}
+
+
+class EquityFactorLong:
+    """Long-only top-quintile equity-factor sleeve."""
+
+    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 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 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 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."""
+
+    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 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/domain/strategies/equity_factor.py

@@ -0,0 +1,101 @@
+"""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."""
+from __future__ import annotations
+
+import math
+from statistics import fmean, median
+
+
+def robust_z(values: list[float | None], k: float = 3.0) -> list[float]:
+    """Cross-sectional robust z-score with winsorization at ±k. Missing/non-finite -> neutral 0.0.
+
+    Uses a median/MAD location-scale (Gaussian-consistent MAD scaled by 1.4826) rather than
+    mean/stdev so a single extreme outlier cannot inflate the scale enough to hide itself below
+    the ±k clamp (the classic non-robust degeneracy). This makes "winsorized at ±k" hold for the
+    outlier instead of leaving its raw z just under k.
+    """
+    present = [v for v in values if v is not None and math.isfinite(v)]
+    if len(present) < 2:
+        return [0.0 for _ in values]
+    med = median(present)
+    mad = median(sorted(abs(v - med) for v in present))
+    scale = 1.4826 * mad
+    if scale == 0.0:
+        return [0.0 for _ in values]
+    out: list[float] = []
+    for v in values:
+        if v is None or not math.isfinite(v):
+            out.append(0.0)
+        else:
+            out.append(max(-k, min(k, (v - med) / scale)))
+    return out
+
+
+def _mean_z(cols: list[list[float]]) -> list[float]:
+    n = len(cols[0])
+    return [fmean([col[i] for col in cols]) for i in range(n)]
+
+
+def value_score(pe: list[float | None], pb: list[float | None]) -> list[float]:
+    pe_clean = [None if (p is None or p <= 0) else -p for p in pe]
+    pb_clean = [None if (p is None or p <= 0) else -p for p in pb]
+    return _mean_z([robust_z(pe_clean), robust_z(pb_clean)])
+
+
+def quality_score(piotroski: list[float | None], roe: list[float | None],
+                  debt_equity: list[float | None], gross_margin: list[float | None]) -> list[float]:
+    neg_de = [None if d is None else -d for d in debt_equity]
+    return _mean_z([robust_z(piotroski), robust_z(roe), robust_z(neg_de), robust_z(gross_margin)])
+
+
+def momentum_score(mom_12_1: list[float | None]) -> list[float]:
+    return robust_z(mom_12_1)
+
+
+def composite(value_z: list[float], quality_z: list[float], momentum_z: list[float]) -> list[float]:
+    return _mean_z([value_z, quality_z, momentum_z])
+
+
+def _quintile_cut(scores: list[float], quantile: float) -> tuple[float, float]:
+    s = sorted(scores)
+    n = len(s)
+    lo = s[max(0, int(quantile * n) - 1)]
+    hi = s[min(n - 1, n - int(quantile * n))]
+    return lo, hi
+
+
+def construction_weights(scores: list[float], construction: str, quantile: float = 0.2) -> list[float]:
+    """Map composite scores -> portfolio weights for one of: 'long', 'ls', 'tilt'."""
+    n = len(scores)
+    if n == 0:
+        return []
+    if construction == "tilt":
+        med = median(scores)
+        pos = [max(0.0, s - med) for s in scores]
+        tot = sum(pos)
+        return [p / tot for p in pos] if tot > 0 else [1.0 / n] * n
+    lo, hi = _quintile_cut(scores, quantile)
+    top = [i for i, s in enumerate(scores) if s >= hi]
+    bot = [i for i, s in enumerate(scores) if s <= lo]
+    w = [0.0] * n
+    if top:
+        for i in top:
+            w[i] += 1.0 / len(top)
+    if construction == "ls" and bot:
+        for i in bot:
+            w[i] += -1.0 / len(bot)
+    return w
+
+
+def book_return(prev_weights: dict[str, float], prev_prices: dict[str, float],
+                cur_prices: dict[str, float], borrow_daily: float = 0.0) -> float:
+    """Realized one-period book return of the held weights, minus per-day borrow cost on short notional."""
+    r = 0.0
+    short_notional = 0.0
+    for sym, wt in prev_weights.items():
+        p0, p1 = prev_prices.get(sym), cur_prices.get(sym)
+        if p0 and p1 and p0 > 0:
+            r += wt * (p1 / p0 - 1.0)
+        if wt < 0:
+            short_notional += -wt
+    return r - borrow_daily * short_notional

src/fxhnt/ports/market_data.py

@@ -39,3 +39,18 @@ class VolIndexClient(Protocol):
     def dvol(self, currency: str) -> dict[int, float]:
         """{ epoch_day: implied-vol-index } for BTC/ETH (Deribit DVOL)."""
         ...
+
+
+class FundamentalsClient(Protocol):
+    def metrics(self, symbol: str) -> dict[str, float]:
+        """Latest daily valuation metrics: {'peRatio','pbRatio','marketCap', ...} (most recent available)."""
+        ...
+    def statement_metrics(self, symbol: str) -> dict[str, float]:
+        """Latest reported statement metrics: {'piotroskiFScore','roe','debtEquity','grossMargin', ...}."""
+        ...
+
+
+class UniverseSource(Protocol):
+    def top_n(self, n: int) -> list[str]:
+        """The n most-liquid US common-stock tickers by trailing dollar-volume (descending)."""
+        ...

src/fxhnt/registry.py

@@ -40,4 +40,19 @@ STRATEGY_REGISTRY: dict[str, dict] = {
         "venue": "glbx", "state_file": "poc_state",
         "gate_spec": {"min_days": 20, "min_total_return": 0.0, "min_sharpe": 0.3},
     },
+    "eqfactor_long": {
+        "display_name": "Equity factor — long-only", "sleeve": "equity-factor",
+        "venue": "us-equity", "state_file": "eqfactor_long_state",
+        "gate_spec": {"min_days": 60, "min_total_return": 0.0, "min_sharpe": 0.3},
+    },
+    "eqfactor_ls": {
+        "display_name": "Equity factor — long-short", "sleeve": "equity-factor",
+        "venue": "us-equity", "state_file": "eqfactor_ls_state",
+        "gate_spec": {"min_days": 60, "min_total_return": 0.0, "min_sharpe": 0.3},
+    },
+    "eqfactor_tilt": {
+        "display_name": "Equity factor — long-tilt", "sleeve": "equity-factor",
+        "venue": "us-equity", "state_file": "eqfactor_tilt_state",
+        "gate_spec": {"min_days": 60, "min_total_return": 0.0, "min_sharpe": 0.3},
+    },
 }

tests/integration/test_equity_factor.py

@@ -0,0 +1,56 @@
+"""Pure equity-factor math: winsorized z-score, family/composite scores, 3 construction weightings, book return."""
+from __future__ import annotations
+
+import pytest
+
+from fxhnt.domain.strategies import equity_factor as ef
+
+
+def test_robust_z_winsorizes_and_standardizes() -> None:
+    z = ef.robust_z([1.0, 2.0, 3.0, 4.0, 100.0], k=2.0)   # 100 is an outlier
+    assert max(z) == pytest.approx(2.0)                    # winsorized at +k
+    assert all(-2.0 - 1e-9 <= v <= 2.0 + 1e-9 for v in z)
+
+
+def test_robust_z_missing_is_neutral_zero() -> None:
+    z = ef.robust_z([1.0, None, 3.0])
+    assert z[1] == 0.0                                     # missing -> neutral
+
+
+def test_composite_is_equal_weight_of_families() -> None:
+    c = ef.composite(value_z=[1.0, -1.0], quality_z=[1.0, -1.0], momentum_z=[1.0, -1.0])
+    assert c == 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)
+    assert sum(w) == pytest.approx(1.0)
+    assert w[9] == pytest.approx(0.5) and w[8] == pytest.approx(0.5)
+    assert all(w[i] == 0.0 for i in range(8))
+
+
+def test_long_short_is_market_neutral_gross_two() -> None:
+    scores = [float(i) for i in range(10)]
+    w = ef.construction_weights(scores, "ls", quantile=0.2)
+    assert sum(w) == pytest.approx(0.0)
+    assert sum(abs(x) for x in w) == pytest.approx(2.0)
+    assert w[9] == pytest.approx(0.5) and w[0] == pytest.approx(-0.5)
+
+
+def test_tilt_is_long_only_rank_weighted_sum_one() -> None:
+    scores = [float(i) for i in range(10)]
+    w = ef.construction_weights(scores, "tilt")
+    assert sum(w) == pytest.approx(1.0)
+    assert all(x >= 0.0 for x in w)
+    assert w[9] > w[5] >= 0.0
+
+
+def test_book_return_weighted_minus_short_borrow() -> None:
+    prev_w = {"A": 0.5, "B": -0.5}
+    prev_p = {"A": 100.0, "B": 100.0}
+    cur_p = {"A": 110.0, "B": 90.0}
+    r = ef.book_return(prev_w, prev_p, cur_p, borrow_daily=0.0)
+    assert r == pytest.approx(0.5 * 0.10 + (-0.5) * (-0.10))   # +0.10
+    r2 = ef.book_return(prev_w, prev_p, cur_p, borrow_daily=0.001)
+    assert r2 == pytest.approx(r - 0.001 * 0.5)

tests/integration/test_equity_factor_strategy.py

@@ -0,0 +1,259 @@
+"""Equity-factor target builder + the three live-booking ForwardStrategy services (long / ls / tilt).
+
+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."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from fxhnt.adapters.persistence.state_reader import ForwardStateReader
+from fxhnt.application.equity_factor_strategy import (
+    EquityFactorLong,
+    EquityFactorLS,
+    EquityFactorTilt,
+    _factor_targets,
+    _momentum_12_1,
+)
+from fxhnt.application.forward_tracker import ForwardTracker
+from fxhnt.domain.strategies import equity_factor as ef
+
+
+# --------------------------------------------------------------------------- fakes
+class FakeUniverse:
+    def __init__(self, tickers: list[str]) -> None:
+        self._tickers = tickers
+
+    def top_n(self, n: int) -> list[str]:
+        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:
+        self._data = closes_by_sym
+
+    def adj_closes(self, symbol: str) -> dict[str, float]:
+        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.
+_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}
+
+_N_DAYS = 300  # >= 252 so 12-1 momentum is computable
+
+
+def _iso(i: int) -> str:
+    import datetime
+    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`."""
+    g = (1.0 + total_drift) ** (1.0 / _N_DAYS)
+    return {_iso(i): 100.0 * (g ** i) 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_universe() -> FakeUniverse:
+    return FakeUniverse(_SYMS)
+
+
+# --------------------------------------------------------------------------- _momentum_12_1
+def test_momentum_12_1_uses_252_21_window() -> None:
+    closes = _price_path(0.40)
+    dates = sorted(closes)
+    expected = closes[dates[-21]] / closes[dates[-252]] - 1.0
+    assert _momentum_12_1(closes) == pytest.approx(expected)
+
+
+def test_momentum_12_1_none_when_too_short() -> None:
+    closes = {_iso(i): 100.0 + i for i in range(100)}  # < 252 points
+    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())
+
+
+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_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())
+
+
+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)
+
+
+# --------------------------------------------------------------------------- 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()
+    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()
+    assert st0.forward_days == 0
+    loaded0 = json.loads(Path(p).read_text())
+    prev_targets = dict(loaded0["extra"]["positions"])
+    prev_prices = dict(loaded0["extra"]["prices"])
+    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
+    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)]
+    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()
+    assert st1.forward_days == 1
+    loaded1 = json.loads(Path(p).read_text())
+    summary, rows = ForwardStateReader().read(p, sid)
+
+    # Hand-compute the realized book return of the PRIOR targets at the new prices.
+    borrow_daily = (0.01 / 252.0) if ls else 0.0
+    expected_ret = ef.book_return(prev_targets, prev_prices,
+                                  {s: cur_prices[s] for s in prev_targets}, borrow_daily)
+    booked = loaded1["days"][-1]["ret"]
+    assert booked == pytest.approx(expected_ret)
+    return st1, loaded0, loaded1, summary, rows, prev_targets
+
+
+def test_equity_factor_long_service_round_trips(tmp_path) -> None:
+    _, loaded0, loaded1, summary, rows, prev_targets = _drive_two_steps(
+        EquityFactorLong, "eqfactor_long", tmp_path, ls=False)
+    assert summary.days == 1 and len(rows) == 1
+    assert "positions" in loaded1["extra"] and "last_rebal" in loaded1["extra"]
+    assert loaded1["extra"]["last_rebal"] == "2026-02-15"     # rebalanced on the month boundary
+    assert all(w >= 0.0 for w in prev_targets.values())       # long-only
+
+
+def test_equity_factor_ls_service_round_trips(tmp_path) -> None:
+    _, loaded0, loaded1, summary, rows, prev_targets = _drive_two_steps(
+        EquityFactorLS, "eqfactor_ls", tmp_path, ls=True)
+    assert summary.days == 1 and len(rows) == 1
+    assert "positions" in loaded1["extra"] and "last_rebal" in loaded1["extra"]
+    # ls targets net ~0 / gross ~2
+    assert sum(prev_targets.values()) == pytest.approx(0.0)
+    assert sum(abs(w) for w in prev_targets.values()) == pytest.approx(2.0)
+
+
+def test_equity_factor_tilt_service_round_trips(tmp_path) -> None:
+    _, loaded0, loaded1, summary, rows, prev_targets = _drive_two_steps(
+        EquityFactorTilt, "eqfactor_tilt", tmp_path, ls=False)
+    assert summary.days == 1 and len(rows) == 1
+    assert "positions" in loaded1["extra"] and "last_rebal" in loaded1["extra"]
+    assert all(w >= 0.0 for w in prev_targets.values())       # tilt is long-only
+    assert sum(prev_targets.values()) == pytest.approx(1.0)
+
+
+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()
+    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
+    day[0] = "2026-01-20"
+    st1 = ForwardTracker(EquityFactorLong(universe, fund, bars, 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)
+    assert loaded1["extra"]["last_rebal"] == "2026-01-10"        # rebalance date unchanged

tests/integration/test_orchestration_definitions.py

@@ -14,6 +14,9 @@ def test_definitions_load_with_assets_and_schedule() -> None:
     # B1: the six paper-track assets are wired into the graph alongside the B0 four
     paper = {"sixtyforty_nav", "multistrat_nav", "gd_nav", "funding_nav", "crossvenue_nav", "poc_nav"}
     assert paper <= asset_keys, f"missing paper-track assets: {paper - asset_keys}"
+    # 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}"
 
     # --- schedule present with the right cron ---
     # defs.schedules is a list[ScheduleDefinition] (or None when empty)
@@ -31,5 +34,5 @@ def test_definitions_load_with_assets_and_schedule() -> None:
     cockpit_def = repo.assets_defs_by_key[AssetKey("cockpit_forward")]
     deps = cockpit_def.asset_deps[AssetKey("cockpit_forward")]
     upstream = {k.to_user_string() for k in deps}
-    expected_upstream = {"combined_forward_nav"} | paper
+    expected_upstream = {"combined_forward_nav"} | paper | eqfactor
     assert expected_upstream <= upstream, f"cockpit_forward missing upstream: {expected_upstream - upstream}"

tests/integration/test_tiingo_fundamentals.py

@@ -0,0 +1,200 @@
+"""TiingoFundamentalsClient — latest daily valuation metrics + latest reported statement metrics.
+NO live network: the `_get` HTTP layer is monkeypatched to return fixed fixtures.
+
+Invariants proven here:
+  (1) metrics() hits the fundamentals/daily endpoint and returns the LATEST (max date) row's
+      valuation fields (peRatio/pbRatio/marketCap, ...) as floats, skipping None.
+  (2) statement_metrics() hits the fundamentals/statements endpoint, picks the LATEST (max date)
+      entry, and flattens its statementData sections into a {dataCode: float(value)} dict.
+  (3) the request window's startDate is computed from `lookback_days` (today - lookback_days).
+"""
+from __future__ import annotations
+
+import datetime as dt
+from typing import Any
+
+from fxhnt.adapters.data.tiingo_fundamentals import TiingoFundamentalsClient
+
+
+def _daily(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Tiingo fundamentals/daily response: a flat list of valuation-metric rows."""
+    return [
+        {
+            "date": f"{r['date']}T00:00:00.000Z",
+            "marketCap": r.get("marketCap"),
+            "enterpriseVal": r.get("enterpriseVal"),
+            "peRatio": r.get("peRatio"),
+            "pbRatio": r.get("pbRatio"),
+            "trailingPEG1Y": r.get("trailingPEG1Y"),
+        }
+        for r in rows
+    ]
+
+
+def _statements(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Tiingo fundamentals/statements response: list of period entries with nested statementData."""
+    return entries
+
+
+def test_metrics_returns_latest_daily_row() -> None:
+    client = TiingoFundamentalsClient(api_key="x")
+    # Out-of-order on purpose: the latest (max date) row must be chosen, not the first/last.
+    rows = [
+        {"date": "2026-01-02", "marketCap": 100.0, "peRatio": 20.0, "pbRatio": 3.0,
+         "enterpriseVal": 110.0, "trailingPEG1Y": 1.5},
+        {"date": "2026-01-06", "marketCap": 120.0, "peRatio": 22.0, "pbRatio": 3.3,
+         "enterpriseVal": 130.0, "trailingPEG1Y": 1.7},
+        {"date": "2026-01-03", "marketCap": 110.0, "peRatio": 21.0, "pbRatio": 3.1,
+         "enterpriseVal": 120.0, "trailingPEG1Y": 1.6},
+    ]
+    captured: dict[str, str] = {}
+
+    def fake_get(url: str, tries: int = 4) -> Any:
+        captured["url"] = url
+        assert "/tiingo/fundamentals/AAPL/daily" in url
+        return _daily(rows)
+
+    client._get = fake_get
+    out = client.metrics("AAPL")
+
+    # Latest row (2026-01-06) chosen.
+    assert out["marketCap"] == 120.0
+    assert out["peRatio"] == 22.0
+    assert out["pbRatio"] == 3.3
+    assert out["enterpriseVal"] == 130.0
+    assert out["trailingPEG1Y"] == 1.7
+    assert "date" not in out
+    assert all(isinstance(v, float) for v in out.values())
+    assert "/statements" not in captured["url"]
+
+
+def test_metrics_skips_none_values() -> None:
+    client = TiingoFundamentalsClient(api_key="x")
+    rows = [
+        {"date": "2026-01-06", "marketCap": 120.0, "peRatio": None, "pbRatio": 3.3,
+         "enterpriseVal": None, "trailingPEG1Y": 1.7},
+    ]
+    client._get = lambda url, tries=4: _daily(rows)  # noqa: E731
+    out = client.metrics("AAPL")
+    assert "peRatio" not in out
+    assert "enterpriseVal" not in out
+    assert out["marketCap"] == 120.0
+    assert out["pbRatio"] == 3.3
+
+
+def test_statement_metrics_flattens_latest_overview() -> None:
+    client = TiingoFundamentalsClient(api_key="x")
+    entries = [
+        {
+            "date": "2025-09-30T00:00:00.000Z", "year": 2025, "quarter": 3,
+            "statementData": {
+                "overview": [
+                    {"dataCode": "roe", "value": 0.10},
+                    {"dataCode": "piotroskiFScore", "value": 5},
+                ],
+            },
+        },
+        {
+            "date": "2025-12-31T00:00:00.000Z", "year": 2025, "quarter": 4,
+            "statementData": {
+                "overview": [
+                    {"dataCode": "roe", "value": 0.42},
+                    {"dataCode": "piotroskiFScore", "value": 8},
+                    {"dataCode": "debtEquity", "value": 1.25},
+                    {"dataCode": "grossMargin", "value": 0.46},
+                ],
+                "incomeStatement": [
+                    {"dataCode": "revenue", "value": 1000.0},
+                ],
+                "balanceSheet": [
+                    {"dataCode": "totalAssets", "value": 5000.0},
+                ],
+                "cashFlow": [
+                    {"dataCode": "freeCashFlow", "value": 250.0},
+                ],
+            },
+        },
+    ]
+    captured: dict[str, str] = {}
+
+    def fake_get(url: str, tries: int = 4) -> Any:
+        captured["url"] = url
+        assert "/tiingo/fundamentals/AAPL/statements" in url
+        return _statements(entries)
+
+    client._get = fake_get
+    out = client.statement_metrics("AAPL")
+
+    # Latest entry (2025-12-31) chosen — NOT the 2025-09-30 one.
+    assert out["roe"] == 0.42
+    assert out["piotroskiFScore"] == 8.0
+    assert out["debtEquity"] == 1.25
+    assert out["grossMargin"] == 0.46
+    # Other sections flattened too.
+    assert out["revenue"] == 1000.0
+    assert out["totalAssets"] == 5000.0
+    assert out["freeCashFlow"] == 250.0
+    assert all(isinstance(v, float) for v in out.values())
+    assert "/daily" not in captured["url"].split("/statements")[0] + "/statements"
+
+
+def test_statement_metrics_skips_none_values() -> None:
+    client = TiingoFundamentalsClient(api_key="x")
+    entries = [
+        {
+            "date": "2025-12-31T00:00:00.000Z",
+            "statementData": {
+                "overview": [
+                    {"dataCode": "roe", "value": 0.42},
+                    {"dataCode": "debtEquity", "value": None},
+                ],
+            },
+        },
+    ]
+    client._get = lambda url, tries=4: entries  # noqa: E731
+    out = client.statement_metrics("AAPL")
+    assert out["roe"] == 0.42
+    assert "debtEquity" not in out
+
+
+def test_metrics_startdate_window_from_lookback_days() -> None:
+    client = TiingoFundamentalsClient(api_key="x", lookback_days=800)
+    captured: dict[str, str] = {}
+
+    def fake_get(url: str, tries: int = 4) -> Any:
+        captured["url"] = url
+        return _daily([{"date": "2026-01-06", "marketCap": 1.0, "peRatio": 1.0, "pbRatio": 1.0}])
+
+    client._get = fake_get
+    client.metrics("AAPL")
+    expected_start = (dt.date.today() - dt.timedelta(days=800)).isoformat()
+    assert f"startDate={expected_start}" in captured["url"]
+
+
+def test_statement_metrics_startdate_window_from_lookback_days() -> None:
+    client = TiingoFundamentalsClient(api_key="x", lookback_days=800)
+    captured: dict[str, str] = {}
+
+    def fake_get(url: str, tries: int = 4) -> Any:
+        captured["url"] = url
+        return _statements([
+            {"date": "2025-12-31T00:00:00.000Z",
+             "statementData": {"overview": [{"dataCode": "roe", "value": 0.1}]}},
+        ])
+
+    client._get = fake_get
+    client.statement_metrics("AAPL")
+    expected_start = (dt.date.today() - dt.timedelta(days=800)).isoformat()
+    assert f"startDate={expected_start}" in captured["url"]
+
+
+def test_empty_metrics_response_returns_empty_dict() -> None:
+    client = TiingoFundamentalsClient(api_key="x")
+    client._get = lambda url, tries=4: []  # noqa: E731
+    assert client.metrics("AAPL") == {}
+
+
+def test_empty_statements_response_returns_empty_dict() -> None:
+    client = TiingoFundamentalsClient(api_key="x")
+    client._get = lambda url, tries=4: []  # noqa: E731
+    assert client.statement_metrics("AAPL") == {}

tests/integration/test_tiingo_universe.py

@@ -0,0 +1,200 @@
+"""TiingoUniverseSource — top-N most-liquid US common stocks by trailing dollar-volume.
+NO live network: the two fetch helpers (`_fetch_candidates` supported-tickers, `_fetch_volumes`
+IEX snapshot) are monkeypatched to return fixed fixtures.
+
+Invariants proven here:
+  (1) the supported-tickers candidate filter drops non-Stock assetType, non-US/foreign-currency,
+      and delisted (stale endDate) rows; valid active US stocks are kept.
+  (2) the FULL filtered candidate set is ranked by dollar-volume (last × volume), descending —
+      no alphabetical/file-order pre-slice (a late-in-file liquid name beats an early illiquid one).
+  (3) top_n(n) returns at most n tickers.
+  (4) max_candidates is a runaway safety guard only — it never trips for a normal set.
+"""
+from __future__ import annotations
+
+import datetime as dt
+from typing import Any
+
+from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource
+
+_TODAY = dt.date.today()
+_RECENT = _TODAY.isoformat()
+_STALE = (_TODAY - dt.timedelta(days=400)).isoformat()
+
+
+def _candidate_rows() -> list[dict[str, str]]:
+    """supported_tickers.csv rows (active US stocks + a non-stock, a foreign-ccy, a delisted)."""
+    return [
+        # valid active US common stocks (max endDate = _RECENT)
+        {"ticker": "A", "exchange": "NYSE", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT},
+        {"ticker": "B", "exchange": "NASDAQ", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT},
+        {"ticker": "C", "exchange": "NYSE ARCA", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT},
+        # DROP: not a Stock (ETF)
+        {"ticker": "SPY", "exchange": "NYSE ARCA", "assetType": "ETF",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT},
+        # DROP: foreign price currency
+        {"ticker": "FOREIGN", "exchange": "NYSE", "assetType": "Stock",
+         "priceCurrency": "EUR", "startDate": "2000-01-01", "endDate": _RECENT},
+        # DROP: delisted (stale endDate, far before the file max)
+        {"ticker": "DEAD", "exchange": "NASDAQ", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _STALE},
+        # DROP: unsupported exchange (OTC)
+        {"ticker": "OTCJUNK", "exchange": "OTC", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT},
+    ]
+
+
+def test_candidate_filter_drops_non_stock_foreign_and_delisted() -> None:
+    src = TiingoUniverseSource(api_key="x")
+    src._fetch_candidates = lambda: _candidate_rows()  # type: ignore[method-assign]  # noqa: E731
+
+    kept = src._candidates()
+
+    assert set(kept) == {"A", "B", "C"}
+    assert "SPY" not in kept       # non-Stock dropped
+    assert "FOREIGN" not in kept   # non-USD dropped
+    assert "DEAD" not in kept      # delisted dropped
+    assert "OTCJUNK" not in kept   # unsupported exchange dropped
+
+
+def test_top_n_ranks_by_dollar_volume_descending() -> None:
+    src = TiingoUniverseSource(api_key="x")
+    src._fetch_candidates = lambda: _candidate_rows()  # type: ignore[method-assign]  # noqa: E731
+
+    # C has highest last×volume, A medium, B lowest.
+    volumes = [
+        {"ticker": "A", "last": 50.0, "volume": 1_000_000},    # 50M
+        {"ticker": "B", "tngoLast": 10.0, "volume": 1_000_000},  # 10M (uses tngoLast)
+        {"ticker": "C", "last": 100.0, "volume": 1_000_000},   # 100M
+    ]
+
+    captured: dict[str, list[str]] = {}
+
+    def fake_volumes(tickers: list[str]) -> list[dict[str, Any]]:
+        captured["tickers"] = tickers
+        return volumes
+
+    src._fetch_volumes = fake_volumes  # type: ignore[method-assign]
+
+    assert src.top_n(2) == ["C", "A"]
+    # only the filtered candidates were priced
+    assert set(captured["tickers"]) == {"A", "B", "C"}
+
+
+def test_top_n_returns_at_most_n() -> None:
+    src = TiingoUniverseSource(api_key="x")
+    src._fetch_candidates = lambda: _candidate_rows()  # type: ignore[method-assign]  # noqa: E731
+    src._fetch_volumes = lambda tickers: [  # type: ignore[method-assign]  # noqa: E731
+        {"ticker": "A", "last": 50.0, "volume": 1_000_000},
+        {"ticker": "B", "last": 10.0, "volume": 1_000_000},
+        {"ticker": "C", "last": 100.0, "volume": 1_000_000},
+    ]
+
+    assert len(src.top_n(2)) == 2
+    assert len(src.top_n(1)) == 1
+    # n larger than the universe → at most the available count
+    assert len(src.top_n(99)) == 3
+
+
+def test_full_filtered_set_is_volume_ranked_not_pre_sliced() -> None:
+    # With the default (high) runaway guard, ALL filtered candidates are priced — no file-order
+    # pre-slice. 5 candidates, guard untouched → all 5 reach the IEX volume fetch.
+    src = TiingoUniverseSource(api_key="x")
+    rows = [
+        {"ticker": t, "exchange": "NYSE", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT}
+        for t in ("A", "B", "C", "D", "E")
+    ]
+    src._fetch_candidates = lambda: rows  # type: ignore[method-assign]  # noqa: E731
+
+    captured: dict[str, list[str]] = {}
+
+    def fake_volumes(tickers: list[str]) -> list[dict[str, Any]]:
+        captured["tickers"] = tickers
+        return [{"ticker": t, "last": 1.0, "volume": 1} for t in tickers]
+
+    src._fetch_volumes = fake_volumes  # type: ignore[method-assign]
+    src.top_n(2)
+    # all 5 priced — the guard did not pre-slice
+    assert set(captured["tickers"]) == {"A", "B", "C", "D", "E"}
+
+
+def test_runaway_guard_only_trips_above_threshold() -> None:
+    # The guard is a safety-only cap: it must NOT trip for a normal-sized filtered set, and it
+    # DOES trip (file-order drop) only when the filtered set exceeds max_candidates.
+    rows = [
+        {"ticker": t, "exchange": "NYSE", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT}
+        for t in ("A", "B", "C", "D", "E")
+    ]
+
+    # guard above the set size → all priced
+    src_ok = TiingoUniverseSource(api_key="x", max_candidates=10)
+    src_ok._fetch_candidates = lambda: rows  # type: ignore[method-assign]  # noqa: E731
+    captured_ok: dict[str, list[str]] = {}
+
+    def fake_ok(tickers: list[str]) -> list[dict[str, Any]]:
+        captured_ok["tickers"] = tickers
+        return [{"ticker": t, "last": 1.0, "volume": 1} for t in tickers]
+
+    src_ok._fetch_volumes = fake_ok  # type: ignore[method-assign]
+    src_ok.top_n(2)
+    assert len(captured_ok["tickers"]) == 5  # guard did not trip
+
+    # guard below the set size → trips, drops by file order down to the cap
+    src_cap = TiingoUniverseSource(api_key="x", max_candidates=2)
+    src_cap._fetch_candidates = lambda: rows  # type: ignore[method-assign]  # noqa: E731
+    captured_cap: dict[str, list[str]] = {}
+
+    def fake_cap(tickers: list[str]) -> list[dict[str, Any]]:
+        captured_cap["tickers"] = tickers
+        return [{"ticker": t, "last": 1.0, "volume": 1} for t in tickers]
+
+    src_cap._fetch_volumes = fake_cap  # type: ignore[method-assign]
+    src_cap.top_n(2)
+    assert len(captured_cap["tickers"]) == 2  # guard tripped
+
+
+def test_late_alphabet_high_volume_beats_early_low_volume() -> None:
+    # ANTI-BIAS GUARANTEE: a late-in-file high-volume ticker (ZM) must out-rank an early-in-file
+    # low-volume ticker (AAA). The old file-order pre-slice would have ranked only the early
+    # names and missed ZM; the fix ranks the full filtered set by dollar-volume.
+    order = ["AAA", "AAB", "TSLA", "WMT", "ZM"]  # alphabetical file order
+    rows = [
+        {"ticker": t, "exchange": "NASDAQ", "assetType": "Stock",
+         "priceCurrency": "USD", "startDate": "2000-01-01", "endDate": _RECENT}
+        for t in order
+    ]
+    # Default (high) runaway guard → full set priced. Under the OLD buggy file-order pre-slice
+    # the liquid late names (TSLA/WMT/ZM) could be dropped before ranking; the fix prices them.
+    src = TiingoUniverseSource(api_key="x")
+    src._fetch_candidates = lambda: rows  # type: ignore[method-assign]  # noqa: E731
+
+    # Late-in-order names carry the liquidity; early names are illiquid.
+    volumes = {
+        "AAA": (1.0, 100),          # tiny dollar-volume
+        "AAB": (1.0, 100),
+        "TSLA": (250.0, 5_000_000),  # huge
+        "WMT": (160.0, 3_000_000),
+        "ZM": (70.0, 4_000_000),
+    }
+
+    captured: dict[str, list[str]] = {}
+
+    def fake_volumes(tickers: list[str]) -> list[dict[str, Any]]:
+        captured["tickers"] = tickers
+        return [{"ticker": t, "last": volumes[t][0], "volume": volumes[t][1]} for t in tickers]
+
+    src._fetch_volumes = fake_volumes  # type: ignore[method-assign]
+
+    top3 = src.top_n(3)
+    # the full filtered set was priced (no file-order pre-slice)
+    assert set(captured["tickers"]) == set(order)
+    # late-alphabet liquid names selected; early illiquid names excluded.
+    # dollar-volume: TSLA 1.25B > WMT 480M > ZM 280M.
+    assert top3 == ["TSLA", "WMT", "ZM"]
+    assert "AAA" not in top3
+    assert "AAB" not in top3