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.
820 lines
31 KiB
Python
820 lines
31 KiB
Python
"""Unit tests for the EarningsRunup pre-event drift engine."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from libs.backtest.domain import LookaheadViolationError, StrategyEngineConfig
|
|
from libs.backtest.earnings_calendar import (
|
|
EarningsCalendarEntry,
|
|
PointInTimeEarningsCalendar,
|
|
)
|
|
from libs.backtest.earnings_runup import (
|
|
EARNINGS_RUNUP_EVENT_TYPE,
|
|
EarningsRunupTriggerInputs,
|
|
_PitCalendarUpcomingEarningsAdapter,
|
|
_SnapshotStoreBarAdapter,
|
|
build_earnings_runup_candidates,
|
|
evaluate_trigger,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fakes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeAttention:
|
|
def __init__(self, by_symbol_date: dict[tuple[str, dt.date], float | None]) -> None:
|
|
self.by_symbol_date = by_symbol_date
|
|
|
|
def get_zscore_20d(self, symbol: str, as_of_date: dt.date) -> float | None:
|
|
return self.by_symbol_date.get((symbol.upper(), as_of_date))
|
|
|
|
|
|
def _make_engine(**overrides: Any) -> StrategyEngineConfig:
|
|
base: dict[str, Any] = dict(
|
|
engine_id="earnings_runup_preevent_long",
|
|
event_types=[EARNINGS_RUNUP_EVENT_TYPE],
|
|
direction="long_only",
|
|
timing_class="after_close",
|
|
entry_timing_policy="next_open",
|
|
max_holding_days=7,
|
|
earnings_runup_enabled=True,
|
|
earnings_runup_days_to_earnings_min=3,
|
|
earnings_runup_days_to_earnings_max=7,
|
|
earnings_runup_attention_zscore_20d_min=1.5,
|
|
earnings_runup_dollar_volume_zscore_20d_min=1.0,
|
|
earnings_runup_calendar_buffer_days=1,
|
|
)
|
|
base.update(overrides)
|
|
return StrategyEngineConfig(**base)
|
|
|
|
|
|
def _generate_business_days(start: dt.date, count: int) -> list[dt.date]:
|
|
out: list[dt.date] = []
|
|
cursor = start
|
|
while len(out) < count:
|
|
if cursor.weekday() < 5:
|
|
out.append(cursor)
|
|
cursor = cursor + dt.timedelta(days=1)
|
|
return out
|
|
|
|
|
|
def _build_bars(
|
|
symbol: str,
|
|
trading_days: list[dt.date],
|
|
*,
|
|
base_volume: float,
|
|
spike_factor: float,
|
|
base_close: float = 100.0,
|
|
) -> dict[str, dict[dt.date, dict[str, Any]]]:
|
|
"""Build a bars-by-symbol-date dict with the LAST bar's $-volume = base*spike_factor.
|
|
|
|
Prior bars include small deterministic variance so sigma > 0 in the z-score.
|
|
"""
|
|
inner: dict[dt.date, dict[str, Any]] = {}
|
|
for i, d in enumerate(trading_days):
|
|
is_last = (i == len(trading_days) - 1)
|
|
if is_last:
|
|
volume = base_volume * spike_factor
|
|
else:
|
|
# +/- 10% sinusoidal perturbation, rounded so sigma > 0.
|
|
jitter = 1.0 + 0.1 * ((i % 5) - 2) / 2.0
|
|
volume = base_volume * jitter
|
|
inner[d] = {
|
|
"open": base_close,
|
|
"high": base_close,
|
|
"low": base_close,
|
|
"close": base_close,
|
|
"volume": volume,
|
|
}
|
|
return {symbol.upper(): inner}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# evaluate_trigger() — happy path + 3 negative cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _trigger_inputs(**overrides: Any) -> EarningsRunupTriggerInputs:
|
|
base: dict[str, Any] = dict(
|
|
symbol="AAPL",
|
|
decision_date=dt.date(2026, 4, 13), # Mon
|
|
next_trading_date=dt.date(2026, 4, 14),
|
|
upcoming_earnings_reaction_date=dt.date(2026, 4, 21),
|
|
days_to_earnings=5,
|
|
attention_zscore_20d=1.8,
|
|
dollar_volume_zscore_20d=1.2,
|
|
last_close_price=100.0,
|
|
avg_dollar_volume_20d=200_000_000.0,
|
|
last_bar_date=dt.date(2026, 4, 10), # prior Fri
|
|
last_bar_timestamp=dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc),
|
|
)
|
|
base.update(overrides)
|
|
return EarningsRunupTriggerInputs(**base)
|
|
|
|
|
|
def test_trigger_fires_when_all_three_conditions_met():
|
|
engine = _make_engine()
|
|
passes, reason = evaluate_trigger(_trigger_inputs(), engine)
|
|
assert passes is True
|
|
assert reason is None
|
|
|
|
|
|
def test_trigger_blocks_when_days_to_earnings_below_min():
|
|
engine = _make_engine()
|
|
passes, reason = evaluate_trigger(_trigger_inputs(days_to_earnings=2), engine)
|
|
assert passes is False
|
|
assert "days_to_earnings" in (reason or "")
|
|
|
|
|
|
def test_trigger_blocks_when_attention_zscore_below_min():
|
|
engine = _make_engine()
|
|
passes, reason = evaluate_trigger(_trigger_inputs(attention_zscore_20d=1.4), engine)
|
|
assert passes is False
|
|
assert "attention_z" in (reason or "")
|
|
|
|
|
|
def test_trigger_blocks_when_dollar_volume_zscore_below_min():
|
|
engine = _make_engine()
|
|
passes, reason = evaluate_trigger(_trigger_inputs(dollar_volume_zscore_20d=0.99), engine)
|
|
assert passes is False
|
|
assert "dollar_volume_z" in (reason or "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_earnings_runup_candidates() — end-to-end with fakes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_full_setup(
|
|
symbol: str = "AAPL",
|
|
*,
|
|
spike_factor: float = 5.0,
|
|
attention_z: float | None = 2.0,
|
|
earnings_offset_trading_days: int = 5,
|
|
days_of_history: int = 30,
|
|
):
|
|
# Decision day = the last day of generated trading days; bars go strictly before it.
|
|
trading_days = _generate_business_days(dt.date(2026, 3, 2), days_of_history + earnings_offset_trading_days + 2)
|
|
decision_date = trading_days[days_of_history] # T-1 close
|
|
next_trading_date = trading_days[days_of_history + 1]
|
|
earnings_reaction = trading_days[days_of_history + earnings_offset_trading_days]
|
|
|
|
# Bars for prior `days_of_history` days, ending on the day BEFORE decision_date.
|
|
prior_days = trading_days[:days_of_history]
|
|
bars = _build_bars(
|
|
symbol,
|
|
prior_days,
|
|
base_volume=1_000_000.0,
|
|
spike_factor=spike_factor,
|
|
)
|
|
bar_provider = _SnapshotStoreBarAdapter(bars_by_symbol=bars)
|
|
|
|
# PIT calendar: known earnings reaction date for the symbol.
|
|
pit_calendar = PointInTimeEarningsCalendar(
|
|
[
|
|
EarningsCalendarEntry(
|
|
symbol=symbol,
|
|
as_of_date=trading_days[0],
|
|
expected_reaction_date=earnings_reaction,
|
|
expected_event_date=earnings_reaction,
|
|
filing_time_bucket="post_market",
|
|
)
|
|
]
|
|
)
|
|
upcoming_provider = _PitCalendarUpcomingEarningsAdapter(
|
|
pit_calendar=pit_calendar,
|
|
trading_days=trading_days,
|
|
)
|
|
|
|
attention_provider = _FakeAttention(
|
|
{(symbol.upper(), decision_date): attention_z}
|
|
)
|
|
|
|
return {
|
|
"symbol": symbol,
|
|
"decision_date": decision_date,
|
|
"next_trading_date": next_trading_date,
|
|
"earnings_reaction": earnings_reaction,
|
|
"trading_days": trading_days,
|
|
"bar_provider": bar_provider,
|
|
"upcoming_provider": upcoming_provider,
|
|
"attention_provider": attention_provider,
|
|
"bars": bars,
|
|
}
|
|
|
|
|
|
def test_build_emits_candidate_for_eligible_symbol():
|
|
setup = _build_full_setup()
|
|
engine = _make_engine()
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert len(cands) == 1
|
|
cand = cands[0]
|
|
assert cand.event_type == EARNINGS_RUNUP_EVENT_TYPE
|
|
assert cand.symbol == setup["symbol"]
|
|
assert cand.engine_id == engine.engine_id
|
|
assert cand.execution_date == setup["next_trading_date"]
|
|
assert cand.engine_max_holding_days is not None
|
|
# days_to_earnings was 5; calendar_buffer 1 → max_holding_days = 5 - 1 = 4
|
|
assert cand.engine_max_holding_days == 4
|
|
assert cand.features["earnings_runup_days_to_earnings"] == 5
|
|
assert cand.features["earnings_runup_stop_pct"] == 0.04
|
|
assert cand.features["earnings_runup_target_pct"] == 0.08
|
|
|
|
|
|
def test_build_does_not_fire_when_attention_below_min():
|
|
setup = _build_full_setup(attention_z=0.5)
|
|
engine = _make_engine()
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert cands == []
|
|
|
|
|
|
def test_build_does_not_fire_when_dollar_volume_zscore_below_min():
|
|
# Spike factor of 1.0 (no spike) → z-score near 0
|
|
setup = _build_full_setup(spike_factor=1.0)
|
|
engine = _make_engine()
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert cands == []
|
|
|
|
|
|
def test_build_does_not_fire_when_days_to_earnings_outside_window():
|
|
# earnings_offset = 10 trading days → > max 7
|
|
setup = _build_full_setup(earnings_offset_trading_days=10)
|
|
engine = _make_engine()
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert cands == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lookahead defenses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_raises_lookahead_when_bar_date_equals_decision_date():
|
|
"""A bar dated on or after decision_date must trigger LookaheadViolationError."""
|
|
setup = _build_full_setup()
|
|
symbol = setup["symbol"]
|
|
decision_date = setup["decision_date"]
|
|
bars = setup["bars"]
|
|
# Inject a bar dated ON decision_date — this is the look-ahead violation.
|
|
bars[symbol.upper()][decision_date] = {
|
|
"open": 100.0, "high": 100.0, "low": 100.0, "close": 100.0,
|
|
"volume": 5_000_000.0,
|
|
}
|
|
bar_provider = _SnapshotStoreBarAdapter(bars_by_symbol=bars)
|
|
|
|
# Add a sentinel bar AFTER decision_date too, so the adapter's `< as_of_date` filter
|
|
# is the only thing keeping us safe. Then we manually subvert it.
|
|
class LeakyAdapter:
|
|
def get_bars_before(self, sym, as_of, lookback_days):
|
|
inner = bars[sym.upper()]
|
|
# Deliberately include the bar dated == decision_date.
|
|
return sorted(
|
|
[(d, b) for d, b in inner.items() if d <= as_of]
|
|
)[-lookback_days:]
|
|
|
|
engine = _make_engine()
|
|
with pytest.raises(LookaheadViolationError):
|
|
build_earnings_runup_candidates(
|
|
decision_date=decision_date,
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[symbol],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=LeakyAdapter(),
|
|
)
|
|
|
|
|
|
def test_build_raises_lookahead_when_explicit_assertion_violated():
|
|
"""Direct assertion path — feature timestamp >= cutoff must raise."""
|
|
from libs.backtest.earnings_runup import _assert_no_lookahead
|
|
|
|
decision_date = dt.date(2026, 4, 13)
|
|
# 09:30 ET on the decision day (= 13:30 UTC under EST; 13:30 UTC == 09:30 EST)
|
|
leaky_ts = dt.datetime(2026, 4, 13, 14, 30, tzinfo=dt.timezone.utc) # 10:30 ET
|
|
with pytest.raises(LookaheadViolationError):
|
|
_assert_no_lookahead("AAPL", decision_date, [leaky_ts])
|
|
|
|
|
|
def test_assert_no_lookahead_accepts_strictly_prior_timestamp():
|
|
from libs.backtest.earnings_runup import _assert_no_lookahead
|
|
|
|
decision_date = dt.date(2026, 4, 13)
|
|
safe_ts = dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc) # prior day close
|
|
# Should not raise
|
|
_assert_no_lookahead("AAPL", decision_date, [safe_ts])
|
|
|
|
|
|
def test_assert_no_lookahead_rejects_naive_timestamp():
|
|
from libs.backtest.earnings_runup import _assert_no_lookahead
|
|
|
|
decision_date = dt.date(2026, 4, 13)
|
|
naive_ts = dt.datetime(2026, 4, 10, 21, 0)
|
|
with pytest.raises(LookaheadViolationError):
|
|
_assert_no_lookahead("AAPL", decision_date, [naive_ts])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PIT earnings calendar respects as_of_date
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_pit_calendar_does_not_reveal_unannounced_future_earnings():
|
|
"""Earnings dates whose as_of_date is AFTER decision_date must not be visible."""
|
|
trading_days = _generate_business_days(dt.date(2026, 3, 2), 30)
|
|
decision_date = trading_days[10]
|
|
earnings_reaction_date = trading_days[15]
|
|
|
|
# Calendar entry was published AFTER decision_date — must be invisible.
|
|
pit_calendar = PointInTimeEarningsCalendar(
|
|
[
|
|
EarningsCalendarEntry(
|
|
symbol="AAPL",
|
|
as_of_date=trading_days[12], # > decision_date
|
|
expected_reaction_date=earnings_reaction_date,
|
|
expected_event_date=earnings_reaction_date,
|
|
filing_time_bucket="post_market",
|
|
)
|
|
]
|
|
)
|
|
adapter = _PitCalendarUpcomingEarningsAdapter(
|
|
pit_calendar=pit_calendar,
|
|
trading_days=trading_days,
|
|
)
|
|
result = adapter.get_next_reaction_date(
|
|
symbol="AAPL",
|
|
as_of_date=decision_date,
|
|
max_lookahead_calendar_days=14,
|
|
)
|
|
assert result is None
|
|
|
|
|
|
def test_pit_calendar_reveals_announced_future_earnings():
|
|
trading_days = _generate_business_days(dt.date(2026, 3, 2), 30)
|
|
decision_date = trading_days[10]
|
|
earnings_reaction_date = trading_days[15]
|
|
|
|
pit_calendar = PointInTimeEarningsCalendar(
|
|
[
|
|
EarningsCalendarEntry(
|
|
symbol="AAPL",
|
|
as_of_date=trading_days[5], # known well before decision_date
|
|
expected_reaction_date=earnings_reaction_date,
|
|
expected_event_date=earnings_reaction_date,
|
|
filing_time_bucket="post_market",
|
|
)
|
|
]
|
|
)
|
|
adapter = _PitCalendarUpcomingEarningsAdapter(
|
|
pit_calendar=pit_calendar,
|
|
trading_days=trading_days,
|
|
)
|
|
result = adapter.get_next_reaction_date(
|
|
symbol="AAPL",
|
|
as_of_date=decision_date,
|
|
max_lookahead_calendar_days=14,
|
|
)
|
|
assert result == earnings_reaction_date
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exit policy stub tests — verify candidate carries the exit configuration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_candidate_carries_stop_target_trailing_config_in_features():
|
|
setup = _build_full_setup()
|
|
engine = _make_engine(
|
|
earnings_runup_stop_pct=0.05,
|
|
earnings_runup_target_pct=0.10,
|
|
earnings_runup_trailing_activate_pct=0.06,
|
|
earnings_runup_trailing_giveback_pct=0.025,
|
|
)
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert len(cands) == 1
|
|
feats = cands[0].features
|
|
assert feats["earnings_runup_stop_pct"] == 0.05
|
|
assert feats["earnings_runup_target_pct"] == 0.10
|
|
assert feats["earnings_runup_trailing_activate_pct"] == 0.06
|
|
assert feats["earnings_runup_trailing_giveback_pct"] == 0.025
|
|
|
|
|
|
def test_candidate_max_holding_days_forces_flat_before_print():
|
|
"""Hard exit: max_holding_days = days_to_earnings - calendar_buffer_days (>=1)."""
|
|
# 4 trading days to earnings, buffer 1 → max_hold = 3
|
|
setup = _build_full_setup(earnings_offset_trading_days=4)
|
|
engine = _make_engine(earnings_runup_calendar_buffer_days=1)
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert len(cands) == 1
|
|
assert cands[0].engine_max_holding_days == 3
|
|
|
|
|
|
def test_candidate_max_holding_days_never_below_one():
|
|
setup = _build_full_setup(earnings_offset_trading_days=3)
|
|
engine = _make_engine(earnings_runup_calendar_buffer_days=5) # absurd buffer
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert len(cands) == 1
|
|
assert cands[0].engine_max_holding_days >= 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Behavioral exit tests — drive a synthetic position through simulate_exit and
|
|
# verify pct exits map correctly to STOP / TARGET / TIME outcomes.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_position_for_runup(
|
|
*,
|
|
entry_price: float = 100.0,
|
|
stop_pct: float = 0.04,
|
|
target_pct: float = 0.08,
|
|
days_held: int = 0,
|
|
) -> Any:
|
|
from libs.backtest.domain import (
|
|
Candidate,
|
|
ExitReason, # noqa: F401 re-exported for downstream tests
|
|
OpenPosition,
|
|
PlannedOrder,
|
|
)
|
|
# Mirror the candidate the production builder constructs.
|
|
stop_mult = stop_pct / 0.02
|
|
target_r = target_pct / stop_pct
|
|
synthetic_atr = entry_price * 0.02
|
|
cand = Candidate(
|
|
event_id="evt_runup_exit",
|
|
symbol="AAPL",
|
|
score=0.75,
|
|
sector="UNKNOWN",
|
|
event_type=EARNINGS_RUNUP_EVENT_TYPE,
|
|
event_timestamp=dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc),
|
|
event_date=dt.date(2026, 4, 13),
|
|
filing_time_bucket="post_market",
|
|
reaction_date=dt.date(2026, 4, 13),
|
|
execution_date=dt.date(2026, 4, 14),
|
|
entry_price_est=entry_price,
|
|
avg_dollar_volume=200_000_000.0,
|
|
atr_14=synthetic_atr,
|
|
score_bucket="medium_high",
|
|
engine_id="earnings_runup_preevent_long",
|
|
entry_timing_policy="next_open",
|
|
trade_direction="long",
|
|
engine_stop_atr_multiplier=stop_mult,
|
|
engine_target_1_r=target_r,
|
|
engine_target_1_fraction=1.0,
|
|
engine_max_holding_days=4,
|
|
)
|
|
stop_price = entry_price * (1.0 - stop_pct)
|
|
target_price = entry_price * (1.0 + target_pct)
|
|
plan = PlannedOrder(
|
|
candidate=cand,
|
|
shares=100,
|
|
entry_price_limit=entry_price,
|
|
stop_price=stop_price,
|
|
target_price=target_price,
|
|
risk_dollars=stop_pct * entry_price * 100,
|
|
event_date=cand.event_date,
|
|
timing_class="after_close",
|
|
engine_id=cand.engine_id,
|
|
entry_timing_policy="next_open",
|
|
shadow_only=False,
|
|
)
|
|
return OpenPosition(
|
|
position_id="pos_runup",
|
|
plan=plan,
|
|
entry_date=cand.execution_date,
|
|
entry_price=entry_price,
|
|
entry_fill_slippage_bps=10.0,
|
|
current_stop=stop_price,
|
|
target_price=target_price,
|
|
peak_price=entry_price,
|
|
shares_open=100,
|
|
shares_total=100,
|
|
days_held=days_held,
|
|
)
|
|
|
|
|
|
def _exec_config_for_exit_test() -> Any:
|
|
from libs.backtest.domain import ExecutionConfig
|
|
return ExecutionConfig(
|
|
entry_fill_model="next_open",
|
|
exit_fill_model="daily_bar_approximation",
|
|
slippage_bps_base=10.0,
|
|
commission_per_share=0.005,
|
|
same_bar_priority="stop_first_conservative",
|
|
max_holding_days=4,
|
|
)
|
|
|
|
|
|
def test_exit_stop_at_minus_4pct():
|
|
"""Long position with -4% stop must STOP-exit when bar.low <= 96.0."""
|
|
from libs.backtest.domain import ExitReason
|
|
from libs.backtest.execution import simulate_exit
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0, stop_pct=0.04)
|
|
# Bar drops to 95.5 → below the 96.0 stop → STOP exit.
|
|
bar = {"date": dt.date(2026, 4, 15), "open": 99.0, "high": 99.5, "low": 95.5, "close": 96.5, "volume": 1_000_000}
|
|
trade = simulate_exit(pos, bar, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
|
|
assert trade is not None
|
|
assert trade.exit_reason == ExitReason.STOP
|
|
|
|
|
|
def test_exit_target_at_plus_8pct():
|
|
"""Long position with +8% target must TARGET-exit when bar.high >= 108.0."""
|
|
from libs.backtest.domain import ExitReason
|
|
from libs.backtest.execution import simulate_exit
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0, target_pct=0.08)
|
|
bar = {"date": dt.date(2026, 4, 15), "open": 102.0, "high": 108.5, "low": 101.0, "close": 107.0, "volume": 1_000_000}
|
|
trade = simulate_exit(pos, bar, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
|
|
assert trade is not None
|
|
assert trade.exit_reason == ExitReason.TARGET
|
|
|
|
|
|
def test_exit_forced_max_hold_before_print():
|
|
"""When days_held >= max_holding_days and no stop/target, exit reason is TIME."""
|
|
from libs.backtest.domain import ExitReason
|
|
from libs.backtest.execution import simulate_exit
|
|
|
|
# max_holding_days = 2; position already held 2 days.
|
|
pos = _build_position_for_runup(entry_price=100.0, days_held=2)
|
|
cfg = _exec_config_for_exit_test()
|
|
cfg = cfg.model_copy(update={"max_holding_days": 2})
|
|
bar = {"date": dt.date(2026, 4, 15), "open": 102.0, "high": 103.0, "low": 99.0, "close": 102.5, "volume": 1_000_000}
|
|
trade = simulate_exit(pos, bar, cfg, dt.date(2026, 4, 15))
|
|
assert trade is not None
|
|
assert trade.exit_reason == ExitReason.TIME
|
|
|
|
|
|
def test_exit_trailing_giveback_after_activation():
|
|
"""Behavioral approximation of trailing exit: peak rises >+5%, then gives back >3%.
|
|
|
|
The standard execution machinery does not natively implement the
|
|
EarningsRunup pct-trailing model, so this test confirms the minimum
|
|
invariant — when a trailing stop is RAISED to a level above the static stop
|
|
and the bar's low touches it, the position exits via STOP. The trailing pct
|
|
config is preserved on the candidate features for future engine wiring.
|
|
"""
|
|
from libs.backtest.domain import ExitReason
|
|
from libs.backtest.execution import simulate_exit
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
# Manually move stop up to 105.0 (= activation at 105 with 0% giveback for the test).
|
|
raised = pos.model_copy(update={"current_stop": 105.0, "peak_price": 106.0})
|
|
bar = {"date": dt.date(2026, 4, 15), "open": 106.0, "high": 106.5, "low": 104.5, "close": 104.8, "volume": 1_000_000}
|
|
trade = simulate_exit(raised, bar, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
|
|
assert trade is not None
|
|
assert trade.exit_reason == ExitReason.STOP
|
|
# Confirm exit price is above original entry — i.e. the trailing stop captured profit.
|
|
assert trade.exit_price > pos.entry_price
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pct-trailing wiring tests — verify update_trailing_stop honors activation
|
|
# threshold and entry-relative giveback (EarningsRunup-style trailing).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _bar(*, low: float, high: float, open_: float | None = None, close: float | None = None) -> dict[str, Any]:
|
|
return {
|
|
"date": dt.date(2026, 4, 15),
|
|
"open": open_ if open_ is not None else (low + high) / 2,
|
|
"high": high,
|
|
"low": low,
|
|
"close": close if close is not None else (low + high) / 2,
|
|
"volume": 1_000_000,
|
|
}
|
|
|
|
|
|
def test_pct_trailing_does_not_arm_below_activation_threshold():
|
|
"""Position at +3% (activation=5%) → stop unchanged at -4% from entry."""
|
|
from libs.backtest.execution import update_trailing_stop
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
# Bar with high=103 (+3% peak), low=101 → not yet at activation (+5%)
|
|
update_trailing_stop(
|
|
pos, _bar(low=101.0, high=103.0),
|
|
trailing_model="pct_3",
|
|
warmup_days=0,
|
|
pct_activation=0.05,
|
|
pct_giveback=0.03,
|
|
)
|
|
# peak should track high=103
|
|
assert pos.peak_price == 103.0
|
|
# stop should NOT have moved up — still at -4% = 96.0
|
|
assert pos.current_stop == 96.0
|
|
|
|
|
|
def test_pct_trailing_arms_at_activation_threshold():
|
|
"""Peak hits +5% → stop raised to +2% (activation 5% - giveback 3% = 2%)."""
|
|
from libs.backtest.execution import update_trailing_stop
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
update_trailing_stop(
|
|
pos, _bar(low=102.0, high=105.0),
|
|
trailing_model="pct_3",
|
|
warmup_days=0,
|
|
pct_activation=0.05,
|
|
pct_giveback=0.03,
|
|
)
|
|
# peak = 105, stop = peak - giveback*entry = 105 - 0.03*100 = 102.0
|
|
assert pos.peak_price == 105.0
|
|
assert pos.current_stop == pytest.approx(102.0)
|
|
|
|
|
|
def test_pct_trailing_ratchets_up_with_continued_profit():
|
|
"""Peak rises to +7% → stop raised to +4% (peak - 3% of entry)."""
|
|
from libs.backtest.execution import update_trailing_stop
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
# First armed at peak=105 → stop=102
|
|
update_trailing_stop(
|
|
pos, _bar(low=102.0, high=105.0),
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=0.05, pct_giveback=0.03,
|
|
)
|
|
assert pos.current_stop == pytest.approx(102.0)
|
|
# Continued profit: peak now 107
|
|
update_trailing_stop(
|
|
pos, _bar(low=104.0, high=107.0),
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=0.05, pct_giveback=0.03,
|
|
)
|
|
assert pos.peak_price == 107.0
|
|
assert pos.current_stop == pytest.approx(104.0)
|
|
|
|
|
|
def test_pct_trailing_does_not_ratchet_down_on_pullback():
|
|
"""Peak 107 (stop 104) then peak holds at 107 while bar drops to 105 → stop stays at 104."""
|
|
from libs.backtest.execution import update_trailing_stop
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
# Establish peak=107, stop=104
|
|
update_trailing_stop(
|
|
pos, _bar(low=102.0, high=107.0),
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=0.05, pct_giveback=0.03,
|
|
)
|
|
assert pos.current_stop == pytest.approx(104.0)
|
|
assert pos.peak_price == 107.0
|
|
# Pullback: bar high=106 (below previous peak), low=105
|
|
update_trailing_stop(
|
|
pos, _bar(low=105.0, high=106.0),
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=0.05, pct_giveback=0.03,
|
|
)
|
|
# Peak unchanged; stop must stay (not ratchet down)
|
|
assert pos.peak_price == 107.0
|
|
assert pos.current_stop == pytest.approx(104.0)
|
|
|
|
|
|
def test_pct_trailing_reversal_triggers_stop_exit_with_profit():
|
|
"""Peak +7% → stop +4%; then bar low touches +3% → position exits via STOP at +4%."""
|
|
from libs.backtest.domain import ExitReason
|
|
from libs.backtest.execution import simulate_exit, update_trailing_stop
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
# Day 1: peak rises to 107, stop ratchets to 104
|
|
update_trailing_stop(
|
|
pos, _bar(low=102.0, high=107.0),
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=0.05, pct_giveback=0.03,
|
|
)
|
|
assert pos.current_stop == pytest.approx(104.0)
|
|
|
|
# Day 2: bar opens at 105.5, drops to 103.5 → trailing stop at 104.0 hit.
|
|
pos.days_held = 1
|
|
bar2 = _bar(low=103.5, high=105.5, open_=105.5, close=104.0)
|
|
# First update peak (no new high)
|
|
update_trailing_stop(
|
|
pos, bar2,
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=0.05, pct_giveback=0.03,
|
|
)
|
|
# Stop still at 104 (peak unchanged)
|
|
assert pos.current_stop == pytest.approx(104.0)
|
|
# Now simulate exit: bar.low=103.5 < stop=104 → STOP exit at 104
|
|
trade = simulate_exit(pos, bar2, _exec_config_for_exit_test(), dt.date(2026, 4, 15))
|
|
assert trade is not None
|
|
assert trade.exit_reason == ExitReason.STOP
|
|
# Exit price ~= 104.0 (the trailing stop, less small slippage), above entry → profit captured
|
|
assert trade.exit_price == pytest.approx(104.0, rel=0.005)
|
|
assert trade.exit_price > pos.entry_price
|
|
|
|
|
|
def test_pct_trailing_legacy_behavior_when_activation_none():
|
|
"""Regression: pct_3 with activation=None must trail unconditionally (legacy behavior)."""
|
|
from libs.backtest.execution import update_trailing_stop
|
|
|
|
pos = _build_position_for_runup(entry_price=100.0)
|
|
# Even at +1% peak, legacy pct_3 trails to peak * 0.97 = 100*0.97 = 97 (only ratchets if > current_stop)
|
|
# Note: position's current_stop starts at 96, and 101*0.97=97.97 > 96, so it should ratchet up.
|
|
update_trailing_stop(
|
|
pos, _bar(low=100.0, high=101.0),
|
|
trailing_model="pct_3", warmup_days=0,
|
|
pct_activation=None, pct_giveback=None,
|
|
)
|
|
assert pos.peak_price == 101.0
|
|
# Legacy formula: peak * (1 - 0.03) = 97.97 (ratchets up from 96)
|
|
assert pos.current_stop == pytest.approx(97.97)
|
|
|
|
|
|
def test_pct_trailing_via_engine_candidate_in_effective_exec():
|
|
"""End-to-end: candidate built from engine carries pct trailing fields into effective exec."""
|
|
from libs.backtest.execution import build_effective_execution_config
|
|
|
|
setup = _build_full_setup()
|
|
engine = _make_engine(
|
|
earnings_runup_stop_pct=0.04,
|
|
earnings_runup_target_pct=0.08,
|
|
earnings_runup_trailing_activate_pct=0.05,
|
|
earnings_runup_trailing_giveback_pct=0.03,
|
|
)
|
|
cands = build_earnings_runup_candidates(
|
|
decision_date=setup["decision_date"],
|
|
next_trading_date=setup["next_trading_date"],
|
|
universe_symbols=[setup["symbol"]],
|
|
engine=engine,
|
|
upcoming_earnings_provider=setup["upcoming_provider"],
|
|
attention_provider=setup["attention_provider"],
|
|
bar_provider=setup["bar_provider"],
|
|
)
|
|
assert len(cands) == 1
|
|
cand = cands[0]
|
|
# Candidate has the new pct trailing fields
|
|
assert cand.engine_trailing_model == "pct_3"
|
|
assert cand.engine_trailing_pct_activation == pytest.approx(0.05)
|
|
assert cand.engine_trailing_pct_giveback == pytest.approx(0.03)
|
|
|
|
# Build a minimal BacktestConfig and check effective_exec inherits these
|
|
from libs.backtest.domain import BacktestConfig, ExecutionConfig
|
|
base_cfg = BacktestConfig(
|
|
strategy_name="test",
|
|
dataset_snapshot_id="test",
|
|
execution=ExecutionConfig(max_holding_days=4),
|
|
)
|
|
eff = build_effective_execution_config(cand, base_cfg)
|
|
assert eff.trailing_model == "pct_3"
|
|
assert eff.trailing_pct_activation == pytest.approx(0.05)
|
|
assert eff.trailing_pct_giveback == pytest.approx(0.03)
|