stonks-oracle/services/intelligence_pipeline_v3/schemas/safety.py

"""Safety-critical field definitions for v3 promotion gates.

Fields marked as safety-critical MUST pass their quality gates before pipeline
outputs are allowed to influence production aggregation or trading decisions.
A single safety-critical failure blocks promotion for the affected document type.

Schema version: 1.0.0
"""

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    pass


class SafetyCriticalField(str, Enum):
    """Fields whose incorrect extraction can materially affect trading decisions."""

    # Company identity — wrong ticker attribution can cause trades on the wrong security
    COMPANY_IDENTITY = "company_identity"

    # Event classification — misclassifying earnings_beat as earnings_miss inverts signals
    EVENT_CLASS = "event_class"

    # Sentiment direction — wrong sentiment directly affects position direction
    SENTIMENT_DIRECTION = "sentiment_direction"

    # Numeric fact accuracy — wrong EPS or revenue magnitude affects impact estimation
    NUMERIC_FACT_VALUE = "numeric_fact_value"

    # Direct effect attribution — attributing an event to the wrong company creates false signals
    DIRECT_EFFECT_ATTRIBUTION = "direct_effect_attribution"

    # Evidence support — claims without valid evidence spans are unverifiable
    EVIDENCE_SUPPORT = "evidence_support"

    # Confidence calibration — overconfident scores bypass appropriate review thresholds
    CONFIDENCE_CALIBRATION = "confidence_calibration"


# Map each safety-critical field to its minimum required quality metric for promotion
SAFETY_CRITICAL_FIELDS: dict[SafetyCriticalField, dict[str, float]] = {
    SafetyCriticalField.COMPANY_IDENTITY: {
        "precision": 0.95,
        "recall": 0.90,
        "f1": 0.92,
    },
    SafetyCriticalField.EVENT_CLASS: {
        "macro_f1": 0.85,
        "per_class_min_f1": 0.70,
    },
    SafetyCriticalField.SENTIMENT_DIRECTION: {
        "macro_f1": 0.85,
        "direction_accuracy": 0.90,
    },
    SafetyCriticalField.NUMERIC_FACT_VALUE: {
        "exact_match": 0.80,
        "tolerance_match_5pct": 0.92,
    },
    SafetyCriticalField.DIRECT_EFFECT_ATTRIBUTION: {
        "precision": 0.93,
        "recall": 0.88,
    },
    SafetyCriticalField.EVIDENCE_SUPPORT: {
        "support_rate": 0.95,
        "offset_validity": 0.98,
    },
    SafetyCriticalField.CONFIDENCE_CALIBRATION: {
        "ece": 0.05,  # Expected Calibration Error — lower is better
        "brier_score": 0.15,  # Lower is better
    },
}


@dataclass
class SafetyGateResult:
    """Result of checking a single safety-critical field against its gate."""

    field: SafetyCriticalField
    passed: bool
    metric_name: str
    required_value: float
    actual_value: float
    is_lower_better: bool = False

    @property
    def margin(self) -> float:
        """How far above (or below for lower-is-better) the threshold."""
        if self.is_lower_better:
            return self.required_value - self.actual_value
        return self.actual_value - self.required_value


def check_safety_gates(
    metrics: dict[SafetyCriticalField, dict[str, float]],
) -> list[SafetyGateResult]:
    """Check all safety-critical fields against their promotion thresholds.

    Args:
        metrics: Measured quality metrics per safety-critical field.
                 Keys match SafetyCriticalField, values are metric_name -> value dicts.

    Returns:
        List of SafetyGateResult for each check performed.
        Any result with passed=False blocks promotion.
    """
    lower_is_better = {"ece", "brier_score"}
    results: list[SafetyGateResult] = []

    for field, thresholds in SAFETY_CRITICAL_FIELDS.items():
        measured = metrics.get(field, {})
        for metric_name, required in thresholds.items():
            actual = measured.get(metric_name)
            if actual is None:
                # Missing metric fails the gate
                results.append(
                    SafetyGateResult(
                        field=field,
                        passed=False,
                        metric_name=metric_name,
                        required_value=required,
                        actual_value=float("nan"),
                        is_lower_better=metric_name in lower_is_better,
                    )
                )
                continue

            is_lower = metric_name in lower_is_better
            if is_lower:
                passed = actual <= required
            else:
                passed = actual >= required

            results.append(
                SafetyGateResult(
                    field=field,
                    passed=passed,
                    metric_name=metric_name,
                    required_value=required,
                    actual_value=actual,
                    is_lower_better=is_lower,
                )
            )

    return results