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

feat(b2.2): equity-factor price-only (momentum+lowvol); drop fundamentals fetch (Tiingo DOW-30-only)

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-16 22:31:31 +02:00
parent d29593e0ed
commit 7fc1790780
3 changed files with 131 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/fxhnt/adapters/orchestration/assets.py
View File

@@ -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": {}}

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

@@ -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

222
tests/integration/test_equity_factor_strategy.py
View File

@@ -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