You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
544 lines
20 KiB
Python
544 lines
20 KiB
Python
"""CrossSectionalMomentum — classic 12-1 academic momentum factor, look-ahead-safe.
|
|
|
|
On the first trading day of each calendar month, rank universe by the
|
|
cumulative return from t-252 to t-21 (i.e. 12 months back, skipping the most
|
|
recent month per Jegadeesh-Titman 1993). Buy the top-N highest-momentum
|
|
symbols at next_open, hold for `holding_days` trading days, then force-flat.
|
|
|
|
This is the canonical academic momentum factor. The 1-month skip removes
|
|
short-term reversal contamination. Top-decile long-only is the cleanest
|
|
specification compatible with the existing long-only execution stack.
|
|
|
|
Look-ahead defenses (NON-NEGOTIABLE):
|
|
* BarHistoryProvider returns bars STRICTLY before decision_date.
|
|
* `_assert_strictly_before` re-checks every used bar date.
|
|
* Rebalance gate uses last_bar_date.month vs decision_date.month — purely
|
|
derived from already-strict bars; no calendar lookup needed.
|
|
|
|
Per advisor pre-commit (Phase 19): STANDALONE POC only — no hybrid scaffolding.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import statistics
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING, Any, Iterable, Protocol
|
|
|
|
if TYPE_CHECKING:
|
|
from libs.backtest.xsmom_cache import XsmomRankCache
|
|
|
|
from libs.backtest.domain import (
|
|
Candidate,
|
|
LookaheadViolationError,
|
|
StrategyEngineConfig,
|
|
)
|
|
from libs.common.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
CROSS_SECTIONAL_MOMENTUM_EVENT_TYPE = "xsmom_12_1"
|
|
|
|
_ET_MARKET_OPEN = dt.time(9, 30)
|
|
_ET_OFFSET = dt.timedelta(hours=-5)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider Protocol (shared shape with VolBreakout52w)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class BarHistoryProvider(Protocol):
|
|
def get_bars_before(
|
|
self,
|
|
symbol: str,
|
|
as_of_date: dt.date,
|
|
lookback_days: int,
|
|
) -> list[tuple[dt.date, dict[str, Any]]]: ...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lookahead defense
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _decision_cutoff_utc(decision_date: dt.date) -> dt.datetime:
|
|
et_naive = dt.datetime.combine(decision_date, _ET_MARKET_OPEN)
|
|
utc_naive = et_naive - _ET_OFFSET
|
|
return utc_naive.replace(tzinfo=dt.timezone.utc)
|
|
|
|
|
|
def _assert_strictly_before(
|
|
symbol: str,
|
|
decision_date: dt.date,
|
|
used_dates: Iterable[dt.date],
|
|
) -> None:
|
|
for d in used_dates:
|
|
if d >= decision_date:
|
|
raise LookaheadViolationError(
|
|
f"CrossSectionalMomentum {symbol}: bar date {d.isoformat()} "
|
|
f"is not strictly before decision_date {decision_date.isoformat()}"
|
|
)
|
|
|
|
|
|
def _bar_close_timestamp(bar_date: dt.date) -> dt.datetime:
|
|
"""16:00 ET close on bar_date, tz-aware UTC."""
|
|
et_naive = dt.datetime.combine(bar_date, dt.time(16, 0))
|
|
utc_naive = et_naive - _ET_OFFSET
|
|
return utc_naive.replace(tzinfo=dt.timezone.utc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pure trigger computations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def is_rebalance_day(
|
|
bars: list[tuple[dt.date, dict[str, Any]]],
|
|
decision_date: dt.date,
|
|
) -> bool:
|
|
"""First trading day of the calendar month iff last_bar's month != decision_date.month.
|
|
|
|
Pure function: derives the rebalance flag from already-strict bar history.
|
|
Returns False on empty bars.
|
|
"""
|
|
if not bars:
|
|
return False
|
|
last_bar_date = bars[-1][0]
|
|
if last_bar_date >= decision_date:
|
|
raise LookaheadViolationError(
|
|
f"is_rebalance_day: last_bar_date {last_bar_date.isoformat()} "
|
|
f"is not strictly before decision_date {decision_date.isoformat()}"
|
|
)
|
|
return last_bar_date.month != decision_date.month
|
|
|
|
|
|
def compute_12_1_momentum(
|
|
bars: list[tuple[dt.date, dict[str, Any]]],
|
|
*,
|
|
lookback_days: int = 252,
|
|
skip_days: int = 21,
|
|
) -> tuple[float | None, list[dt.date]]:
|
|
"""Cumulative return from t-(lookback+skip) to t-skip. Skips most recent `skip_days`.
|
|
|
|
Returns (momentum_score, used_dates). None when insufficient history.
|
|
The window is `bars[-(lookback+skip):-skip]` — last bar is t-skip-1, first
|
|
bar is t-(lookback+skip).
|
|
"""
|
|
needed = lookback_days + skip_days
|
|
if len(bars) < needed:
|
|
return None, []
|
|
window = bars[-needed:-skip_days] if skip_days > 0 else bars[-needed:]
|
|
if len(window) < 2:
|
|
return None, []
|
|
start_close = float(window[0][1].get("close", 0.0))
|
|
end_close = float(window[-1][1].get("close", 0.0))
|
|
if start_close <= 0 or end_close <= 0:
|
|
return None, []
|
|
momentum = (end_close / start_close) - 1.0
|
|
used_dates = [d for d, _ in window]
|
|
return momentum, used_dates
|
|
|
|
|
|
def compute_20d_volatility(
|
|
bars: list[tuple[dt.date, dict[str, Any]]],
|
|
) -> float | None:
|
|
"""Stdev of daily log returns over the last 20 bars. Returns None on insufficient data."""
|
|
if len(bars) < 21:
|
|
return None
|
|
closes = [float(b.get("close", 0.0)) for _, b in bars[-21:]]
|
|
if any(c <= 0 for c in closes):
|
|
return None
|
|
rets = []
|
|
for i in range(1, len(closes)):
|
|
rets.append((closes[i] / closes[i - 1]) - 1.0)
|
|
if len(rets) < 2:
|
|
return None
|
|
return float(statistics.stdev(rets))
|
|
|
|
|
|
def compute_avg_dollar_volume_20d(
|
|
bars: list[tuple[dt.date, dict[str, Any]]],
|
|
) -> float:
|
|
if not bars:
|
|
return 0.0
|
|
tail = bars[-20:]
|
|
return statistics.fmean(
|
|
float(b.get("close", 0.0)) * float(b.get("volume", 0.0))
|
|
for _, b in tail
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-symbol trigger inputs + evaluation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CrossSectionalMomentumInputs:
|
|
symbol: str
|
|
decision_date: dt.date
|
|
next_trading_date: dt.date
|
|
last_bar_date: dt.date
|
|
last_bar_timestamp: dt.datetime
|
|
last_close: float
|
|
momentum_12_1: float
|
|
volatility_20d: float
|
|
avg_dollar_volume_20d: float
|
|
|
|
|
|
def evaluate_universe_gates(
|
|
inputs: CrossSectionalMomentumInputs,
|
|
engine: StrategyEngineConfig,
|
|
) -> tuple[bool, str | None]:
|
|
"""Reject low-quality candidates BEFORE ranking. Pure function."""
|
|
if inputs.last_bar_date >= inputs.decision_date:
|
|
raise LookaheadViolationError(
|
|
f"CrossSectionalMomentum {inputs.symbol}: last_bar_date "
|
|
f"{inputs.last_bar_date.isoformat()} not strictly before decision_date"
|
|
)
|
|
price_min = float(getattr(engine, "xsmom_min_price", 5.0) or 0.0)
|
|
if price_min > 0 and inputs.last_close < price_min:
|
|
return False, f"last_close {inputs.last_close:.2f} < min {price_min:.2f}"
|
|
adv_min = float(getattr(engine, "xsmom_min_avg_dollar_volume", 10_000_000.0) or 0.0)
|
|
if adv_min > 0 and inputs.avg_dollar_volume_20d < adv_min:
|
|
return False, (
|
|
f"avg_dollar_volume_20d {inputs.avg_dollar_volume_20d:,.0f} < min {adv_min:,.0f}"
|
|
)
|
|
vol_max = float(getattr(engine, "xsmom_volatility_20d_max", 0.08) or 0.0)
|
|
if vol_max > 0 and inputs.volatility_20d > vol_max:
|
|
return False, f"volatility_20d {inputs.volatility_20d:.4f} > max {vol_max:.4f}"
|
|
return True, None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Build candidates (top-N selection across universe)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_candidates_from_cache(
|
|
rows: list[dict[str, Any]],
|
|
decision_date: dt.date,
|
|
next_trading_date: dt.date,
|
|
engine: StrategyEngineConfig,
|
|
top_n: int,
|
|
) -> list[Candidate]:
|
|
"""Build top-N candidates from cached ranked rows (already post-quality-gate)."""
|
|
rows_sorted = sorted(rows, key=lambda r: float(r["momentum_12_1"]), reverse=True)
|
|
top = rows_sorted[:top_n]
|
|
total_ranked = len(rows_sorted)
|
|
|
|
candidates: list[Candidate] = []
|
|
for rank, row in enumerate(top, start=1):
|
|
last_bar_date = dt.date.fromisoformat(str(row["last_bar_date"]))
|
|
_assert_strictly_before(str(row["symbol"]), decision_date, [last_bar_date])
|
|
inputs = CrossSectionalMomentumInputs(
|
|
symbol=str(row["symbol"]),
|
|
decision_date=decision_date,
|
|
next_trading_date=next_trading_date,
|
|
last_bar_date=last_bar_date,
|
|
last_bar_timestamp=dt.datetime.fromisoformat(str(row["last_bar_timestamp_iso"])),
|
|
last_close=float(row["last_close"]),
|
|
momentum_12_1=float(row["momentum_12_1"]),
|
|
volatility_20d=float(row["volatility_20d"]),
|
|
avg_dollar_volume_20d=float(row["avg_dollar_volume_20d"]),
|
|
)
|
|
candidates.append(_build_candidate(inputs, engine, rank=rank, total_ranked=total_ranked))
|
|
|
|
if candidates:
|
|
logger.info(
|
|
"cross_sectional_momentum_rebalance",
|
|
decision_date=decision_date.isoformat(),
|
|
universe_scanned=total_ranked,
|
|
top_n_emitted=len(candidates),
|
|
top_score=float(top[0]["momentum_12_1"]) if top else None,
|
|
bottom_score=float(top[-1]["momentum_12_1"]) if top else None,
|
|
cache_hit=True,
|
|
)
|
|
return candidates
|
|
|
|
|
|
def build_candidates(
|
|
decision_date: dt.date,
|
|
next_trading_date: dt.date,
|
|
universe_symbols: Iterable[str],
|
|
engine: StrategyEngineConfig,
|
|
bar_provider: BarHistoryProvider,
|
|
*,
|
|
cache: "XsmomRankCache | None" = None,
|
|
) -> list[Candidate]:
|
|
"""Emit top-N synthetic momentum candidates on rebalance days only.
|
|
|
|
On non-rebalance days, returns []. The universe is scanned once per
|
|
rebalance day, ranked, and the top-N pass through.
|
|
|
|
If ``cache`` is provided: on rebalance days, tries to serve from the disk
|
|
cache (populated on prior runs). Cache miss triggers the full universe scan
|
|
and saves the ranked universe for future runs.
|
|
"""
|
|
if not getattr(engine, "xsmom_enabled", False):
|
|
return []
|
|
if next_trading_date <= decision_date:
|
|
raise LookaheadViolationError(
|
|
f"CrossSectionalMomentum next_trading_date {next_trading_date.isoformat()} "
|
|
f"must be strictly after decision_date {decision_date.isoformat()}"
|
|
)
|
|
|
|
lookback = int(getattr(engine, "xsmom_lookback_days", 252) or 252)
|
|
skip = int(getattr(engine, "xsmom_skip_days", 21) or 21)
|
|
top_n = int(getattr(engine, "xsmom_top_n", 20) or 20)
|
|
momentum_min = float(getattr(engine, "xsmom_momentum_min", 0.0) or 0.0)
|
|
|
|
fetch_lookback = lookback + skip + 5
|
|
|
|
scored: list[tuple[float, CrossSectionalMomentumInputs]] = []
|
|
seen: set[str] = set()
|
|
rebalance_checked = False
|
|
|
|
for raw_symbol in universe_symbols:
|
|
symbol = str(raw_symbol).strip().upper()
|
|
if not symbol or symbol in seen:
|
|
continue
|
|
seen.add(symbol)
|
|
|
|
bars = bar_provider.get_bars_before(symbol, decision_date, lookback_days=fetch_lookback)
|
|
if not bars:
|
|
continue
|
|
|
|
last_bar_date, last_bar = bars[-1]
|
|
if last_bar_date >= decision_date:
|
|
raise LookaheadViolationError(
|
|
f"CrossSectionalMomentum bar for {symbol} on {last_bar_date.isoformat()} "
|
|
f"is not strictly before decision_date {decision_date.isoformat()}"
|
|
)
|
|
|
|
# Rebalance gate — check once per call, derived from bars (no calendar).
|
|
if not rebalance_checked:
|
|
is_rebalance = is_rebalance_day(bars, decision_date)
|
|
rebalance_checked = True
|
|
if not is_rebalance:
|
|
return []
|
|
# Rebalance confirmed: try cache before scanning remaining universe.
|
|
if cache is not None:
|
|
cached_rows = cache.get_date(decision_date)
|
|
if cached_rows is not None:
|
|
return _build_candidates_from_cache(
|
|
cached_rows, decision_date, next_trading_date, engine, top_n
|
|
)
|
|
|
|
last_close = float(last_bar.get("close", 0.0))
|
|
if last_close <= 0:
|
|
continue
|
|
|
|
momentum, mom_used_dates = compute_12_1_momentum(
|
|
bars, lookback_days=lookback, skip_days=skip
|
|
)
|
|
if momentum is None:
|
|
continue
|
|
if momentum < momentum_min:
|
|
continue
|
|
|
|
_assert_strictly_before(symbol, decision_date, mom_used_dates)
|
|
|
|
vol_20d = compute_20d_volatility(bars)
|
|
if vol_20d is None:
|
|
continue
|
|
adv_20d = compute_avg_dollar_volume_20d(bars)
|
|
|
|
inputs = CrossSectionalMomentumInputs(
|
|
symbol=symbol,
|
|
decision_date=decision_date,
|
|
next_trading_date=next_trading_date,
|
|
last_bar_date=last_bar_date,
|
|
last_bar_timestamp=_bar_close_timestamp(last_bar_date),
|
|
last_close=last_close,
|
|
momentum_12_1=momentum,
|
|
volatility_20d=vol_20d,
|
|
avg_dollar_volume_20d=adv_20d,
|
|
)
|
|
|
|
passes, reason = evaluate_universe_gates(inputs, engine)
|
|
if not passes:
|
|
continue
|
|
|
|
scored.append((momentum, inputs))
|
|
|
|
# Sort by momentum descending, take top N.
|
|
scored.sort(key=lambda t: t[0], reverse=True)
|
|
top = scored[:top_n]
|
|
|
|
# Save full ranked universe to cache (pre-top_n) for future runs.
|
|
if cache is not None and scored:
|
|
cache.save_date(decision_date, [
|
|
{
|
|
"decision_date": decision_date.isoformat(),
|
|
"symbol": inp.symbol,
|
|
"momentum_12_1": inp.momentum_12_1,
|
|
"volatility_20d": inp.volatility_20d,
|
|
"avg_dollar_volume_20d": inp.avg_dollar_volume_20d,
|
|
"last_close": inp.last_close,
|
|
"last_bar_date": inp.last_bar_date.isoformat(),
|
|
"last_bar_timestamp_iso": inp.last_bar_timestamp.isoformat(),
|
|
}
|
|
for _, inp in scored
|
|
])
|
|
|
|
candidates: list[Candidate] = []
|
|
for rank, (_, inputs) in enumerate(top, start=1):
|
|
candidates.append(_build_candidate(inputs, engine, rank=rank, total_ranked=len(scored)))
|
|
|
|
if candidates:
|
|
logger.info(
|
|
"cross_sectional_momentum_rebalance",
|
|
decision_date=decision_date.isoformat(),
|
|
universe_scanned=len(scored),
|
|
top_n_emitted=len(candidates),
|
|
top_score=top[0][0] if top else None,
|
|
bottom_score=top[-1][0] if top else None,
|
|
)
|
|
|
|
return candidates
|
|
|
|
|
|
def _build_candidate(
|
|
inputs: CrossSectionalMomentumInputs,
|
|
engine: StrategyEngineConfig,
|
|
*,
|
|
rank: int,
|
|
total_ranked: int,
|
|
) -> Candidate:
|
|
holding_days = int(getattr(engine, "xsmom_holding_days", 21) or 21)
|
|
stop_pct = float(getattr(engine, "xsmom_stop_pct", 0.10) or 0.10)
|
|
target_pct = float(getattr(engine, "xsmom_target_pct", 0.30) or 0.30)
|
|
|
|
synthetic_atr = max(inputs.last_close * 0.02, 0.01)
|
|
stop_mult = stop_pct / 0.02 if stop_pct > 0 else 5.0
|
|
target_r = target_pct / stop_pct if stop_pct > 0 else 3.0
|
|
|
|
# Score: higher momentum → higher score, monotonic, clipped to [0, 0.99].
|
|
# Map momentum [0%, +60%] → score [0.5, 0.99] (linear).
|
|
score = 0.5 + 0.49 * min(1.0, max(0.0, inputs.momentum_12_1 / 0.60))
|
|
score_bucket = (
|
|
"high" if score >= 0.8
|
|
else "medium_high" if score >= 0.6
|
|
else "medium"
|
|
)
|
|
|
|
event_id = (
|
|
f"synth_xsmom_{inputs.symbol.lower()}_"
|
|
f"{inputs.decision_date.isoformat()}"
|
|
)
|
|
|
|
features = {
|
|
"xsmom_decision_date": inputs.decision_date.isoformat(),
|
|
"xsmom_last_close": inputs.last_close,
|
|
"xsmom_momentum_12_1": round(inputs.momentum_12_1, 6),
|
|
"xsmom_volatility_20d": round(inputs.volatility_20d, 6),
|
|
"xsmom_avg_dollar_volume_20d": inputs.avg_dollar_volume_20d,
|
|
"xsmom_rank": rank,
|
|
"xsmom_total_ranked": total_ranked,
|
|
"xsmom_stop_pct": stop_pct,
|
|
"xsmom_target_pct": target_pct,
|
|
"xsmom_holding_days": holding_days,
|
|
}
|
|
|
|
return Candidate(
|
|
event_id=event_id,
|
|
symbol=inputs.symbol,
|
|
source_symbol=inputs.symbol,
|
|
score=score,
|
|
sector="UNKNOWN",
|
|
event_type=CROSS_SECTIONAL_MOMENTUM_EVENT_TYPE,
|
|
event_timestamp=inputs.last_bar_timestamp,
|
|
event_date=inputs.decision_date,
|
|
filing_time_bucket="post_market",
|
|
timing_class="after_close",
|
|
reaction_date=inputs.decision_date,
|
|
execution_date=inputs.next_trading_date,
|
|
entry_price_est=inputs.last_close,
|
|
avg_dollar_volume=inputs.avg_dollar_volume_20d,
|
|
atr_14=synthetic_atr,
|
|
score_bucket=score_bucket,
|
|
engine_id=engine.engine_id,
|
|
entry_timing_policy="next_open",
|
|
trade_direction="long",
|
|
engine_max_holding_days=holding_days,
|
|
engine_risk_budget_pct=engine.engine_risk_budget_pct,
|
|
engine_capital_bucket_id=(
|
|
(engine.capital_bucket_id or engine.engine_id)
|
|
if engine.capital_bucket_allocation_pct is not None
|
|
else None
|
|
),
|
|
engine_capital_bucket_allocation_pct=engine.capital_bucket_allocation_pct,
|
|
engine_per_trade_risk_pct=engine.per_trade_risk_pct_override,
|
|
engine_target_1_r=target_r,
|
|
engine_target_1_fraction=1.0,
|
|
engine_stop_atr_multiplier=stop_mult,
|
|
engine_next_open_gap_cap_pct=engine.next_open_gap_cap_pct,
|
|
engine_use_reaction_day_low_stop=False,
|
|
engine_early_failure_close_below_entry_and_reaction_close=False,
|
|
# xsmom needs full holding period — explicitly disable NO_PROGRESS / trailing
|
|
# at the candidate level so v7.356 base config's tight gates don't fire on
|
|
# synthetic candidates in hybrid runs.
|
|
engine_early_failure_no_progress_days=0,
|
|
engine_early_failure_no_progress_r=0.0,
|
|
engine_early_failure_no_progress_fraction=0.0,
|
|
engine_trailing_model="none",
|
|
engine_trailing_warmup_days=999,
|
|
shadow_only=engine.shadow_only,
|
|
features=features,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter (reuse VolBreakout52w's cached-bars pattern)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class _SnapshotStoreBarAdapter:
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]]
|
|
_sorted_cache: dict[str, list[tuple[dt.date, dict[str, Any]]]] = field(default_factory=dict)
|
|
|
|
def _sorted_for(self, symbol: str) -> list[tuple[dt.date, dict[str, Any]]]:
|
|
sym_upper = symbol.upper()
|
|
cached = self._sorted_cache.get(sym_upper)
|
|
if cached is not None:
|
|
return cached
|
|
sym_bars = self.bars_by_symbol.get(sym_upper)
|
|
if not sym_bars:
|
|
self._sorted_cache[sym_upper] = []
|
|
return self._sorted_cache[sym_upper]
|
|
ordered = sorted(sym_bars.items(), key=lambda kv: kv[0])
|
|
self._sorted_cache[sym_upper] = ordered
|
|
return ordered
|
|
|
|
def get_bars_before(
|
|
self,
|
|
symbol: str,
|
|
as_of_date: dt.date,
|
|
lookback_days: int,
|
|
) -> list[tuple[dt.date, dict[str, Any]]]:
|
|
ordered = self._sorted_for(symbol)
|
|
if not ordered:
|
|
return []
|
|
eligible: list[tuple[dt.date, dict[str, Any]]] = []
|
|
for d, b in ordered:
|
|
if d >= as_of_date:
|
|
break
|
|
eligible.append((d, b))
|
|
return eligible[-lookback_days:]
|
|
|
|
|
|
__all__ = [
|
|
"CROSS_SECTIONAL_MOMENTUM_EVENT_TYPE",
|
|
"BarHistoryProvider",
|
|
"CrossSectionalMomentumInputs",
|
|
"_SnapshotStoreBarAdapter",
|
|
"build_candidates",
|
|
"compute_12_1_momentum",
|
|
"compute_20d_volatility",
|
|
"compute_avg_dollar_volume_20d",
|
|
"evaluate_universe_gates",
|
|
"is_rebalance_day",
|
|
]
|