gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(b2): shared batched-concurrent factor scores; services consume precomputed scores

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-16 21:27:53 +02:00
parent 547074aebb
commit 6d2acf7956
2 changed files with 184 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
src/fxhnt/application/equity_factor_strategy.py
View File

@@ -1,12 +1,17 @@
"""Equity-factor target builder + the three live-booking ForwardStrategy services (long / ls / tilt).
"""Shared batched-concurrent factor-score builder + the three live-booking ForwardStrategy services.
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`."""
The universe + per-ticker fundamentals + daily bars are fetched ONCE (concurrently, not per-sleeve) into a
single precomputed `scores` dict via `compute_factor_scores`; the three sleeves (long / ls / tilt) each
consume that SAME shared dict and do NO I/O of their own. This removes the ~3× redundant sequential fetch
(every sleeve re-pulling the whole universe + fundamentals + prices) that made the live run ~37+ min.
Each service marks the prior target book to today's precomputed prices, books the realized daily return
(net of short-borrow for the long-short construction), and rebalances to a fresh factor book on the monthly
boundary. On the FIRST run the prior book is empty so the booked return is 0 and the tracker freezes; it
just seeds the initial target positions + their marking prices + the rebalance month into `extra`."""
from __future__ import annotations
import concurrent.futures
import logging
from collections.abc import Callable
from typing import Any
@@ -38,120 +43,125 @@ def _momentum_12_1(closes: dict[str, float]) -> float | None:
return p_recent / p_then - 1.0
def _gather_ticker(sym: str, fund: FundamentalsClient, bars: DailyBarClient) -> dict[str, Any] | None:
"""Fetch one ticker's value metrics (pe/pb), statement metrics (piotroski/roe/debtEquity/grossMargin),
12-1 momentum and latest adjusted close. Per-ticker resilient: ANY exception (the Tiingo fundamentals
endpoint 400/404s on a no-coverage name, no bar history, ...) returns None so the name is dropped from
the cross-section entirely (a name we can't score must not be in the book) rather than aborting the
whole sleeve. Pure read — safe to run concurrently across a thread pool."""
try:
m = fund.metrics(sym)
s = fund.statement_metrics(sym)
closes = bars.adj_closes(sym)
price = closes[max(closes)] if closes else None
return {
"sym": sym,
"pe": m.get("peRatio"),
"pb": m.get("pbRatio"),
"pio": s.get("piotroskiFScore"),
"roe": s.get("roe"),
"de": s.get("debtEquity"),
"gm": s.get("grossMargin"),
"mom": _momentum_12_1(closes),
"price": price,
}
except Exception:
return None
def _gather_universe(universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient,
n: int, max_workers: int) -> list[dict[str, Any]]:
"""Pull the top-n universe and gather each ticker's data CONCURRENTLY over a thread pool, dropping
(and counting) any ticker whose gather failed. Returns the survivors' per-ticker records in the
deterministic top-n order (concurrency fans out the I/O; results are re-ordered to the universe order
so the cross-section + downstream weights are reproducible regardless of completion order)."""
syms = universe.top_n(n)
records: dict[str, dict[str, Any]] = {}
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {pool.submit(_gather_ticker, sym, fund, bars): sym for sym in syms}
for fut in concurrent.futures.as_completed(futures):
rec = fut.result()
if rec is not None:
records[rec["sym"]] = rec
survivors = [records[sym] for sym in syms if sym in records]
_logger.info("equity-factor: gathered %d/%d, dropped %d (no data)",
len(survivors), len(syms), len(syms) - len(survivors))
return survivors
def compute_factor_scores(universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient,
n: int = 150, max_workers: int = 12) -> dict[str, Any]:
"""Fetch the universe + fundamentals + prices ONCE (batched-concurrent) and compute the SHARED factor
cross-section the three sleeves consume. Returns {'date', 'syms', 'scores', 'prices'} over survivors
only — gather the per-ticker data concurrently, run the survivors through the pure value/quality/
momentum -> composite pipeline, and emit the aligned syms/scores plus a {sym: latest_close} price map."""
survivors = _gather_universe(universe, fund, bars, n, max_workers)
syms = [r["sym"] for r in survivors]
v = ef.value_score([r["pe"] for r in survivors], [r["pb"] for r in survivors])
q = ef.quality_score([r["pio"] for r in survivors], [r["roe"] for r in survivors],
[r["de"] for r in survivors], [r["gm"] for r in survivors])
mm = ef.momentum_score([r["mom"] for r in survivors])
c = ef.composite(v, q, mm)
prices = {r["sym"]: r["price"] for r in survivors if r["price"] is not None}
return {"date": _today_iso(), "syms": syms, "scores": c, "prices": prices}
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.
Per-ticker resilient: in the live ~8000-name universe a fraction of tickers lack Tiingo
fundamentals coverage (the endpoint 400/404s) or have no bar history. Any failure gathering a
ticker's data drops that name from the cross-section entirely (a name we can't score must not be
in the book) and is counted; the factor scores + composite + construction weights are built over
only the successfully-fetched survivors. One bad ticker no longer aborts the whole sleeve."""
syms = universe.top_n(n)
survivors: list[str] = []
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] = []
dropped = 0
for sym in syms:
try:
m = fund.metrics(sym)
s = fund.statement_metrics(sym)
mm_sym = _momentum_12_1(bars.adj_closes(sym))
except Exception:
dropped += 1
continue
survivors.append(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(mm_sym)
_logger.info("equity-factor: scored %d/%d, dropped %d (no data)", len(survivors), len(syms), dropped)
v = ef.value_score(pe, pb)
q = ef.quality_score(pio, roe, de, gm)
mm = ef.momentum_score(mom)
Reuses the same concurrent per-ticker gather as `compute_factor_scores`; maps the survivors' composite
scores to construction weights and returns {sym: weight} dropping zero-weight names."""
survivors = _gather_universe(universe, fund, bars, n, max_workers=12)
syms = [r["sym"] for r in survivors]
v = ef.value_score([r["pe"] for r in survivors], [r["pb"] for r in survivors])
q = ef.quality_score([r["pio"] for r in survivors], [r["roe"] for r in survivors],
[r["de"] for r in survivors], [r["gm"] for r in survivors])
mm = ef.momentum_score([r["mom"] for r in survivors])
c = ef.composite(v, q, mm)
w = ef.construction_weights(c, construction, quantile)
return {sym: weight for sym, weight in zip(survivors, w) if weight != 0}
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`."""
"""Live-booking equity-factor ForwardStrategy over a SHARED precomputed `scores` dict (no I/O). Marks
the prior target book to today's precomputed 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,
def __init__(self, construction: str, scores: dict[str, Any], 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._scores = scores
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
cur: dict[str, float] = self._scores["prices"]
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
r = ef.book_return(prev, prev_prices, cur, borrow_daily) if prev else 0.0
if rebalancing:
new_positions = _factor_targets(self._universe, self._fund, self._bars, self._construction, self._n)
if last_rebal is None or today[:7] != last_rebal[:7]:
w = ef.construction_weights(self._scores["scores"], self._construction)
new = {s: wt for s, wt in zip(self._scores["syms"], w) if wt != 0}
last_rebal = today
else:
new_positions = dict(prev)
new = 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}
new_prices = {s: cur[s] for s in new if s in cur}
return [(today, r)], {"positions": new, "prices": new_prices, "last_rebal": last_rebal}
class EquityFactorLong:
"""Long-only top-quintile equity-factor sleeve."""
"""Long-only top-quintile equity-factor sleeve over a shared precomputed `scores` dict."""
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 __init__(self, scores: dict[str, Any], clock: Callable[[], str] = _today_iso) -> None:
self._impl = _EquityFactorStrategy("long", scores, 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)
@@ -160,20 +170,19 @@ class EquityFactorLong:
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 __init__(self, scores: dict[str, Any], borrow_annual: float = 0.01,
clock: Callable[[], str] = _today_iso) -> None:
self._impl = _EquityFactorStrategy("ls", scores, 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."""
"""Long-only rank-weighted (tilt) equity-factor sleeve over a shared precomputed `scores` dict."""
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 __init__(self, scores: dict[str, Any], clock: Callable[[], str] = _today_iso) -> None:
self._impl = _EquityFactorStrategy("tilt", scores, 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)

116
tests/integration/test_equity_factor_strategy.py
View File

@@ -1,10 +1,11 @@
"""Equity-factor target builder + the three live-booking ForwardStrategy services (long / ls / tilt).
"""Shared batched-concurrent factor-score builder + the three live-booking ForwardStrategy services.
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."""
debtEquity/grossMargin and enough adj-close history for the 12-1 momentum. `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. The target builder is also asserted directly against the pure domain's top/bottom names."""
from __future__ import annotations
import json
@@ -20,6 +21,7 @@ from fxhnt.application.equity_factor_strategy import (
EquityFactorTilt,
_factor_targets,
_momentum_12_1,
compute_factor_scores,
)
from fxhnt.application.forward_tracker import ForwardTracker
from fxhnt.domain.strategies import equity_factor as ef
@@ -120,6 +122,30 @@ def _make_universe() -> FakeUniverse:
return FakeUniverse(_SYMS)
def _domain_scores(syms: list[str]) -> list[float]:
"""The pure-domain composite for the given names, fed the same aligned inputs the builder would —
asserted against directly (not circularly via the builder)."""
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]
return ef.composite(ef.value_score(pe, pb), ef.quality_score(pio, roe, de, gm), ef.momentum_score(mom))
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])[_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)
@@ -133,6 +159,35 @@ def test_momentum_12_1_none_when_too_short() -> None:
assert _momentum_12_1(closes) is None
# --------------------------------------------------------------------------- compute_factor_scores
def test_compute_factor_scores_matches_domain_and_emits_prices() -> None:
"""The shared builder must fetch ONCE and emit aligned syms/scores equal to the pure-domain composite
(asserted directly, not circularly) plus a {sym: latest_close} price map."""
s = compute_factor_scores(_make_universe(), _make_fund(), _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])[_iso(_N_DAYS - 1)])
assert "date" in s
def test_compute_factor_scores_drops_raising_ticker() -> None:
"""A ticker whose fundamentals 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."""
fund = FakeFundamentals(
_METRICS, _STMTS,
raise_for="C",
exc=urllib.error.HTTPError("http://tiingo/fundamentals/C", 404, "Not Found", {}, None), # type: ignore[arg-type]
)
s = compute_factor_scores(_make_universe(), fund, _make_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
# --------------------------------------------------------------------------- _factor_targets composition
def test_factor_targets_long_selects_top_name() -> None:
targets = _factor_targets(_make_universe(), _make_fund(), _make_bars(), "long", n=6)
@@ -162,15 +217,7 @@ def test_factor_targets_tilt_is_long_only_overweights_best() -> None:
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)
w = ef.construction_weights(_domain_scores(syms), "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)
@@ -216,17 +263,17 @@ def test_factor_targets_drops_failing_ticker_with_plain_exception() -> None:
# --------------------------------------------------------------------------- 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()
"""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"]
st0 = ForwardTracker(strategy_factory(universe, fund, bars, clock=lambda: day[0]), p).step()
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"])
@@ -234,18 +281,14 @@ def _drive_two_steps(strategy_factory, sid: str, tmp_path, *, ls: bool):
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
# 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}
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)]
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(universe, fund, bars, clock=lambda: day[0]), p).step()
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)
@@ -289,22 +332,19 @@ def test_equity_factor_tilt_service_round_trips(tmp_path) -> None:
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()
scores0 = _fixed_scores(_SYMS)
ForwardTracker(EquityFactorLong(scores0, 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
# 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(universe, fund, bars, clock=lambda: day[0]), p).step()
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)