Capstone - Complete Analytics System

# Python: Real-Time Analytics Framework
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import pandas as pd
from datetime import datetime

@dataclass
class LiveMatchState:
    """Manages live match state and real-time analytics."""
    match_id: str
    home_team: str
    away_team: str
    events: List[Dict] = field(default_factory=list)
    current_minute: int = 0
    score: Dict[str, int] = field(default_factory=lambda: {"home": 0, "away": 0})
    xg: Dict[str, float] = field(default_factory=lambda: {"home": 0.0, "away": 0.0})
    metrics: Dict = field(default_factory=dict)
    alerts: List[str] = field(default_factory=list)

    def add_event(self, event: Dict) -> List[str]:
        """Add new event and update all metrics."""
        self.events.append(event)
        self.current_minute = max(event.get("minute", 0), self.current_minute)

        # Update metrics
        self._update_metrics(event)

        # Check for alerts
        new_alerts = self._check_alerts(event)
        self.alerts.extend(new_alerts)

        return new_alerts

    def _update_metrics(self, event: Dict):
        """Update score and xG based on new event."""
        # Update score
        if event.get("type") == "Shot" and event.get("outcome") == "Goal":
            team_key = "home" if event["team"] == self.home_team else "away"
            self.score[team_key] += 1

        # Update xG
        if event.get("type") == "Shot":
            xg_value = event.get("xg", 0)
            team_key = "home" if event["team"] == self.home_team else "away"
            self.xg[team_key] += xg_value

        # Calculate rolling metrics
        self._calculate_rolling_metrics()

    def _calculate_rolling_metrics(self, window_minutes: int = 10):
        """Calculate rolling window metrics."""
        cutoff = self.current_minute - window_minutes
        recent = [e for e in self.events if e.get("minute", 0) >= cutoff]

        if not recent:
            return

        home_events = [e for e in recent if e.get("team") == self.home_team]
        away_events = [e for e in recent if e.get("team") == self.away_team]

        self.metrics["rolling"] = {
            "home_possession": len(home_events) / len(recent) * 100 if recent else 50,
            "home_shots_10min": sum(1 for e in home_events if e.get("type") == "Shot"),
            "away_shots_10min": sum(1 for e in away_events if e.get("type") == "Shot"),
            "home_pressure": sum(1 for e in home_events if e.get("type") == "Pressure"),
            "away_pressure": sum(1 for e in away_events if e.get("type") == "Pressure"),
        }

    def _check_alerts(self, event: Dict) -> List[str]:
        """Check for tactical alerts based on new event."""
        alerts = []

        # High xG shot alert
        if event.get("type") == "Shot" and event.get("xg", 0) > 0.3:
            alerts.append(f"HIGH xG CHANCE ({event['xg']:.2f}) - {event.get('player')}")

        # Momentum shift alert
        rolling = self.metrics.get("rolling", {})
        if rolling.get("away_shots_10min", 0) >= 4 and rolling.get("home_shots_10min", 0) <= 1:
            alerts.append("MOMENTUM SHIFT: Opponent dominating last 10 min")

        return alerts

    def get_state(self) -> Dict:
        """Get current match state summary."""
        return {
            "minute": self.current_minute,
            "score": self.score,
            "xg": self.xg,
            "metrics": self.metrics,
            "n_events": len(self.events),
            "recent_alerts": self.alerts[-5:] if self.alerts else []
        }

# Usage
match_state = LiveMatchState(
    match_id="match_001",
    home_team="Barcelona",
    away_team="Real Madrid"
)
print(match_state.get_state())

# Python: Simulated Streaming Data Handler
import asyncio
from typing import Callable, Optional, Generator
import pandas as pd

class EventStream:
    """Simulated event stream for live match data."""

    def __init__(self, events_df: pd.DataFrame, speed_factor: float = 1.0):
        self.events = events_df.sort_values(["minute", "second"]).reset_index(drop=True)
        self.events["game_time"] = self.events["minute"] * 60 + self.events["second"]
        self.current_index = 0
        self.speed_factor = speed_factor

    def get_next_event(self) -> Optional[dict]:
        """Get the next event in the stream."""
        if self.current_index >= len(self.events):
            return None

        event = self.events.iloc[self.current_index].to_dict()
        self.current_index += 1
        return event

    def has_more(self) -> bool:
        """Check if there are more events."""
        return self.current_index < len(self.events)

    def __iter__(self) -> Generator:
        """Iterate through events."""
        while self.has_more():
            yield self.get_next_event()

async def process_live_match(match_state: LiveMatchState,
                              event_stream: EventStream,
                              callback: Optional[Callable] = None):
    """Process live match events asynchronously."""
    for event in event_stream:
        if event:
            alerts = match_state.add_event(event)

            if callback:
                await callback(match_state.get_state(), alerts)

            # Simulate real-time delay
            await asyncio.sleep(0.1 / event_stream.speed_factor)

    return match_state

# Example callback
async def live_update_callback(state: dict, alerts: list):
    """Print live updates."""
    print(f"\rMinute {state['minute']} | "
          f"Score: {state['score']['home']}-{state['score']['away']} | "
          f"xG: {state['xg']['home']:.2f}-{state['xg']['away']:.2f}",
          end="")

    for alert in alerts:
        print(f"\n*** ALERT: {alert} ***")

# Run the live match processor
# asyncio.run(process_live_match(match_state, event_stream, live_update_callback))

# Python: Live Win Probability Model
import numpy as np
from scipy import stats
from typing import Dict

def calculate_win_probability(home_goals: int, away_goals: int,
                               home_xg: float, away_xg: float,
                               minute: int, n_sims: int = 10000) -> Dict[str, float]:
    """Calculate win probability using Poisson simulation."""
    # Remaining time factor
    remaining_pct = (90 - minute) / 90

    # Calculate xG rate and regress to mean
    safe_minute = max(minute, 1)
    xg_rate_home = home_xg / safe_minute * 90
    xg_rate_away = away_xg / safe_minute * 90

    # Regress to league average
    league_avg_xg = 1.3
    regression_factor = 0.3

    projected_home_xg = xg_rate_home * (1 - regression_factor) + league_avg_xg * regression_factor
    projected_away_xg = xg_rate_away * (1 - regression_factor) + league_avg_xg * regression_factor

    # Expected goals remaining
    remaining_home_xg = projected_home_xg * remaining_pct
    remaining_away_xg = projected_away_xg * remaining_pct

    # Simulate outcomes
    np.random.seed(42)  # For reproducibility in examples
    home_remaining = np.random.poisson(remaining_home_xg, n_sims)
    away_remaining = np.random.poisson(remaining_away_xg, n_sims)

    home_final = home_goals + home_remaining
    away_final = away_goals + away_remaining

    # Calculate probabilities
    home_win_prob = (home_final > away_final).mean() * 100
    draw_prob = (home_final == away_final).mean() * 100
    away_win_prob = (home_final < away_final).mean() * 100

    return {
        "home_win": round(home_win_prob, 1),
        "draw": round(draw_prob, 1),
        "away_win": round(away_win_prob, 1)
    }

def track_win_probability(match_state: LiveMatchState) -> Dict[str, float]:
    """Get current win probability from match state."""
    state = match_state.get_state()

    return calculate_win_probability(
        home_goals=state["score"]["home"],
        away_goals=state["score"]["away"],
        home_xg=state["xg"]["home"],
        away_xg=state["xg"]["away"],
        minute=state["minute"]
    )

# Example
probs = calculate_win_probability(
    home_goals=1, away_goals=0,
    home_xg=1.2, away_xg=0.8,
    minute=60
)

print(f"Win Probability at minute 60 (1-0):")
print(f"Home Win: {probs['home_win']}%")
print(f"Draw: {probs['draw']}%")
print(f"Away Win: {probs['away_win']}%")

# Python: Momentum Detection System
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import numpy as np

@dataclass
class MomentumTracker:
    """Tracks and analyzes match momentum."""
    events: List[Dict] = field(default_factory=list)
    momentum_history: List[Dict] = field(default_factory=list)
    current_momentum: float = 0.0  # -1 to 1 scale

    # Event weights for momentum calculation
    EVENT_WEIGHTS = {
        "Shot": 3.0,
        "Key Pass": 2.0,
        "Progressive Pass": 1.0,
        "Pressure": 0.5,
        "Ball Recovery": 1.0,
        "Dribble": 1.0,
    }

    def add_events(self, new_events: List[Dict], home_team: str) -> float:
        """Add events and recalculate momentum."""
        self.events.extend(new_events)
        return self.calculate_momentum(home_team)

    def calculate_momentum(self, home_team: str, window_minutes: int = 5) -> float:
        """Calculate current momentum based on recent events."""
        if not self.events:
            return 0.0

        current_minute = max(e.get("minute", 0) for e in self.events)
        cutoff = current_minute - window_minutes
        recent = [e for e in self.events if e.get("minute", 0) >= cutoff]

        if not recent:
            return 0.0

        home_score = 0.0
        away_score = 0.0

        for event in recent:
            weight = self.EVENT_WEIGHTS.get(event.get("type"), 0.5)

            # Bonus for final third actions
            if event.get("location_x", 0) > 80:
                weight *= 1.5

            if event.get("team") == home_team:
                home_score += weight
            else:
                away_score += weight

        # Normalize to -1 to 1
        total = home_score + away_score
        if total > 0:
            self.current_momentum = (home_score - away_score) / total
        else:
            self.current_momentum = 0.0

        # Track history
        self.momentum_history.append({
            "minute": current_minute,
            "momentum": self.current_momentum
        })

        return self.current_momentum

    def detect_momentum_shift(self, threshold: float = 0.3) -> Optional[str]:
        """Detect significant momentum shifts."""
        if len(self.momentum_history) < 2:
            return None

        recent = self.momentum_history[-2:]
        change = recent[1]["momentum"] - recent[0]["momentum"]

        if abs(change) > threshold:
            direction = "HOME" if change > 0 else "AWAY"
            return f"MOMENTUM SHIFT: {direction} gaining control ({change:+.2f})"

        return None

    def get_momentum_label(self) -> str:
        """Get human-readable momentum label."""
        m = self.current_momentum
        if m > 0.3:
            return "HOME DOMINANT"
        elif m < -0.3:
            return "AWAY DOMINANT"
        elif m > 0.1:
            return "HOME SLIGHT EDGE"
        elif m < -0.1:
            return "AWAY SLIGHT EDGE"
        else:
            return "BALANCED"

# Usage
momentum_tracker = MomentumTracker()
print("Momentum tracking initialized")

# Python: Live Match Dashboard with Streamlit
import streamlit as st
import plotly.graph_objects as go
import plotly.express as px
import pandas as pd
import time

def create_live_dashboard():
    """Create a Streamlit live match dashboard."""
    st.set_page_config(page_title="Live Match Analytics", layout="wide")
    st.title("Live Match Analytics Dashboard")

    # Initialize session state
    if "match_state" not in st.session_state:
        st.session_state.match_state = None

    # Create placeholder for auto-refresh
    placeholder = st.empty()

    # Simulate live updates (in production, fetch from API)
    while True:
        with placeholder.container():
            state = st.session_state.match_state

            # Row 1: Key Metrics
            col1, col2, col3, col4 = st.columns(4)

            with col1:
                if state:
                    st.metric(
                        "Score",
                        f"{state['score']['home']} - {state['score']['away']}",
                        f"Minute {state['minute']}"
                    )
                else:
                    st.metric("Score", "--", "Waiting...")

            with col2:
                if state:
                    st.metric(
                        "xG",
                        f"{state['xg']['home']:.2f} - {state['xg']['away']:.2f}"
                    )
                else:
                    st.metric("xG", "--")

            with col3:
                if state:
                    probs = calculate_win_probability(
                        state["score"]["home"], state["score"]["away"],
                        state["xg"]["home"], state["xg"]["away"],
                        state["minute"]
                    )
                    st.metric("Home Win Prob", f"{probs['home_win']}%")
                else:
                    st.metric("Home Win Prob", "--")

            with col4:
                momentum_label = state.get("momentum_label", "BALANCED") if state else "--"
                st.metric("Momentum", momentum_label)

            # Row 2: Charts
            col_left, col_right = st.columns(2)

            with col_left:
                st.subheader("xG Flow")
                if state and "xg_timeline" in state:
                    fig = create_xg_flow_chart(state["xg_timeline"])
                    st.plotly_chart(fig, use_container_width=True)
                else:
                    st.info("Waiting for match data...")

            with col_right:
                st.subheader("Win Probability")
                if state and "win_prob_history" in state:
                    fig = create_win_prob_chart(state["win_prob_history"])
                    st.plotly_chart(fig, use_container_width=True)
                else:
                    st.info("Waiting for match data...")

            # Row 3: Alerts and Stats
            col_a, col_b, col_c = st.columns(3)

            with col_a:
                st.subheader("Live Alerts")
                if state and state.get("recent_alerts"):
                    for alert in state["recent_alerts"]:
                        st.warning(alert)
                else:
                    st.info("No recent alerts")

            with col_b:
                st.subheader("Rolling Stats (10 min)")
                if state and state.get("metrics", {}).get("rolling"):
                    rolling = state["metrics"]["rolling"]
                    st.write(f"Home Possession: {rolling['home_possession']:.1f}%")
                    st.write(f"Home Shots: {rolling['home_shots_10min']}")
                    st.write(f"Away Shots: {rolling['away_shots_10min']}")

            with col_c:
                st.subheader("Recent Events")
                if state and "recent_events" in state:
                    for event in state["recent_events"][-5:]:
                        st.text(f"{event['minute']}' - {event['type']} ({event['team']})")

        # Refresh interval
        time.sleep(1)

def create_xg_flow_chart(timeline_data):
    """Create xG flow chart."""
    fig = go.Figure()
    fig.add_trace(go.Scatter(
        x=timeline_data["minute"],
        y=timeline_data["home_xg"],
        name="Home xG",
        line=dict(color="#2E7D32", width=2)
    ))
    fig.add_trace(go.Scatter(
        x=timeline_data["minute"],
        y=timeline_data["away_xg"],
        name="Away xG",
        line=dict(color="#D32F2F", width=2)
    ))
    fig.update_layout(
        xaxis_title="Minute",
        yaxis_title="Cumulative xG",
        height=300
    )
    return fig

# Run: streamlit run live_dashboard.py

# Python: Optimized Live xG Calculator
import numpy as np
import pickle
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class ShotEvent:
    """Represents a shot event for xG calculation."""
    location_x: float
    location_y: float
    body_part: str = "foot"
    play_pattern: str = "open play"
    under_pressure: bool = False
    first_touch: bool = False

class LiveXGCalculator:
    """Optimized xG calculator for live match use."""

    def __init__(self, model_path: Optional[str] = None):
        self.model = None
        self.feature_means = None
        self.feature_sds = None

        if model_path:
            self._load_model(model_path)

    def _load_model(self, model_path: str):
        """Load pre-trained model and normalization params."""
        import xgboost as xgb
        self.model = xgb.Booster()
        self.model.load_model(model_path)

        with open(f"{model_path}_params.pkl", "rb") as f:
            params = pickle.load(f)
            self.feature_means = params["means"]
            self.feature_sds = params["sds"]

    def extract_features(self, shot: ShotEvent, match_context: Dict) -> np.ndarray:
        """Extract features from shot event."""
        x, y = shot.location_x, shot.location_y

        # Distance and angle to goal
        goal_x, goal_y = 120, 40
        distance = np.sqrt((goal_x - x)**2 + (goal_y - y)**2)
        angle = np.degrees(np.arctan2(abs(y - goal_y), goal_x - x))

        features = np.array([
            distance,
            angle,
            float(shot.body_part == "foot"),
            float(shot.body_part == "head"),
            float(shot.play_pattern == "open play"),
            float(shot.play_pattern == "counter"),
            float(shot.play_pattern in ["corner", "free kick"]),
            float(shot.under_pressure),
            float(shot.first_touch),
            float(match_context.get("home_trailing", False)),
            match_context.get("minute", 45),
            match_context.get("minute", 45) ** 2
        ])

        return features

    def normalize_features(self, features: np.ndarray) -> np.ndarray:
        """Normalize features using stored parameters."""
        if self.feature_means is not None:
            return (features - self.feature_means) / self.feature_sds
        return features

    def predict_xg(self, shot: ShotEvent, match_context: Dict) -> float:
        """Predict xG for a single shot."""
        if self.model is None:
            return self._simple_xg(shot)

        features = self.extract_features(shot, match_context)
        features_norm = self.normalize_features(features)

        import xgboost as xgb
        dmatrix = xgb.DMatrix(features_norm.reshape(1, -1))
        xg = self.model.predict(dmatrix)[0]

        return np.clip(xg, 0.01, 0.99)

    def _simple_xg(self, shot: ShotEvent) -> float:
        """Simple fallback xG calculation."""
        x, y = shot.location_x, shot.location_y
        goal_x, goal_y = 120, 40

        distance = np.sqrt((goal_x - x)**2 + (goal_y - y)**2)
        base_xg = np.exp(-0.1 * distance)

        if shot.body_part == "head":
            base_xg *= 0.7

        return round(float(np.clip(base_xg, 0.01, 0.99)), 3)

    def batch_predict(self, shots: List[ShotEvent], match_context: Dict) -> List[float]:
        """Efficiently predict xG for multiple shots."""
        if self.model is None:
            return [self._simple_xg(s) for s in shots]

        features = np.array([
            self.extract_features(s, match_context) for s in shots
        ])
        features_norm = np.apply_along_axis(self.normalize_features, 1, features)

        import xgboost as xgb
        dmatrix = xgb.DMatrix(features_norm)
        xg_values = self.model.predict(dmatrix)

        return [float(np.clip(xg, 0.01, 0.99)) for xg in xg_values]

# Example usage
calculator = LiveXGCalculator()
shot = ShotEvent(location_x=105, location_y=42, body_part="foot")
xg = calculator._simple_xg(shot)
print(f"Simple xG for shot at (105, 42): {xg:.3f}")

# Python: Robust Event Processing
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)

@dataclass
class RobustEventProcessor:
    """Handles event validation, buffering, and error recovery."""
    buffer_window: float = 5.0  # seconds
    heartbeat_timeout: float = 30.0  # seconds
    event_buffer: List[Dict] = field(default_factory=list)
    last_heartbeat: datetime = field(default_factory=datetime.now)

    VALID_EVENT_TYPES = {
        "Shot", "Pass", "Pressure", "Foul", "Dribble",
        "Ball Recovery", "Clearance", "Goal Keeper"
    }

    REQUIRED_FIELDS = {"type", "minute", "team"}

    def process_event(self, event: Dict) -> List[Dict]:
        """Process incoming event with validation and buffering."""
        # Validate
        valid, reason = self.validate_event(event)
        if not valid:
            logger.warning(f"Invalid event: {reason}")
            return []

        # Buffer for reordering
        self.add_to_buffer(event)

        # Return ready events
        return self.get_ready_events()

    def validate_event(self, event: Dict) -> Tuple[bool, Optional[str]]:
        """Validate event structure and values."""
        # Required fields
        missing = self.REQUIRED_FIELDS - set(event.keys())
        if missing:
            return False, f"Missing fields: {missing}"

        # Type validation
        if event.get("type") not in self.VALID_EVENT_TYPES:
            return False, f"Invalid event type: {event.get('type')}"

        # Range validation
        minute = event.get("minute", -1)
        if minute < 0 or minute > 130:
            return False, f"Invalid minute: {minute}"

        return True, None

    def add_to_buffer(self, event: Dict):
        """Add event to buffer and sort by game time."""
        event["received_at"] = datetime.now()
        self.event_buffer.append(event)

        # Sort by game time
        self.event_buffer.sort(
            key=lambda e: e.get("minute", 0) * 60 + e.get("second", 0)
        )

    def get_ready_events(self) -> List[Dict]:
        """Return events that have waited long enough in buffer."""
        if not self.event_buffer:
            return []

        current_time = datetime.now()
        ready = []
        remaining = []

        for event in self.event_buffer:
            elapsed = (current_time - event["received_at"]).total_seconds()
            if elapsed >= self.buffer_window:
                ready.append(event)
            else:
                remaining.append(event)

        self.event_buffer = remaining
        return ready

    def check_heartbeat(self) -> bool:
        """Check if data feed is still active."""
        elapsed = (datetime.now() - self.last_heartbeat).total_seconds()

        if elapsed > self.heartbeat_timeout:
            logger.warning("Data feed heartbeat timeout - connection may be lost")
            return False

        return True

    def receive_heartbeat(self):
        """Update last heartbeat time."""
        self.last_heartbeat = datetime.now()

processor = RobustEventProcessor()
print("Robust event processor initialized")

# Python: Substitution Recommendation System
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import pandas as pd

@dataclass
class SubstitutionAnalyzer:
    """Analyzes player fatigue and recommends substitutions."""
    fatigue_thresholds: Dict[str, float] = field(default_factory=lambda: {
        "high_intensity_runs": 0.7,
        "sprint_count": 0.6,
        "pass_completion_drop": 10
    })
    player_metrics: pd.DataFrame = field(default_factory=pd.DataFrame)

    def update_player_metrics(self, player_tracking: pd.DataFrame, minute: int) -> pd.DataFrame:
        """Update fatigue metrics for all players."""
        df = player_tracking.copy()
        df["current_minute"] = minute

        # Calculate current rates (per 90)
        df["hir_rate"] = df["high_intensity_runs"] / max(minute, 1) * 90
        df["sprint_rate"] = df["sprints"] / max(minute, 1) * 90
        df["pass_completion_current"] = (
            df["passes_completed"] / df["passes_attempted"].clip(lower=1) * 100
        )

        # For simplicity, use first values as baseline
        # In production, would track baseline from first 30 minutes
        df["baseline_hir"] = df["hir_rate"].mean()
        df["baseline_sprints"] = df["sprint_rate"].mean()
        df["baseline_pass_pct"] = df["pass_completion_current"].mean()

        # Calculate retention
        df["hir_retention"] = df["hir_rate"] / df["baseline_hir"].clip(lower=0.1)
        df["sprint_retention"] = df["sprint_rate"] / df["baseline_sprints"].clip(lower=0.1)
        df["pass_pct_change"] = df["pass_completion_current"] - df["baseline_pass_pct"]

        self.player_metrics = df
        return df

    def get_fatigue_alerts(self) -> pd.DataFrame:
        """Identify players showing fatigue."""
        if self.player_metrics.empty:
            return pd.DataFrame()

        df = self.player_metrics.copy()

        df["hir_fatigued"] = df["hir_retention"] < self.fatigue_thresholds["high_intensity_runs"]
        df["sprint_fatigued"] = df["sprint_retention"] < self.fatigue_thresholds["sprint_count"]
        df["passing_degraded"] = df["pass_pct_change"] < -self.fatigue_thresholds["pass_completion_drop"]

        df["fatigue_score"] = (
            df["hir_fatigued"].astype(int) +
            df["sprint_fatigued"].astype(int) +
            df["passing_degraded"].astype(int)
        )

        fatigued = df[df["fatigue_score"] >= 2].sort_values(
            "fatigue_score", ascending=False
        )

        return fatigued[["player_id", "player_name", "position", "fatigue_score",
                        "hir_retention", "sprint_retention", "pass_pct_change"]]

    def recommend_substitution(self, game_state: Dict, bench_players: pd.DataFrame) -> Dict:
        """Generate substitution recommendation."""
        fatigued = self.get_fatigue_alerts()

        if fatigued.empty:
            return {
                "recommendation": "No urgent substitutions needed",
                "priority": "low"
            }

        # Get most fatigued player
        target = fatigued.iloc[0]

        # Find replacement
        suitable = bench_players[bench_players["position"] == target["position"]]
        if not suitable.empty:
            replacement = suitable.sort_values("minutes_rest", ascending=False).iloc[0]
            replacement_name = replacement["player_name"]
        else:
            replacement_name = "available substitute"

        # Determine urgency
        score_diff = game_state.get("score_diff", 0)
        minute = game_state.get("minute", 0)

        if score_diff < 0 and minute > 70:
            urgency = "high"
        elif target["fatigue_score"] >= 3:
            urgency = "high"
        elif minute > 75:
            urgency = "medium"
        else:
            urgency = "low"

        return {
            "recommendation": f"Replace {target['player_name']} with {replacement_name} (fatigue score: {target['fatigue_score']})",
            "target_player": target["player_name"],
            "replacement": replacement_name,
            "priority": urgency,
            "reasoning": {
                "hir_retention": target["hir_retention"],
                "sprint_retention": target["sprint_retention"],
                "pass_degradation": target["pass_pct_change"]
            }
        }

analyzer = SubstitutionAnalyzer()
print("Substitution analyzer initialized")

# Python: Complete Real-Time Match Analysis
import pandas as pd
import numpy as np
from typing import Dict, List

async def run_complete_match_analysis(events_df: pd.DataFrame,
                                       home_team: str,
                                       away_team: str) -> Dict:
    """Run complete real-time analysis on a match."""

    # Initialize components
    match_state = LiveMatchState(
        match_id="match_001",
        home_team=home_team,
        away_team=away_team
    )
    momentum_tracker = MomentumTracker()
    alert_system = TacticalAlertSystem()
    event_processor = RobustEventProcessor()

    # Results tracking
    timeline = []
    all_alerts = []

    # Process events chronologically
    events_sorted = events_df.sort_values(["minute", "second"])

    for _, event in events_sorted.iterrows():
        event_dict = event.to_dict()

        # Process through robust handler
        validated = event_processor.process_event(event_dict)

        if validated:
            for valid_event in validated:
                # Update match state
                match_state.add_event(valid_event)

                # Update momentum
                momentum_tracker.add_events([valid_event], home_team)

                # Get current state
                state = match_state.get_state()
                state["momentum_label"] = momentum_tracker.get_momentum_label()

                # Check for momentum shift
                shift = momentum_tracker.detect_momentum_shift()
                if shift:
                    state["momentum_change"] = 0.5

                # Check alerts
                alerts = alert_system.check_alerts(valid_event, state)
                all_alerts.extend(alerts)

                # Calculate win probability
                win_prob = calculate_win_probability(
                    state["score"]["home"], state["score"]["away"],
                    state["xg"]["home"], state["xg"]["away"],
                    state["minute"]
                )

                # Record timeline
                timeline.append({
                    "minute": state["minute"],
                    "home_score": state["score"]["home"],
                    "away_score": state["score"]["away"],
                    "home_xg": state["xg"]["home"],
                    "away_xg": state["xg"]["away"],
                    "momentum": momentum_tracker.current_momentum,
                    "home_win_prob": win_prob["home_win"],
                    "n_alerts": len(alerts)
                })

    # Summary
    final_state = match_state.get_state()

    print("\n=== MATCH SUMMARY ===")
    print(f"Final Score: {final_state['score']['home']} - {final_state['score']['away']}")
    print(f"Final xG: {final_state['xg']['home']:.2f} - {final_state['xg']['away']:.2f}")
    print(f"Total Alerts Generated: {len(all_alerts)}")

    if all_alerts:
        print("\nKey Alerts:")
        high_priority = [a for a in all_alerts if a["priority"] == "high"]
        for alert in high_priority[:5]:
            print(f"  [{alert['minute']}'] {alert['message']}")

    return {
        "timeline": pd.DataFrame(timeline),
        "alerts": all_alerts,
        "final_state": final_state
    }

print("Complete match analysis system ready")

# Python: Tactical Alert System
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Any
import pandas as pd

@dataclass
class AlertRule:
    """Defines a tactical alert rule."""
    id: str
    condition: Callable[[Dict, Dict], bool]
    message: Callable[[Dict, Dict], str]
    priority: str = "medium"

class TacticalAlertSystem:
    """System for generating tactical alerts during live matches."""

    def __init__(self, cooldown_minutes: int = 5):
        self.alert_rules: Dict[str, AlertRule] = {}
        self.alert_history: List[Dict] = []
        self.cooldown_minutes = cooldown_minutes
        self._define_default_rules()

    def _define_default_rules(self):
        """Define default tactical alert rules."""

        # High xG chance
        self.add_rule(AlertRule(
            id="high_xg_chance",
            condition=lambda e, s: e.get("type") == "Shot" and e.get("xg", 0) > 0.25,
            message=lambda e, s: f"BIG CHANCE: {e.get('xg', 0)*100:.0f}% xG shot by {e.get('player')}",
            priority="high"
        ))

        # Momentum swing
        self.add_rule(AlertRule(
            id="momentum_swing",
            condition=lambda e, s: abs(s.get("momentum_change", 0)) > 0.4,
            message=lambda e, s: f"MOMENTUM SWING: {'HOME' if s.get('momentum_change', 0) > 0 else 'AWAY'} taking control",
            priority="high"
        ))

        # Pressing intensity drop
        self.add_rule(AlertRule(
            id="pressing_drop",
            condition=lambda e, s: (
                s.get("metrics", {}).get("rolling", {}).get("home_pressure", 10) < 5 and
                s.get("minute", 0) > 60
            ),
            message=lambda e, s: "WARNING: Pressing intensity dropped - consider energy management",
            priority="medium"
        ))

        # xG underperformance
        self.add_rule(AlertRule(
            id="xg_underperform",
            condition=lambda e, s: (
                s.get("minute", 0) >= 60 and
                (s.get("xg", {}).get("home", 0) - s.get("score", {}).get("home", 0)) > 1.0
            ),
            message=lambda e, s: f"xG GAP: Creating chances ({s.get('xg', {}).get('home', 0):.1f} xG) but only {s.get('score', {}).get('home', 0)} goals",
            priority="medium"
        ))

        # Dangerous free kick
        self.add_rule(AlertRule(
            id="set_piece_zone",
            condition=lambda e, s: (
                e.get("type") == "Foul" and
                e.get("location_x", 0) > 85 and
                20 < e.get("location_y", 0) < 60
            ),
            message=lambda e, s: "DANGEROUS FREE KICK: Good scoring position",
            priority="high"
        ))

    def add_rule(self, rule: AlertRule):
        """Add a new alert rule."""
        self.alert_rules[rule.id] = rule

    def check_alerts(self, event: Dict, state: Dict) -> List[Dict]:
        """Check all rules and return triggered alerts."""
        alerts = []
        current_minute = state.get("minute", 0)

        for rule_id, rule in self.alert_rules.items():
            # Check cooldown
            recent_same = [
                a for a in self.alert_history
                if a["rule_id"] == rule_id and
                a["minute"] >= current_minute - self.cooldown_minutes
            ]
            if recent_same:
                continue

            # Check condition
            try:
                if rule.condition(event, state):
                    alert = {
                        "rule_id": rule_id,
                        "message": rule.message(event, state),
                        "priority": rule.priority,
                        "minute": current_minute
                    }
                    alerts.append(alert)
                    self.alert_history.append({
                        "rule_id": rule_id,
                        "minute": current_minute
                    })
            except Exception:
                pass  # Skip failed rules

        return alerts

alert_system = TacticalAlertSystem()
print(f"Tactical alert system initialized with {len(alert_system.alert_rules)} rules")