"""Tests for the stock-specific impact model layer. Covers: - Task 36: Event-time feature snapshots - Task 37: Outcome labels - Task 38: Deterministic impact baseline - Task 39: Trained tabular impact model - Task 40: Impact output integration """ from __future__ import annotations import math from datetime import datetime, timedelta, timezone import pytest from hypothesis import given, settings from hypothesis import strategies as st from services.intelligence_pipeline_v3.impact.baseline import ( EVENT_CLASS_BASE_MAGNITUDE, EVENT_CLASS_DIRECTION, DeterministicImpactBaseline, ) from services.intelligence_pipeline_v3.impact.features import ( ImpactFeatureSet, clear_feature_snapshots, get_feature_snapshot, persist_feature_snapshot, validate_no_future_leakage, ) from services.intelligence_pipeline_v3.impact.integration import ( ComparisonMetric, DirectionProbabilities, HorizonProbabilities, ImpactPipelineConfig, ImpactPredictionOutput, clear_comparison_metrics, filter_generative_scores, get_comparison_metrics, map_to_legacy_impact, record_comparison_metric, ) from services.intelligence_pipeline_v3.impact.labels import ( LABEL_GENERATOR_VERSION, compute_abnormal_return, compute_abnormal_volume, compute_time_to_peak, generate_outcome_labels, ) from services.intelligence_pipeline_v3.impact.trained_model import ( ImpactModelTrainer, TrainingExample, clear_artifact_registry, create_walk_forward_splits, get_approved_model, get_model_artifact, register_model_artifact, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- EVENT_TIME = datetime(2024, 6, 15, 14, 30, 0, tzinfo=timezone.utc) def _make_feature_set(**overrides) -> ImpactFeatureSet: """Create a valid ImpactFeatureSet with sensible defaults.""" defaults = { "event_class_probabilities": {"earnings_beat": 0.7, "guidance_raise": 0.2, "other": 0.1}, "sentiment_positive": 0.6, "sentiment_negative": 0.1, "sentiment_neutral": 0.3, "magnitude": 0.05, "surprise": 0.7, "source_credibility": 0.8, "novelty_score": 0.6, "evidence_coverage": 0.9, "company_sector": "Technology", "company_industry": "Software", "market_cap_bucket": "large", "beta": 1.2, "pre_event_volatility": 0.25, "volume_regime": "normal", "broad_market_regime": "bull", "event_directness": "direct", "document_type": "news", "event_time": EVENT_TIME, "feature_version": "1.0.0", } defaults.update(overrides) return ImpactFeatureSet(**defaults) def _make_price_series( start: datetime, n_points: int = 100, base_price: float = 100.0, daily_return: float = 0.001 ) -> list[tuple[datetime, float]]: """Generate a simple price series.""" series = [] price = base_price for i in range(n_points): ts = start + timedelta(hours=i) series.append((ts, price)) price *= 1.0 + daily_return return series # =========================================================================== # Task 36: Event-time feature snapshots # =========================================================================== class TestImpactFeatureSet: """Tests for ImpactFeatureSet model and snapshot persistence.""" def test_create_valid_feature_set(self): fs = _make_feature_set() assert fs.sentiment_positive == 0.6 assert fs.company_sector == "Technology" assert fs.event_time == EVENT_TIME def test_missing_numeric_as_nan(self): fs = _make_feature_set(magnitude=float("nan"), surprise=float("nan")) assert math.isnan(fs.magnitude) assert math.isnan(fs.surprise) def test_unknown_categorical_defaults(self): fs = _make_feature_set( company_sector="unknown", volume_regime="unknown", broad_market_regime="unknown", ) assert fs.company_sector == "unknown" assert fs.volume_regime == "unknown" assert fs.broad_market_regime == "unknown" def test_invalid_categorical_becomes_unknown(self): fs = _make_feature_set(market_cap_bucket="supermassive") assert fs.market_cap_bucket == "unknown" def test_to_numeric_vector_returns_list(self): fs = _make_feature_set() vec = fs.to_numeric_vector() assert isinstance(vec, list) assert all(isinstance(v, float) for v in vec) def test_persist_feature_snapshot_immutable(self): clear_feature_snapshots() fs = _make_feature_set() pred_time = EVENT_TIME + timedelta(minutes=5) snapshot_id = persist_feature_snapshot(fs, pred_time) assert snapshot_id assert get_feature_snapshot(snapshot_id) is not None # Same content produces same ID (content-addressed) snapshot_id_2 = persist_feature_snapshot(fs, pred_time) assert snapshot_id == snapshot_id_2 def test_persist_rejects_future_prediction_time(self): clear_feature_snapshots() fs = _make_feature_set() # prediction_time before event_time is invalid pred_time = EVENT_TIME - timedelta(hours=1) with pytest.raises(ValueError, match="cannot be before"): persist_feature_snapshot(fs, pred_time) def test_validate_no_future_leakage_clean(self): fs = _make_feature_set() timestamps = [EVENT_TIME - timedelta(hours=i) for i in range(1, 5)] violations = validate_no_future_leakage(fs, timestamps) assert violations == [] def test_validate_no_future_leakage_detects_post_event(self): fs = _make_feature_set() timestamps = [ EVENT_TIME - timedelta(hours=1), EVENT_TIME + timedelta(hours=1), # LEAKAGE ] violations = validate_no_future_leakage(fs, timestamps) assert len(violations) == 1 assert "after" in violations[0] # =========================================================================== # Task 36.4: Property test — no post-event timestamps in features # =========================================================================== # Hypothesis strategy for valid feature sets @st.composite def feature_set_strategy(draw): event_time = draw( st.datetimes( min_value=datetime(2020, 1, 1), max_value=datetime(2025, 1, 1), timezones=st.just(timezone.utc), ) ) return _make_feature_set(event_time=event_time) @st.composite def pre_event_timestamps_strategy(draw, event_time: datetime): """Generate timestamps all before event_time.""" n = draw(st.integers(min_value=1, max_value=10)) timestamps = [] for _ in range(n): offset_seconds = draw(st.integers(min_value=1, max_value=86400 * 30)) timestamps.append(event_time - timedelta(seconds=offset_seconds)) return timestamps @given( event_time=st.datetimes( min_value=datetime(2020, 1, 1), max_value=datetime(2025, 1, 1), timezones=st.just(timezone.utc), ), n_timestamps=st.integers(min_value=1, max_value=10), ) @settings(max_examples=100) def test_pbt_no_post_event_leakage(event_time, n_timestamps): """**Validates: Requirements 12.2, 12.10** Property: When market_data_timestamps are all before event_time, validate_no_future_leakage returns no violations. """ fs = _make_feature_set(event_time=event_time) # All timestamps strictly before event_time timestamps = [ event_time - timedelta(seconds=i + 1) for i in range(n_timestamps) ] violations = validate_no_future_leakage(fs, timestamps) assert violations == [], f"Expected no leakage but found: {violations}" @given( event_time=st.datetimes( min_value=datetime(2020, 1, 1), max_value=datetime(2024, 12, 31), timezones=st.just(timezone.utc), ), post_offset_seconds=st.integers(min_value=0, max_value=86400), ) @settings(max_examples=100) def test_pbt_detects_post_event_leakage(event_time, post_offset_seconds): """**Validates: Requirements 12.2, 12.10** Property: When any market_data_timestamp is at or after event_time, validate_no_future_leakage detects the violation. """ fs = _make_feature_set(event_time=event_time) # Include one timestamp at or after event_time timestamps = [event_time + timedelta(seconds=post_offset_seconds)] violations = validate_no_future_leakage(fs, timestamps) assert len(violations) >= 1 # =========================================================================== # Task 37: Outcome labels # =========================================================================== class TestOutcomeLabels: """Tests for abnormal return computation and label generation.""" def test_compute_abnormal_return_basic(self): """Asset goes up 5%, benchmark goes up 2% → abnormal = 3%.""" event = EVENT_TIME price_series = [ (event, 100.0), (event + timedelta(days=1), 105.0), ] bench_series = [ (event, 100.0), (event + timedelta(days=1), 102.0), ] result = compute_abnormal_return( price_series, bench_series, event, timedelta(days=1) ) assert abs(result - 0.03) < 1e-10 def test_compute_abnormal_return_negative(self): """Asset goes down 3%, benchmark goes up 1% → abnormal = -4%.""" event = EVENT_TIME price_series = [ (event, 100.0), (event + timedelta(days=1), 97.0), ] bench_series = [ (event, 100.0), (event + timedelta(days=1), 101.0), ] result = compute_abnormal_return( price_series, bench_series, event, timedelta(days=1) ) assert abs(result - (-0.04)) < 1e-10 def test_compute_abnormal_return_empty_series_raises(self): with pytest.raises(ValueError, match="empty"): compute_abnormal_return([], [(EVENT_TIME, 100.0)], EVENT_TIME, timedelta(days=1)) def test_compute_abnormal_return_zero_price_raises(self): event = EVENT_TIME price_series = [(event, 0.0), (event + timedelta(days=1), 5.0)] bench_series = [(event, 100.0), (event + timedelta(days=1), 101.0)] with pytest.raises(ValueError, match="zero"): compute_abnormal_return(price_series, bench_series, event, timedelta(days=1)) def test_compute_abnormal_volume(self): event = EVENT_TIME # Trailing: 20 days of volume=1000 volume_series = [ (event - timedelta(days=i), 1000.0) for i in range(1, 21) ] # Event day: volume=3000 (3x normal) volume_series.append((event, 3000.0)) volume_series.sort(key=lambda x: x[0]) result = compute_abnormal_volume(volume_series, event, timedelta(days=1)) assert result is not None assert abs(result - 3.0) < 0.1 def test_compute_time_to_peak(self): event = EVENT_TIME # Price spikes 2 hours after event price_series = [ (event, 100.0), (event + timedelta(hours=1), 101.0), (event + timedelta(hours=2), 105.0), # Peak (event + timedelta(hours=3), 103.0), (event + timedelta(hours=4), 102.0), ] result = compute_time_to_peak(price_series, event, timedelta(hours=6)) assert result is not None assert abs(result - 2.0) < 0.1 def test_generate_outcome_labels_all_horizons(self): event = EVENT_TIME # Generate enough price data for 90d horizon price_series = _make_price_series(event - timedelta(hours=1), n_points=2200) bench_series = _make_price_series(event - timedelta(hours=1), n_points=2200, daily_return=0.0005) labels = generate_outcome_labels( ticker="AAPL", event_time=event, price_series=price_series, benchmark_series=bench_series, ) assert labels.ticker == "AAPL" assert labels.label_generator_version == LABEL_GENERATOR_VERSION assert len(labels.labels) == 5 # All 5 horizons def test_label_generator_version_tracked(self): assert LABEL_GENERATOR_VERSION == "1.0.0" # =========================================================================== # Task 38: Deterministic impact baseline # =========================================================================== class TestDeterministicImpactBaseline: """Tests for the rule-based baseline model.""" def setup_method(self): self.baseline = DeterministicImpactBaseline() def test_predict_returns_impact_prediction(self): fs = _make_feature_set() prediction = self.baseline.predict(fs) assert prediction.direction_probabilities is not None assert prediction.expected_magnitude > 0 assert prediction.model_source.startswith("deterministic_baseline") def test_earnings_beat_positive_direction(self): """Earnings beat should have predominantly positive direction.""" fs = _make_feature_set( event_class_probabilities={"earnings_beat": 0.9, "other": 0.1}, sentiment_positive=0.7, sentiment_negative=0.1, sentiment_neutral=0.2, ) prediction = self.baseline.predict(fs) assert prediction.direction_probabilities["positive"] > prediction.direction_probabilities["negative"] def test_earnings_miss_negative_direction(self): """Earnings miss should have predominantly negative direction.""" fs = _make_feature_set( event_class_probabilities={"earnings_miss": 0.9, "other": 0.1}, sentiment_positive=0.1, sentiment_negative=0.7, sentiment_neutral=0.2, ) prediction = self.baseline.predict(fs) assert prediction.direction_probabilities["negative"] > prediction.direction_probabilities["positive"] def test_guidance_raise_positive(self): fs = _make_feature_set( event_class_probabilities={"guidance_raise": 0.8, "other": 0.2}, sentiment_positive=0.6, sentiment_negative=0.1, sentiment_neutral=0.3, ) prediction = self.baseline.predict(fs) assert prediction.direction_probabilities["positive"] > 0.4 def test_guidance_cut_negative(self): fs = _make_feature_set( event_class_probabilities={"guidance_cut": 0.8, "other": 0.2}, sentiment_positive=0.1, sentiment_negative=0.6, sentiment_neutral=0.3, ) prediction = self.baseline.predict(fs) assert prediction.direction_probabilities["negative"] > 0.4 def test_ma_announcement_high_magnitude(self): """M&A has higher base magnitude than dividend changes.""" fs_ma = _make_feature_set( event_class_probabilities={"ma_announcement": 0.9, "other": 0.1}, ) fs_div = _make_feature_set( event_class_probabilities={"dividend_change": 0.9, "other": 0.1}, ) pred_ma = self.baseline.predict(fs_ma) pred_div = self.baseline.predict(fs_div) assert pred_ma.expected_magnitude > pred_div.expected_magnitude def test_novelty_amplifies_magnitude(self): """Higher novelty should increase expected magnitude.""" fs_novel = _make_feature_set(novelty_score=0.9) fs_stale = _make_feature_set(novelty_score=0.1) pred_novel = self.baseline.predict(fs_novel) pred_stale = self.baseline.predict(fs_stale) assert pred_novel.expected_magnitude > pred_stale.expected_magnitude def test_low_evidence_discounts_magnitude(self): """Low evidence coverage should reduce magnitude.""" fs_strong = _make_feature_set(evidence_coverage=0.95) fs_weak = _make_feature_set(evidence_coverage=0.1) pred_strong = self.baseline.predict(fs_strong) pred_weak = self.baseline.predict(fs_weak) assert pred_strong.expected_magnitude > pred_weak.expected_magnitude def test_direct_event_shorter_horizon(self): """Direct events should have more weight on shorter horizons.""" fs_direct = _make_feature_set(event_directness="direct") fs_spec = _make_feature_set(event_directness="speculative") pred_direct = self.baseline.predict(fs_direct) pred_spec = self.baseline.predict(fs_spec) assert pred_direct.horizon_probabilities["intraday"] > pred_spec.horizon_probabilities["intraday"] def test_unknown_event_high_uncertainty(self): """Unknown events should produce higher uncertainty.""" fs_known = _make_feature_set( event_class_probabilities={"earnings_beat": 0.95, "other": 0.05}, ) fs_unknown = _make_feature_set( event_class_probabilities={}, ) pred_known = self.baseline.predict(fs_known) pred_unknown = self.baseline.predict(fs_unknown) assert pred_unknown.uncertainty > pred_known.uncertainty def test_all_event_classes_have_mappings(self): """Every event class in the lookup tables should produce valid output.""" for event_class in EVENT_CLASS_BASE_MAGNITUDE: fs = _make_feature_set( event_class_probabilities={event_class: 0.9, "other": 0.1}, ) prediction = self.baseline.predict(fs) assert prediction.expected_magnitude > 0 total_dir = sum(prediction.direction_probabilities.values()) assert abs(total_dir - 1.0) < 0.01 total_hor = sum(prediction.horizon_probabilities.values()) assert abs(total_hor - 1.0) < 0.01 def test_direction_probabilities_sum_to_one(self): """Direction probs should always sum to approximately 1.0.""" for event_class in EVENT_CLASS_DIRECTION: fs = _make_feature_set( event_class_probabilities={event_class: 0.8, "other": 0.2}, ) prediction = self.baseline.predict(fs) total = sum(prediction.direction_probabilities.values()) assert abs(total - 1.0) < 0.01, f"Failed for {event_class}: sum={total}" def test_horizon_probabilities_sum_to_one(self): """Horizon probs should always sum to approximately 1.0.""" for event_class in EVENT_CLASS_DIRECTION: fs = _make_feature_set( event_class_probabilities={event_class: 0.8, "other": 0.2}, ) prediction = self.baseline.predict(fs) total = sum(prediction.horizon_probabilities.values()) assert abs(total - 1.0) < 0.01, f"Failed for {event_class}: sum={total}" def test_magnitude_bounded(self): """Magnitude should never exceed 2x base (conservative cap).""" for event_class, base_mag in EVENT_CLASS_BASE_MAGNITUDE.items(): fs = _make_feature_set( event_class_probabilities={event_class: 0.95, "other": 0.05}, novelty_score=1.0, surprise=1.0, evidence_coverage=1.0, ) prediction = self.baseline.predict(fs) assert prediction.expected_magnitude <= base_mag * 2.0 + 0.001 # =========================================================================== # Task 39: Trained tabular impact model # =========================================================================== class TestTrainedImpactModel: """Tests for the trained tabular impact model.""" def setup_method(self): clear_artifact_registry() def _make_training_examples(self, n: int = 50) -> list[TrainingExample]: """Generate synthetic training examples.""" from services.intelligence_pipeline_v3.impact.labels import ( OutcomeLabel, OutcomeLabelSet, ) examples = [] base_time = datetime(2023, 1, 1, tzinfo=timezone.utc) for i in range(n): event_time = base_time + timedelta(days=i) features = _make_feature_set( event_time=event_time, event_class_probabilities={"earnings_beat": 0.7, "other": 0.3}, sentiment_positive=0.5 + 0.3 * (i % 2), sentiment_negative=0.2 - 0.1 * (i % 2), sentiment_neutral=0.3 - 0.2 * (i % 2), ) labels = OutcomeLabelSet( event_time=event_time, ticker="AAPL", labels=[ OutcomeLabel( horizon="1d", signed_return=0.02 * (1 if i % 2 == 0 else -1), absolute_return=0.02, ), ], ) examples.append(TrainingExample( features=features, labels=labels, ticker="AAPL", event_time=event_time )) return examples def test_train_produces_model_card(self): examples = self._make_training_examples(50) trainer = ImpactModelTrainer() card = trainer.train(examples) assert card.model_id assert card.method == "gradient_boosted" assert card.feature_version == "1.0.0" assert card.total_training_samples > 0 def test_train_empty_raises(self): trainer = ImpactModelTrainer() with pytest.raises(ValueError, match="empty"): trainer.train([]) def test_walk_forward_splits_temporal_ordering(self): start = datetime(2023, 1, 1, tzinfo=timezone.utc) end = datetime(2024, 1, 1, tzinfo=timezone.utc) splits = create_walk_forward_splits(start, end, n_splits=3) assert len(splits) == 3 for split in splits: assert split.train_start <= split.train_end assert split.train_end <= split.calibration_end assert split.calibration_end <= split.validation_end def test_trainer_predict_untrained_falls_to_baseline(self): trainer = ImpactModelTrainer() fs = _make_feature_set() prediction = trainer.predict(fs) assert "baseline" in prediction.model_source def test_trainer_predict_after_training(self): examples = self._make_training_examples(50) trainer = ImpactModelTrainer() trainer.train(examples) assert trainer.is_trained fs = _make_feature_set() prediction = trainer.predict(fs) assert prediction.direction_probabilities is not None assert prediction.expected_magnitude >= 0 def test_register_and_retrieve_artifact(self): examples = self._make_training_examples(30) trainer = ImpactModelTrainer() card = trainer.train(examples) model_id = register_model_artifact(card) assert get_model_artifact(model_id) is not None def test_no_approved_model_initially(self): assert get_approved_model() is None def test_model_card_has_segment_metrics(self): examples = self._make_training_examples(50) trainer = ImpactModelTrainer() card = trainer.train(examples) # Should have metrics by event and sector assert isinstance(card.metrics_by_event, list) assert isinstance(card.metrics_by_sector, list) # =========================================================================== # Task 40: Impact output integration # =========================================================================== class TestImpactIntegration: """Tests for impact output integration and legacy compatibility.""" def setup_method(self): clear_comparison_metrics() def test_impact_prediction_output_model(self): pred = ImpactPredictionOutput( direction_probs=DirectionProbabilities(positive=0.6, negative=0.2, neutral=0.2), expected_magnitude=0.03, signed_magnitude=0.02, horizon_probs=HorizonProbabilities( intraday=0.3, one_day=0.3, seven_day=0.2, thirty_day=0.1, ninety_day=0.1 ), uncertainty=0.3, model_source="deterministic_baseline_v1.0.0", ) assert pred.direction_probs.positive == 0.6 assert pred.expected_magnitude == 0.03 def test_map_to_legacy_impact_score(self): pred = ImpactPredictionOutput( direction_probs=DirectionProbabilities(positive=0.7, negative=0.1, neutral=0.2), expected_magnitude=0.04, signed_magnitude=0.03, horizon_probs=HorizonProbabilities( intraday=0.1, one_day=0.4, seven_day=0.3, thirty_day=0.15, ninety_day=0.05 ), uncertainty=0.3, model_source="test", ) legacy = map_to_legacy_impact(pred) assert legacy.impact_score == 0.03 assert legacy.impact_horizon == "1d" def test_map_to_legacy_clamps_score(self): pred = ImpactPredictionOutput( direction_probs=DirectionProbabilities(positive=0.9, negative=0.0, neutral=0.1), expected_magnitude=2.0, signed_magnitude=1.5, # Exceeds 1.0 horizon_probs=HorizonProbabilities(intraday=0.5, one_day=0.3), uncertainty=0.2, model_source="test", ) legacy = map_to_legacy_impact(pred) assert legacy.impact_score == 1.0 # Clamped def test_filter_generative_scores_v3_mode(self): config = ImpactPipelineConfig(v3_mode_enabled=True) signal = { "impact_score": 0.5, "impact_horizon": "7d", "novelty_score": 0.8, "confidence": 0.7, "sentiment": "positive", "ticker": "AAPL", } filtered = filter_generative_scores(signal, config) assert "impact_score" not in filtered assert "impact_horizon" not in filtered assert "novelty_score" not in filtered assert "confidence" not in filtered # Non-generative fields preserved assert filtered["sentiment"] == "positive" assert filtered["ticker"] == "AAPL" def test_filter_generative_scores_disabled(self): config = ImpactPipelineConfig(v3_mode_enabled=False) signal = {"impact_score": 0.5, "ticker": "AAPL"} filtered = filter_generative_scores(signal, config) assert filtered == signal # No filtering when disabled def test_comparison_metrics_disabled_by_default(self): metric = ComparisonMetric( ticker="AAPL", event_time=EVENT_TIME, prediction_source="baseline", predicted_direction="positive", predicted_magnitude=0.03, predicted_horizon="1d", ) # Default config has comparison disabled record_comparison_metric(metric) assert get_comparison_metrics() == [] def test_comparison_metrics_when_enabled(self): import os os.environ["IMPACT_COMPARISON_METRICS"] = "true" try: metric = ComparisonMetric( ticker="AAPL", event_time=EVENT_TIME, prediction_source="baseline", predicted_direction="positive", predicted_magnitude=0.03, predicted_horizon="1d", ) record_comparison_metric(metric) metrics = get_comparison_metrics() assert len(metrics) == 1 assert metrics[0].ticker == "AAPL" finally: os.environ.pop("IMPACT_COMPARISON_METRICS", None) clear_comparison_metrics()