""" Squad Analysis Engine - V9 Feature Kadro ve oyuncu bazlı analiz. Analiz Edilen Metrikler: - İlk 11 kalitesi (golcü formu, key player) - Yedek gücü - Eksik oyuncu etkisi - Pozisyon bazlı güç - Takım içi golcü dağılımı """ import os from typing import Dict, Optional, List, Tuple from dataclasses import dataclass, field from datetime import datetime from collections import defaultdict try: import psycopg2 from psycopg2.extras import RealDictCursor except ImportError: psycopg2 = None @dataclass class PlayerForm: """Oyuncu form bilgisi""" player_id: str player_name: str goals_last_5: int = 0 assists_last_5: int = 0 minutes_last_5: int = 0 cards_last_5: int = 0 is_key_player: bool = False # Golcü veya sık oynayan @dataclass class SquadAnalysis: """Takım kadro analizi""" team_id: str team_name: str = "" # İlk 11 bilgisi starting_count: int = 0 sub_count: int = 0 total_squad: int = 0 # Pozisyon dağılımı goalkeeper_count: int = 0 defender_count: int = 0 midfielder_count: int = 0 forward_count: int = 0 # Form metrikleri total_goals_last_5: int = 0 # Kadrodaki oyuncuların son 5 maçtaki golleri total_assists_last_5: int = 0 key_players_count: int = 0 # Golcü sayısı key_player_missing: int = 0 # Eksik golcü # Kalite metrikleri avg_minutes_per_player: float = 0.0 # Ortalama oynama süresi squad_experience: float = 0.0 # 0-1, takımla oynama deneyimi rotation_rate: float = 0.0 # Kadro rotasyonu oranı @dataclass class SquadFeatures: """Model için kadro feature'ları""" # Home team features home_starting_11: int = 11 home_sub_count: int = 7 home_total_squad: int = 18 home_goalkeepers: int = 1 home_defenders: int = 4 home_midfielders: int = 4 home_forwards: int = 2 home_goals_last_5: int = 0 home_assists_last_5: int = 0 home_key_players: int = 0 home_squad_experience: float = 0.5 # Away team features away_starting_11: int = 11 away_sub_count: int = 7 away_total_squad: int = 18 away_goalkeepers: int = 1 away_defenders: int = 4 away_midfielders: int = 4 away_forwards: int = 2 away_goals_last_5: int = 0 away_assists_last_5: int = 0 away_key_players: int = 0 away_squad_experience: float = 0.5 # Comparison features squad_strength_diff: float = 0.0 # + = home stronger goals_form_diff: float = 0.0 key_players_diff: int = 0 def to_dict(self) -> Dict[str, float]: return { # Home 'home_starting_11': float(self.home_starting_11), 'home_sub_count': float(self.home_sub_count), 'home_total_squad': float(self.home_total_squad), 'home_goalkeepers': float(self.home_goalkeepers), 'home_defenders': float(self.home_defenders), 'home_midfielders': float(self.home_midfielders), 'home_forwards': float(self.home_forwards), 'home_goals_last_5': float(self.home_goals_last_5), 'home_assists_last_5': float(self.home_assists_last_5), 'home_key_players': float(self.home_key_players), 'home_squad_experience': self.home_squad_experience, # Away 'away_starting_11': float(self.away_starting_11), 'away_sub_count': float(self.away_sub_count), 'away_total_squad': float(self.away_total_squad), 'away_goalkeepers': float(self.away_goalkeepers), 'away_defenders': float(self.away_defenders), 'away_midfielders': float(self.away_midfielders), 'away_forwards': float(self.away_forwards), 'away_goals_last_5': float(self.away_goals_last_5), 'away_assists_last_5': float(self.away_assists_last_5), 'away_key_players': float(self.away_key_players), 'away_squad_experience': self.away_squad_experience, # Diffs 'squad_strength_diff': self.squad_strength_diff, 'goals_form_diff': self.goals_form_diff, 'key_players_diff': float(self.key_players_diff), } class SquadAnalysisEngine: """ Kadro ve oyuncu analiz motoru. Beşiktaş-Galatasaray maçı için: - İlk 11'deki oyuncuların son 5 maçtaki gol/asist - Key player tespiti (çok gol atan oyuncular) - Pozisyon dağılımı (4-3-3, 4-4-2 vb.) - Yedek kalitesi hesaplar. """ # Pozisyon mapping POSITION_MAP = { 'goalkeeper': 'GK', 'gk': 'GK', 'kaleci': 'GK', 'defender': 'DEF', 'def': 'DEF', 'defans': 'DEF', 'savunma': 'DEF', 'midfielder': 'MID', 'mid': 'MID', 'orta saha': 'MID', 'forward': 'FWD', 'fwd': 'FWD', 'forvet': 'FWD', 'striker': 'FWD', } def __init__(self): self.conn = None self._player_form_cache: Dict[str, PlayerForm] = {} def _connect_db(self): if psycopg2 is None: return None try: from data.db import get_clean_dsn self.conn = psycopg2.connect(get_clean_dsn()) return self.conn except Exception as e: print(f"[SquadEngine] DB connection failed: {e}") return None def get_conn(self): if self.conn is None or self.conn.closed: self._connect_db() return self.conn def _normalize_position(self, position: Optional[str]) -> str: """Pozisyonu normalize et""" if not position: return 'UNK' pos_lower = position.lower().strip() for key, val in self.POSITION_MAP.items(): if key in pos_lower: return val return 'UNK' def get_player_form(self, player_id: str, before_date_ms: int = None) -> PlayerForm: """Oyuncunun son 5 maçtaki formunu hesapla""" if player_id in self._player_form_cache: return self._player_form_cache[player_id] form = PlayerForm(player_id=player_id, player_name="") conn = self.get_conn() if conn is None: return form try: with conn.cursor(cursor_factory=RealDictCursor) as cur: # Oyuncu adını al cur.execute("SELECT name FROM players WHERE id = %s", (player_id,)) player_row = cur.fetchone() if player_row: form.player_name = player_row['name'] # Son 5 maçtaki gol ve asist cur.execute(""" SELECT COUNT(*) FILTER (WHERE event_type = 'goal' AND event_subtype NOT ILIKE '%%penaltı kaçırma%%') as goals, COUNT(*) FILTER (WHERE event_type = 'goal' AND assist_player_id IS NOT NULL) as assists_given FROM match_player_events WHERE player_id = %s AND match_id IN ( SELECT match_id FROM match_player_participation WHERE player_id = %s ORDER BY match_id DESC LIMIT 5 ) """, (player_id, player_id)) stats = cur.fetchone() if stats: form.goals_last_5 = stats['goals'] or 0 # Asist hesapla (assist_player_id olarak geçen) cur.execute(""" SELECT COUNT(*) as assists FROM match_player_events WHERE assist_player_id = %s AND match_id IN ( SELECT match_id FROM match_player_participation WHERE player_id = %s ORDER BY match_id DESC LIMIT 5 ) """, (player_id, player_id)) assist_row = cur.fetchone() if assist_row: form.assists_last_5 = assist_row['assists'] or 0 # Kart sayısı cur.execute(""" SELECT COUNT(*) as cards FROM match_player_events WHERE player_id = %s AND event_type = 'card' AND match_id IN ( SELECT match_id FROM match_player_participation WHERE player_id = %s ORDER BY match_id DESC LIMIT 5 ) """, (player_id, player_id)) card_row = cur.fetchone() if card_row: form.cards_last_5 = card_row['cards'] or 0 # Key player mi? (Son 10 maçta 3+ gol) cur.execute(""" SELECT COUNT(*) as total_goals FROM match_player_events WHERE player_id = %s AND event_type = 'goal' AND event_subtype NOT ILIKE '%%penaltı kaçırma%%' """, (player_id,)) total_row = cur.fetchone() form.is_key_player = (total_row['total_goals'] or 0) >= 3 self._player_form_cache[player_id] = form return form except Exception as e: import traceback traceback.print_exc() print(f"[SquadEngine] Error getting player form: {e}") return form def analyze_squad(self, match_id: str, team_id: str) -> SquadAnalysis: """Takımın maç kadrosunu analiz et""" analysis = SquadAnalysis(team_id=team_id) conn = self.get_conn() if conn is None: return analysis try: with conn.cursor(cursor_factory=RealDictCursor) as cur: # Takım adını al cur.execute("SELECT name FROM teams WHERE id = %s", (team_id,)) team_row = cur.fetchone() if team_row: analysis.team_name = team_row['name'] # Maç kadrosunu al cur.execute(""" SELECT player_id, position, is_starting FROM match_player_participation WHERE match_id = %s AND team_id = %s """, (match_id, team_id)) players = cur.fetchall() for p in players: if p['is_starting']: analysis.starting_count += 1 else: analysis.sub_count += 1 pos = self._normalize_position(p['position']) if pos == 'GK': analysis.goalkeeper_count += 1 elif pos == 'DEF': analysis.defender_count += 1 elif pos == 'MID': analysis.midfielder_count += 1 elif pos == 'FWD': analysis.forward_count += 1 # İlk 11'in formunu topluca hesapla if p['is_starting']: form = self.get_player_form(p['player_id']) analysis.total_goals_last_5 += form.goals_last_5 analysis.total_assists_last_5 += form.assists_last_5 if form.is_key_player: analysis.key_players_count += 1 analysis.total_squad = analysis.starting_count + analysis.sub_count # Takım deneyimi (bu takımla kaç maç oynamışlar) if analysis.starting_count > 0: cur.execute(""" SELECT AVG(match_count) as avg_exp FROM ( SELECT player_id, COUNT(*) as match_count FROM match_player_participation WHERE team_id = %s AND is_starting = true GROUP BY player_id ) sub """, (team_id,)) exp_row = cur.fetchone() if exp_row and exp_row['avg_exp']: # Normalize: 50+ maç = 1.0 analysis.squad_experience = min(exp_row['avg_exp'] / 50, 1.0) return analysis except Exception as e: print(f"[SquadEngine] Error analyzing squad: {e}") return analysis def analyze_squad_from_list(self, player_ids: List[str], team_id: str) -> SquadAnalysis: """ Memory'deki oyuncu listesinden kadro analizi yap. DB'de olmayan canlı maçlar için kullanılır. """ analysis = SquadAnalysis(team_id=team_id) # Varsayılan: İlk 11 oyuncu (listede genellikle ilk 11 verilir) # Eğer liste boşsa if not player_ids: return analysis # Varsayımlar: Mackolik API'den gelen liste sıralıdır. # İlk 11 genellikle as kadrodur. Ancak burada sadece 'starting' oyuncuları alıyoruz varsayalım. # User calling uses explicit starting 11 list. analysis.starting_count = len(player_ids) analysis.total_squad = len(player_ids) # Subs unknown usually unless separate list # Position tahmini zor, default dağıt? Veya oyuncu detayına git? # Hız için: Oyuncu ID'sinden DB'ye bakıp pozisyon öğrenmeye çalışabiliriz. conn = self.get_conn() if conn is None: return analysis try: with conn.cursor(cursor_factory=RealDictCursor) as cur: # Calculate stats for these specific players for pid in player_ids: # Get Form form = self.get_player_form(pid) analysis.total_goals_last_5 += form.goals_last_5 analysis.total_assists_last_5 += form.assists_last_5 if form.is_key_player: analysis.key_players_count += 1 # Get Position/Exp history attempt cur.execute(""" SELECT position, COUNT(*) as match_count FROM match_player_participation WHERE player_id = %s AND team_id = %s GROUP BY position ORDER BY match_count DESC LIMIT 1 """, (pid, team_id)) row = cur.fetchone() if row: pos = self._normalize_position(row.get('position', 'UNK')) if pos == 'GK': analysis.goalkeeper_count += 1 elif pos == 'DEF': analysis.defender_count += 1 elif pos == 'MID': analysis.midfielder_count += 1 elif pos == 'FWD': analysis.forward_count += 1 # Experience contribution exp = min(row['match_count'] / 50.0, 1.0) analysis.squad_experience += exp # Average experience if analysis.starting_count > 0: analysis.squad_experience /= analysis.starting_count except Exception as e: print(f"[SquadEngine] Live analyze error: {e}") return analysis def get_features( self, match_id: str, home_team_id: str, away_team_id: str ) -> Dict[str, float]: """ Maç için kadro feature'larını hesapla. Args: match_id: Maç ID'si home_team_id: Ev sahibi takım ID away_team_id: Deplasman takım ID Returns: Kadro feature'ları dict olarak """ features = SquadFeatures() # Ev sahibi analizi home = self.analyze_squad(match_id, home_team_id) features.home_starting_11 = home.starting_count features.home_sub_count = home.sub_count features.home_total_squad = home.total_squad features.home_goalkeepers = home.goalkeeper_count features.home_defenders = home.defender_count features.home_midfielders = home.midfielder_count features.home_forwards = home.forward_count features.home_goals_last_5 = home.total_goals_last_5 features.home_assists_last_5 = home.total_assists_last_5 features.home_key_players = home.key_players_count features.home_squad_experience = home.squad_experience # Deplasman analizi away = self.analyze_squad(match_id, away_team_id) features.away_starting_11 = away.starting_count features.away_sub_count = away.sub_count features.away_total_squad = away.total_squad features.away_goalkeepers = away.goalkeeper_count features.away_defenders = away.defender_count features.away_midfielders = away.midfielder_count features.away_forwards = away.forward_count features.away_goals_last_5 = away.total_goals_last_5 features.away_assists_last_5 = away.total_assists_last_5 features.away_key_players = away.key_players_count features.away_squad_experience = away.squad_experience # Karşılaştırma feature'ları home_strength = ( home.total_goals_last_5 * 2 + home.total_assists_last_5 + home.key_players_count * 3 + home.squad_experience * 10 ) away_strength = ( away.total_goals_last_5 * 2 + away.total_assists_last_5 + away.key_players_count * 3 + away.squad_experience * 10 ) features.squad_strength_diff = home_strength - away_strength features.goals_form_diff = home.total_goals_last_5 - away.total_goals_last_5 features.key_players_diff = home.key_players_count - away.key_players_count return features.to_dict() def get_features_without_match( self, home_team_id: str, away_team_id: str ) -> Dict[str, float]: """ Maç ID olmadan takım bazlı feature'ları hesapla. Son maçtaki kadroyu referans alır. """ features = SquadFeatures() conn = self.get_conn() if conn is None: return features.to_dict() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: for team_id, prefix in [(home_team_id, 'home'), (away_team_id, 'away')]: # Son maçı bul cur.execute(""" SELECT mpp.match_id FROM match_player_participation mpp JOIN matches m ON mpp.match_id = m.id WHERE mpp.team_id = %s ORDER BY m.mst_utc DESC LIMIT 1 """, (team_id,)) row = cur.fetchone() if row: analysis = self.analyze_squad(row['match_id'], team_id) if prefix == 'home': features.home_starting_11 = analysis.starting_count features.home_sub_count = analysis.sub_count features.home_total_squad = analysis.total_squad features.home_goals_last_5 = analysis.total_goals_last_5 features.home_assists_last_5 = analysis.total_assists_last_5 features.home_key_players = analysis.key_players_count features.home_squad_experience = analysis.squad_experience else: features.away_starting_11 = analysis.starting_count features.away_sub_count = analysis.sub_count features.away_total_squad = analysis.total_squad features.away_goals_last_5 = analysis.total_goals_last_5 features.away_assists_last_5 = analysis.total_assists_last_5 features.away_key_players = analysis.key_players_count features.away_squad_experience = analysis.squad_experience # Karşılaştırma features.goals_form_diff = features.home_goals_last_5 - features.away_goals_last_5 features.key_players_diff = features.home_key_players - features.away_key_players return features.to_dict() except Exception as e: print(f"[SquadEngine] Error: {e}") return features.to_dict() # Singleton instance _engine: Optional[SquadAnalysisEngine] = None def get_squad_analysis_engine() -> SquadAnalysisEngine: """Singleton squad analysis engine instance döndür""" global _engine if _engine is None: _engine = SquadAnalysisEngine() return _engine if __name__ == "__main__": # Test engine = get_squad_analysis_engine() print("\n🧪 Squad Analysis Engine Test") print("=" * 50) # Test with known team IDs (Galatasaray, Fenerbahce) features = engine.get_features_without_match( home_team_id="test_gs", away_team_id="test_fb" ) print("\n📊 Features:") for key, value in features.items(): print(f" {key}: {value:.2f}")