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