"""Unit tests for the PeerSympathy engine.""" from __future__ import annotations import datetime as dt import math 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 _PitCalendarUpcomingEarningsAdapter from libs.backtest.peer_sympathy import ( PEER_SYMPATHY_EVENT_TYPE, LeaderPrint, PeerSympathyTriggerInputs, _assert_correlation_window_safe, build_peer_sympathy_candidates, compute_correlation, evaluate_trigger, ) # --------------------------------------------------------------------------- # Fakes & helpers # --------------------------------------------------------------------------- class _FakeBarHistory: """In-memory BarHistoryProvider stub. ``bars`` is keyed by symbol → {date: bar_dict}. ``get_bars_before`` returns the chronologically-ordered subset strictly before ``as_of_date``. """ def __init__(self, bars: dict[str, dict[dt.date, dict[str, Any]]]) -> None: self.bars = {k.upper(): dict(v) for k, v in bars.items()} 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.get(symbol.upper()) if not sym_bars: return [] eligible = sorted( (d, sym_bars[d]) for d in sym_bars if d < as_of_date ) return eligible[-lookback_days:] class _StaticPeerResolver: def __init__(self, peers_by_leader: dict[str, list[str]]) -> None: self.peers_by_leader = {k.upper(): list(v) for k, v in peers_by_leader.items()} def peers_for_leader( self, engine: StrategyEngineConfig, leader_symbol: str, leader_sector: str, ) -> list[str]: return self.peers_by_leader.get(leader_symbol.upper(), []) 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 _make_engine(**overrides: Any) -> StrategyEngineConfig: base: dict[str, Any] = dict( engine_id="peer_sympathy_long", event_types=[PEER_SYMPATHY_EVENT_TYPE], direction="long_only", timing_class="after_close", entry_timing_policy="next_open", peer_sympathy_enabled=True, peer_sympathy_leader_event_types=["earnings_release", "guidance_update", "material_contract"], peer_sympathy_leader_reaction_min=0.05, peer_sympathy_correlation_min=0.55, peer_sympathy_correlation_window_start=65, peer_sympathy_correlation_window_end_skip=5, peer_sympathy_top_n_peers=2, peer_sympathy_blackout_days_to_peer_event=3, peer_sympathy_stop_pct=0.035, peer_sympathy_target_pct=0.06, peer_sympathy_max_holding_days=3, ) base.update(overrides) return StrategyEngineConfig(**base) def _build_correlated_series( leader_symbol: str, peer_symbol: str, trading_days: list[dt.date], *, correlation: float, base: float = 100.0, seed: int = 0, ) -> dict[str, dict[dt.date, dict[str, Any]]]: """Construct two synthetic price series with approximate ``correlation`` between their consecutive log-returns. peer_return[t] = correlation * leader_return[t] + sqrt(1-rho^2) * noise[t] """ import random rng = random.Random(seed) leader_returns = [rng.gauss(0.001, 0.015) for _ in trading_days] noise = [rng.gauss(0, 0.015) for _ in trading_days] leader_closes: list[float] = [base] peer_closes: list[float] = [base] rho = float(correlation) sqrt_term = math.sqrt(max(0.0, 1.0 - rho * rho)) for i in range(1, len(trading_days)): l_ret = leader_returns[i] p_ret = rho * l_ret + sqrt_term * noise[i] leader_closes.append(leader_closes[-1] * math.exp(l_ret)) peer_closes.append(peer_closes[-1] * math.exp(p_ret)) bars: dict[str, dict[dt.date, dict[str, Any]]] = {leader_symbol: {}, peer_symbol: {}} for i, d in enumerate(trading_days): bars[leader_symbol][d] = { "open": leader_closes[i], "high": leader_closes[i] * 1.01, "low": leader_closes[i] * 0.99, "close": leader_closes[i], "volume": 1_000_000.0, } bars[peer_symbol][d] = { "open": peer_closes[i], "high": peer_closes[i] * 1.01, "low": peer_closes[i] * 0.99, "close": peer_closes[i], "volume": 2_000_000.0, } return bars def _trigger_inputs(**overrides: Any) -> PeerSympathyTriggerInputs: base: dict[str, Any] = dict( leader_symbol="NVDA", leader_sector="Technology", leader_event_type="earnings_release", leader_reaction=0.08, peer_symbol="AVGO", decision_date=dt.date(2026, 4, 13), next_trading_date=dt.date(2026, 4, 14), correlation=0.72, peer_last_close=110.0, peer_avg_dollar_volume_20d=300_000_000.0, peer_last_bar_date=dt.date(2026, 4, 10), peer_last_bar_timestamp=dt.datetime(2026, 4, 10, 21, 0, tzinfo=dt.timezone.utc), peer_upcoming_earnings_reaction_date=None, peer_trading_days_to_own_earnings=None, ) base.update(overrides) return PeerSympathyTriggerInputs(**base) # --------------------------------------------------------------------------- # evaluate_trigger() — happy path + 3 negative + blackout # --------------------------------------------------------------------------- def test_trigger_fires_when_all_conditions_met(): engine = _make_engine() passes, reason = evaluate_trigger(_trigger_inputs(), engine) assert passes is True, reason def test_trigger_blocks_when_event_type_not_qualifying(): engine = _make_engine() passes, reason = evaluate_trigger( _trigger_inputs(leader_event_type="other_material_event"), engine, ) assert passes is False assert "leader_event_type" in (reason or "") def test_trigger_blocks_when_leader_reaction_below_min(): engine = _make_engine() passes, reason = evaluate_trigger(_trigger_inputs(leader_reaction=0.03), engine) assert passes is False assert "leader_reaction" in (reason or "") def test_trigger_blocks_when_correlation_below_min(): engine = _make_engine() passes, reason = evaluate_trigger(_trigger_inputs(correlation=0.50), engine) assert passes is False assert "correlation" in (reason or "") def test_trigger_blocks_on_peer_earnings_blackout(): engine = _make_engine() passes, reason = evaluate_trigger( _trigger_inputs(peer_trading_days_to_own_earnings=2), engine, ) assert passes is False assert "blackout" in (reason or "") or "earnings" in (reason or "") # --------------------------------------------------------------------------- # compute_correlation() — basic invariants # --------------------------------------------------------------------------- def test_compute_correlation_returns_high_value_for_correlated_series(): trading_days = _generate_business_days(dt.date(2026, 1, 5), 100) decision_date = trading_days[-1] bars = _build_correlated_series("NVDA", "AVGO", trading_days[:-1], correlation=0.85, seed=1) leader_bars = sorted((d, b) for d, b in bars["NVDA"].items()) peer_bars = sorted((d, b) for d, b in bars["AVGO"].items()) rho, used = compute_correlation( leader_bars, peer_bars, decision_date=decision_date, window_start=65, window_end_skip=5, trading_days=trading_days, ) assert rho is not None assert rho > 0.6 # roughly tracks the imposed correlation assert used # non-empty def test_compute_correlation_skips_last_n_days(): trading_days = _generate_business_days(dt.date(2026, 1, 5), 100) decision_date = trading_days[-1] bars = _build_correlated_series("NVDA", "AVGO", trading_days[:-1], correlation=0.85, seed=2) leader_bars = sorted((d, b) for d, b in bars["NVDA"].items()) peer_bars = sorted((d, b) for d, b in bars["AVGO"].items()) _, used = compute_correlation( leader_bars, peer_bars, decision_date=decision_date, window_start=65, window_end_skip=5, trading_days=trading_days, ) assert used skip_idx = trading_days.index(decision_date) - 5 forbidden_floor = trading_days[skip_idx] assert max(used) < forbidden_floor def test_assert_correlation_window_safe_raises_when_recent_date_used(): trading_days = _generate_business_days(dt.date(2026, 1, 5), 80) decision_date = trading_days[-1] # used_dates includes a date from within the skip window (T-2) leaky_date = trading_days[-3] with pytest.raises(LookaheadViolationError): _assert_correlation_window_safe( leader_symbol="NVDA", peer_symbol="AVGO", decision_date=decision_date, window_end_skip=5, used_dates=[leaky_date], trading_days=trading_days, ) def test_assert_correlation_window_safe_accepts_safely_old_dates(): trading_days = _generate_business_days(dt.date(2026, 1, 5), 80) decision_date = trading_days[-1] safe_date = trading_days[-20] # Should not raise. _assert_correlation_window_safe( leader_symbol="NVDA", peer_symbol="AVGO", decision_date=decision_date, window_end_skip=5, used_dates=[safe_date], trading_days=trading_days, ) # --------------------------------------------------------------------------- # build_peer_sympathy_candidates() — end-to-end # --------------------------------------------------------------------------- def _build_full_setup( *, correlation: float = 0.85, leader_reaction: float = 0.08, leader_event_type: str = "earnings_release", peer_symbol: str = "AVGO", days_of_history: int = 90, ): # Pad the calendar with extra trailing days so PIT-calendar tests can place # peer earnings dates AFTER ``next_trading_date`` without IndexError. trading_days = _generate_business_days(dt.date(2026, 1, 5), days_of_history + 15) decision_date = trading_days[days_of_history] # leader event day = T next_trading_date = trading_days[days_of_history + 1] history_days = trading_days[:days_of_history] bars = _build_correlated_series( "NVDA", peer_symbol, history_days, correlation=correlation, seed=11 ) bar_provider = _FakeBarHistory(bars) peer_resolver = _StaticPeerResolver({"NVDA": [peer_symbol]}) leader = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt_nvda_2026q1", event_type=leader_event_type, event_date=decision_date, event_timestamp=dt.datetime.combine( decision_date, dt.time(16, 0), tzinfo=dt.timezone.utc ), reaction_day_return=leader_reaction, score=0.85, ) return { "trading_days": trading_days, "decision_date": decision_date, "next_trading_date": next_trading_date, "leader": leader, "peer_symbol": peer_symbol, "bar_provider": bar_provider, "peer_resolver": peer_resolver, } def test_build_emits_candidate_for_correlated_peer(): setup = _build_full_setup(correlation=0.90) engine = _make_engine() cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 cand = cands[0] assert cand.event_type == PEER_SYMPATHY_EVENT_TYPE assert cand.symbol == setup["peer_symbol"] assert cand.source_symbol == "NVDA" assert cand.engine_id == engine.engine_id assert cand.execution_date == setup["next_trading_date"] assert cand.engine_max_holding_days is not None assert cand.features["peer_sympathy_leader_symbol"] == "NVDA" assert cand.features["peer_sympathy_correlation"] >= 0.55 def test_build_skips_uncorrelated_peer(): setup = _build_full_setup(correlation=0.10) engine = _make_engine() cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert cands == [] def test_build_skips_when_leader_reaction_below_min(): setup = _build_full_setup(correlation=0.90, leader_reaction=0.02) engine = _make_engine() cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert cands == [] def test_build_skips_when_event_type_not_qualifying(): setup = _build_full_setup(correlation=0.90, leader_event_type="other_material_event") engine = _make_engine() cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert cands == [] def test_build_takes_top_n_peers_only(): """With 3 peers (corr 0.95, 0.75, 0.40), top_n=2 → only first two emitted.""" days_of_history = 90 trading_days = _generate_business_days(dt.date(2026, 1, 5), days_of_history + 2) decision_date = trading_days[days_of_history] next_trading_date = trading_days[days_of_history + 1] history_days = trading_days[:days_of_history] # Build leader and 3 peers with controlled correlation. bars: dict[str, dict[dt.date, dict[str, Any]]] = {} for peer, rho, seed in [("AVGO", 0.95, 100), ("AMD", 0.75, 200), ("MU", 0.40, 300)]: sub = _build_correlated_series("NVDA", peer, history_days, correlation=rho, seed=seed) # Only NVDA appears once — re-use leader from first iteration. if "NVDA" not in bars: bars["NVDA"] = sub["NVDA"] bars[peer] = sub[peer] bar_provider = _FakeBarHistory(bars) peer_resolver = _StaticPeerResolver({"NVDA": ["AVGO", "AMD", "MU"]}) leader = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=decision_date, event_timestamp=dt.datetime.combine(decision_date, dt.time(16, 0), tzinfo=dt.timezone.utc), reaction_day_return=0.08, ) engine = _make_engine(peer_sympathy_top_n_peers=2) cands = build_peer_sympathy_candidates( decision_date=decision_date, next_trading_date=next_trading_date, leaders=[leader], peer_resolver=peer_resolver, engine=engine, bar_provider=bar_provider, trading_days=trading_days, ) # MU (0.40) is below correlation_min anyway; AVGO + AMD survive. assert len(cands) <= 2 chosen = {c.symbol for c in cands} assert "MU" not in chosen # --------------------------------------------------------------------------- # Look-ahead defenses # --------------------------------------------------------------------------- def test_build_raises_when_next_trading_date_not_after_decision(): setup = _build_full_setup() engine = _make_engine() with pytest.raises(LookaheadViolationError): build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["decision_date"], # equal → violation leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) def test_build_raises_when_leader_event_timestamp_naive(): setup = _build_full_setup() bad_leader = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=setup["decision_date"], event_timestamp=dt.datetime.combine(setup["decision_date"], dt.time(16, 0)), # naive reaction_day_return=0.08, ) engine = _make_engine() with pytest.raises(LookaheadViolationError): build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[bad_leader], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) def test_build_raises_when_leader_event_timestamp_after_peer_open(): """Leader event timestamp at-or-after T+1 09:30 ET cutoff is a look-ahead violation.""" setup = _build_full_setup() # 09:30 ET on next_trading_date == 14:30 UTC under EST. leak_ts = dt.datetime.combine(setup["next_trading_date"], dt.time(15, 0), tzinfo=dt.timezone.utc) leak_leader = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=setup["decision_date"], event_timestamp=leak_ts, reaction_day_return=0.08, ) engine = _make_engine() with pytest.raises(LookaheadViolationError): build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[leak_leader], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) def test_build_does_not_consult_peer_t0_reaction(): """Bars dated == decision_date (peer T+0) must NOT enter selection. Concrete check: inject a peer bar dated ON decision_date with an extreme return; if the engine were reading T+0 it would either crash on look-ahead (preferred) or produce a different correlation. We use a permissive bar provider that lets ``< as_of`` filter run; the engine must produce a candidate consistent with PRE-T data alone. """ setup = _build_full_setup(correlation=0.90) bars = setup["bar_provider"].bars # Inject a wild peer bar on decision_date (not strictly before). bars[setup["peer_symbol"]][setup["decision_date"]] = { "open": 50.0, "high": 50.0, "low": 50.0, "close": 50.0, "volume": 99_000_000.0, } engine = _make_engine() # The default _FakeBarHistory.get_bars_before filters strictly < decision_date, # so the injected T+0 bar is not visible. Candidate features must therefore # NOT reference any T+0 quantity. We assert by snapshotting the candidate # features and confirming the only price reference is the LAST bar < T. cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 cand = cands[0] # entry_price_est must equal the last close strictly BEFORE decision_date. last_pre_t = max(d for d in bars[setup["peer_symbol"]] if d < setup["decision_date"]) expected_close = float(bars[setup["peer_symbol"]][last_pre_t]["close"]) assert math.isclose(cand.entry_price_est, expected_close, rel_tol=1e-6) # --------------------------------------------------------------------------- # Peer-resolver integration with existing leader-follower infra # --------------------------------------------------------------------------- def test_peer_set_sourced_from_leader_follower_extra_peer_symbols_by_sector(): """Engine config's leader_follower_extra_peer_symbols_by_sector must surface peers.""" from libs.backtest.proxies import peer_candidates_for_symbol # Sanity: the underlying helper recognizes NVDA → has tech peers from the curated map. peers = peer_candidates_for_symbol("NVDA", "Technology") assert "AVGO" in peers assert "AMD" in peers def test_peer_set_extra_by_sector_is_consumed_by_resolver_protocol(): """The runner's _RunnerPeerResolver wraps the existing _leader_follower_peer_candidates; the static fake must equally honor curated peers by sector.""" resolver = _StaticPeerResolver({"NVDA": ["AVGO", "AMD"]}) engine = _make_engine( leader_follower_extra_peer_symbols_by_sector={"Technology": ["MU"]} ) peers = resolver.peers_for_leader(engine, "NVDA", "Technology") # Static fake returns the supplied list; this confirms the Protocol shape. assert peers == ["AVGO", "AMD"] # --------------------------------------------------------------------------- # Exit policy stub tests — verify candidate carries correct exit configuration # --------------------------------------------------------------------------- def test_candidate_carries_stop_target_max_hold_in_engine_overrides(): setup = _build_full_setup(correlation=0.90) engine = _make_engine( peer_sympathy_stop_pct=0.035, peer_sympathy_target_pct=0.06, peer_sympathy_max_holding_days=3, ) cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 cand = cands[0] # stop_pct 0.035 / 0.02 = 1.75 ATR multiplier assert math.isclose(cand.engine_stop_atr_multiplier, 1.75, rel_tol=1e-6) # target_pct / stop_pct = 0.06 / 0.035 = 1.714... assert math.isclose(cand.engine_target_1_r, 0.06 / 0.035, rel_tol=1e-6) assert cand.engine_target_1_fraction == 1.0 assert cand.engine_max_holding_days == 3 def test_candidate_max_hold_capped_by_peer_earnings_blackout(): """If peer's own earnings are 4 trading days out and blackout=3 → max_hold = max(1, 4-3) = 1.""" setup = _build_full_setup(correlation=0.90) # Build a PIT calendar: peer has earnings 4 trading days after next_trading_date. trading_days = setup["trading_days"] next_idx = trading_days.index(setup["next_trading_date"]) peer_earnings_date = trading_days[next_idx + 4] pit_calendar = PointInTimeEarningsCalendar( [ EarningsCalendarEntry( symbol=setup["peer_symbol"], as_of_date=trading_days[0], expected_reaction_date=peer_earnings_date, expected_event_date=peer_earnings_date, filing_time_bucket="post_market", ) ] ) upcoming = _PitCalendarUpcomingEarningsAdapter( pit_calendar=pit_calendar, trading_days=trading_days, ) engine = _make_engine( peer_sympathy_blackout_days_to_peer_event=3, peer_sympathy_max_holding_days=3, ) cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], upcoming_earnings_provider=upcoming, trading_days=trading_days, ) # 4 days to event > blackout 3 → not blocked. Hold = max(1, 4-3) = 1. assert len(cands) == 1 assert cands[0].engine_max_holding_days == 1 def test_candidate_blocked_when_peer_earnings_within_blackout_window(): """Peer earnings 2 trading days out, blackout=3 → trigger BLOCKS (no candidate).""" setup = _build_full_setup(correlation=0.90) trading_days = setup["trading_days"] next_idx = trading_days.index(setup["next_trading_date"]) peer_earnings_date = trading_days[next_idx + 2] # 2 trading days out pit_calendar = PointInTimeEarningsCalendar( [ EarningsCalendarEntry( symbol=setup["peer_symbol"], as_of_date=trading_days[0], expected_reaction_date=peer_earnings_date, expected_event_date=peer_earnings_date, filing_time_bucket="post_market", ) ] ) upcoming = _PitCalendarUpcomingEarningsAdapter( pit_calendar=pit_calendar, trading_days=trading_days, ) engine = _make_engine(peer_sympathy_blackout_days_to_peer_event=3) cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], upcoming_earnings_provider=upcoming, trading_days=trading_days, ) assert cands == [] # --------------------------------------------------------------------------- # Behavioral exit tests — drive synthetic position through simulate_exit and # confirm pct exits map correctly to STOP / TARGET / TIME outcomes. # --------------------------------------------------------------------------- def _build_position_for_peer_sympathy( *, entry_price: float = 100.0, stop_pct: float = 0.035, target_pct: float = 0.06, days_held: int = 0, ) -> Any: from libs.backtest.domain import Candidate, OpenPosition, PlannedOrder stop_mult = stop_pct / 0.02 target_r = target_pct / stop_pct synthetic_atr = entry_price * 0.02 cand = Candidate( event_id="evt_peer_sympathy", symbol="AVGO", source_symbol="NVDA", score=0.8, sector="Technology", event_type=PEER_SYMPATHY_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=300_000_000.0, atr_14=synthetic_atr, score_bucket="high", engine_id="peer_sympathy_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=3, ) 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_peer", 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=3, ) def test_exit_stop_at_minus_3_5_pct(): from libs.backtest.domain import ExitReason from libs.backtest.execution import simulate_exit pos = _build_position_for_peer_sympathy(entry_price=100.0, stop_pct=0.035) # Bar drops to 96.0 < 96.5 stop → STOP. bar = {"date": dt.date(2026, 4, 15), "open": 99.0, "high": 99.5, "low": 96.0, "close": 96.7, "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_6_pct(): from libs.backtest.domain import ExitReason from libs.backtest.execution import simulate_exit pos = _build_position_for_peer_sympathy(entry_price=100.0, target_pct=0.06) # Bar high reaches 106.5 > target 106.0 → TARGET. bar = {"date": dt.date(2026, 4, 15), "open": 102.0, "high": 106.5, "low": 101.0, "close": 105.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.TARGET def test_exit_time_at_max_holding_days(): from libs.backtest.domain import ExitReason from libs.backtest.execution import simulate_exit cfg = _exec_config_for_exit_test().model_copy(update={"max_holding_days": 3}) pos = _build_position_for_peer_sympathy(entry_price=100.0, days_held=3) bar = {"date": dt.date(2026, 4, 17), "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, 17)) assert trade is not None assert trade.exit_reason == ExitReason.TIME def test_exit_blackout_caps_max_hold_via_engine_max_holding_days(): """When peer's own earnings are within blackout window, candidate's engine_max_holding_days is capped to (days_to_event - blackout) ≥ 1. The execution machinery then treats this as the effective hold ceiling.""" setup = _build_full_setup(correlation=0.90) trading_days = setup["trading_days"] next_idx = trading_days.index(setup["next_trading_date"]) peer_earnings_date = trading_days[next_idx + 5] pit_calendar = PointInTimeEarningsCalendar( [ EarningsCalendarEntry( symbol=setup["peer_symbol"], as_of_date=trading_days[0], expected_reaction_date=peer_earnings_date, expected_event_date=peer_earnings_date, filing_time_bucket="post_market", ) ] ) upcoming = _PitCalendarUpcomingEarningsAdapter( pit_calendar=pit_calendar, trading_days=trading_days, ) engine = _make_engine( peer_sympathy_blackout_days_to_peer_event=3, peer_sympathy_max_holding_days=3, ) cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], upcoming_earnings_provider=upcoming, trading_days=trading_days, ) assert len(cands) == 1 # 5 days to event - blackout 3 = 2; min(default_max_hold=3, 2) = 2. assert cands[0].engine_max_holding_days == 2 # --------------------------------------------------------------------------- # Regression: runner-adapter select_candidates must NOT pass the peer_sympathy # strategy_engine, because that engine declares event_types=['peer_sympathy'] # (a synthetic downstream type) which would filter out every real leader row # (earnings_release / guidance_update / material_contract). This is the bug # that produced 0 trades over 1051 days in PoC v1. # --------------------------------------------------------------------------- def _make_leader_raw_row(**overrides: Any) -> dict[str, Any]: """Minimal real-shape PEAD candidate row representing a leader print.""" base: dict[str, Any] = { "event_id": "EVT::NVDA::2024-02-22", "symbol": "NVDA", "issuer_id": "ISSUER::0001045810", "score": 0.85, "sector": "Technology", "event_type": "earnings_release", "event_timestamp": "2024-02-21T21:00:00+00:00", "filing_time_bucket": "post_market", "entry_convention": "next_open_after_reaction_close", "reaction_date": "2024-02-22", "entry_date": "2024-02-23", "entry_price": 730.0, "avg_dollar_volume": 25_000_000_000.0, "avg_dollar_volume_20d": 25_000_000_000.0, "atr_14": 25.0, "reaction_day_return": 0.164, "reaction_day_open": 680.0, "reaction_day_close": 791.0, "reaction_day_high": 800.0, "reaction_day_low": 670.0, "exchange_proxy": "NASDAQ", "volume_ratio_20d": 3.5, "gap_size": 0.10, } base.update(overrides) return base def test_select_candidates_with_peer_sympathy_engine_drops_real_leaders(): """Demonstrates the bug: passing the peer_sympathy engine to select_candidates drops every real leader print, because the engine's event_types=['peer_sympathy'] does not include 'earnings_release' etc. This test pins down the unsafe interaction so a future dev cannot silently re-introduce ``strategy_engine=engine`` in ``_schedule_peer_sympathy_candidates`` without it failing here. """ from libs.backtest.domain import SignalConfig, UniverseConfig from libs.backtest.peer_sympathy import PEER_SYMPATHY_EVENT_TYPE from libs.backtest.selector import select_candidates rows = [ _make_leader_raw_row(symbol="NVDA", event_type="earnings_release", reaction_day_return=0.164), _make_leader_raw_row(symbol="MRNA", event_type="earnings_release", reaction_day_return=0.135, event_id="EVT::MRNA::2024-02-22", issuer_id="ISSUER::0001682852"), _make_leader_raw_row(symbol="MU", event_type="guidance_update", reaction_day_return=0.086, event_id="EVT::MU::2023-12-21", issuer_id="ISSUER::0000723125"), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v13e", score_threshold=0.0, max_candidates_per_day=18, ) peer_sympathy_engine = _make_engine( event_types=[PEER_SYMPATHY_EVENT_TYPE], score_threshold_override=0.0, ) # Bug reproduction: engine event_types filter rejects all real leader rows. selected_with_engine = select_candidates( rows, universe, signal, strategy_engine=peer_sympathy_engine, truncate_to=90, ) assert selected_with_engine == [], ( "Bug regression: select_candidates with peer_sympathy strategy_engine " "must drop real leader rows because their event_type ('earnings_release', " "'guidance_update') is not in the engine's event_types=['peer_sympathy']. " "If this assertion stops holding, the runner adapter contract has shifted " "and the no-engine call in _schedule_peer_sympathy_candidates may need " "to be revisited." ) def test_select_candidates_without_engine_retains_real_leaders(): """The fixed runner-adapter call path: ``select_candidates`` is invoked WITHOUT the peer_sympathy strategy_engine, so real leader rows are retained and can feed the manual peer_sympathy_leader_event_types filter downstream. """ from libs.backtest.domain import SignalConfig, UniverseConfig from libs.backtest.selector import select_candidates rows = [ _make_leader_raw_row(symbol="NVDA", event_type="earnings_release", reaction_day_return=0.164), _make_leader_raw_row(symbol="MRNA", event_type="earnings_release", reaction_day_return=0.135, event_id="EVT::MRNA::2024-02-22", issuer_id="ISSUER::0001682852"), _make_leader_raw_row(symbol="MU", event_type="guidance_update", reaction_day_return=0.086, event_id="EVT::MU::2023-12-21", issuer_id="ISSUER::0000723125"), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v13e", score_threshold=0.0, max_candidates_per_day=18, ) # Fix: NO strategy_engine kwarg. Real leader rows survive selection. selected = select_candidates( rows, universe, signal, truncate_to=90, ) selected_symbols = {c.symbol.upper() for c in selected} assert "NVDA" in selected_symbols assert "MRNA" in selected_symbols assert "MU" in selected_symbols, ( "Fix regression: select_candidates without strategy_engine must retain " "real leader rows so that _schedule_peer_sympathy_candidates can apply " "its manual peer_sympathy_leader_event_types filter and emit synthetic " "peer candidates. If this fails, the runner adapter is once again " "starving downstream peer-sympathy logic." ) # --------------------------------------------------------------------------- # Salvage variant (v2): entry_timing_policy="reaction_close" # --------------------------------------------------------------------------- # # Hypothesis: peers move in the SAME intraday session as the leader's print, # not on T+1 gap. Entry at peer's T 16:00 ET close instead of T+1 09:30 ET. # Look-ahead defenses: # - leader.event_timestamp must be strictly < peer's T 16:00 ET close. # - peer T+0 reaction is NEVER referenced (still uses bars strictly < T). # Filing-time bucket allow-list (e.g. ["pre_market","regular_hours"]) is the # intended way to skip AMC prints that can't be sympathy-traded same-day. def _bmo_leader_setup( *, filing_time_bucket: str = "pre_market", correlation: float = 0.90 ) -> dict[str, Any]: """Build a setup whose leader timestamp is BEFORE peer's T 16:00 ET close.""" setup = _build_full_setup(correlation=correlation) decision_date = setup["decision_date"] # 08:00 ET == 13:00 UTC (EST offset). Strictly before 21:00 UTC (16:00 ET). bmo_ts = dt.datetime.combine(decision_date, dt.time(13, 0), tzinfo=dt.timezone.utc) setup["leader"] = LeaderPrint( symbol="NVDA", sector="Technology", event_id=setup["leader"].event_id, event_type=setup["leader"].event_type, event_date=decision_date, event_timestamp=bmo_ts, reaction_day_return=setup["leader"].reaction_day_return, score=setup["leader"].score, filing_time_bucket=filing_time_bucket, ) return setup def test_reaction_close_bmo_leader_emits_same_day_peer_entry(): """BMO leader print → peer enters at peer's T reaction_close (today).""" setup = _bmo_leader_setup(filing_time_bucket="pre_market") engine = _make_engine( peer_sympathy_entry_timing_policy="reaction_close", peer_sympathy_leader_filing_time_buckets=["pre_market", "regular_hours"], ) cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 cand = cands[0] assert cand.entry_timing_policy == "reaction_close" assert cand.timing_class == "same_day" # Entry happens TODAY, not T+1. assert cand.execution_date == setup["decision_date"] assert cand.reaction_date == setup["decision_date"] def test_reaction_close_regular_hours_leader_emits_same_day_peer_entry(): """Regular-hours filing (e.g. 11:30 ET) → still strictly before peer's T 16:00 ET close.""" setup = _build_full_setup(correlation=0.90) decision_date = setup["decision_date"] rh_ts = dt.datetime.combine(decision_date, dt.time(16, 30), tzinfo=dt.timezone.utc) # 11:30 ET setup["leader"] = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=decision_date, event_timestamp=rh_ts, reaction_day_return=0.08, filing_time_bucket="regular_hours", ) engine = _make_engine( peer_sympathy_entry_timing_policy="reaction_close", peer_sympathy_leader_filing_time_buckets=["pre_market", "regular_hours"], ) cands = build_peer_sympathy_candidates( decision_date=decision_date, next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 assert cands[0].entry_timing_policy == "reaction_close" assert cands[0].execution_date == decision_date def test_reaction_close_post_market_leader_excluded_by_bucket_filter(): """AMC filing in the bucket-restricted config → no candidate emitted. Note: under PEAD's reaction-date convention, an AMC filing on calendar day X has reaction_date = X+1, so leader.event_timestamp is X 21:00 UTC (16:00 ET on X) and peer's reaction_close cutoff on decision_date X+1 is X+1 21:00 UTC. The raw timestamp guard PASSES (X 21:00 < X+1 21:00). The ``peer_sympathy_leader_filing_time_buckets`` allow-list is what actually excludes AMC — that is the explicit knob in the salvage-variant config. """ setup = _build_full_setup(correlation=0.90) decision_date = setup["decision_date"] amc_ts = dt.datetime.combine( decision_date - dt.timedelta(days=1), dt.time(21, 0), tzinfo=dt.timezone.utc, ) # 16:00 ET previous day setup["leader"] = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=decision_date, event_timestamp=amc_ts, reaction_day_return=0.08, filing_time_bucket="post_market", ) # Bucket allow-list excludes post_market — engine drops the leader. engine = _make_engine( peer_sympathy_entry_timing_policy="reaction_close", peer_sympathy_leader_filing_time_buckets=["pre_market", "regular_hours"], ) cands = build_peer_sympathy_candidates( decision_date=decision_date, next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert cands == [] def test_reaction_close_post_market_falls_back_when_buckets_unrestricted(): """If the engine config does NOT restrict filing buckets, AMC still produces a candidate (timestamp guard alone passes). This documents that the bucket filter is the load-bearing knob, not the timestamp check. """ setup = _build_full_setup(correlation=0.90) decision_date = setup["decision_date"] amc_ts = dt.datetime.combine( decision_date - dt.timedelta(days=1), dt.time(21, 0), tzinfo=dt.timezone.utc, ) setup["leader"] = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=decision_date, event_timestamp=amc_ts, reaction_day_return=0.08, filing_time_bucket="post_market", ) engine = _make_engine( peer_sympathy_entry_timing_policy="reaction_close", peer_sympathy_leader_filing_time_buckets=None, # no restriction ) cands = build_peer_sympathy_candidates( decision_date=decision_date, next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 assert cands[0].entry_timing_policy == "reaction_close" def test_reaction_close_raises_when_leader_timestamp_at_or_after_peer_close(): """If the leader's print is AT-or-AFTER peer's 16:00 ET close on T, entering peer at that close is a look-ahead violation. """ setup = _build_full_setup(correlation=0.90) decision_date = setup["decision_date"] # 16:00 ET on T == 21:00 UTC. AT cutoff is a violation. leak_ts = dt.datetime.combine(decision_date, dt.time(21, 0), tzinfo=dt.timezone.utc) setup["leader"] = LeaderPrint( symbol="NVDA", sector="Technology", event_id="evt", event_type="earnings_release", event_date=decision_date, event_timestamp=leak_ts, reaction_day_return=0.08, filing_time_bucket="post_market", # we don't filter to surface the timestamp guard ) engine = _make_engine( peer_sympathy_entry_timing_policy="reaction_close", peer_sympathy_leader_filing_time_buckets=None, ) with pytest.raises(LookaheadViolationError): build_peer_sympathy_candidates( decision_date=decision_date, next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) def test_reaction_close_does_not_consult_peer_t0_reaction(): """Same invariant as the v1 next_open path: peer T+0 bars must NEVER be referenced in selection. Inject a wild T+0 peer bar and verify the candidate's entry_price_est is computed from PRE-T data alone. """ setup = _bmo_leader_setup(correlation=0.90) bars = setup["bar_provider"].bars bars[setup["peer_symbol"]][setup["decision_date"]] = { "open": 50.0, "high": 50.0, "low": 50.0, "close": 50.0, "volume": 99_000_000.0, } engine = _make_engine( peer_sympathy_entry_timing_policy="reaction_close", peer_sympathy_leader_filing_time_buckets=["pre_market", "regular_hours"], ) cands = build_peer_sympathy_candidates( decision_date=setup["decision_date"], next_trading_date=setup["next_trading_date"], leaders=[setup["leader"]], peer_resolver=setup["peer_resolver"], engine=engine, bar_provider=setup["bar_provider"], trading_days=setup["trading_days"], ) assert len(cands) == 1 last_pre_t = max(d for d in bars[setup["peer_symbol"]] if d < setup["decision_date"]) expected_close = float(bars[setup["peer_symbol"]][last_pre_t]["close"]) assert math.isclose(cands[0].entry_price_est, expected_close, rel_tol=1e-6)