yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
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.
main
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'main'
${ noResults }
fithia2/libs/backtest/low_vol_anomaly.py

512 lines
18 KiB
Python
Raw Permalink Blame History

"""LowVolAnomaly — Frazzini-Pedersen (2014) "Betting Against Beta" / Low-Vol Anomaly.
On the first trading day of each calendar month, rank universe by ASCENDING
realized volatility over `lookback_days` (default 60). Buy the top-N
LOWEST-vol symbols at next_open, hold for `holding_days` trading days, then
force-flat.
Long-only, large-cap-friendly. The lowest-vol slice of a liquid universe is
expected to be dominated by stable mega-caps — explicitly orthogonal to the
small-cap moonshot tail of the form4 silo (HYMC etc).
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.
Mirrors libs.backtest.cross_sectional_momentum but ranks on inverse-vol.
"""
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.low_vol_cache import LowVolRankCache
from libs.backtest.domain import (
Candidate,
LookaheadViolationError,
StrategyEngineConfig,
)
from libs.common.logging import get_logger
logger = get_logger(__name__)
LOW_VOL_ANOMALY_EVENT_TYPE = "low_vol_anomaly"
_ET_MARKET_OPEN = dt.time(9, 30)
_ET_OFFSET = dt.timedelta(hours=-5)
# ---------------------------------------------------------------------------
# Provider Protocol (shared shape with xsmom / 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 _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"LowVolAnomaly {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."""
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_realized_volatility(
bars: list[tuple[dt.date, dict[str, Any]]],
*,
lookback_days: int = 60,
) -> tuple[float | None, list[dt.date]]:
"""Stdev of daily simple returns over the trailing ``lookback_days`` bars.
Returns (volatility, used_dates). None when insufficient history.
"""
if len(bars) < lookback_days + 1:
return None, []
window = bars[-(lookback_days + 1):]
closes = [float(b.get("close", 0.0)) for _, b in window]
if any(c <= 0 for c in closes):
return None, []
rets: list[float] = []
for i in range(1, len(closes)):
rets.append((closes[i] / closes[i - 1]) - 1.0)
if len(rets) < 2:
return None, []
vol = float(statistics.stdev(rets))
used_dates = [d for d, _ in window]
return vol, used_dates
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 LowVolAnomalyInputs:
symbol: str
decision_date: dt.date
next_trading_date: dt.date
last_bar_date: dt.date
last_bar_timestamp: dt.datetime
last_close: float
realized_volatility: float
avg_dollar_volume_20d: float
def evaluate_universe_gates(
inputs: LowVolAnomalyInputs,
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"LowVolAnomaly {inputs.symbol}: last_bar_date "
f"{inputs.last_bar_date.isoformat()} not strictly before decision_date"
)
price_min = float(getattr(engine, "lowvol_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, "lowvol_min_avg_dollar_volume", 5_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_min = float(getattr(engine, "lowvol_volatility_min", 0.0) or 0.0)
if vol_min > 0 and inputs.realized_volatility < vol_min:
return False, f"realized_volatility {inputs.realized_volatility:.6f} < min {vol_min:.6f}"
return True, None
# ---------------------------------------------------------------------------
# Build candidates (top-N ascending vol 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).
Sorts ASCENDING by realized_volatility (lowest vol first).
"""
rows_sorted = sorted(rows, key=lambda r: float(r["realized_volatility"]))
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 = LowVolAnomalyInputs(
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"]),
realized_volatility=float(row["realized_volatility"]),
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(
"low_vol_anomaly_rebalance",
decision_date=decision_date.isoformat(),
universe_scanned=total_ranked,
top_n_emitted=len(candidates),
top_score=float(top[0]["realized_volatility"]) if top else None,
bottom_score=float(top[-1]["realized_volatility"]) 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: "LowVolRankCache | None" = None,
) -> list[Candidate]:
"""Emit top-N synthetic LOW-VOL candidates on rebalance days only.
On non-rebalance days, returns []. The universe is scanned once per
rebalance day, ranked ASCENDING by realized_volatility, and the top-N
(lowest vol) 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, "lowvol_enabled", False):
return []
if next_trading_date <= decision_date:
raise LookaheadViolationError(
f"LowVolAnomaly next_trading_date {next_trading_date.isoformat()} "
f"must be strictly after decision_date {decision_date.isoformat()}"
)
lookback = int(getattr(engine, "lowvol_lookback_days", 60) or 60)
top_n = int(getattr(engine, "lowvol_top_n", 20) or 20)
vol_min = float(getattr(engine, "lowvol_volatility_min", 0.0) or 0.0)
max_mcap = getattr(engine, "lowvol_max_market_cap_proxy", None)
# max_market_cap_proxy is None by default — low-vol = large-cap-friendly.
# The runner-level filter (engine.max_market_cap_proxy) still applies.
# +5 day buffer for off-by-one and weekend gaps
fetch_lookback = lookback + 25
scored: list[tuple[float, LowVolAnomalyInputs]] = []
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"LowVolAnomaly 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
vol, vol_used_dates = compute_realized_volatility(bars, lookback_days=lookback)
if vol is None:
continue
if vol_min > 0 and vol < vol_min:
continue
_assert_strictly_before(symbol, decision_date, vol_used_dates)
adv_20d = compute_avg_dollar_volume_20d(bars)
inputs = LowVolAnomalyInputs(
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,
realized_volatility=vol,
avg_dollar_volume_20d=adv_20d,
)
passes, _reason = evaluate_universe_gates(inputs, engine)
if not passes:
continue
scored.append((vol, inputs))
# Sort by volatility ASCENDING (lowest vol = best), take top N.
scored.sort(key=lambda t: t[0])
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,
"realized_volatility": inp.realized_volatility,
"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(
"low_vol_anomaly_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: LowVolAnomalyInputs,
engine: StrategyEngineConfig,
*,
rank: int,
total_ranked: int,
) -> Candidate:
holding_days = int(getattr(engine, "lowvol_holding_days", 21) or 21)
stop_pct = float(getattr(engine, "lowvol_stop_pct", 0.10) or 0.10)
target_pct = float(getattr(engine, "lowvol_target_pct", 0.20) or 0.20)
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 2.0
# Score: lower vol → higher score, monotonic, clipped to [0.5, 0.99].
# Map vol [0%, 4%] → score [0.99, 0.5] (linear, inverted).
vol_ref = 0.04
inv = 1.0 - min(1.0, max(0.0, inputs.realized_volatility / vol_ref))
score = 0.5 + 0.49 * inv
score_bucket = (
"high" if score >= 0.8
else "medium_high" if score >= 0.6
else "medium"
)
event_id = (
f"synth_lowvol_{inputs.symbol.lower()}_"
f"{inputs.decision_date.isoformat()}"
)
features = {
"lowvol_decision_date": inputs.decision_date.isoformat(),
"lowvol_last_close": inputs.last_close,
"lowvol_realized_volatility": round(inputs.realized_volatility, 6),
"lowvol_avg_dollar_volume_20d": inputs.avg_dollar_volume_20d,
"lowvol_rank": rank,
"lowvol_total_ranked": total_ranked,
"lowvol_stop_pct": stop_pct,
"lowvol_target_pct": target_pct,
"lowvol_holding_days": holding_days,
}
return Candidate(
event_id=event_id,
symbol=inputs.symbol,
source_symbol=inputs.symbol,
score=score,
sector="UNKNOWN",
event_type=LOW_VOL_ANOMALY_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,
# low-vol needs full holding period — explicitly disable NO_PROGRESS / trailing
# so v9.x base config's tight gates don't fire on synthetic candidates.
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 xsmom'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__ = [
"LOW_VOL_ANOMALY_EVENT_TYPE",
"BarHistoryProvider",
"LowVolAnomalyInputs",
"_SnapshotStoreBarAdapter",
"build_candidates",
"compute_avg_dollar_volume_20d",
"compute_realized_volatility",
"evaluate_universe_gates",
"is_rebalance_day",
]
Reference in New Issue View Git Blame Copy Permalink