From 7fc1790780f2bb04e1fdc50b8c6dcf123b163b12 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 16 Jun 2026 22:31:31 +0200 Subject: [PATCH] feat(b2.2): equity-factor price-only (momentum+lowvol); drop fundamentals fetch (Tiingo DOW-30-only) Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fxhnt/adapters/orchestration/assets.py | 13 +- .../application/equity_factor_strategy.py | 108 +++++---- .../test_equity_factor_strategy.py | 222 ++++++------------ 3 files changed, 131 insertions(+), 212 deletions(-) diff --git a/src/fxhnt/adapters/orchestration/assets.py b/src/fxhnt/adapters/orchestration/assets.py index 866cb7e..e25ca31 100644 --- a/src/fxhnt/adapters/orchestration/assets.py +++ b/src/fxhnt/adapters/orchestration/assets.py @@ -211,18 +211,19 @@ def poc_nav(context: AssetExecutionContext) -> dict: # type: ignore[type-arg] @asset def eqfactor_scores(context: AssetExecutionContext) -> dict: # type: ignore[type-arg] - """Shared factor cross-section fetched ONCE (Tiingo universe + fundamentals + daily bars, batched-concurrent). + """Shared PRICE-ONLY factor cross-section fetched ONCE (Tiingo universe + daily bars, batched-concurrent). - The three equity-factor sleeves consume this SAME precomputed scores dict, removing the prior ~3× redundant - sequential fetch (each sleeve re-pulling the whole universe). A transient network/API error logs a warning - and yields an EMPTY cross-section (sleeves then book nothing) rather than crashing the nightly run.""" + The sleeve is price-only (momentum + low-vol): one adj_closes fetch per ticker yields both signals, so + prices alone — which Tiingo serves broadly — score the full universe. Fundamentals are NOT fetched (the + Tiingo plan covers DOW-30 only, which silently dropped ~270/300 names). The three equity-factor sleeves + consume this SAME precomputed scores dict. A transient network/API error logs a warning and yields an + EMPTY cross-section (sleeves then book nothing) rather than crashing the nightly run.""" from fxhnt.adapters.data.tiingo_daily import TiingoDailyClient - from fxhnt.adapters.data.tiingo_fundamentals import TiingoFundamentalsClient from fxhnt.adapters.data.tiingo_universe import TiingoUniverseSource from fxhnt.application.equity_factor_strategy import compute_factor_scores try: - s = compute_factor_scores(TiingoUniverseSource(), TiingoFundamentalsClient(), TiingoDailyClient(), n=150) + s = compute_factor_scores(TiingoUniverseSource(), TiingoDailyClient(), n=150) except (OSError, TimeoutError, ConnectionError) as e: # transient: don't crash the run context.log.warning(f"eqfactor_scores fetch failed: {e}") s = {"date": "", "syms": [], "scores": [], "prices": {}} diff --git a/src/fxhnt/application/equity_factor_strategy.py b/src/fxhnt/application/equity_factor_strategy.py index 9396648..debc23c 100644 --- a/src/fxhnt/application/equity_factor_strategy.py +++ b/src/fxhnt/application/equity_factor_strategy.py @@ -1,9 +1,15 @@ -"""Shared batched-concurrent factor-score builder + the three live-booking ForwardStrategy services. +"""Shared batched-concurrent PRICE-ONLY factor-score builder + the three live-booking ForwardStrategy services. -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. +The universe + per-ticker 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. + +The sleeve is PRICE-ONLY (momentum + low-vol): a single `adj_closes` fetch per ticker yields BOTH the +12-1 momentum and a trailing realized vol, so the whole cross-section needs nothing but prices — which +Tiingo serves broadly. Fundamentals are intentionally NOT fetched: the current Tiingo plan exposes +fundamentals for DOW-30 only, so a fundamentals fetch silently dropped ~270/300 names. The +FundamentalsClient port + TiingoFundamentalsClient adapter are retained (unused here) for a future +fundamentals upgrade. 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 @@ -13,18 +19,21 @@ from __future__ import annotations import concurrent.futures import logging +import math from collections.abc import Callable +from statistics import fmean 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 +from fxhnt.ports.market_data import DailyBarClient, UniverseSource _logger = logging.getLogger(__name__) # Trading-day offsets for the 12-1 momentum window: skip the most recent ~21d, look back ~252d (1y). _LOOKBACK = 252 _SKIP = 21 +_VOL_WINDOW = 90 def _momentum_12_1(closes: dict[str, float]) -> float | None: @@ -43,42 +52,59 @@ 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.""" +def _realized_vol(closes: dict[str, float], window: int = _VOL_WINDOW) -> float | None: + """Trailing realized vol = sample stdev of the last `window` daily log returns. + + Sorts the dates, takes the most-recent `window+1` closes (-> `window` returns), and returns their + population/sample stdev. Returns None when there is not enough history for a full window or when any + close in the window is non-positive (log return undefined).""" + dates = sorted(closes) + if len(dates) < window + 1: + return None + tail = [closes[d] for d in dates[-(window + 1):]] + rets: list[float] = [] + for p0, p1 in zip(tail, tail[1:]): + if p0 is None or p1 is None or p0 <= 0 or p1 <= 0: + return None + rets.append(math.log(p1 / p0)) + if len(rets) < 2: + return None + mu = fmean(rets) + return math.sqrt(fmean([(r - mu) ** 2 for r in rets])) + + +def _gather_ticker(sym: str, bars: DailyBarClient) -> dict[str, Any] | None: + """Fetch one ticker's adjusted-close history ONCE and derive its 12-1 momentum, trailing realized vol + and latest close. Per-ticker resilient: ANY exception (the Tiingo daily endpoint 400/404s on a + no-coverage name, ...) returns None. A name with insufficient history for momentum is also dropped (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) + mom = _momentum_12_1(closes) + if mom is None: + return None 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), + "mom": mom, + "vol": _realized_vol(closes), "price": price, } except Exception: return None -def _gather_universe(universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient, +def _gather_universe(universe: UniverseSource, 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 + """Pull the top-n universe and gather each ticker's prices 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} + futures = {pool.submit(_gather_ticker, sym, bars): sym for sym in syms} for fut in concurrent.futures.as_completed(futures): rec = fut.result() if rec is not None: @@ -89,40 +115,22 @@ def _gather_universe(universe: UniverseSource, fund: FundamentalsClient, bars: D return survivors -def compute_factor_scores(universe: UniverseSource, fund: FundamentalsClient, bars: DailyBarClient, +def compute_factor_scores(universe: UniverseSource, 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 + """Fetch the universe + prices ONCE (batched-concurrent) and compute the SHARED PRICE-ONLY 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) + only — gather each ticker's adj_closes concurrently (deriving momentum AND trailing realized vol from + that single fetch), run the survivors through the pure momentum + low-vol -> composite_price pipeline, + and emit the aligned syms/scores plus a {sym: latest_close} price map.""" + survivors = _gather_universe(universe, 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) + lv = ef.lowvol_score([r["vol"] for r in survivors]) + c = ef.composite_price(mm, lv) 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. - - 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(syms, w) if weight != 0} - - class _EquityFactorStrategy: """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 diff --git a/tests/integration/test_equity_factor_strategy.py b/tests/integration/test_equity_factor_strategy.py index 2977c21..587ebbc 100644 --- a/tests/integration/test_equity_factor_strategy.py +++ b/tests/integration/test_equity_factor_strategy.py @@ -1,11 +1,11 @@ """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. `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.""" +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 @@ -19,8 +19,8 @@ from fxhnt.application.equity_factor_strategy import ( EquityFactorLong, EquityFactorLS, EquityFactorTilt, - _factor_targets, _momentum_12_1, + _realized_vol, compute_factor_scores, ) from fxhnt.application.forward_tracker import ForwardTracker @@ -36,64 +36,32 @@ class FakeUniverse: 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]], - raise_for: str | None = None, - exc: Exception | None = None) -> None: - self._metrics = metrics_by_sym - self._stmts = stmts_by_sym - # When set, metrics()/statement_metrics() raise for this one ticker — models a name - # the Tiingo fundamentals endpoint 400/404s on (no coverage / odd symbol). +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 _maybe_raise(self, symbol: str) -> None: + 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}") - - def metrics(self, symbol: str) -> dict[str, float]: - self._maybe_raise(symbol) - return dict(self._metrics.get(symbol, {})) - - def statement_metrics(self, symbol: str) -> dict[str, float]: - self._maybe_raise(symbol) - 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. +# 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"] -_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} +# 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 +_N_DAYS = 300 # >= 252 so 12-1 momentum is computable; > 90 so the realized vol window is full def _iso(i: int) -> str: @@ -101,21 +69,18 @@ def _iso(i: int) -> str: 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`.""" +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) for i in range(_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() -> FakeDailyBars: - return FakeDailyBars({s: _price_path(_MOM[s]) for s in _SYMS}) - - -def _make_fund() -> FakeFundamentals: - return FakeFundamentals(_METRICS, _STMTS) +def _make_bars(**kwargs) -> FakeDailyBars: + return FakeDailyBars({s: _price_path(_MOM[s], _VOL[s]) for s in _SYMS}, **kwargs) def _make_universe() -> FakeUniverse: @@ -123,16 +88,11 @@ def _make_universe() -> FakeUniverse: 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)) + """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: @@ -142,7 +102,7 @@ def _fixed_scores(syms: list[str]) -> dict: "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}, + "prices": {s: _price_path(_MOM[s], _VOL[s])[_iso(_N_DAYS - 1)] for s in syms}, } @@ -159,28 +119,41 @@ def test_momentum_12_1_none_when_too_short() -> None: 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 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) + """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])[_iso(_N_DAYS - 1)]) + 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 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, + """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/fundamentals/C", 404, "Not Found", {}, None), # type: ignore[arg-type] + exc=urllib.error.HTTPError("http://tiingo/daily/C", 404, "Not Found", {}, None), # type: ignore[arg-type] ) - s = compute_factor_scores(_make_universe(), fund, _make_bars(), n=6) + s = compute_factor_scores(_make_universe(), bars, n=6) survivors = ["A", "B", "D", "E", "F"] assert s["syms"] == survivors # C dropped, order preserved @@ -188,77 +161,14 @@ def test_compute_factor_scores_drops_raising_ticker() -> None: 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) - # 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) - 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) - - -# --------------------------------------------------------------------------- per-ticker resilience -def test_factor_targets_skips_ticker_lacking_coverage() -> None: - """One ticker whose fundamentals endpoint raises (e.g. Tiingo 400/404) must be dropped from the - cross-section without aborting the whole book; the good names still get weights.""" - universe = _make_universe() - bars = _make_bars() - # "C" 404s on the Tiingo fundamentals endpoint; A/B/D/E/F have normal coverage. - fund = FakeFundamentals( - _METRICS, _STMTS, - raise_for="C", - exc=urllib.error.HTTPError("http://tiingo/fundamentals/C", 404, "Not Found", {}, None), # type: ignore[arg-type] - ) - - targets = _factor_targets(universe, fund, bars, "ls", n=6) - - # (a) did not raise — we got here. (b) the failing ticker is excluded. - assert "C" not in targets - # (c) still scored the good names: with C dropped, the surviving cross-section is A,B,D,E,F and - # the ls construction still longs the best survivor (A) and shorts the worst survivor (F). - assert targets["A"] == pytest.approx(1.0) - assert targets["F"] == pytest.approx(-1.0) - 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_drops_failing_ticker_with_plain_exception() -> None: - """A plain Exception (not just HTTPError) on a ticker is also tolerated and the name dropped.""" - universe = _make_universe() - bars = _make_bars() - fund = FakeFundamentals(_METRICS, _STMTS, raise_for="A") # default RuntimeError - - targets = _factor_targets(universe, fund, bars, "long", n=6) - - # A (normally the sole long-only pick) raised → dropped; long-only now picks the best survivor B. - assert "A" not in targets - assert set(targets) == {"B"} - assert targets["B"] == pytest.approx(1.0) +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