260 lines
12 KiB
Python
260 lines
12 KiB
Python
"""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
|