"""Deterministic impact baseline model. Provides conservative, rule-based impact predictions when no trained model is available or approved. Maps event class + sentiment + magnitude + novelty to signed impact and horizon predictions. Design reference: Section I (Impact and Horizon Model) — Model family. Requirement 12.4, 12.8. """ from __future__ import annotations import math from services.intelligence_pipeline_v3.impact.features import ImpactFeatureSet # --------------------------------------------------------------------------- # Impact prediction output (shared with trained model) # --------------------------------------------------------------------------- class ImpactPrediction: """Result of an impact model prediction. Attributes ---------- direction_probabilities : dict Probabilities for positive, negative, neutral outcomes. expected_magnitude : float Expected absolute move magnitude. signed_magnitude : float Direction-weighted expected magnitude. horizon_probabilities : dict Probability distribution over horizons. uncertainty : float Model uncertainty estimate (higher = less confident). model_source : str Which model produced this prediction. """ def __init__( self, direction_probabilities: dict[str, float], expected_magnitude: float, signed_magnitude: float, horizon_probabilities: dict[str, float], uncertainty: float, model_source: str = "deterministic_baseline", ) -> None: self.direction_probabilities = direction_probabilities self.expected_magnitude = expected_magnitude self.signed_magnitude = signed_magnitude self.horizon_probabilities = horizon_probabilities self.uncertainty = uncertainty self.model_source = model_source def to_dict(self) -> dict: return { "direction_probabilities": self.direction_probabilities, "expected_magnitude": self.expected_magnitude, "signed_magnitude": self.signed_magnitude, "horizon_probabilities": self.horizon_probabilities, "uncertainty": self.uncertainty, "model_source": self.model_source, } # --------------------------------------------------------------------------- # Event class impact mappings (conservative) # --------------------------------------------------------------------------- # Base magnitude for each event class (conservative estimates) EVENT_CLASS_BASE_MAGNITUDE: dict[str, float] = { "earnings_beat": 0.04, "earnings_miss": 0.05, "guidance_raise": 0.03, "guidance_cut": 0.04, "product_launch": 0.02, "legal_regulatory": 0.03, "ma_announcement": 0.06, "supply_chain": 0.02, "rating_change": 0.02, "macro_event": 0.01, "management_change": 0.02, "dividend_change": 0.01, "buyback": 0.01, } # Default direction bias for event classes (positive, negative, neutral) EVENT_CLASS_DIRECTION: dict[str, tuple[float, float, float]] = { "earnings_beat": (0.70, 0.10, 0.20), "earnings_miss": (0.10, 0.70, 0.20), "guidance_raise": (0.65, 0.10, 0.25), "guidance_cut": (0.10, 0.65, 0.25), "product_launch": (0.50, 0.15, 0.35), "legal_regulatory": (0.15, 0.55, 0.30), "ma_announcement": (0.40, 0.25, 0.35), "supply_chain": (0.15, 0.50, 0.35), "rating_change": (0.45, 0.30, 0.25), "macro_event": (0.30, 0.30, 0.40), "management_change": (0.30, 0.30, 0.40), "dividend_change": (0.50, 0.20, 0.30), "buyback": (0.55, 0.15, 0.30), } # Default horizon distribution for event classes EVENT_CLASS_HORIZON: dict[str, dict[str, float]] = { "earnings_beat": {"intraday": 0.40, "1d": 0.30, "7d": 0.15, "30d": 0.10, "90d": 0.05}, "earnings_miss": {"intraday": 0.45, "1d": 0.30, "7d": 0.15, "30d": 0.07, "90d": 0.03}, "guidance_raise": {"intraday": 0.25, "1d": 0.30, "7d": 0.20, "30d": 0.15, "90d": 0.10}, "guidance_cut": {"intraday": 0.30, "1d": 0.30, "7d": 0.20, "30d": 0.13, "90d": 0.07}, "product_launch": {"intraday": 0.15, "1d": 0.20, "7d": 0.25, "30d": 0.25, "90d": 0.15}, "legal_regulatory": {"intraday": 0.20, "1d": 0.20, "7d": 0.20, "30d": 0.20, "90d": 0.20}, "ma_announcement": {"intraday": 0.35, "1d": 0.30, "7d": 0.20, "30d": 0.10, "90d": 0.05}, "supply_chain": {"intraday": 0.10, "1d": 0.15, "7d": 0.25, "30d": 0.30, "90d": 0.20}, "rating_change": {"intraday": 0.35, "1d": 0.30, "7d": 0.20, "30d": 0.10, "90d": 0.05}, "macro_event": {"intraday": 0.15, "1d": 0.20, "7d": 0.25, "30d": 0.25, "90d": 0.15}, "management_change": {"intraday": 0.15, "1d": 0.20, "7d": 0.25, "30d": 0.25, "90d": 0.15}, "dividend_change": {"intraday": 0.20, "1d": 0.25, "7d": 0.25, "30d": 0.20, "90d": 0.10}, "buyback": {"intraday": 0.15, "1d": 0.20, "7d": 0.25, "30d": 0.25, "90d": 0.15}, } # Default for unknown event classes _DEFAULT_DIRECTION = (0.30, 0.30, 0.40) _DEFAULT_MAGNITUDE = 0.015 _DEFAULT_HORIZON = {"intraday": 0.20, "1d": 0.20, "7d": 0.20, "30d": 0.20, "90d": 0.20} # --------------------------------------------------------------------------- # Baseline model # --------------------------------------------------------------------------- class DeterministicImpactBaseline: """Rule-based impact prediction using event class + sentiment + magnitude + novelty. This is the fallback model used when no trained model has been approved. It produces conservative, explainable predictions based on fixed mappings. The baseline NEVER uses a generative model's self-scored impact. """ MODEL_VERSION = "1.0.0" def predict(self, features: ImpactFeatureSet) -> ImpactPrediction: """Produce an impact prediction from pre-event features. Parameters ---------- features The event-time feature snapshot. Returns ------- ImpactPrediction Conservative direction, magnitude, and horizon prediction. """ # Determine primary event class (highest probability) primary_event = self._get_primary_event_class(features) # Get base predictions from event class base_direction = EVENT_CLASS_DIRECTION.get(primary_event, _DEFAULT_DIRECTION) base_magnitude = EVENT_CLASS_BASE_MAGNITUDE.get(primary_event, _DEFAULT_MAGNITUDE) base_horizon = EVENT_CLASS_HORIZON.get(primary_event, _DEFAULT_HORIZON) # Adjust direction by sentiment direction = self._adjust_direction_by_sentiment(base_direction, features) # Adjust magnitude by surprise, novelty, and evidence coverage magnitude = self._adjust_magnitude(base_magnitude, features) # Compute signed magnitude signed_magnitude = magnitude * (direction[0] - direction[1]) # Adjust horizon by event directness horizon = self._adjust_horizon(base_horizon, features) # Compute uncertainty (higher for unknown/speculative events) uncertainty = self._compute_uncertainty(features, primary_event) return ImpactPrediction( direction_probabilities={ "positive": direction[0], "negative": direction[1], "neutral": direction[2], }, expected_magnitude=magnitude, signed_magnitude=signed_magnitude, horizon_probabilities=horizon, uncertainty=uncertainty, model_source=f"deterministic_baseline_v{self.MODEL_VERSION}", ) def _get_primary_event_class(self, features: ImpactFeatureSet) -> str: """Get the highest-probability event class.""" if not features.event_class_probabilities: return "unknown" return max( features.event_class_probabilities, key=lambda k: features.event_class_probabilities[k], ) def _adjust_direction_by_sentiment( self, base_direction: tuple[float, float, float], features: ImpactFeatureSet, ) -> tuple[float, float, float]: """Blend event-class direction with calibrated sentiment. Uses a 60/40 split: 60% event class prior, 40% sentiment signal. """ event_weight = 0.6 sentiment_weight = 0.4 pos = event_weight * base_direction[0] + sentiment_weight * features.sentiment_positive neg = event_weight * base_direction[1] + sentiment_weight * features.sentiment_negative neu = event_weight * base_direction[2] + sentiment_weight * features.sentiment_neutral # Normalize to sum to 1.0 total = pos + neg + neu if total > 0: pos, neg, neu = pos / total, neg / total, neu / total else: pos, neg, neu = 0.33, 0.33, 0.34 return (pos, neg, neu) def _adjust_magnitude( self, base_magnitude: float, features: ImpactFeatureSet, ) -> float: """Adjust base magnitude by surprise, novelty, and evidence coverage. Higher surprise/novelty/evidence → higher magnitude. Conservative: never more than 2x base. """ multiplier = 1.0 # Surprise amplification (NaN means no surprise data → neutral) if not math.isnan(features.surprise): # surprise is normalized, values > 0.5 indicate above-average surprise multiplier *= 1.0 + 0.5 * max(0.0, features.surprise - 0.5) # Novelty amplification (novel events have more impact) multiplier *= 1.0 + 0.3 * features.novelty_score # Evidence coverage: less evidence → discount magnitude multiplier *= 0.5 + 0.5 * features.evidence_coverage # Cap at 2x base (conservative) multiplier = min(multiplier, 2.0) return base_magnitude * multiplier def _adjust_horizon( self, base_horizon: dict[str, float], features: ImpactFeatureSet, ) -> dict[str, float]: """Adjust horizon by event directness. - Direct events: shift probability toward shorter horizons. - Second-order/speculative: shift toward longer horizons. """ horizon = dict(base_horizon) if features.event_directness == "direct": # Shift mass toward shorter horizons shift = 0.05 horizon["intraday"] = horizon.get("intraday", 0.2) + shift horizon["1d"] = horizon.get("1d", 0.2) + shift * 0.5 horizon["90d"] = max(0.0, horizon.get("90d", 0.2) - shift) horizon["30d"] = max(0.0, horizon.get("30d", 0.2) - shift * 0.5) elif features.event_directness in ("second_order", "speculative"): # Shift mass toward longer horizons shift = 0.05 horizon["90d"] = horizon.get("90d", 0.2) + shift horizon["30d"] = horizon.get("30d", 0.2) + shift * 0.5 horizon["intraday"] = max(0.0, horizon.get("intraday", 0.2) - shift) horizon["1d"] = max(0.0, horizon.get("1d", 0.2) - shift * 0.5) # Normalize to sum to 1.0 total = sum(horizon.values()) if total > 0: horizon = {k: v / total for k, v in horizon.items()} return horizon def _compute_uncertainty( self, features: ImpactFeatureSet, primary_event: str, ) -> float: """Compute prediction uncertainty. Higher uncertainty when: - Event class is unknown or low-confidence - Low evidence coverage - Source credibility is low - Market regime is unknown """ uncertainty = 0.5 # Base uncertainty for deterministic model # Unknown event class increases uncertainty if primary_event == "unknown": uncertainty += 0.2 # Low event class confidence increases uncertainty max_event_prob = max(features.event_class_probabilities.values()) if features.event_class_probabilities else 0.0 uncertainty += 0.1 * (1.0 - max_event_prob) # Low evidence coverage increases uncertainty uncertainty += 0.1 * (1.0 - features.evidence_coverage) # Low source credibility increases uncertainty if not math.isnan(features.source_credibility): uncertainty += 0.05 * (1.0 - features.source_credibility) # Unknown market regime increases uncertainty if features.broad_market_regime == "unknown": uncertainty += 0.05 # Clamp to [0, 1] return max(0.0, min(1.0, uncertainty))