317 lines
10 KiB
Python
Executable File
317 lines
10 KiB
Python
Executable File
"""
|
||
Head-to-Head (H2H) Feature Engine
|
||
Takımların birbirine karşı geçmiş performansını analiz eder.
|
||
"""
|
||
|
||
import os
|
||
import psycopg2
|
||
from typing import Dict, Optional, Tuple
|
||
from dataclasses import dataclass
|
||
from functools import lru_cache
|
||
|
||
import sys
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from data.db import get_clean_dsn
|
||
|
||
|
||
@dataclass
|
||
class H2HProfile:
|
||
"""Head-to-Head analiz sonucu"""
|
||
total_matches: int
|
||
home_wins: int
|
||
draws: int
|
||
away_wins: int
|
||
home_goals_total: int
|
||
away_goals_total: int
|
||
btts_count: int # Both teams to score
|
||
over25_count: int
|
||
|
||
@property
|
||
def home_win_rate(self) -> float:
|
||
return self.home_wins / self.total_matches if self.total_matches > 0 else 0.33
|
||
|
||
@property
|
||
def draw_rate(self) -> float:
|
||
return self.draws / self.total_matches if self.total_matches > 0 else 0.33
|
||
|
||
@property
|
||
def away_win_rate(self) -> float:
|
||
return self.away_wins / self.total_matches if self.total_matches > 0 else 0.33
|
||
|
||
@property
|
||
def avg_total_goals(self) -> float:
|
||
return (self.home_goals_total + self.away_goals_total) / self.total_matches if self.total_matches > 0 else 2.5
|
||
|
||
@property
|
||
def btts_rate(self) -> float:
|
||
return self.btts_count / self.total_matches if self.total_matches > 0 else 0.5
|
||
|
||
@property
|
||
def over25_rate(self) -> float:
|
||
return self.over25_count / self.total_matches if self.total_matches > 0 else 0.5
|
||
|
||
@property
|
||
def home_dominance(self) -> float:
|
||
"""Ev sahibinin üstünlük skoru (-1 ile 1 arası)"""
|
||
if self.total_matches == 0:
|
||
return 0
|
||
return (self.home_wins - self.away_wins) / self.total_matches
|
||
|
||
def to_features(self) -> Dict[str, float]:
|
||
"""Feature dictionary döndür"""
|
||
return {
|
||
'h2h_total_matches': self.total_matches,
|
||
'h2h_home_win_rate': self.home_win_rate,
|
||
'h2h_draw_rate': self.draw_rate,
|
||
'h2h_away_win_rate': self.away_win_rate,
|
||
'h2h_avg_goals': self.avg_total_goals,
|
||
'h2h_btts_rate': self.btts_rate,
|
||
'h2h_over25_rate': self.over25_rate,
|
||
'h2h_home_dominance': self.home_dominance,
|
||
}
|
||
|
||
|
||
class H2HFeatureEngine:
|
||
"""
|
||
Head-to-Head Feature Engine
|
||
|
||
İki takım arasındaki geçmiş karşılaşmaları analiz eder.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.conn = None
|
||
self._cache: Dict[Tuple[str, str], H2HProfile] = {}
|
||
|
||
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_h2h_profile(self, home_team_id: str, away_team_id: str,
|
||
before_date: Optional[int] = None,
|
||
limit: int = 20) -> H2HProfile:
|
||
"""
|
||
İki takım arasındaki geçmiş karşılaşmaları analiz et.
|
||
|
||
Args:
|
||
home_team_id: Ev sahibi takım ID
|
||
away_team_id: Deplasman takım ID
|
||
before_date: Bu tarihten önceki maçlar (mst_utc, milliseconds)
|
||
limit: Kaç maç geriye bakılacak
|
||
|
||
Returns:
|
||
H2HProfile: Head-to-head analiz sonucu
|
||
"""
|
||
cache_key = (home_team_id, away_team_id)
|
||
|
||
# Cache kontrolü (before_date yoksa)
|
||
if before_date is None and cache_key in self._cache:
|
||
return self._cache[cache_key]
|
||
|
||
conn = self.get_conn()
|
||
cur = conn.cursor()
|
||
|
||
# Her iki yöndeki karşılaşmaları al
|
||
# (A evde B deplasman + B evde A deplasman)
|
||
query = """
|
||
SELECT
|
||
home_team_id, away_team_id,
|
||
score_home, score_away
|
||
FROM matches
|
||
WHERE (
|
||
(home_team_id = %s AND away_team_id = %s)
|
||
OR
|
||
(home_team_id = %s AND away_team_id = %s)
|
||
)
|
||
AND score_home IS NOT NULL
|
||
AND score_away IS NOT NULL
|
||
"""
|
||
|
||
params = [home_team_id, away_team_id, away_team_id, home_team_id]
|
||
|
||
if before_date:
|
||
query += " AND mst_utc < %s"
|
||
params.append(before_date)
|
||
|
||
query += " ORDER BY mst_utc DESC LIMIT %s"
|
||
params.append(limit)
|
||
|
||
cur.execute(query, params)
|
||
matches = cur.fetchall()
|
||
|
||
if not matches:
|
||
return H2HProfile(
|
||
total_matches=0, home_wins=0, draws=0, away_wins=0,
|
||
home_goals_total=0, away_goals_total=0,
|
||
btts_count=0, over25_count=0
|
||
)
|
||
|
||
# İstatistikleri hesapla
|
||
home_wins = 0
|
||
draws = 0
|
||
away_wins = 0
|
||
home_goals = 0
|
||
away_goals = 0
|
||
btts = 0
|
||
over25 = 0
|
||
|
||
for match in matches:
|
||
m_home_id, m_away_id, score_h, score_a = match
|
||
|
||
# Perspektifi normalize et (istenen takım açısından)
|
||
if m_home_id == home_team_id:
|
||
# Normal sıralama
|
||
h_score, a_score = score_h, score_a
|
||
else:
|
||
# Ters sıralama (rakip evde oynamış)
|
||
h_score, a_score = score_a, score_h
|
||
|
||
# Sonuç
|
||
if h_score > a_score:
|
||
home_wins += 1
|
||
elif h_score < a_score:
|
||
away_wins += 1
|
||
else:
|
||
draws += 1
|
||
|
||
# Goller
|
||
home_goals += h_score
|
||
away_goals += a_score
|
||
|
||
# BTTS
|
||
if h_score > 0 and a_score > 0:
|
||
btts += 1
|
||
|
||
# Over 2.5
|
||
if h_score + a_score > 2.5:
|
||
over25 += 1
|
||
|
||
profile = H2HProfile(
|
||
total_matches=len(matches),
|
||
home_wins=home_wins,
|
||
draws=draws,
|
||
away_wins=away_wins,
|
||
home_goals_total=home_goals,
|
||
away_goals_total=away_goals,
|
||
btts_count=btts,
|
||
over25_count=over25
|
||
)
|
||
|
||
# Cache'e kaydet
|
||
if before_date is None:
|
||
self._cache[cache_key] = profile
|
||
|
||
return profile
|
||
|
||
def get_features(self, home_team_id: str, away_team_id: str,
|
||
before_date: Optional[int] = None) -> Dict[str, float]:
|
||
"""Feature dictionary döndür"""
|
||
profile = self.get_h2h_profile(home_team_id, away_team_id, before_date)
|
||
return profile.to_features()
|
||
|
||
def get_momentum(self, home_team_id: str, away_team_id: str,
|
||
before_date: Optional[int] = None) -> Dict[str, float]:
|
||
"""
|
||
Son karşılaşmalardaki momentum/trend analizi.
|
||
Son 5 maçtaki trend'e bakar.
|
||
"""
|
||
profile = self.get_h2h_profile(home_team_id, away_team_id, before_date, limit=5)
|
||
|
||
# Streak hesapla (ardışık sonuçlar)
|
||
conn = self.get_conn()
|
||
cur = conn.cursor()
|
||
|
||
query = """
|
||
SELECT home_team_id, score_home, score_away
|
||
FROM matches
|
||
WHERE (
|
||
(home_team_id = %s AND away_team_id = %s)
|
||
OR
|
||
(home_team_id = %s AND away_team_id = %s)
|
||
)
|
||
AND score_home IS NOT NULL
|
||
"""
|
||
params = [home_team_id, away_team_id, away_team_id, home_team_id]
|
||
if before_date:
|
||
query += " AND mst_utc < %s"
|
||
params.append(before_date)
|
||
query += " ORDER BY mst_utc DESC LIMIT 5"
|
||
|
||
cur.execute(query, params)
|
||
recent = cur.fetchall()
|
||
|
||
streak = 0
|
||
streak_type = None # 'home', 'away', 'draw'
|
||
|
||
for match in recent:
|
||
m_home_id, score_h, score_a = match
|
||
|
||
# Perspektifi normalize et
|
||
if m_home_id == home_team_id:
|
||
result = 'home' if score_h > score_a else ('away' if score_h < score_a else 'draw')
|
||
else:
|
||
result = 'away' if score_h > score_a else ('home' if score_h < score_a else 'draw')
|
||
|
||
if streak_type is None:
|
||
streak_type = result
|
||
streak = 1
|
||
elif result == streak_type:
|
||
streak += 1
|
||
else:
|
||
break
|
||
|
||
return {
|
||
'h2h_recent_home_dominance': profile.home_dominance,
|
||
'h2h_streak_length': streak,
|
||
'h2h_streak_home': 1 if streak_type == 'home' else 0,
|
||
'h2h_streak_away': 1 if streak_type == 'away' else 0,
|
||
'h2h_streak_draw': 1 if streak_type == 'draw' else 0,
|
||
}
|
||
|
||
|
||
# Singleton
|
||
_engine = None
|
||
|
||
def get_h2h_engine() -> H2HFeatureEngine:
|
||
global _engine
|
||
if _engine is None:
|
||
_engine = H2HFeatureEngine()
|
||
return _engine
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Test
|
||
engine = get_h2h_engine()
|
||
|
||
# Örnek: Fenerbahçe vs Galatasaray (ID'leri bulunmalı)
|
||
# Test için veritabanından bir karşılaşma çekelim
|
||
conn = engine.get_conn()
|
||
cur = conn.cursor()
|
||
cur.execute("""
|
||
SELECT home_team_id, away_team_id, match_name
|
||
FROM matches
|
||
WHERE score_home IS NOT NULL
|
||
LIMIT 1
|
||
""")
|
||
result = cur.fetchone()
|
||
|
||
if result:
|
||
home_id, away_id, name = result
|
||
print(f"\n🧪 Test: {name}")
|
||
print(f" Home ID: {home_id}")
|
||
print(f" Away ID: {away_id}")
|
||
|
||
profile = engine.get_h2h_profile(home_id, away_id)
|
||
print(f"\n📊 H2H Profil:")
|
||
print(f" Toplam Maç: {profile.total_matches}")
|
||
print(f" Ev Sahibi Kazanma: {profile.home_win_rate:.1%}")
|
||
print(f" Beraberlik: {profile.draw_rate:.1%}")
|
||
print(f" Deplasman Kazanma: {profile.away_win_rate:.1%}")
|
||
print(f" Ortalama Gol: {profile.avg_total_goals:.2f}")
|
||
print(f" BTTS Oranı: {profile.btts_rate:.1%}")
|
||
print(f" Üst 2.5 Oranı: {profile.over25_rate:.1%}")
|
||
print(f" Ev Dominance: {profile.home_dominance:+.2f}")
|
||
|
||
features = engine.get_features(home_id, away_id)
|
||
print(f"\n🔧 Features: {features}")
|