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.
665 lines
26 KiB
Python
665 lines
26 KiB
Python
"""EarningsRunup pre-event drift engine.
|
|
|
|
Buys stocks 3-7 trading days before scheduled earnings when both attention and
|
|
dollar-volume z-scores rise above their 20-day baselines. Exits before the print.
|
|
|
|
This module is the *pure* logic — `BacktestRunner` calls into
|
|
``build_earnings_runup_candidates`` from a thin scheduling hook. The pure
|
|
function takes provider Protocols so it can be unit-tested with stubs.
|
|
|
|
Architectural choice (a): synthetic Candidate emission into the existing
|
|
`_scheduled_delayed_entries` queue, mirroring `_schedule_leader_follower_candidates`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import math
|
|
import statistics
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING, Any, Iterable, Protocol
|
|
|
|
if TYPE_CHECKING:
|
|
from libs.backtest.earnings_runup_cache import ErInputsCache
|
|
|
|
from libs.backtest.domain import (
|
|
Candidate,
|
|
LookaheadViolationError,
|
|
StrategyEngineConfig,
|
|
)
|
|
from libs.common.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
EARNINGS_RUNUP_EVENT_TYPE = "earnings_runup_preevent"
|
|
|
|
# Eastern-time market open used as the leakage cutoff. Decision date features
|
|
# must be timestamped strictly before this instant.
|
|
_ET_MARKET_OPEN = dt.time(9, 30)
|
|
# Naive UTC offset is fine for ordering checks because every timestamp we
|
|
# produce is normalized to the same convention (timezone-aware UTC).
|
|
_ET_OFFSET = dt.timedelta(hours=-5) # EST; DST is irrelevant for an ordering bound
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider Protocols (test-friendly seams)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class AttentionZscoreProvider(Protocol):
|
|
"""Returns the 20-day attention z-score for ``symbol`` as of ``as_of_date``.
|
|
|
|
Implementations must guarantee that no information from on/after ``as_of_date``
|
|
is incorporated into the returned value. ``None`` if data is unavailable.
|
|
"""
|
|
|
|
def get_zscore_20d(self, symbol: str, as_of_date: dt.date) -> float | None: ...
|
|
|
|
|
|
class UpcomingEarningsProvider(Protocol):
|
|
"""Returns the next-known scheduled earnings reaction date for ``symbol`` as of ``as_of_date``."""
|
|
|
|
def get_next_reaction_date(
|
|
self,
|
|
symbol: str,
|
|
as_of_date: dt.date,
|
|
max_lookahead_calendar_days: int,
|
|
) -> dt.date | None: ...
|
|
|
|
|
|
class BarHistoryProvider(Protocol):
|
|
"""Returns chronologically-ordered (date, bar_dict) pairs for ``symbol`` strictly before ``as_of_date``."""
|
|
|
|
def get_bars_before(
|
|
self,
|
|
symbol: str,
|
|
as_of_date: dt.date,
|
|
lookback_days: int,
|
|
) -> list[tuple[dt.date, dict[str, Any]]]: ...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapters: bridge BacktestRunner state to the Protocols above.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class _PitCalendarUpcomingEarningsAdapter:
|
|
"""Adapt PointInTimeEarningsCalendar to UpcomingEarningsProvider."""
|
|
|
|
pit_calendar: Any # libs.backtest.earnings_calendar.PointInTimeEarningsCalendar
|
|
trading_days: list[dt.date]
|
|
|
|
def get_next_reaction_date(
|
|
self,
|
|
symbol: str,
|
|
as_of_date: dt.date,
|
|
max_lookahead_calendar_days: int,
|
|
) -> dt.date | None:
|
|
try:
|
|
idx = self.trading_days.index(as_of_date)
|
|
except ValueError:
|
|
return None
|
|
# Look at every trading day strictly after as_of_date up to lookahead window.
|
|
cutoff = as_of_date + dt.timedelta(days=max_lookahead_calendar_days)
|
|
future = [d for d in self.trading_days[idx + 1:] if d <= cutoff]
|
|
if not future:
|
|
return None
|
|
result = self.pit_calendar.get_known_upcoming_reaction_dates(
|
|
as_of_date=as_of_date,
|
|
allowed_reaction_dates=future,
|
|
symbols=[symbol],
|
|
)
|
|
return result.get(symbol.upper())
|
|
|
|
|
|
@dataclass
|
|
class _SnapshotStoreBarAdapter:
|
|
"""Adapt SnapshotStore (or any object exposing ._bars) to BarHistoryProvider."""
|
|
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]]
|
|
|
|
def get_bars_before(
|
|
self,
|
|
symbol: str,
|
|
as_of_date: dt.date,
|
|
lookback_days: int,
|
|
) -> list[tuple[dt.date, dict[str, Any]]]:
|
|
sym_bars = self.bars_by_symbol.get(symbol.upper())
|
|
if not sym_bars:
|
|
return []
|
|
eligible = sorted(
|
|
(d, sym_bars[d])
|
|
for d in sym_bars
|
|
if d < as_of_date # strict; T-1 is the latest allowed
|
|
)
|
|
return eligible[-lookback_days:]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lookahead defense
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _decision_cutoff_utc(decision_date: dt.date) -> dt.datetime:
|
|
"""09:30 ET on decision_date, expressed as a UTC-aware timestamp.
|
|
|
|
Any feature timestamp >= this instant carries information from inside the
|
|
entry day and constitutes a look-ahead violation.
|
|
"""
|
|
et_naive = dt.datetime.combine(decision_date, _ET_MARKET_OPEN)
|
|
# Convert to UTC by subtracting the (negative) ET offset.
|
|
utc_naive = et_naive - _ET_OFFSET
|
|
return utc_naive.replace(tzinfo=dt.timezone.utc)
|
|
|
|
|
|
def _assert_no_lookahead(
|
|
symbol: str,
|
|
decision_date: dt.date,
|
|
feature_timestamps: Iterable[dt.datetime],
|
|
) -> None:
|
|
cutoff = _decision_cutoff_utc(decision_date)
|
|
for ts in feature_timestamps:
|
|
if ts is None:
|
|
continue
|
|
if ts.tzinfo is None:
|
|
raise LookaheadViolationError(
|
|
f"EarningsRunup feature timestamp for {symbol} is naive ({ts.isoformat()}); "
|
|
"all timestamps must be timezone-aware to compare against the cutoff"
|
|
)
|
|
if ts >= cutoff:
|
|
raise LookaheadViolationError(
|
|
f"EarningsRunup feature timestamp {ts.isoformat()} for {symbol} is "
|
|
f">= decision_date cutoff {cutoff.isoformat()}; this is a look-ahead violation"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Trigger evaluation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EarningsRunupTriggerInputs:
|
|
"""Bundle of T-1 close inputs for one (symbol, decision_date) candidate.
|
|
|
|
Every field whose source bears a timestamp must be timestamped strictly
|
|
before ``decision_date`` 09:30 ET; the candidate builder enforces this.
|
|
"""
|
|
|
|
symbol: str
|
|
decision_date: dt.date
|
|
next_trading_date: dt.date
|
|
upcoming_earnings_reaction_date: dt.date
|
|
days_to_earnings: int # trading-day count from decision_date to event
|
|
attention_zscore_20d: float
|
|
dollar_volume_zscore_20d: float
|
|
last_close_price: float
|
|
avg_dollar_volume_20d: float
|
|
last_bar_date: dt.date
|
|
last_bar_timestamp: dt.datetime # tz-aware
|
|
momentum_20d: float | None = None # (close[-1] / close[-21] - 1); None if insufficient bars
|
|
|
|
|
|
def evaluate_trigger(
|
|
inputs: EarningsRunupTriggerInputs,
|
|
engine: StrategyEngineConfig,
|
|
) -> tuple[bool, str | None]:
|
|
"""Pure trigger check. Returns (passes, reject_reason)."""
|
|
dmin = engine.earnings_runup_days_to_earnings_min
|
|
dmax = engine.earnings_runup_days_to_earnings_max
|
|
if dmin is not None and inputs.days_to_earnings < dmin:
|
|
return False, f"days_to_earnings {inputs.days_to_earnings} < min {dmin}"
|
|
if dmax is not None and inputs.days_to_earnings > dmax:
|
|
return False, f"days_to_earnings {inputs.days_to_earnings} > max {dmax}"
|
|
|
|
az_min = engine.earnings_runup_attention_zscore_20d_min
|
|
if az_min is not None and inputs.attention_zscore_20d < az_min:
|
|
return False, f"attention_z {inputs.attention_zscore_20d:.3f} < min {az_min}"
|
|
|
|
dvz_min = engine.earnings_runup_dollar_volume_zscore_20d_min
|
|
if dvz_min is not None and inputs.dollar_volume_zscore_20d < dvz_min:
|
|
return False, f"dollar_volume_z {inputs.dollar_volume_zscore_20d:.3f} < min {dvz_min}"
|
|
|
|
adv_min = engine.earnings_runup_min_avg_dollar_volume
|
|
if adv_min is not None and inputs.avg_dollar_volume_20d < adv_min:
|
|
return False, (
|
|
f"avg_dollar_volume_20d {inputs.avg_dollar_volume_20d:,.0f} "
|
|
f"< min {adv_min:,.0f}"
|
|
)
|
|
|
|
mom_min = engine.earnings_runup_momentum_20d_min
|
|
if mom_min is not None:
|
|
if inputs.momentum_20d is None:
|
|
return False, "momentum_20d insufficient bars"
|
|
if inputs.momentum_20d < mom_min:
|
|
return False, f"momentum_20d {inputs.momentum_20d:.3f} < min {mom_min}"
|
|
|
|
return True, None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dollar-volume z-score from bar history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _dollar_volume_zscore_20d(bars: list[tuple[dt.date, dict[str, Any]]]) -> tuple[float | None, float | None]:
|
|
"""Compute (zscore_20d, avg_dollar_volume_20d) from the last 21 bars.
|
|
|
|
The most recent bar (T-1) is the observation; the prior 20 form the baseline.
|
|
Returns (None, None) if insufficient history.
|
|
"""
|
|
if len(bars) < 21:
|
|
return None, None
|
|
recent = bars[-1][1]
|
|
prior_20 = bars[-21:-1]
|
|
recent_dv = float(recent.get("close", 0.0)) * float(recent.get("volume", 0.0))
|
|
prior_dv = [
|
|
float(b.get("close", 0.0)) * float(b.get("volume", 0.0))
|
|
for _, b in prior_20
|
|
]
|
|
if len(prior_dv) < 2:
|
|
return None, None
|
|
mu = statistics.fmean(prior_dv)
|
|
sigma = statistics.pstdev(prior_dv)
|
|
if sigma <= 0 or math.isnan(sigma):
|
|
return None, mu
|
|
z = (recent_dv - mu) / sigma
|
|
return z, mu
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def build_earnings_runup_candidates(
|
|
decision_date: dt.date,
|
|
next_trading_date: dt.date,
|
|
universe_symbols: Iterable[str],
|
|
engine: StrategyEngineConfig,
|
|
upcoming_earnings_provider: UpcomingEarningsProvider,
|
|
attention_provider: AttentionZscoreProvider,
|
|
bar_provider: BarHistoryProvider,
|
|
*,
|
|
cache: "ErInputsCache | None" = None,
|
|
) -> list[Candidate]:
|
|
"""Construct synthetic EarningsRunup candidates for ``next_trading_date`` execution.
|
|
|
|
Decision logic runs at T-1 close (=decision_date close); orders fill at
|
|
T+1 next_open. Every input must satisfy ``timestamp < decision_date 09:30 ET``.
|
|
|
|
If ``cache`` is provided, per-symbol-per-day inputs (attention z, dollar
|
|
volume z, momentum, days-to-earnings, reaction date) are looked up from
|
|
disk on HIT and re-thresholded by the current engine's filters. On MISS,
|
|
the full universe is scanned and the cache populated for future runs.
|
|
Threshold filters are NOT in the cache key, so a single cache file serves
|
|
runs with different ER threshold configs.
|
|
"""
|
|
if not engine.earnings_runup_enabled:
|
|
return []
|
|
|
|
dmax = int(engine.earnings_runup_days_to_earnings_max or 0)
|
|
if dmax <= 0:
|
|
return []
|
|
|
|
# HIT path: re-apply engine threshold filters to cached rows.
|
|
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
|
|
)
|
|
|
|
# MISS path: populate cache for ALL upcoming-earnings symbols within the
|
|
# fixed wide window (CACHE_MAX_DAYS_TO_EARNINGS), independent of this
|
|
# engine's dmax. Threshold filters are applied AFTER caching so future
|
|
# runs with different thresholds get a clean HIT.
|
|
from libs.backtest.earnings_runup_cache import CACHE_MAX_DAYS_TO_EARNINGS
|
|
population_dmax = CACHE_MAX_DAYS_TO_EARNINGS
|
|
|
|
# PIT calendar adapter looks ``population_dmax`` trading days ahead. Pad in
|
|
# calendar days to cover weekends/holidays.
|
|
calendar_lookahead = population_dmax * 2 + 7
|
|
|
|
cached_buffer: list[dict[str, Any]] = []
|
|
inputs_list: list[EarningsRunupTriggerInputs] = []
|
|
seen_symbols: set[str] = set()
|
|
|
|
for raw_symbol in universe_symbols:
|
|
symbol = str(raw_symbol).strip().upper()
|
|
if not symbol or symbol in seen_symbols:
|
|
continue
|
|
seen_symbols.add(symbol)
|
|
|
|
upcoming_reaction = upcoming_earnings_provider.get_next_reaction_date(
|
|
symbol=symbol,
|
|
as_of_date=decision_date,
|
|
max_lookahead_calendar_days=calendar_lookahead,
|
|
)
|
|
if upcoming_reaction is None:
|
|
continue
|
|
|
|
# Trading-day distance from decision_date close to event reaction.
|
|
days_to_earnings = _trading_days_between(
|
|
decision_date, upcoming_reaction, getattr(upcoming_earnings_provider, "trading_days", None)
|
|
)
|
|
if days_to_earnings is None:
|
|
continue
|
|
# Skip symbols whose earnings are beyond the cache population window.
|
|
if days_to_earnings > population_dmax:
|
|
continue
|
|
|
|
bars = bar_provider.get_bars_before(symbol, decision_date, lookback_days=65)
|
|
if not bars:
|
|
continue
|
|
last_bar_date, last_bar = bars[-1]
|
|
|
|
# Strict T-1 check: most recent allowed bar is the day BEFORE decision_date.
|
|
if last_bar_date >= decision_date:
|
|
raise LookaheadViolationError(
|
|
f"EarningsRunup bar for {symbol} on {last_bar_date.isoformat()} is not "
|
|
f"strictly before decision_date {decision_date.isoformat()}"
|
|
)
|
|
|
|
last_bar_ts = _bar_close_timestamp(last_bar_date)
|
|
# Run the lookahead assertion early — it MUST be on the hot path.
|
|
_assert_no_lookahead(symbol, decision_date, [last_bar_ts])
|
|
|
|
dv_z, adv_20d = _dollar_volume_zscore_20d(bars)
|
|
if dv_z is None or adv_20d is None:
|
|
continue
|
|
|
|
attention_z = attention_provider.get_zscore_20d(symbol, decision_date)
|
|
if attention_z is None:
|
|
continue
|
|
|
|
last_close = float(last_bar.get("close", 0.0))
|
|
if last_close <= 0:
|
|
continue
|
|
|
|
# 20-day price momentum: (close[-1] / close[-21] - 1). Requires ≥ 21 bars.
|
|
momentum_20d: float | None = None
|
|
if len(bars) >= 21:
|
|
close_20d_ago = float(bars[-21][1].get("close", 0.0))
|
|
if close_20d_ago > 0:
|
|
momentum_20d = last_close / close_20d_ago - 1.0
|
|
|
|
inputs = EarningsRunupTriggerInputs(
|
|
symbol=symbol,
|
|
decision_date=decision_date,
|
|
next_trading_date=next_trading_date,
|
|
upcoming_earnings_reaction_date=upcoming_reaction,
|
|
days_to_earnings=days_to_earnings,
|
|
attention_zscore_20d=float(attention_z),
|
|
dollar_volume_zscore_20d=float(dv_z),
|
|
last_close_price=last_close,
|
|
avg_dollar_volume_20d=float(adv_20d),
|
|
last_bar_date=last_bar_date,
|
|
last_bar_timestamp=last_bar_ts,
|
|
momentum_20d=momentum_20d,
|
|
)
|
|
inputs_list.append(inputs)
|
|
|
|
# Buffer for cache regardless of trigger pass/fail — thresholds applied
|
|
# at read time so the same cache serves different configs.
|
|
cached_buffer.append({
|
|
"decision_date": decision_date.isoformat(),
|
|
"symbol": symbol,
|
|
"days_to_earnings": int(days_to_earnings),
|
|
"reaction_date": upcoming_reaction.isoformat(),
|
|
"momentum_20d": momentum_20d if momentum_20d is not None else float("nan"),
|
|
"dollar_volume_zscore_20d": float(dv_z),
|
|
"attention_zscore_20d": float(attention_z),
|
|
"avg_dollar_volume_20d": float(adv_20d),
|
|
"last_close": last_close,
|
|
"last_bar_date": last_bar_date.isoformat(),
|
|
"last_bar_timestamp_iso": last_bar_ts.isoformat(),
|
|
})
|
|
|
|
# Save to cache (only on MISS path).
|
|
if cache is not None:
|
|
cache.save_date(decision_date, cached_buffer)
|
|
|
|
# Apply engine threshold filters and emit candidates.
|
|
candidates: list[Candidate] = []
|
|
for inputs in inputs_list:
|
|
passes, reason = evaluate_trigger(inputs, engine)
|
|
if not passes:
|
|
logger.debug(
|
|
"earnings_runup_trigger_skipped",
|
|
symbol=inputs.symbol,
|
|
decision_date=decision_date.isoformat(),
|
|
reason=reason,
|
|
)
|
|
continue
|
|
candidate = _build_candidate_from_inputs(inputs, engine)
|
|
candidates.append(candidate)
|
|
|
|
return candidates
|
|
|
|
|
|
def _build_candidates_from_cache(
|
|
rows: list[dict[str, Any]],
|
|
decision_date: dt.date,
|
|
next_trading_date: dt.date,
|
|
engine: StrategyEngineConfig,
|
|
) -> list[Candidate]:
|
|
"""Re-apply engine threshold filters to cached rows + emit candidates.
|
|
|
|
Look-ahead defense: every cached row's ``last_bar_date`` MUST be strictly
|
|
before ``decision_date``. This mirrors xsmom's HIT-path defense.
|
|
"""
|
|
candidates: list[Candidate] = []
|
|
for row in rows:
|
|
symbol = str(row["symbol"])
|
|
last_bar_date = dt.date.fromisoformat(str(row["last_bar_date"]))
|
|
if last_bar_date >= decision_date:
|
|
raise LookaheadViolationError(
|
|
f"EarningsRunup cached bar for {symbol} on "
|
|
f"{last_bar_date.isoformat()} is not strictly before "
|
|
f"decision_date {decision_date.isoformat()}"
|
|
)
|
|
last_bar_ts = dt.datetime.fromisoformat(str(row["last_bar_timestamp_iso"]))
|
|
_assert_no_lookahead(symbol, decision_date, [last_bar_ts])
|
|
|
|
# Handle NaN-encoded None for momentum_20d
|
|
mom_raw = row.get("momentum_20d")
|
|
if mom_raw is None or (isinstance(mom_raw, float) and math.isnan(mom_raw)):
|
|
momentum_20d: float | None = None
|
|
else:
|
|
momentum_20d = float(mom_raw)
|
|
|
|
inputs = EarningsRunupTriggerInputs(
|
|
symbol=symbol,
|
|
decision_date=decision_date,
|
|
next_trading_date=next_trading_date,
|
|
upcoming_earnings_reaction_date=dt.date.fromisoformat(str(row["reaction_date"])),
|
|
days_to_earnings=int(row["days_to_earnings"]),
|
|
attention_zscore_20d=float(row["attention_zscore_20d"]),
|
|
dollar_volume_zscore_20d=float(row["dollar_volume_zscore_20d"]),
|
|
last_close_price=float(row["last_close"]),
|
|
avg_dollar_volume_20d=float(row["avg_dollar_volume_20d"]),
|
|
last_bar_date=last_bar_date,
|
|
last_bar_timestamp=last_bar_ts,
|
|
momentum_20d=momentum_20d,
|
|
)
|
|
passes, reason = evaluate_trigger(inputs, engine)
|
|
if not passes:
|
|
logger.debug(
|
|
"earnings_runup_trigger_skipped",
|
|
symbol=symbol,
|
|
decision_date=decision_date.isoformat(),
|
|
reason=reason,
|
|
cache_hit=True,
|
|
)
|
|
continue
|
|
candidates.append(_build_candidate_from_inputs(inputs, engine))
|
|
return candidates
|
|
|
|
|
|
def _trading_days_between(
|
|
decision_date: dt.date,
|
|
target_date: dt.date,
|
|
trading_days: list[dt.date] | None,
|
|
) -> int | None:
|
|
if trading_days:
|
|
try:
|
|
i0 = trading_days.index(decision_date)
|
|
i1 = trading_days.index(target_date)
|
|
return i1 - i0
|
|
except ValueError:
|
|
return None
|
|
# Fallback: business-day approximation if trading_days not available.
|
|
# Counts weekdays strictly after decision_date up to target_date.
|
|
if target_date <= decision_date:
|
|
return None
|
|
count = 0
|
|
cursor = decision_date
|
|
while cursor < target_date:
|
|
cursor = cursor + dt.timedelta(days=1)
|
|
if cursor.weekday() < 5:
|
|
count += 1
|
|
return count
|
|
|
|
|
|
def _bar_close_timestamp(bar_date: dt.date) -> dt.datetime:
|
|
"""Timestamp the daily-close bar at 16:00 ET on its trading day, in 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)
|
|
|
|
|
|
def _build_candidate_from_inputs(
|
|
inputs: EarningsRunupTriggerInputs,
|
|
engine: StrategyEngineConfig,
|
|
) -> Candidate:
|
|
# Hard hold: forced flat by close of the trading day BEFORE the print.
|
|
buffer = max(0, int(engine.earnings_runup_calendar_buffer_days))
|
|
max_holding_days = max(1, inputs.days_to_earnings - buffer)
|
|
|
|
# Translate pct exits → existing ATR-multiplier / R-multiple machinery in
|
|
# `libs.backtest.allocator.compute_stop_price` and `compute_target_price`.
|
|
# Synthetic ATR := 2% of last close (the same fallback compute_stop_price
|
|
# uses when atr_14 is missing, but we materialize it so target/stop
|
|
# downstream consumers see a non-null ATR).
|
|
# stop_atr_multiplier := stop_pct / 0.02 → produces a stop_distance of
|
|
# ``stop_pct * close`` for the default dynamic_scaler == 1.0.
|
|
# target_1_r := target_pct / stop_pct → fixed-R target sits at +target_pct.
|
|
# target_1_fraction := engine.earnings_runup_target_fraction (default 1.0 = full exit;
|
|
# 0.5 = exit half / let half run with trailing + breakeven floor).
|
|
# Trailing pct exits ARE wired end-to-end via engine_trailing_pct_activation /
|
|
# engine_trailing_pct_giveback on Candidate → ExecutionConfig overrides →
|
|
# update_trailing_stop activation gate. The trailing model name encodes the
|
|
# giveback for legacy logging; the engine fields override the actual stop.
|
|
synthetic_atr = max(inputs.last_close_price * 0.02, 0.01)
|
|
stop_pct = float(engine.earnings_runup_stop_pct)
|
|
target_pct = float(engine.earnings_runup_target_pct)
|
|
stop_mult = stop_pct / 0.02 if stop_pct > 0 else 2.0
|
|
target_r = target_pct / stop_pct if stop_pct > 0 else 2.0
|
|
|
|
# Pct-trailing wiring: pick a "pct_<int>" model whose name reflects the
|
|
# giveback (used as fallback if pct_giveback is somehow None at runtime,
|
|
# and shown in trade diagnostics).
|
|
trailing_giveback = float(engine.earnings_runup_trailing_giveback_pct)
|
|
trailing_activation = float(engine.earnings_runup_trailing_activate_pct)
|
|
trailing_giveback_name = max(1, int(round(trailing_giveback * 100)))
|
|
pct_trailing_model = f"pct_{trailing_giveback_name}"
|
|
|
|
# Score is a deterministic function of the two z-scores so it ranks
|
|
# candidates without leaking future information.
|
|
z_sum = inputs.attention_zscore_20d + inputs.dollar_volume_zscore_20d
|
|
score = 0.5 + 0.05 * z_sum
|
|
score = max(0.0, min(0.99, score))
|
|
score_bucket = (
|
|
"high" if score >= 0.8
|
|
else "medium_high" if score >= 0.6
|
|
else "medium"
|
|
)
|
|
|
|
event_id = (
|
|
f"synth_earnings_runup_{inputs.symbol.lower()}_"
|
|
f"{inputs.decision_date.isoformat()}"
|
|
)
|
|
|
|
features = {
|
|
"earnings_runup_decision_date": inputs.decision_date.isoformat(),
|
|
"earnings_runup_upcoming_reaction_date": inputs.upcoming_earnings_reaction_date.isoformat(),
|
|
"earnings_runup_days_to_earnings": inputs.days_to_earnings,
|
|
"earnings_runup_attention_zscore_20d": round(inputs.attention_zscore_20d, 4),
|
|
"earnings_runup_dollar_volume_zscore_20d": round(inputs.dollar_volume_zscore_20d, 4),
|
|
"earnings_runup_avg_dollar_volume_20d": inputs.avg_dollar_volume_20d,
|
|
"earnings_runup_max_holding_days": max_holding_days,
|
|
"earnings_runup_stop_pct": engine.earnings_runup_stop_pct,
|
|
"earnings_runup_target_pct": engine.earnings_runup_target_pct,
|
|
"earnings_runup_target_fraction": float(engine.earnings_runup_target_fraction),
|
|
"earnings_runup_trailing_activate_pct": engine.earnings_runup_trailing_activate_pct,
|
|
"earnings_runup_trailing_giveback_pct": engine.earnings_runup_trailing_giveback_pct,
|
|
}
|
|
|
|
return Candidate(
|
|
event_id=event_id,
|
|
symbol=inputs.symbol,
|
|
source_symbol=inputs.symbol,
|
|
score=score,
|
|
sector="UNKNOWN",
|
|
event_type=EARNINGS_RUNUP_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_price,
|
|
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=max_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,
|
|
# Map pct-based EarningsRunup exits → engine_*-prefixed overrides on the candidate.
|
|
engine_target_1_r=target_r,
|
|
engine_target_1_fraction=float(engine.earnings_runup_target_fraction),
|
|
# Trailing: prefer engine-level override; fall back to pct trailing model
|
|
# derived from the engine's giveback config so the activation-gated
|
|
# trailing path is used in update_trailing_stop.
|
|
engine_trailing_model=engine.trailing_model_override or pct_trailing_model,
|
|
engine_trailing_warmup_days=(
|
|
engine.trailing_warmup_days_override
|
|
if engine.trailing_warmup_days_override is not None
|
|
else 0
|
|
),
|
|
engine_trailing_pct_activation=trailing_activation,
|
|
engine_trailing_pct_giveback=trailing_giveback,
|
|
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,
|
|
engine_early_failure_no_progress_days=engine.early_failure_no_progress_days_override,
|
|
engine_early_failure_no_progress_r=engine.early_failure_no_progress_r_override,
|
|
engine_early_failure_no_progress_fraction=engine.early_failure_no_progress_fraction_override,
|
|
shadow_only=engine.shadow_only,
|
|
features=features,
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"EARNINGS_RUNUP_EVENT_TYPE",
|
|
"AttentionZscoreProvider",
|
|
"BarHistoryProvider",
|
|
"EarningsRunupTriggerInputs",
|
|
"UpcomingEarningsProvider",
|
|
"_PitCalendarUpcomingEarningsAdapter",
|
|
"_SnapshotStoreBarAdapter",
|
|
"build_earnings_runup_candidates",
|
|
"evaluate_trigger",
|
|
]
|