This commit is contained in:
Executable
+194
@@ -0,0 +1,194 @@
|
||||
"""
|
||||
Team Stats Engine
|
||||
Takımların oyun tarzı istatistiklerini analiz eder.
|
||||
football_team_stats tablosundaki kayıtlardan possession, şut, korner verilerini kullanır.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import psycopg2
|
||||
from typing import Dict
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from data.db import get_clean_dsn
|
||||
|
||||
|
||||
class TeamStatsEngine:
|
||||
"""
|
||||
Takım istatistikleri için feature engine.
|
||||
|
||||
Analiz edilen metrikler:
|
||||
- Ortalama top hakimiyeti (possession)
|
||||
- Ortalama isabetli şut
|
||||
- Ortalama korner
|
||||
- Şut/Gol dönüşüm oranı (xG benzeri)
|
||||
- Savunma gücü
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.conn = None
|
||||
|
||||
def get_conn(self):
|
||||
if self.conn is None or self.conn.closed:
|
||||
self.conn = psycopg2.connect(get_clean_dsn())
|
||||
return self.conn
|
||||
|
||||
def get_features(self, team_id: str, before_date: int,
|
||||
limit: int = 10, max_days: int = 180) -> Dict[str, float]:
|
||||
"""
|
||||
Takımın oyun tarzı feature'larını hesapla.
|
||||
|
||||
Args:
|
||||
team_id: Takım ID
|
||||
before_date: Bu tarihten önceki maçlara bak (ms timestamp)
|
||||
limit: Kaç maç analiz edilecek
|
||||
max_days: Maksimum kaç gün geriye gidilecek
|
||||
|
||||
Returns:
|
||||
Dict: Team stats feature'ları
|
||||
"""
|
||||
if not team_id or len(team_id) < 5:
|
||||
return self._default_features()
|
||||
|
||||
try:
|
||||
conn = self.get_conn()
|
||||
cur = conn.cursor()
|
||||
|
||||
min_date = before_date - (max_days * 24 * 60 * 60 * 1000)
|
||||
|
||||
# Bu takımın son N maçındaki istatistikleri çek
|
||||
cur.execute("""
|
||||
SELECT
|
||||
mts.possession_percentage,
|
||||
mts.shots_on_target,
|
||||
mts.shots_off_target,
|
||||
mts.total_shots,
|
||||
mts.corners,
|
||||
mts.fouls,
|
||||
m.score_home,
|
||||
m.score_away,
|
||||
m.home_team_id
|
||||
FROM football_team_stats mts
|
||||
JOIN matches m ON mts.match_id = m.id
|
||||
WHERE mts.team_id = %s
|
||||
AND m.mst_utc < %s
|
||||
AND m.mst_utc > %s
|
||||
AND m.score_home IS NOT NULL
|
||||
AND m.sport = 'football'
|
||||
ORDER BY m.mst_utc DESC
|
||||
LIMIT %s
|
||||
""", (team_id, before_date, min_date, limit))
|
||||
|
||||
stats = cur.fetchall()
|
||||
|
||||
if not stats:
|
||||
return self._default_features()
|
||||
|
||||
# İstatistikleri hesapla
|
||||
total_matches = len(stats)
|
||||
|
||||
possession_sum = 0
|
||||
shots_on_target_sum = 0
|
||||
shots_total_sum = 0
|
||||
corners_sum = 0
|
||||
fouls_sum = 0
|
||||
goals_scored = 0
|
||||
valid_possession_count = 0
|
||||
|
||||
for stat in stats:
|
||||
poss, sot, soff, total_shots, corners, fouls, sh, sa, home_id = stat
|
||||
|
||||
if poss and poss > 0:
|
||||
possession_sum += poss
|
||||
valid_possession_count += 1
|
||||
|
||||
if sot:
|
||||
shots_on_target_sum += sot
|
||||
if total_shots:
|
||||
shots_total_sum += total_shots
|
||||
if corners:
|
||||
corners_sum += corners
|
||||
if fouls:
|
||||
fouls_sum += fouls
|
||||
|
||||
# Gol hesaplama
|
||||
is_home = (home_id == team_id)
|
||||
goals_scored += sh if is_home else sa
|
||||
|
||||
avg_possession = possession_sum / valid_possession_count if valid_possession_count > 0 else 50.0
|
||||
avg_shots_on_target = shots_on_target_sum / total_matches if total_matches > 0 else 3.0
|
||||
avg_shots_total = shots_total_sum / total_matches if total_matches > 0 else 10.0
|
||||
avg_corners = corners_sum / total_matches if total_matches > 0 else 4.0
|
||||
avg_fouls = fouls_sum / total_matches if total_matches > 0 else 12.0
|
||||
|
||||
# Shot conversion rate (xG benzeri)
|
||||
shot_conversion = goals_scored / shots_total_sum if shots_total_sum > 0 else 0.1
|
||||
|
||||
# Shot accuracy
|
||||
shot_accuracy = shots_on_target_sum / shots_total_sum if shots_total_sum > 0 else 0.35
|
||||
|
||||
return {
|
||||
'avg_possession': avg_possession / 100, # Normalize to 0-1
|
||||
'avg_shots_on_target': avg_shots_on_target,
|
||||
'avg_shots_total': avg_shots_total,
|
||||
'avg_corners': avg_corners,
|
||||
'avg_fouls': avg_fouls,
|
||||
'shot_conversion_rate': shot_conversion,
|
||||
'shot_accuracy': shot_accuracy,
|
||||
'attacking_intensity': (avg_shots_total + avg_corners) / 2
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[TeamStatsEngine] Error: {e}")
|
||||
return self._default_features()
|
||||
|
||||
def _default_features(self) -> Dict[str, float]:
|
||||
return {
|
||||
'avg_possession': 0.50,
|
||||
'avg_shots_on_target': 3.5,
|
||||
'avg_shots_total': 11.0,
|
||||
'avg_corners': 4.5,
|
||||
'avg_fouls': 12.0,
|
||||
'shot_conversion_rate': 0.10,
|
||||
'shot_accuracy': 0.35,
|
||||
'attacking_intensity': 7.5
|
||||
}
|
||||
|
||||
|
||||
# Singleton
|
||||
_engine = None
|
||||
|
||||
def get_team_stats_engine() -> TeamStatsEngine:
|
||||
global _engine
|
||||
if _engine is None:
|
||||
_engine = TeamStatsEngine()
|
||||
return _engine
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
engine = get_team_stats_engine()
|
||||
|
||||
print("\n🧪 Team Stats Engine Test")
|
||||
print("=" * 50)
|
||||
|
||||
# Test için örnek takım ID'si al
|
||||
conn = engine.get_conn()
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT DISTINCT mts.team_id, t.name
|
||||
FROM match_team_stats mts
|
||||
JOIN teams t ON mts.team_id = t.id
|
||||
LIMIT 1
|
||||
""")
|
||||
result = cur.fetchone()
|
||||
|
||||
if result:
|
||||
team_id, team_name = result
|
||||
print(f"Test Takımı: {team_name}")
|
||||
|
||||
import time
|
||||
features = engine.get_features(team_id, int(time.time() * 1000))
|
||||
|
||||
print(f"\n📊 Feature'lar:")
|
||||
for k, v in features.items():
|
||||
print(f" {k}: {v:.3f}")
|
||||
Reference in New Issue
Block a user