262 lines
13 KiB
Python
262 lines
13 KiB
Python
"""Shared batched-concurrent factor-score builder + the three live-booking ForwardStrategy services.
|
|
|
|
Deterministic fakes only (no network): a fixed >=6-name cross-section with enough adj-close history for
|
|
both the 12-1 momentum (~252d) and the trailing realized vol (~90d). The sleeve is PRICE-ONLY
|
|
(momentum + low-vol); fundamentals are NOT fetched. `compute_factor_scores` is asserted against the pure
|
|
domain pipeline directly (incl. dropping a raising ticker). Each construction is then driven through TWO
|
|
ForwardTracker steps from a SHARED fixed `scores` dict (no clients): first freezes inception + seeds
|
|
positions; second books one row after a month-boundary rebalance, round-tripped through ForwardStateReader."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fxhnt.adapters.persistence.state_reader import ForwardStateReader
|
|
from fxhnt.application.equity_factor_strategy import (
|
|
EquityFactorLong,
|
|
EquityFactorLS,
|
|
EquityFactorTilt,
|
|
_momentum_12_1,
|
|
_realized_vol,
|
|
compute_factor_scores,
|
|
)
|
|
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 FakeDailyBars:
|
|
def __init__(self, closes_by_sym: dict[str, dict[str, float]],
|
|
raise_for: str | None = None, exc: Exception | None = None) -> None:
|
|
self._data = closes_by_sym
|
|
# When set, adj_closes() raises for this one ticker — models a name the Tiingo daily endpoint
|
|
# 400/404s on (no coverage / odd symbol) so the name is dropped from the cross-section.
|
|
self._raise_for = raise_for
|
|
self._exc = exc
|
|
|
|
def adj_closes(self, symbol: str) -> dict[str, float]:
|
|
if self._raise_for is not None and symbol == self._raise_for:
|
|
raise self._exc if self._exc is not None else RuntimeError(f"no coverage for {symbol}")
|
|
return dict(self._data.get(symbol, {}))
|
|
|
|
|
|
# --------------------------------------------------------------------------- cross-section fixture
|
|
# Six names A..F, monotone best->worst on the price-only families (momentum + low-vol) so the composite
|
|
# ranking is unambiguous: A best (highest 12-1 momentum, lowest realized vol), F worst. With n=6,
|
|
# quantile=0.2 the top/bottom quintile each select the single extreme name.
|
|
_SYMS = ["A", "B", "C", "D", "E", "F"]
|
|
# 12-1 momentum strength per name (the total cumulative drift baked into the price path).
|
|
_MOM = {"A": 0.40, "B": 0.30, "C": 0.20, "D": 0.10, "E": 0.00, "F": -0.10}
|
|
# trailing realized-vol amplitude per name (zig-zag size on top of the drift): A calmest, F most volatile.
|
|
_VOL = {"A": 0.00, "B": 0.01, "C": 0.02, "D": 0.03, "E": 0.04, "F": 0.05}
|
|
|
|
_N_DAYS = 300 # >= 252 so 12-1 momentum is computable; > 90 so the realized vol window is full
|
|
|
|
|
|
def _iso(i: int) -> str:
|
|
import datetime
|
|
return (datetime.date(2024, 1, 1) + datetime.timedelta(days=i)).isoformat()
|
|
|
|
|
|
def _price_path(total_drift: float, vol_amp: float = 0.0) -> dict[str, float]:
|
|
"""Monotone-drift geometric path over _N_DAYS dates whose 12-1 window ratio encodes `total_drift`,
|
|
with a deterministic alternating ±`vol_amp` wiggle so the trailing realized vol is rank-ordered by
|
|
`vol_amp`. The 12-1 momentum reads closes[d_-21]/closes[d_-252]-1; the ±vol_amp wiggle nets out across
|
|
the 21d/252d offsets (it lands on the same parity) so momentum stays deterministic and rank-preserving
|
|
in `total_drift` regardless of the vol amplitude."""
|
|
g = (1.0 + total_drift) ** (1.0 / _N_DAYS)
|
|
return {_iso(i): 100.0 * (g ** i) * (1.0 + (vol_amp if i % 2 == 0 else -vol_amp)) for i in range(_N_DAYS)}
|
|
|
|
|
|
def _make_bars(**kwargs) -> FakeDailyBars:
|
|
return FakeDailyBars({s: _price_path(_MOM[s], _VOL[s]) for s in _SYMS}, **kwargs)
|
|
|
|
|
|
def _make_universe() -> FakeUniverse:
|
|
return FakeUniverse(_SYMS)
|
|
|
|
|
|
def _domain_scores(syms: list[str]) -> list[float]:
|
|
"""The pure-domain PRICE-ONLY composite for the given names, fed the same aligned inputs the builder
|
|
would (momentum + low-vol) — asserted against directly (not circularly via the builder)."""
|
|
moms: list[float | None] = [_momentum_12_1(_price_path(_MOM[s], _VOL[s])) for s in syms]
|
|
vols: list[float | None] = [_realized_vol(_price_path(_MOM[s], _VOL[s])) for s in syms]
|
|
return ef.composite_price(ef.momentum_score(moms), ef.lowvol_score(vols))
|
|
|
|
|
|
def _fixed_scores(syms: list[str]) -> dict:
|
|
"""A precomputed `scores` dict (the shared product of compute_factor_scores) for the given names, with
|
|
the latest close of each name's price path as its marking price — the form the services consume."""
|
|
return {
|
|
"date": "2026-01-15",
|
|
"syms": list(syms),
|
|
"scores": _domain_scores(syms),
|
|
"prices": {s: _price_path(_MOM[s], _VOL[s])[_iso(_N_DAYS - 1)] for s in syms},
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- _momentum_12_1
|
|
def test_momentum_12_1_uses_252_21_window() -> None:
|
|
closes = _price_path(0.40)
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------- _realized_vol
|
|
def test_realized_vol_orders_by_amplitude() -> None:
|
|
"""A calmer price path (smaller wiggle) has strictly lower trailing realized vol."""
|
|
calm = _realized_vol(_price_path(0.10, 0.01))
|
|
choppy = _realized_vol(_price_path(0.10, 0.05))
|
|
assert calm is not None and choppy is not None
|
|
assert calm < choppy
|
|
|
|
|
|
def test_realized_vol_none_when_too_short() -> None:
|
|
closes = {_iso(i): 100.0 + i for i in range(30)} # < 90 returns available
|
|
assert _realized_vol(closes, window=90) is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- compute_factor_scores
|
|
def test_compute_factor_scores_matches_domain_and_emits_prices() -> None:
|
|
"""The shared builder must fetch each ticker's adj_closes ONCE and emit aligned syms/scores equal to
|
|
the pure-domain PRICE-ONLY composite (asserted directly, not circularly) plus a {sym: latest_close}."""
|
|
s = compute_factor_scores(_make_universe(), _make_bars(), n=6)
|
|
assert s["syms"] == _SYMS
|
|
assert s["scores"] == pytest.approx(_domain_scores(_SYMS))
|
|
# latest adjusted close of each name's price path is its marking price
|
|
for sym in _SYMS:
|
|
assert s["prices"][sym] == pytest.approx(_price_path(_MOM[sym], _VOL[sym])[_iso(_N_DAYS - 1)])
|
|
assert "date" in s
|
|
|
|
|
|
def test_compute_factor_scores_drops_raising_ticker() -> None:
|
|
"""A ticker whose daily endpoint raises (Tiingo 400/404) is dropped from the cross-section without
|
|
aborting the build; survivors-only, and their scores equal the domain over just the survivors."""
|
|
bars = _make_bars(
|
|
raise_for="C",
|
|
exc=urllib.error.HTTPError("http://tiingo/daily/C", 404, "Not Found", {}, None), # type: ignore[arg-type]
|
|
)
|
|
s = compute_factor_scores(_make_universe(), bars, n=6)
|
|
|
|
survivors = ["A", "B", "D", "E", "F"]
|
|
assert s["syms"] == survivors # C dropped, order preserved
|
|
assert "C" not in s["prices"]
|
|
assert s["scores"] == pytest.approx(_domain_scores(survivors)) # scored over survivors only
|
|
|
|
|
|
def test_compute_factor_scores_drops_ticker_with_insufficient_history() -> None:
|
|
"""A ticker without enough price history for momentum/vol is dropped (no momentum signal -> not scored)."""
|
|
data = {s: _price_path(_MOM[s], _VOL[s]) for s in _SYMS}
|
|
data["C"] = {_iso(i): 100.0 + i for i in range(50)} # < 252: momentum None -> dropped
|
|
bars = FakeDailyBars(data)
|
|
s = compute_factor_scores(_make_universe(), bars, n=6)
|
|
assert s["syms"] == ["A", "B", "D", "E", "F"]
|
|
assert "C" not in s["prices"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- live-booking services
|
|
def _drive_two_steps(strategy_factory, sid: str, tmp_path, *, ls: bool):
|
|
"""First step freezes inception + seeds target positions; cross a month boundary via a clock so the
|
|
second step rebalances and books exactly one row off the SHARED precomputed scores. The second step's
|
|
scores carry BUMPED prices so the prior book marks to a non-trivial realized return.
|
|
Returns (st1, loaded0, loaded1, summary, rows, prev_targets)."""
|
|
p = str(tmp_path / f"{sid}_state.json")
|
|
|
|
# clock: inception in month M, second run in month M+1 (forces a rebalance + a strictly-later date).
|
|
day = ["2026-01-15"]
|
|
|
|
scores0 = _fixed_scores(_SYMS)
|
|
st0 = ForwardTracker(strategy_factory(scores0, clock=lambda: day[0]), p).step()
|
|
assert st0.forward_days == 0
|
|
loaded0 = json.loads(Path(p).read_text())
|
|
prev_targets = dict(loaded0["extra"]["positions"])
|
|
prev_prices = dict(loaded0["extra"]["prices"])
|
|
assert prev_targets # seeded a real target book
|
|
assert loaded0["extra"]["last_rebal"] == "2026-01-15"
|
|
|
|
# Second run: bump every name's marking price (a new precomputed `scores` snapshot) and cross the month.
|
|
bumps = {"A": 1.10, "B": 1.05, "C": 1.00, "D": 0.98, "E": 0.95, "F": 0.90}
|
|
scores1 = _fixed_scores(_SYMS)
|
|
cur_prices = {s: scores1["prices"][s] * bumps[s] for s in _SYMS}
|
|
scores1["prices"] = dict(cur_prices)
|
|
day[0] = "2026-02-15" # next month → month-changed rebalance, strictly later date → booked
|
|
|
|
st1 = ForwardTracker(strategy_factory(scores1, clock=lambda: day[0]), p).step()
|
|
assert st1.forward_days == 1
|
|
loaded1 = json.loads(Path(p).read_text())
|
|
summary, rows = ForwardStateReader().read(p, sid)
|
|
|
|
# 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."""
|
|
p = str(tmp_path / "eqfactor_long_state.json")
|
|
|
|
day = ["2026-01-10"]
|
|
scores0 = _fixed_scores(_SYMS)
|
|
ForwardTracker(EquityFactorLong(scores0, clock=lambda: day[0]), p).step()
|
|
loaded0 = json.loads(Path(p).read_text())
|
|
pos0 = loaded0["extra"]["positions"]
|
|
|
|
# new price snapshot, same month, strictly-later date
|
|
scores1 = _fixed_scores(_SYMS)
|
|
scores1["prices"] = {s: scores1["prices"][s] * 1.01 for s in _SYMS}
|
|
day[0] = "2026-01-20"
|
|
st1 = ForwardTracker(EquityFactorLong(scores1, clock=lambda: day[0]), p).step()
|
|
assert st1.forward_days == 1
|
|
loaded1 = json.loads(Path(p).read_text())
|
|
assert loaded1["extra"]["positions"] == pos0 # positions held (no rebalance)
|
|
assert loaded1["extra"]["last_rebal"] == "2026-01-10" # rebalance date unchanged
|