583 lines
22 KiB
Python
Executable File
583 lines
22 KiB
Python
Executable File
"""
|
||
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}")
|