diff --git a/src/fxhnt/application/equity_factor_strategy.py b/src/fxhnt/application/equity_factor_strategy.py index a61935c..9396648 100644 --- a/src/fxhnt/application/equity_factor_strategy.py +++ b/src/fxhnt/application/equity_factor_strategy.py @@ -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) diff --git a/tests/integration/test_equity_factor_strategy.py b/tests/integration/test_equity_factor_strategy.py index d85fad4..2977c21 100644 --- a/tests/integration/test_equity_factor_strategy.py +++ b/tests/integration/test_equity_factor_strategy.py @@ -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)