feat(b2): equity-factor target builder + 3 ForwardStrategy services

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

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)

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