"""Shadow mode runner — live v2/v3 comparison without production impact. Runs v3 alongside production v2, collects comparison data, and tracks stability metrics. V3 results are stored but never influence aggregation or trading until shadow requirements are met. """ from __future__ import annotations import enum from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Any from uuid import UUID, uuid4 class DisagreementLevel(str, enum.Enum): """Severity of v2/v3 disagreement.""" NONE = "none" MINOR = "minor" # Different confidence/probabilities within tolerance MODERATE = "moderate" # Different sentiment or secondary entities MAJOR = "major" # Different primary company or event classification CRITICAL = "critical" # Opposite direction or missing safety-critical field @dataclass class ShadowComparison: """Comparison result between v2 and v3 outputs for one document.""" comparison_id: UUID document_id: str timestamp: datetime disagreement_level: DisagreementLevel v2_output: dict[str, Any] v3_output: dict[str, Any] field_differences: dict[str, Any] = field(default_factory=dict) risk_score: float = 0.0 # 0-1, higher = more concerning reviewed: bool = False reviewer_notes: str = "" @classmethod def create( cls, document_id: str, v2_output: dict[str, Any], v3_output: dict[str, Any], field_differences: dict[str, Any] | None = None, disagreement_level: DisagreementLevel = DisagreementLevel.NONE, risk_score: float = 0.0, ) -> ShadowComparison: return cls( comparison_id=uuid4(), document_id=document_id, timestamp=datetime.now(timezone.utc), disagreement_level=disagreement_level, v2_output=v2_output, v3_output=v3_output, field_differences=field_differences or {}, risk_score=risk_score, ) @dataclass class ShadowConfig: """Configuration for shadow mode operation.""" enabled: bool = False min_duration: timedelta = field(default_factory=lambda: timedelta(days=7)) min_documents: int = 500 max_critical_disagreements: int = 5 max_major_disagreement_rate: float = 0.10 auto_disable_on_errors: bool = True error_threshold: int = 50 sample_review_rate: float = 0.05 # Review 5% of disagreements def is_valid_duration(self, started_at: datetime) -> bool: """Check if minimum shadow duration has elapsed.""" elapsed = datetime.now(timezone.utc) - started_at return elapsed >= self.min_duration @dataclass class ShadowRunner: """Manages shadow mode execution and stability tracking. Tracks comparisons, disagreements, and operational metrics. Enforces minimum duration and document count before allowing promotion. """ config: ShadowConfig started_at: datetime | None = None _comparisons: list[ShadowComparison] = field(default_factory=list) _error_count: int = 0 _documents_processed: int = 0 _fast_path_count: int = 0 _gpu_seconds_total: float = 0.0 def start(self) -> None: """Activate shadow mode.""" self.config.enabled = True self.started_at = datetime.now(timezone.utc) def stop(self) -> None: """Deactivate shadow mode.""" self.config.enabled = False @property def is_active(self) -> bool: return self.config.enabled and self.started_at is not None def record_comparison(self, comparison: ShadowComparison) -> None: """Record a v2/v3 comparison.""" self._comparisons.append(comparison) self._documents_processed += 1 def record_error(self) -> None: """Record a v3 processing error.""" self._error_count += 1 if ( self.config.auto_disable_on_errors and self._error_count >= self.config.error_threshold ): self.stop() def record_processing( self, fast_path: bool = True, gpu_seconds: float = 0.0 ) -> None: """Record processing metrics.""" self._documents_processed += 1 if fast_path: self._fast_path_count += 1 self._gpu_seconds_total += gpu_seconds @property def documents_processed(self) -> int: return self._documents_processed @property def fast_path_rate(self) -> float: if self._documents_processed == 0: return 0.0 return self._fast_path_count / self._documents_processed @property def gpu_reduction_ratio(self) -> float: """Placeholder — needs baseline comparison.""" return 0.0 @property def critical_disagreements(self) -> int: return sum( 1 for c in self._comparisons if c.disagreement_level == DisagreementLevel.CRITICAL ) @property def major_disagreement_rate(self) -> float: if not self._comparisons: return 0.0 major_or_critical = sum( 1 for c in self._comparisons if c.disagreement_level in (DisagreementLevel.MAJOR, DisagreementLevel.CRITICAL) ) return major_or_critical / len(self._comparisons) def meets_promotion_criteria(self) -> bool: """Check if all shadow mode requirements are met for promotion.""" if not self.is_active or self.started_at is None: return False # Minimum duration if not self.config.is_valid_duration(self.started_at): return False # Minimum document count if self._documents_processed < self.config.min_documents: return False # Critical disagreement limit if self.critical_disagreements > self.config.max_critical_disagreements: return False # Major disagreement rate if self.major_disagreement_rate > self.config.max_major_disagreement_rate: return False return True def get_review_sample(self) -> list[ShadowComparison]: """Get disagreements needing human review, prioritized by risk.""" unreviewed = [c for c in self._comparisons if not c.reviewed] # Prioritize by disagreement severity and risk score unreviewed.sort( key=lambda c: ( -_disagreement_priority(c.disagreement_level), -c.risk_score, ) ) sample_size = max( 1, int(len(unreviewed) * self.config.sample_review_rate) ) return unreviewed[:sample_size] def summary(self) -> dict[str, Any]: """Generate shadow mode status summary.""" return { "active": self.is_active, "started_at": self.started_at.isoformat() if self.started_at else None, "documents_processed": self._documents_processed, "fast_path_rate": self.fast_path_rate, "error_count": self._error_count, "critical_disagreements": self.critical_disagreements, "major_disagreement_rate": self.major_disagreement_rate, "meets_promotion_criteria": self.meets_promotion_criteria(), "gpu_seconds_total": self._gpu_seconds_total, } def _disagreement_priority(level: DisagreementLevel) -> int: """Priority ordering for disagreement review.""" return { DisagreementLevel.CRITICAL: 4, DisagreementLevel.MAJOR: 3, DisagreementLevel.MODERATE: 2, DisagreementLevel.MINOR: 1, DisagreementLevel.NONE: 0, }.get(level, 0)