"""Unit tests for libs/backtest/selector.py.""" from __future__ import annotations import datetime as dt import json from zoneinfo import ZoneInfo import pytest from libs.backtest.domain import ( EventTypeProfile, SignalConfig, StrategyEngineConfig, UniverseConfig, ) _UTC = ZoneInfo("UTC") def _make_raw_row(**kwargs) -> dict: defaults = { "event_id": "EVT::TEST::001", "symbol": "AAPL", "issuer_id": "ISSUER::0000320193", "score": 0.75, "sector": "Technology", "event_type": "earnings", "event_timestamp": "2026-01-05T21:00:00+00:00", "filing_time_bucket": "post_market", "entry_convention": "next_open_after_reaction_close", "reaction_date": "2026-01-06", "entry_date": "2026-01-07", "entry_price": 150.0, "avg_dollar_volume": 5_000_000.0, "atr_14": 3.5, } defaults.update(kwargs) return defaults class TestBuildCandidate: def test_basic(self): from libs.backtest.selector import build_candidate row = _make_raw_row() c = build_candidate(row) assert c is not None assert c.symbol == "AAPL" assert c.score == 0.75 assert c.execution_date == dt.date(2026, 1, 7) assert c.event_timestamp.tzinfo is not None assert c.timing_class == "after_close" def test_null_timestamp_returns_none(self): from libs.backtest.selector import build_candidate row = _make_raw_row(event_timestamp=None) assert build_candidate(row) is None def test_zero_entry_price_returns_none(self): from libs.backtest.selector import build_candidate row = _make_raw_row(entry_price=0.0) assert build_candidate(row) is None def test_missing_entry_price_returns_none(self): from libs.backtest.selector import build_candidate row = _make_raw_row() del row["entry_price"] assert build_candidate(row) is None def test_null_exec_date_returns_none(self): from libs.backtest.selector import build_candidate row = _make_raw_row() del row["entry_date"] assert build_candidate(row) is None def test_sector_defaults_to_unknown(self): from libs.backtest.selector import build_candidate row = _make_raw_row(sector=None) c = build_candidate(row) assert c is not None assert c.sector == "UNKNOWN" def test_prefers_avg_dollar_volume_20d_when_present(self): from libs.backtest.selector import build_candidate row = _make_raw_row( avg_dollar_volume=999_000_000.0, avg_dollar_volume_20d=12_345_678.0, ) c = build_candidate(row) assert c is not None assert c.avg_dollar_volume == 12_345_678.0 def test_score_bucket_classification(self): from libs.backtest.selector import build_candidate c = build_candidate(_make_raw_row(score=0.85)) assert c.score_bucket == "high" c = build_candidate(_make_raw_row(score=0.65)) assert c.score_bucket == "medium_high" c = build_candidate(_make_raw_row(score=0.45)) assert c.score_bucket == "medium" c = build_candidate(_make_raw_row(score=0.25)) assert c.score_bucket == "medium_low" c = build_candidate(_make_raw_row(score=0.10)) assert c.score_bucket == "low" def test_same_day_timing_and_direction_from_reaction(self): from libs.backtest.selector import build_candidate row = _make_raw_row( event_date="2026-01-06", reaction_date="2026-01-06", execution_date="2026-01-07", trade_direction="", reaction_day_return=-0.12, ) c = build_candidate(row) assert c is not None assert c.event_date == dt.date(2026, 1, 6) assert c.timing_class == "same_day" assert c.trade_direction == "short" def test_sector_etf_proxy_uses_proxy_trade_fields(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="sector_etf_proxy", event_types=["earnings"], trade_symbol_mode="sector_etf", ) row = _make_raw_row( score=0.9, event_close=150.0, reaction_day_low=145.0, reaction_day_high=153.0, sector_etf_proxy="XLK", sector_etf_event_close=210.0, sector_etf_entry_price=211.5, sector_etf_reaction_day_low=206.0, sector_etf_reaction_day_high=212.0, sector_etf_avg_dollar_volume=250_000_000.0, sector_etf_atr_14=4.2, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.symbol == "XLK" assert c.source_symbol == "AAPL" assert c.trade_symbol_mode == "sector_etf" assert c.entry_price_est == 211.5 assert c.avg_dollar_volume == 250_000_000.0 assert c.atr_14 == 4.2 assert c.features["event_close"] == 210.0 assert c.features["reaction_day_low"] == 206.0 assert c.features["source_event_close"] == 150.0 def test_peer_proxy_uses_proxy_trade_fields(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="peer_proxy", event_types=["earnings"], trade_symbol_mode="peer_proxy", ) row = _make_raw_row( symbol="AVGO", score=0.9, event_close=220.0, reaction_day_low=214.0, reaction_day_high=224.0, peer_proxy_symbol="NVDA", peer_proxy_event_close=132.0, peer_proxy_entry_price=133.5, peer_proxy_reaction_day_low=128.0, peer_proxy_reaction_day_high=134.0, peer_proxy_avg_dollar_volume=600_000_000.0, peer_proxy_atr_14=5.1, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.symbol == "NVDA" assert c.source_symbol == "AVGO" assert c.trade_symbol_mode == "peer_proxy" assert c.entry_price_est == 133.5 assert c.avg_dollar_volume == 600_000_000.0 assert c.atr_14 == 5.1 assert c.features["event_close"] == 132.0 assert c.features["reaction_day_low"] == 128.0 assert c.features["source_event_close"] == 220.0 def test_allowed_sectors_filters_source_row(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="healthcare_only", event_types=["earnings"], allowed_sectors=["Healthcare"], ) blocked = build_candidate(_make_raw_row(sector="Technology"), strategy_engine=engine) allowed = build_candidate(_make_raw_row(sector="Healthcare"), strategy_engine=engine) assert blocked is None assert allowed is not None def test_excluded_symbols_filters_source_row(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="symbol_blacklist", event_types=["earnings"], excluded_symbols=["WY", "DPZ"], ) blocked = build_candidate(_make_raw_row(symbol="WY"), strategy_engine=engine) allowed = build_candidate(_make_raw_row(symbol="CBRE"), strategy_engine=engine) assert blocked is None assert allowed is not None def test_peer_proxy_quality_filters_use_proxy_fields(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="peer_proxy_quality", event_types=["earnings"], trade_symbol_mode="peer_proxy", proxy_reaction_day_return_max=0.01, proxy_gap_size_max=0.005, ) blocked = build_candidate( _make_raw_row( symbol="SNOW", peer_proxy_symbol="CRM", peer_proxy_event_close=250.0, peer_proxy_entry_price=251.0, peer_proxy_reaction_day_low=247.0, peer_proxy_reaction_day_high=252.0, peer_proxy_avg_dollar_volume=1_500_000_000.0, peer_proxy_atr_14=4.5, peer_proxy_reaction_day_return=0.017, peer_proxy_gap_size=0.009, ), strategy_engine=engine, ) allowed = build_candidate( _make_raw_row( symbol="URI", peer_proxy_symbol="CAT", peer_proxy_event_close=420.0, peer_proxy_entry_price=421.0, peer_proxy_reaction_day_low=417.0, peer_proxy_reaction_day_high=422.0, peer_proxy_avg_dollar_volume=1_000_000_000.0, peer_proxy_atr_14=6.0, peer_proxy_reaction_day_return=0.004, peer_proxy_gap_size=-0.001, ), strategy_engine=engine, ) assert blocked is None assert allowed is not None def test_engine_can_force_long_direction_for_negative_reaction(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="next_open_long_reversal_micro", event_types=["guidance_update"], timing_class="after_close", direction="long_only", forced_trade_direction_override="long", entry_timing_policy="next_open", ) row = _make_raw_row( event_type="guidance_update", event_date="2026-01-06", reaction_date="2026-01-07", entry_date="2026-01-08", trade_direction="", reaction_day_return=-0.05, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.trade_direction == "long" def test_engine_macro_scaler_fields_are_carried_to_candidate(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="macro_scaled_engine", event_types=["earnings"], macro_vix_size_scaler_low=22.0, macro_vix_size_scaler_high=30.0, macro_vix_size_scaler_min=0.7, macro_hy_spread_size_scaler_low=3.8, macro_hy_spread_size_scaler_high=4.5, macro_hy_spread_size_scaler_min=0.8, ) c = build_candidate(_make_raw_row(), strategy_engine=engine) assert c is not None assert c.engine_macro_vix_size_scaler_low == 22.0 assert c.engine_macro_vix_size_scaler_high == 30.0 assert c.engine_macro_vix_size_scaler_min == 0.7 assert c.engine_macro_hy_spread_size_scaler_low == 3.8 assert c.engine_macro_hy_spread_size_scaler_high == 4.5 assert c.engine_macro_hy_spread_size_scaler_min == 0.8 def test_engine_can_override_per_trade_risk(self): from libs.backtest.selector import build_candidate row = _make_raw_row( event_date="2026-01-06", reaction_day_return=0.09, event_direction="mixed", guidance_status="inline_or_maintained", filing_time_bucket="regular_hours", gap_size=0.08, close_location=0.66, volume_ratio_20d=2.7, event_close=150.0, ) engine = StrategyEngineConfig( engine_id="mixed_inline_lowrisk", event_types=["earnings"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", per_trade_risk_pct_override=0.01, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.engine_per_trade_risk_pct == pytest.approx(0.01) assert c.engine_id == "mixed_inline_lowrisk" def test_engine_can_override_sector_limit(self): from libs.backtest.selector import build_candidate row = _make_raw_row( event_type="other_material_event", event_direction="unknown", guidance_status="not_provided", filing_time_bucket="post_market", reaction_day_return=0.03, close_location=0.78, gap_size=0.01, volume_ratio_20d=1.2, ) engine = StrategyEngineConfig( engine_id="other_material_unknown_orderly", event_types=["other_material_event"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", max_positions_per_sector_override=4, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.engine_max_positions_per_sector == 4 assert c.engine_id == "other_material_unknown_orderly" def test_engine_can_override_position_caps(self): from libs.backtest.selector import build_candidate row = _make_raw_row( event_type="other_material_event", event_direction="unknown", guidance_status="not_provided", filing_time_bucket="post_market", reaction_day_return=0.03, close_location=0.78, gap_size=0.01, volume_ratio_20d=1.2, ) engine = StrategyEngineConfig( engine_id="other_material_unknown_capped", event_types=["other_material_event"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", max_position_value_pct_override=0.35, max_adv_fraction_override=0.015, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.engine_max_position_value_pct == pytest.approx(0.35) assert c.engine_max_adv_fraction == pytest.approx(0.015) assert c.engine_id == "other_material_unknown_capped" def test_engine_can_inherit_parent_filters(self): from libs.backtest.selector import build_candidate parent = StrategyEngineConfig( engine_id="core_parent", event_types=["guidance_update"], guidance_statuses=["raised"], filing_time_buckets=["regular_hours"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", close_location_min=0.55, ) child = StrategyEngineConfig( engine_id="core_child", inherits_from_engine_id="core_parent", close_location_min=0.75, ) row = _make_raw_row( event_type="guidance_update", event_direction="bullish", guidance_status="raised", filing_time_bucket="regular_hours", event_date="2026-01-06", reaction_date="2026-01-06", close_location=0.8, reaction_day_return=0.1, event_close=150.0, ) c = build_candidate( row, strategy_engine=child, engine_lookup={"core_parent": parent, "core_child": child}, ) assert c is not None assert c.engine_id == "core_child" assert c.entry_timing_policy == "reaction_close" def test_engine_can_exclude_if_matches_sibling(self): from libs.backtest.selector import build_candidate cool = StrategyEngineConfig( engine_id="core_cool", event_types=["guidance_update"], guidance_statuses=["raised"], filing_time_buckets=["regular_hours"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", close_location_min=0.75, ) hot = StrategyEngineConfig( engine_id="core_hot", inherits_from_engine_id="core_cool", exclude_if_matches_engine_id="core_cool", close_location_min=0.55, ) cool_row = _make_raw_row( event_type="guidance_update", event_direction="bullish", guidance_status="raised", filing_time_bucket="regular_hours", event_date="2026-01-06", reaction_date="2026-01-06", close_location=0.8, reaction_day_return=0.1, event_close=150.0, ) hot_only_row = _make_raw_row( event_type="guidance_update", event_direction="bullish", guidance_status="raised", filing_time_bucket="regular_hours", event_date="2026-01-06", reaction_date="2026-01-06", close_location=0.6, reaction_day_return=0.1, event_close=150.0, ) lookup = {"core_cool": cool, "core_hot": hot} assert build_candidate(cool_row, strategy_engine=hot, engine_lookup=lookup) is None assert build_candidate(hot_only_row, strategy_engine=hot, engine_lookup=lookup) is not None def test_reaction_close_engine_uses_event_close(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="earnings_same_day_long_close_v1", event_types=["earnings"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", max_holding_days=3, engine_risk_budget_pct=0.35, ) row = _make_raw_row( event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, entry_date="2026-01-07", reaction_day_return=0.11, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.execution_date == dt.date(2026, 1, 6) assert c.entry_price_est == pytest.approx(149.5) assert c.engine_id == "earnings_same_day_long_close_v1" assert c.entry_timing_policy == "reaction_close" def test_engine_execution_overrides_are_copied_to_candidate(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="earnings_same_day_long_trend_v1", event_types=["earnings"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", max_holding_days=12, engine_risk_budget_pct=0.25, target_atr_multiplier_override=2.5, target_1_r_override=3.25, target_1_fraction_override=0.33, trailing_model_override="pct_10", trailing_warmup_days_override=2, early_failure_close_below_entry_and_reaction_close_override=True, early_failure_no_progress_days_override=1, early_failure_no_progress_r_override=0.2, early_failure_no_progress_fraction_override=1.0, veto_oneoff_penalty_override=0.95, allow_oneoff_downsizing_override=True, oneoff_downsize_floor_override=0.6, veto_parse_confidence_min_override=0.25, ) row = _make_raw_row( event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, entry_date="2026-01-07", reaction_day_return=0.11, ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.engine_max_holding_days == 12 assert c.engine_risk_budget_pct == pytest.approx(0.25) assert c.engine_target_atr_multiplier == pytest.approx(2.5) assert c.engine_target_1_r == pytest.approx(3.25) assert c.engine_target_1_fraction == pytest.approx(0.33) assert c.engine_trailing_model == "pct_10" assert c.engine_trailing_warmup_days == 2 assert c.engine_early_failure_close_below_entry_and_reaction_close is True assert c.engine_early_failure_no_progress_days == 1 assert c.engine_early_failure_no_progress_r == pytest.approx(0.2) assert c.engine_early_failure_no_progress_fraction == pytest.approx(1.0) assert c.engine_veto_oneoff_penalty == pytest.approx(0.95) assert c.engine_allow_oneoff_downsizing is True assert c.engine_oneoff_downsize_floor == pytest.approx(0.6) assert c.engine_veto_parse_confidence_min == pytest.approx(0.25) def test_mixed_inline_execution_overrides_override_generic_engine_exit(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="same_day_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", early_failure_no_progress_days_override=1, early_failure_no_progress_r_override=0.15, mixed_inline_early_failure_no_progress_days_override=1, mixed_inline_early_failure_no_progress_r_override=0.30, mixed_inline_early_failure_no_progress_fraction_override=1.0, ) row = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, reaction_day_return=0.11, event_direction="mixed", guidance_status="inline_or_maintained", ) c = build_candidate(row, strategy_engine=engine) assert c is not None assert c.engine_early_failure_no_progress_days == 1 assert c.engine_early_failure_no_progress_r == pytest.approx(0.30) assert c.engine_early_failure_no_progress_fraction == pytest.approx(1.0) def test_unknown_inline_execution_overrides_apply_only_to_weak_gap_high_close_subset(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="same_day_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", early_failure_no_progress_days_override=1, early_failure_no_progress_r_override=0.15, unknown_inline_exit_close_location_min=0.90, unknown_inline_exit_gap_size_max=0.02, unknown_inline_early_failure_no_progress_days_override=1, unknown_inline_early_failure_no_progress_r_override=0.30, unknown_inline_early_failure_no_progress_fraction_override=1.0, ) weak_hot_row = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, reaction_day_return=0.11, event_direction="unknown", guidance_status="inline_or_maintained", close_location=0.93, gap_size=0.01, ) c = build_candidate(weak_hot_row, strategy_engine=engine) assert c is not None assert c.engine_early_failure_no_progress_days == 1 assert c.engine_early_failure_no_progress_r == pytest.approx(0.30) assert c.engine_early_failure_no_progress_fraction == pytest.approx(1.0) stronger_gap_row = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, reaction_day_return=0.11, event_direction="unknown", guidance_status="inline_or_maintained", close_location=0.93, gap_size=0.04, ) c2 = build_candidate(stronger_gap_row, strategy_engine=engine) assert c2 is not None assert c2.engine_early_failure_no_progress_days == 1 assert c2.engine_early_failure_no_progress_r == pytest.approx(0.15) def test_engine_route_skips_non_matching_direction(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="short_only_engine", event_types=["earnings"], timing_class="after_close", direction="short_only", ) row = _make_raw_row(reaction_day_return=0.09, trade_direction="long") assert build_candidate(row, strategy_engine=engine) is None def test_engine_route_respects_event_direction_filter(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="bullish_only_engine", event_types=["earnings"], event_directions=["bullish"], timing_class="after_close", direction="long_only", ) row = _make_raw_row( reaction_day_return=0.09, trade_direction="long", event_direction="mixed", ) assert build_candidate(row, strategy_engine=engine) is None def test_engine_route_respects_guidance_status_filter(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="raised_only_engine", event_types=["earnings"], guidance_statuses=["raised"], timing_class="after_close", direction="long_only", ) row = _make_raw_row( reaction_day_return=0.09, trade_direction="long", guidance_status="inline_or_maintained", ) assert build_candidate(row, strategy_engine=engine) is None def test_engine_route_respects_generic_feature_ranges(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="mixed_largecap_priority", event_types=["earnings_release"], event_directions=["mixed"], guidance_statuses=["inline_or_maintained"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", min_market_cap_proxy=8_000_000_000.0, max_market_cap_proxy=25_000_000_000.0, document_quality_score_max=0.71, signal_strength_score_max=0.2, parse_confidence_overall_max=0.70, ) allowed = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, reaction_day_return=0.11, event_direction="mixed", guidance_status="inline_or_maintained", market_cap_proxy=12_000_000_000.0, document_quality_score=0.68, signal_strength_score=0.2, parse_confidence_overall=0.65, ) assert build_candidate(allowed, strategy_engine=engine) is not None too_small = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, reaction_day_return=0.11, event_direction="mixed", guidance_status="inline_or_maintained", market_cap_proxy=5_000_000_000.0, document_quality_score=0.68, signal_strength_score=0.2, parse_confidence_overall=0.65, ) assert build_candidate(too_small, strategy_engine=engine) is None too_clean = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=149.5, reaction_day_return=0.11, event_direction="mixed", guidance_status="inline_or_maintained", market_cap_proxy=12_000_000_000.0, document_quality_score=0.75, signal_strength_score=0.2, parse_confidence_overall=0.65, ) assert build_candidate(too_clean, strategy_engine=engine) is None def test_engine_route_respects_oneoff_and_prior_event_ranges(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="selective_short_proxy", event_types=["earnings_release"], event_directions=["bearish", "mixed"], timing_class="after_close", direction="short_only", forced_trade_direction_override="short", oneoff_penalty_min=0.5, prior_event_fwd5d_max=0.01, ) allowed = _make_raw_row( event_type="earnings_release", event_direction="bearish", oneoff_penalty=0.666667, prior_event_fwd5d=-0.02, reaction_day_return=-0.14, trade_direction="short", ) assert build_candidate(allowed, strategy_engine=engine) is not None low_oneoff = _make_raw_row( event_type="earnings_release", event_direction="bearish", oneoff_penalty=0.2, prior_event_fwd5d=-0.02, reaction_day_return=-0.14, trade_direction="short", ) assert build_candidate(low_oneoff, strategy_engine=engine) is None high_prior = _make_raw_row( event_type="earnings_release", event_direction="bearish", oneoff_penalty=0.666667, prior_event_fwd5d=0.08, reaction_day_return=-0.14, trade_direction="short", ) assert build_candidate(high_prior, strategy_engine=engine) is None def test_engine_route_respects_sentiment_and_earnings_surprise_ranges(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="selective_short_text", event_types=["earnings_release"], event_directions=["bearish", "mixed"], timing_class="after_close", direction="short_only", forced_trade_direction_override="short", lm_net_sentiment_max=0.0, earnings_surprise_pct_max=5.0, ) allowed = _make_raw_row( event_type="earnings_release", event_direction="bearish", lm_net_sentiment=-0.0015, earnings_surprise_pct=-3.0, reaction_day_return=-0.12, trade_direction="short", ) assert build_candidate(allowed, strategy_engine=engine) is not None positive_tone = _make_raw_row( event_type="earnings_release", event_direction="bearish", lm_net_sentiment=0.002, earnings_surprise_pct=-3.0, reaction_day_return=-0.12, trade_direction="short", ) assert build_candidate(positive_tone, strategy_engine=engine) is None too_positive_surprise = _make_raw_row( event_type="earnings_release", event_direction="bearish", lm_net_sentiment=-0.0015, earnings_surprise_pct=12.0, reaction_day_return=-0.12, trade_direction="short", ) assert build_candidate(too_positive_surprise, strategy_engine=engine) is None def test_engine_route_respects_sentiment_surprise_and_price_text_dislocation(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="selective_short_text_dislocation", event_types=["earnings_release"], event_directions=["bearish", "mixed"], timing_class="after_close", direction="short_only", forced_trade_direction_override="short", sentiment_surprise_min=0.02, price_text_dislocation_min=0.02, ) allowed = _make_raw_row( event_type="earnings_release", event_direction="mixed", lm_net_sentiment=-0.03, earnings_surprise_pct=0.0, reaction_day_return=0.005, trade_direction="short", ) assert build_candidate(allowed, strategy_engine=engine) is not None low_sentiment_surprise = _make_raw_row( event_type="earnings_release", event_direction="mixed", lm_net_sentiment=-0.005, earnings_surprise_pct=0.0, reaction_day_return=0.005, trade_direction="short", ) assert build_candidate(low_sentiment_surprise, strategy_engine=engine) is None low_price_text_dislocation = _make_raw_row( event_type="earnings_release", event_direction="mixed", lm_net_sentiment=-0.03, earnings_surprise_pct=0.0, reaction_day_return=0.025, trade_direction="short", ) assert build_candidate(low_price_text_dislocation, strategy_engine=engine) is None def test_engine_route_respects_positive_price_text_dislocation_fields(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="positive_underreaction_ranked", event_types=["earnings_release"], event_directions=["unknown", "mixed", "bullish"], timing_class="after_close", direction="long_only", positive_price_text_dislocation_min=0.05, positive_price_text_dislocation_rank_min=0.8, ) allowed = _make_raw_row( event_type="earnings_release", event_direction="bullish", guidance_status="raised", lm_net_sentiment=0.02, earnings_surprise_pct=12.0, reaction_day_return=0.01, positive_price_text_dislocation=0.13, positive_price_text_dislocation_rank=1.0, trade_direction="long", ) assert build_candidate(allowed, strategy_engine=engine) is not None low_dislocation = _make_raw_row( event_type="earnings_release", event_direction="bullish", guidance_status="raised", lm_net_sentiment=0.02, earnings_surprise_pct=12.0, reaction_day_return=0.01, positive_price_text_dislocation=0.03, positive_price_text_dislocation_rank=1.0, trade_direction="long", ) assert build_candidate(low_dislocation, strategy_engine=engine) is None low_rank = _make_raw_row( event_type="earnings_release", event_direction="bullish", guidance_status="raised", lm_net_sentiment=0.02, earnings_surprise_pct=12.0, reaction_day_return=0.01, positive_price_text_dislocation=0.13, positive_price_text_dislocation_rank=0.4, trade_direction="long", ) assert build_candidate(low_rank, strategy_engine=engine) is None def test_engine_route_respects_peer_relative_surprise_filters(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="peer_relative_surprise_proxy", event_types=["earnings"], earnings_surprise_pct_min=10.0, peer_sector_event_count_365d_min=4.0, peer_relative_surprise_pct_365d_min=15.0, peer_relative_sue_hist_mean_4q_365d_min=0.0, ) allowed = _make_raw_row( earnings_surprise_pct=18.0, peer_sector_event_count_365d=6.0, peer_relative_surprise_pct_365d=22.0, peer_relative_sue_hist_mean_4q_365d=0.8, ) assert build_candidate(allowed, strategy_engine=engine) is not None low_peer_relative_surprise = _make_raw_row( earnings_surprise_pct=18.0, peer_sector_event_count_365d=6.0, peer_relative_surprise_pct_365d=10.0, peer_relative_sue_hist_mean_4q_365d=0.8, ) assert build_candidate(low_peer_relative_surprise, strategy_engine=engine) is None low_peer_history = _make_raw_row( earnings_surprise_pct=18.0, peer_sector_event_count_365d=2.0, peer_relative_surprise_pct_365d=22.0, peer_relative_sue_hist_mean_4q_365d=0.8, ) assert build_candidate(low_peer_history, strategy_engine=engine) is None def test_engine_route_respects_catalyst_persistence_filters(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="catalyst_persistence", event_types=["earnings"], prior_catalyst_count_60d_min=1.0, prior_catalyst_type_diversity_60d_min=1.0, ) allowed = _make_raw_row( prior_catalyst_count_60d=2.0, prior_catalyst_type_diversity_60d=2.0, ) assert build_candidate(allowed, strategy_engine=engine) is not None low_count = _make_raw_row( prior_catalyst_count_60d=0.0, prior_catalyst_type_diversity_60d=2.0, ) assert build_candidate(low_count, strategy_engine=engine) is None low_diversity = _make_raw_row( prior_catalyst_count_60d=2.0, prior_catalyst_type_diversity_60d=0.0, ) assert build_candidate(low_diversity, strategy_engine=engine) is None def test_engine_route_respects_filing_exchange_and_bar_shape_filters(self): from libs.backtest.selector import build_candidate engine = StrategyEngineConfig( engine_id="regular_hours_nyse_gap_conflict", event_types=["earnings_release"], timing_class="same_day", direction="long_only", filing_time_buckets=["regular_hours"], allowed_exchanges=["NYSE"], avg_dollar_volume_max=150_000_000.0, reaction_day_range_pct_max=0.09, upper_wick_pct_max=0.03, ) allowed = _make_raw_row( event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", filing_time_bucket="regular_hours", exchange_proxy="NYSE", event_close=100.0, reaction_day_low=94.0, reaction_day_high=102.0, reaction_day_return=0.08, avg_dollar_volume_20d=120_000_000.0, ) assert build_candidate(allowed, strategy_engine=engine) is not None wrong_bucket = dict(allowed, filing_time_bucket="pre_market") assert build_candidate(wrong_bucket, strategy_engine=engine) is None wrong_exchange = dict(allowed, exchange_proxy="NASDAQ") assert build_candidate(wrong_exchange, strategy_engine=engine) is None too_wide = dict(allowed, reaction_day_low=90.0, reaction_day_high=102.0) assert build_candidate(too_wide, strategy_engine=engine) is None too_wicky = dict(allowed, reaction_day_low=97.0, reaction_day_high=104.5) assert build_candidate(too_wicky, strategy_engine=engine) is None too_liquid = dict(allowed, avg_dollar_volume_20d=220_000_000.0) assert build_candidate(too_liquid, strategy_engine=engine) is None def test_select_candidates_recomputes_pead_score_for_engine_thresholds(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="LOWVOL", score=0.8, event_type="earnings_release", reaction_day_return=0.14, volume_ratio_20d=2.4, gap_size=0.01, ), _make_raw_row( symbol="HIGHVOL", score=0.8, event_type="earnings_release", reaction_day_return=0.14, volume_ratio_20d=4.5, gap_size=0.01, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="pead", score_threshold=0.65, pead_reaction_threshold=0.10, pead_volume_threshold=2.0, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="after_close_long_quality", event_types=["earnings_release"], timing_class="after_close", direction="long_only", pead_volume_threshold_override=3.0, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["HIGHVOL"] def test_select_candidates_uses_engine_score_threshold_override(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="PASS", score=0.78, event_type="earnings_release", reaction_day_return=0.16, volume_ratio_20d=4.0, gap_size=0.01, ), _make_raw_row( symbol="FAIL", score=0.72, event_type="earnings_release", reaction_day_return=0.14, volume_ratio_20d=4.0, gap_size=0.01, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="pead", score_threshold=0.65, pead_reaction_threshold=0.10, pead_volume_threshold=2.0, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="after_close_long_quality", event_types=["earnings_release"], timing_class="after_close", direction="long_only", score_threshold_override=0.75, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["PASS"] def test_select_candidates_filters_weak_reaction_gap_chase(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.018, close_location=0.85, volume_ratio_20d=1.4, gap_size=0.03, ), _make_raw_row( symbol="KEPT", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.018, close_location=0.85, volume_ratio_20d=1.4, gap_size=0.01, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", weak_reaction_threshold=0.03, weak_reaction_gap_max=0.02, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_filters_weak_unknown_direction_reactions(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.021, close_location=0.85, volume_ratio_20d=1.4, event_direction="unknown", ), _make_raw_row( symbol="KEPT", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.051, close_location=0.85, volume_ratio_20d=1.4, event_direction="unknown", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", unknown_direction_reaction_min=0.04, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_filters_unknown_direction_low_close_location(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.061, close_location=0.58, volume_ratio_20d=1.4, event_direction="unknown", ), _make_raw_row( symbol="KEPT", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.061, close_location=0.78, volume_ratio_20d=1.4, event_direction="unknown", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", unknown_direction_close_location_min=0.70, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_filters_unknown_direction_high_close_location(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.061, close_location=0.94, volume_ratio_20d=1.6, event_direction="unknown", ), _make_raw_row( symbol="KEPT", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.061, close_location=0.84, volume_ratio_20d=1.6, event_direction="unknown", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", unknown_direction_close_location_max=0.90, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_filters_unknown_direction_low_gap_size(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.061, close_location=0.78, gap_size=0.01, volume_ratio_20d=1.6, event_direction="unknown", ), _make_raw_row( symbol="KEPT", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", event_close=151.0, reaction_day_return=0.061, close_location=0.78, gap_size=0.03, volume_ratio_20d=1.6, event_direction="unknown", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", unknown_direction_gap_size_min=0.02, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_engine_reaction_day_return_bounds(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="CRASH", score=0.85, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=-0.52, volume_ratio_20d=8.0, trade_direction="short", ), _make_raw_row( symbol="NORMAL", score=0.82, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=-0.18, volume_ratio_20d=5.0, trade_direction="short", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="pead", score_threshold=0.65, pead_reaction_threshold=0.10, pead_volume_threshold=2.0, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_short_filtered", event_types=["earnings_release"], timing_class="same_day", direction="short_only", reaction_day_return_min=-0.45, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["NORMAL"] def test_select_candidates_caps_mixed_inline_close_location(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="TOO_WEAK", score=0.78, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.11, event_close=151.0, close_location=0.55, gap_size=0.03, volume_ratio_20d=2.4, event_direction="mixed", guidance_status="inline_or_maintained", trade_direction="long", ), _make_raw_row( symbol="TOO_HOT", score=0.78, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.11, event_close=151.0, close_location=0.94, gap_size=0.03, volume_ratio_20d=2.4, event_direction="mixed", guidance_status="inline_or_maintained", trade_direction="long", ), _make_raw_row( symbol="KEPT", score=0.77, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.10, event_close=151.0, close_location=0.84, gap_size=0.03, volume_ratio_20d=2.2, event_direction="mixed", guidance_status="inline_or_maintained", trade_direction="long", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", mixed_inline_close_location_min=0.60, mixed_inline_close_location_max=0.90, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_caps_mixed_inline_gap_size(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="TOO_GAPPY", score=0.78, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.11, event_close=151.0, close_location=0.81, gap_size=0.10, volume_ratio_20d=2.4, event_direction="mixed", guidance_status="inline_or_maintained", trade_direction="long", ), _make_raw_row( symbol="KEPT", score=0.77, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.10, event_close=151.0, close_location=0.82, gap_size=0.04, volume_ratio_20d=2.2, event_direction="mixed", guidance_status="inline_or_maintained", trade_direction="long", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_core", event_types=["earnings_release"], timing_class="same_day", direction="long_only", entry_timing_policy="reaction_close", mixed_inline_gap_size_max=0.05, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_raw_macro_vix_bounds(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="LOWVIX", score=0.78, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="bullish", guidance_status="raised", reaction_day_return=0.08, close_location=0.82, gap_size=0.02, volume_ratio_20d=2.0, macro_vix=17.5, ), _make_raw_row( symbol="MIDVIX", score=0.77, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="bullish", guidance_status="raised", reaction_day_return=0.08, close_location=0.82, gap_size=0.02, volume_ratio_20d=2.0, macro_vix=19.5, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="after_close_vix_gate", event_types=["earnings_release"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", macro_vix_min=18.0, macro_vix_max=22.0, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["MIDVIX"] def test_select_candidates_respects_pre_event_noise_feature_bounds(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.78, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="bullish", guidance_status="raised", reaction_day_return=0.08, close_location=0.82, gap_size=0.02, volume_ratio_20d=2.0, pre_event_hurst_60d=0.78, pre_event_entropy_60d=2.04, pre_event_market_temperature=1.42, ), _make_raw_row( symbol="KEPT", score=0.77, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="bullish", guidance_status="raised", reaction_day_return=0.08, close_location=0.82, gap_size=0.02, volume_ratio_20d=2.0, pre_event_hurst_60d=0.58, pre_event_entropy_60d=1.78, pre_event_market_temperature=0.92, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="after_close_noise_gate", event_types=["earnings_release"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", pre_event_hurst_60d_max=0.70, pre_event_entropy_60d_max=1.90, pre_event_market_temperature_max=1.20, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_pre_event_bb_position_bounds(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.78, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="mixed", guidance_status="not_provided", reaction_day_return=0.02, close_location=0.84, gap_size=0.01, volume_ratio_20d=0.9, pre_event_bb_position=0.18, ), _make_raw_row( symbol="KEPT", score=0.77, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="mixed", guidance_status="not_provided", reaction_day_return=0.02, close_location=0.84, gap_size=0.01, volume_ratio_20d=0.9, pre_event_bb_position=0.42, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="mixed_ome_bb_gate", event_types=["other_material_event"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", event_directions=["mixed"], guidance_statuses=["not_provided"], pre_event_bb_position_min=0.25, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_pre_event_gravitational_pull_max(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.77, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="unknown", guidance_status="not_provided", reaction_day_return=0.02, close_location=0.84, gap_size=0.01, volume_ratio_20d=1.1, pre_event_gravitational_pull=3.4, ), _make_raw_row( symbol="KEPT", score=0.76, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="unknown", guidance_status="not_provided", reaction_day_return=0.02, close_location=0.84, gap_size=0.01, volume_ratio_20d=1.1, pre_event_gravitational_pull=2.4, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="unknown_ome_pull_gate", event_types=["other_material_event"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", event_directions=["unknown"], guidance_statuses=["not_provided"], pre_event_gravitational_pull_max=2.7, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_pre_event_short_ratio_max(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.77, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", reaction_day_return=0.12, close_location=0.78, volume_ratio_20d=1.2, event_direction="unknown", guidance_status="not_provided", pre_event_short_ratio=0.64, ), _make_raw_row( symbol="KEPT", score=0.76, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", reaction_day_return=0.12, close_location=0.78, volume_ratio_20d=1.2, event_direction="unknown", guidance_status="not_provided", pre_event_short_ratio=0.48, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="unknown_strong_short_ratio_gate", event_types=["other_material_event"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", event_directions=["unknown"], guidance_statuses=["not_provided"], pre_event_short_ratio_max=0.56, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_pre_event_sector_momentum_20d_min(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="FILTERED", score=0.77, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="unknown", guidance_status="not_provided", pre_event_sector_momentum_20d=-0.03, ), _make_raw_row( symbol="KEPT", score=0.76, event_type="other_material_event", event_date="2026-01-06", reaction_date="2026-01-07", event_direction="unknown", guidance_status="not_provided", pre_event_sector_momentum_20d=0.02, ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="return_max_long_v2", score_threshold=0.45, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="othermat_sector_momentum_gate", event_types=["other_material_event"], timing_class="after_close", direction="long_only", entry_timing_policy="next_open", event_directions=["unknown"], guidance_statuses=["not_provided"], pre_event_sector_momentum_20d_min=-0.01, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["KEPT"] def test_select_candidates_respects_engine_reaction_day_return_upper_bound(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="TOO_HOT", score=0.90, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.42, volume_ratio_20d=6.0, trade_direction="long", ), _make_raw_row( symbol="OK", score=0.80, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.18, volume_ratio_20d=4.0, trade_direction="long", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="pead", score_threshold=0.65, pead_reaction_threshold=0.10, pead_volume_threshold=2.0, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_capped", event_types=["earnings_release"], timing_class="same_day", direction="long_only", reaction_day_return_max=0.30, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["OK"] def test_select_candidates_filters_by_entry_convention(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="EVENTDAY", event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", entry_convention="next_open_after_reaction_close", reaction_day_return=0.18, volume_ratio_20d=4.0, trade_direction="long", ), _make_raw_row( symbol="CONTINUATION", event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", entry_convention="next_open_after_continuation_signal", reaction_day_return=0.18, volume_ratio_20d=4.0, trade_direction="long", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="pead", score_threshold=0.65, pead_reaction_threshold=0.10, pead_volume_threshold=2.0, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="next_open_continuation_only", event_types=["earnings_release"], entry_conventions=["next_open_after_continuation_signal"], timing_class="same_day", direction="long_only", reaction_day_return_max=0.30, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["CONTINUATION"] def test_select_candidates_respects_engine_gap_size_bounds(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="TIGHT", score=0.82, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.18, volume_ratio_20d=4.0, gap_size=0.04, trade_direction="long", ), _make_raw_row( symbol="WIDE", score=0.84, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.18, volume_ratio_20d=4.0, gap_size=0.16, trade_direction="long", ), _make_raw_row( symbol="TOO_WIDE", score=0.86, event_type="earnings_release", event_date="2026-01-06", reaction_date="2026-01-06", reaction_day_return=0.18, volume_ratio_20d=4.0, gap_size=0.34, trade_direction="long", ), ] universe = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0.0) signal = SignalConfig( scoring_model="pead", score_threshold=0.65, pead_reaction_threshold=0.10, pead_volume_threshold=2.0, max_candidates_per_day=5, ) engine = StrategyEngineConfig( engine_id="same_day_long_gapped", event_types=["earnings_release"], timing_class="same_day", direction="long_only", gap_size_min=0.10, gap_size_max=0.30, ) selected = select_candidates(rows, universe, signal, strategy_engine=engine) assert [candidate.symbol for candidate in selected] == ["WIDE"] class TestRankCandidates: def test_sorted_by_score_desc(self): from libs.backtest.selector import build_candidate, rank_candidates rows = [ _make_raw_row(symbol="A", score=0.5, avg_dollar_volume=1e6), _make_raw_row(symbol="B", score=0.8, avg_dollar_volume=1e6), _make_raw_row(symbol="C", score=0.6, avg_dollar_volume=1e6), ] candidates = [build_candidate(r) for r in rows] ranked = rank_candidates([c for c in candidates if c]) assert ranked[0].symbol == "B" assert ranked[1].symbol == "C" assert ranked[2].symbol == "A" def test_tiebreak_by_avg_dollar_volume(self): from libs.backtest.selector import build_candidate, rank_candidates rows = [ _make_raw_row(symbol="A", score=0.7, avg_dollar_volume=1e6), _make_raw_row(symbol="B", score=0.7, avg_dollar_volume=5e6), ] candidates = [build_candidate(r) for r in rows] ranked = rank_candidates([c for c in candidates if c]) assert ranked[0].symbol == "B" # higher avg_dollar_volume def test_tiebreak_by_symbol_asc(self): from libs.backtest.selector import build_candidate, rank_candidates rows = [ _make_raw_row(symbol="Z", score=0.7, avg_dollar_volume=1e6), _make_raw_row(symbol="A", score=0.7, avg_dollar_volume=1e6), ] candidates = [build_candidate(r) for r in rows] ranked = rank_candidates([c for c in candidates if c]) assert ranked[0].symbol == "A" def test_deterministic(self): from libs.backtest.selector import build_candidate, rank_candidates rows = [ _make_raw_row(symbol="C", score=0.9), _make_raw_row(symbol="A", score=0.7), _make_raw_row(symbol="B", score=0.8), ] candidates = [build_candidate(r) for r in rows] r1 = rank_candidates([c for c in candidates if c]) r2 = rank_candidates([c for c in candidates if c]) assert [c.symbol for c in r1] == [c.symbol for c in r2] def test_custom_score_band_tiebreak_can_prefer_positive_prior(self): from libs.backtest.selector import build_candidate, rank_candidates rows = [ _make_raw_row( symbol="A", score=0.621, avg_dollar_volume=1e6, prior_event_fwd5d=0.05, ), _make_raw_row( symbol="B", score=0.629, avg_dollar_volume=1e6, prior_event_fwd5d=-0.04, ), ] candidates = [build_candidate(r) for r in rows] ranked = rank_candidates( [c for c in candidates if c], ["-score_band_2dp", "-prior_positive_flag", "-score"], ) assert [c.symbol for c in ranked] == ["A", "B"] def test_custom_score_band_tiebreak_can_prefer_macro_favorable(self): from libs.backtest.selector import build_candidate, rank_candidates rows = [ _make_raw_row( symbol="A", score=0.621, avg_dollar_volume=1e6, macro_vix=20.0, macro_hy_spread=3.4, ), _make_raw_row( symbol="B", score=0.629, avg_dollar_volume=1e6, macro_vix=14.0, macro_hy_spread=2.8, ), ] candidates = [build_candidate(r) for r in rows] ranked = rank_candidates( [c for c in candidates if c], ["-score_band_2dp", "-macro_favorable_flag", "-score"], ) assert [c.symbol for c in ranked] == ["A", "B"] class TestFilterCandidates: def test_score_threshold(self): from libs.backtest.selector import build_candidate, filter_by_score rows = [ _make_raw_row(symbol="A", score=0.3), _make_raw_row(symbol="B", score=0.7), _make_raw_row(symbol="C", score=0.5), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] filtered = filter_by_score(candidates, score_threshold=0.5) assert len(filtered) == 2 assert all(c.score >= 0.5 for c in filtered) def test_min_price_filter(self): from libs.backtest.selector import build_candidate, filter_by_universe u = UniverseConfig(min_price=100.0, min_avg_dollar_volume=0) rows = [ _make_raw_row(symbol="CHEAP", entry_price=50.0), _make_raw_row(symbol="OK", entry_price=150.0), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] filtered = filter_by_universe(candidates, u) assert len(filtered) == 1 assert filtered[0].symbol == "OK" def test_engine_min_price_override_bypasses_global_floor(self): from libs.backtest.selector import build_candidate, filter_by_universe u = UniverseConfig(min_price=15.0, min_avg_dollar_volume=0) engine = StrategyEngineConfig( engine_id="low_price_pocket", min_entry_price_override=5.0, ) rows = [ _make_raw_row(symbol="LOW", entry_price=9.62), _make_raw_row(symbol="HIGH", entry_price=20.0), ] candidates = [build_candidate(r, strategy_engine=engine) for r in rows] filtered = filter_by_universe([c for c in candidates if c], u) assert {c.symbol for c in filtered} == {"LOW", "HIGH"} def test_engine_max_price_override_filters_high_names(self): from libs.backtest.selector import build_candidate, filter_by_universe u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=0) engine = StrategyEngineConfig( engine_id="low_price_only", max_entry_price_override=15.0, ) rows = [ _make_raw_row(symbol="LOW", entry_price=9.62), _make_raw_row(symbol="HIGH", entry_price=20.0), ] candidates = [build_candidate(r, strategy_engine=engine) for r in rows] filtered = filter_by_universe([c for c in candidates if c], u) assert {c.symbol for c in filtered} == {"LOW"} def test_min_adv_filter(self): from libs.backtest.selector import build_candidate, filter_by_universe u = UniverseConfig(min_price=0, min_avg_dollar_volume=2_000_000) rows = [ _make_raw_row(symbol="ILLIQUID", avg_dollar_volume=500_000), _make_raw_row(symbol="LIQUID", avg_dollar_volume=5_000_000), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] filtered = filter_by_universe(candidates, u) assert len(filtered) == 1 assert filtered[0].symbol == "LIQUID" def test_truncate(self): from libs.backtest.selector import build_candidate, rank_candidates, truncate_candidates rows = [_make_raw_row(symbol=s, score=0.9 - i * 0.1) for i, s in enumerate("ABCDE")] candidates = rank_candidates([build_candidate(r) for r in rows if build_candidate(r)]) truncated = truncate_candidates(candidates, max_per_day=3) assert len(truncated) == 3 class TestFilterByEventType: def test_disabled_event_type_filtered(self): from libs.backtest.selector import build_candidate, filter_by_event_type rows = [ _make_raw_row(symbol="A", event_type="earnings_release"), _make_raw_row(symbol="B", event_type="management_change"), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] profiles = { "earnings_release": EventTypeProfile(enabled=True), "management_change": EventTypeProfile(enabled=False), } filtered = filter_by_event_type(candidates, profiles) assert len(filtered) == 1 assert filtered[0].symbol == "A" def test_per_type_score_threshold(self): from libs.backtest.selector import build_candidate, filter_by_event_type rows = [ _make_raw_row(symbol="A", event_type="earnings_release", score=0.55), _make_raw_row(symbol="B", event_type="earnings_release", score=0.75), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] profiles = { "earnings_release": EventTypeProfile(score_threshold_override=0.6), } filtered = filter_by_event_type(candidates, profiles) assert len(filtered) == 1 assert filtered[0].symbol == "B" def test_unknown_event_type_blocked(self): """Event types not in profiles dict are blocked (default deny).""" from libs.backtest.selector import build_candidate, filter_by_event_type rows = [ _make_raw_row(symbol="A", event_type="earnings_release"), _make_raw_row(symbol="B", event_type="unknown_type"), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] profiles = { "earnings_release": EventTypeProfile(enabled=True), } filtered = filter_by_event_type(candidates, profiles) assert len(filtered) == 1 assert filtered[0].symbol == "A" def test_unknown_event_type_passes_when_in_profiles(self): """Event type 'unknown' passes through when explicitly enabled in profiles.""" from libs.backtest.selector import build_candidate, filter_by_event_type rows = [ _make_raw_row(symbol="A", event_type="earnings_release"), _make_raw_row(symbol="B", event_type="unknown"), ] candidates = [build_candidate(r) for r in rows if build_candidate(r)] profiles = { "earnings_release": EventTypeProfile(enabled=True), "unknown": EventTypeProfile(enabled=True, direction_filter="any"), } filtered = filter_by_event_type(candidates, profiles) assert len(filtered) == 2 symbols = [c.symbol for c in filtered] assert "A" in symbols assert "B" in symbols def test_no_profiles_passthrough(self): from libs.backtest.selector import build_candidate, filter_by_event_type rows = [_make_raw_row(symbol="A")] candidates = [build_candidate(r) for r in rows if build_candidate(r)] assert filter_by_event_type(candidates, {}) == candidates class TestSelectCandidates: def test_full_pipeline(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row(symbol="A", score=0.9, avg_dollar_volume=5e6, entry_price=100.0), _make_raw_row(symbol="B", score=0.3, avg_dollar_volume=5e6, entry_price=100.0), # below threshold _make_raw_row(symbol="C", score=0.8, avg_dollar_volume=1e4, entry_price=100.0), # low ADV _make_raw_row(symbol="D", score=0.7, avg_dollar_volume=5e6, entry_price=2.0), # below min_price ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig(score_threshold=0.5, max_candidates_per_day=10) result = select_candidates(rows, u, s) symbols = [c.symbol for c in result] assert "A" in symbols assert "B" not in symbols # below threshold assert "C" not in symbols # low ADV assert "D" not in symbols # below min_price def test_sector_etf_proxy_dedupes_by_trade_symbol(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( event_id="EVT::TEST::001", symbol="AAPL", score=0.9, sector="Technology", sector_etf_proxy="XLK", sector_etf_event_close=210.0, sector_etf_entry_price=211.0, sector_etf_avg_dollar_volume=250_000_000.0, sector_etf_atr_14=4.0, ), _make_raw_row( event_id="EVT::TEST::002", symbol="MSFT", score=0.8, sector="Technology", sector_etf_proxy="XLK", sector_etf_event_close=210.0, sector_etf_entry_price=211.0, sector_etf_avg_dollar_volume=250_000_000.0, sector_etf_atr_14=4.0, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5) engine = StrategyEngineConfig( engine_id="sector_etf_proxy", event_types=["earnings"], trade_symbol_mode="sector_etf", ) result = select_candidates(rows, u, s, strategy_engine=engine) assert len(result) == 1 assert result[0].symbol == "XLK" assert result[0].source_symbol == "AAPL" def test_sector_etf_proxy_respects_excluded_trade_symbol(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( event_id="EVT::TEST::001", symbol="AAPL", score=0.9, sector="Technology", sector_etf_proxy="XLK", sector_etf_event_close=210.0, sector_etf_entry_price=211.0, sector_etf_avg_dollar_volume=250_000_000.0, sector_etf_atr_14=4.0, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5) engine = StrategyEngineConfig( engine_id="sector_etf_proxy", event_types=["earnings"], trade_symbol_mode="sector_etf", ) result = select_candidates( rows, u, s, strategy_engine=engine, excluded_symbols={"XLK"}, ) assert result == [] def test_peer_proxy_dedupes_by_trade_symbol(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( event_id="EVT::TEST::101", symbol="AVGO", score=0.9, peer_proxy_symbol="NVDA", peer_proxy_event_close=132.0, peer_proxy_entry_price=133.0, peer_proxy_avg_dollar_volume=500_000_000.0, peer_proxy_atr_14=5.0, ), _make_raw_row( event_id="EVT::TEST::102", symbol="AMD", score=0.8, peer_proxy_symbol="NVDA", peer_proxy_event_close=132.0, peer_proxy_entry_price=133.0, peer_proxy_avg_dollar_volume=500_000_000.0, peer_proxy_atr_14=5.0, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig(score_threshold=0.1, max_candidates_per_day=5) engine = StrategyEngineConfig( engine_id="peer_proxy", event_types=["earnings"], trade_symbol_mode="peer_proxy", ) result = select_candidates(rows, u, s, strategy_engine=engine) assert len(result) == 1 assert result[0].symbol == "NVDA" assert result[0].source_symbol == "AVGO" def test_pipeline_with_event_type_profiles(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row(symbol="A", score=0.9, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release"), _make_raw_row(symbol="B", score=0.7, avg_dollar_volume=5e6, entry_price=100.0, event_type="management_change"), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig(score_threshold=0.5, max_candidates_per_day=10) profiles = { "earnings_release": EventTypeProfile(enabled=True), "management_change": EventTypeProfile(enabled=False), } result = select_candidates(rows, u, s, event_type_profiles=profiles) assert len(result) == 1 assert result[0].symbol == "A" def test_pipeline_honors_custom_ranking_fields(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="WIDEGAP", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", gap_size=0.06, close_location=0.85, ), _make_raw_row( symbol="TIGHTGAP", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", gap_size=0.01, close_location=0.70, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig( score_threshold=0.5, max_candidates_per_day=10, ranking_fields=["gap_size", "-close_location"], ) result = select_candidates(rows, u, s) assert [candidate.symbol for candidate in result] == ["TIGHTGAP", "WIDEGAP"] def test_pipeline_honors_engine_ranking_fields_override(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="WIDEGAP", score=0.8, avg_dollar_volume=9e6, entry_price=100.0, event_type="earnings_release", gap_size=0.06, close_location=0.85, ), _make_raw_row( symbol="TIGHTGAP", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", gap_size=0.01, close_location=0.70, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig( score_threshold=0.5, max_candidates_per_day=10, ) engine = StrategyEngineConfig( engine_id="close_location_first", event_types=["earnings_release"], ranking_fields_override=["close_location", "gap_size"], ) result = select_candidates(rows, u, s, strategy_engine=engine) assert [candidate.symbol for candidate in result] == ["TIGHTGAP", "WIDEGAP"] def test_pipeline_computes_positive_price_text_dislocation_ranks(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="TOP", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", event_direction="bullish", guidance_status="raised", lm_net_sentiment=0.02, earnings_surprise_pct=15.0, reaction_day_return=-0.02, ), _make_raw_row( symbol="LOW", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", event_direction="bullish", guidance_status="raised", lm_net_sentiment=0.003, earnings_surprise_pct=5.0, reaction_day_return=0.01, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig(score_threshold=0.5, max_candidates_per_day=10) engine = StrategyEngineConfig( engine_id="ranked_positive_underreaction", event_types=["earnings_release"], event_directions=["bullish"], positive_price_text_dislocation_rank_min=0.8, ranking_fields_override=["-positive_price_text_dislocation_rank", "-score"], ) result = select_candidates(rows, u, s, strategy_engine=engine) assert [candidate.symbol for candidate in result] == ["TOP"] assert result[0].features["positive_price_text_dislocation_rank"] == pytest.approx(1.0) def test_pipeline_honors_guidance_rank_flags(self): from libs.backtest.selector import select_candidates rows = [ _make_raw_row( symbol="INLINE", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", guidance_status="inline_or_maintained", close_location=0.90, ), _make_raw_row( symbol="RAISED", score=0.8, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", guidance_status="raised", close_location=0.70, ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig( score_threshold=0.5, max_candidates_per_day=10, ranking_fields=["-guidance_raised_flag", "-close_location"], ) result = select_candidates(rows, u, s) assert [candidate.symbol for candidate in result] == ["RAISED", "INLINE"] def test_pipeline_honors_learned_ranking_model(self, tmp_path): from libs.backtest.selector import select_candidates model_path = tmp_path / "ranker.json" model_path.write_text(json.dumps({ "model_type": "bucket_blend_v1", "global_mean": 0.01, "features": [ { "name": "direction_guidance_combo", "weight": 1.0, "values": { "bullish|raised": 0.08, "unknown|inline_or_maintained": 0.02, }, }, ], })) rows = [ _make_raw_row( symbol="UNKNOWN", score=0.95, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", event_direction="unknown", guidance_status="inline_or_maintained", ), _make_raw_row( symbol="RAISED", score=0.70, avg_dollar_volume=5e6, entry_price=100.0, event_type="earnings_release", event_direction="bullish", guidance_status="raised", ), ] u = UniverseConfig(min_price=5.0, min_avg_dollar_volume=1_000_000) s = SignalConfig( score_threshold=0.5, max_candidates_per_day=10, ranking_model_path=str(model_path), ranking_fields=["-ranking_model_score", "-score"], ) result = select_candidates(rows, u, s) assert [candidate.symbol for candidate in result] == ["RAISED", "UNKNOWN"] assert result[0].features["ranking_model_score"] == pytest.approx(0.08)