435 lines
15 KiB
Python
Executable File
435 lines
15 KiB
Python
Executable File
"""
|
||
Momentum Engine - Son Maç Trendleri
|
||
V9 Model için takımların anlık form trendini analiz eder.
|
||
|
||
Faktörler:
|
||
1. Gol atma trendi (artan/azalan/stabil)
|
||
2. Yenilmezlik/yenilgi serisi
|
||
3. Son maç psikolojisi (büyük galibiyet/mağlubiyet etkisi)
|
||
4. Ev/Deplasman momentum farkı
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from typing import Dict, List, Tuple, Optional
|
||
from dataclasses import dataclass, field
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
try:
|
||
import psycopg2
|
||
from psycopg2.extras import RealDictCursor
|
||
except ImportError:
|
||
psycopg2 = None
|
||
|
||
|
||
@dataclass
|
||
class MomentumData:
|
||
"""Takım momentum verileri"""
|
||
goals_trend: float = 0.0 # -1 (azalan) to +1 (artan)
|
||
conceded_trend: float = 0.0 # -1 (azalan) to +1 (artan) [negatif iyi]
|
||
unbeaten_streak: int = 0 # Yenilmezlik serisi
|
||
losing_streak: int = 0 # Yenilgi serisi
|
||
winning_streak: int = 0 # Galibiyet serisi
|
||
last_match_impact: float = 0.0 # Son maç psikolojik etkisi (-1 to +1)
|
||
momentum_score: float = 0.0 # Toplam momentum (-1 to +1)
|
||
form_direction: str = "stable" # "improving", "declining", "stable"
|
||
xg_underperformance: float = 0.0 # (xG_For - Real_Goals) in last matches (>0 means underperforming)
|
||
xg_conceded_diff: float = 0.0 # (Real_Conceded - xG_Against) in last matches
|
||
|
||
|
||
class MomentumEngine:
|
||
"""
|
||
Son maçlardaki trendi analiz eder.
|
||
Form yükselişi/düşüşü, seriler ve psikolojik etki.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.conn = None
|
||
self._connect_db()
|
||
|
||
def _connect_db(self):
|
||
"""Veritabanına bağlan"""
|
||
if psycopg2 is None:
|
||
return
|
||
|
||
try:
|
||
from data.db import get_clean_dsn
|
||
self.conn = psycopg2.connect(get_clean_dsn())
|
||
except Exception as e:
|
||
print(f"[MomentumEngine] DB connection failed: {e}")
|
||
self.conn = None
|
||
|
||
def _get_conn(self):
|
||
"""Bağlantıyı kontrol et ve döndür"""
|
||
if self.conn is None or self.conn.closed:
|
||
self._connect_db()
|
||
return self.conn
|
||
|
||
def get_recent_matches(
|
||
self,
|
||
team_id: str,
|
||
before_date_ms: int,
|
||
limit: int = 5,
|
||
home_only: bool = False,
|
||
away_only: bool = False
|
||
) -> List[Dict]:
|
||
"""
|
||
Takımın son maçlarını getir.
|
||
|
||
Returns:
|
||
List of matches with scores and home/away info
|
||
"""
|
||
conn = self._get_conn()
|
||
if conn is None:
|
||
return []
|
||
|
||
try:
|
||
cursor = conn.cursor(cursor_factory=RealDictCursor)
|
||
|
||
conditions = ["mst_utc < %s", "score_home IS NOT NULL"]
|
||
params = [before_date_ms]
|
||
|
||
if home_only:
|
||
conditions.append("home_team_id = %s")
|
||
params.append(team_id)
|
||
elif away_only:
|
||
conditions.append("away_team_id = %s")
|
||
params.append(team_id)
|
||
else:
|
||
conditions.append("(home_team_id = %s OR away_team_id = %s)")
|
||
params.extend([team_id, team_id])
|
||
query = f"""
|
||
SELECT
|
||
id, home_team_id, away_team_id,
|
||
score_home, score_away, mst_utc
|
||
FROM matches
|
||
WHERE {' AND '.join(conditions)}
|
||
ORDER BY mst_utc DESC
|
||
LIMIT %s
|
||
"""
|
||
params.append(limit)
|
||
|
||
cursor.execute(query, params)
|
||
return cursor.fetchall()
|
||
|
||
except Exception as e:
|
||
print(f"[MomentumEngine] Query error: {e}")
|
||
return []
|
||
|
||
def calculate_goals_trend(self, matches: List[Dict], team_id: str) -> Tuple[float, float]:
|
||
"""
|
||
Gol atma ve yeme trendini hesapla.
|
||
Son 3 maç vs önceki 2 maç karşılaştırması.
|
||
|
||
Returns:
|
||
(goals_trend, conceded_trend) - -1 to +1
|
||
"""
|
||
if len(matches) < 3:
|
||
return 0.0, 0.0
|
||
|
||
# Her maç için gol ve yenilen gol hesapla
|
||
goals = []
|
||
conceded = []
|
||
|
||
for match in matches:
|
||
if match['home_team_id'] == team_id:
|
||
goals.append(match['score_home'])
|
||
conceded.append(match['score_away'])
|
||
else:
|
||
goals.append(match['score_away'])
|
||
conceded.append(match['score_home'])
|
||
|
||
# Son 3 vs önceki maçlar
|
||
recent_goals = sum(goals[:3]) / 3 if len(goals) >= 3 else 0
|
||
older_goals = sum(goals[3:]) / len(goals[3:]) if len(goals) > 3 else recent_goals
|
||
|
||
recent_conceded = sum(conceded[:3]) / 3 if len(conceded) >= 3 else 0
|
||
older_conceded = sum(conceded[3:]) / len(conceded[3:]) if len(conceded) > 3 else recent_conceded
|
||
|
||
# Trend hesapla (-1 to +1)
|
||
goals_trend = min(max((recent_goals - older_goals) / 2, -1), 1)
|
||
conceded_trend = min(max((recent_conceded - older_conceded) / 2, -1), 1)
|
||
|
||
return goals_trend, conceded_trend
|
||
|
||
def calculate_streaks(self, matches: List[Dict], team_id: str) -> Tuple[int, int, int]:
|
||
"""
|
||
Galibiyet, yenilmezlik ve yenilgi serilerini hesapla.
|
||
|
||
Returns:
|
||
(winning_streak, unbeaten_streak, losing_streak)
|
||
"""
|
||
winning = 0
|
||
unbeaten = 0
|
||
losing = 0
|
||
|
||
for match in matches:
|
||
# Sonucu belirle
|
||
if match['home_team_id'] == team_id:
|
||
goals_for = match['score_home']
|
||
goals_against = match['score_away']
|
||
else:
|
||
goals_for = match['score_away']
|
||
goals_against = match['score_home']
|
||
|
||
if goals_for > goals_against: # Galibiyet
|
||
if losing == 0: # Henüz yenilgi serisi başlamamış
|
||
winning += 1
|
||
unbeaten += 1
|
||
else:
|
||
break
|
||
elif goals_for == goals_against: # Beraberlik
|
||
if losing == 0:
|
||
winning = 0 # Galibiyet serisi bitti
|
||
unbeaten += 1
|
||
else:
|
||
break
|
||
else: # Yenilgi
|
||
if winning > 0 or unbeaten > 0:
|
||
winning = 0
|
||
unbeaten = 0
|
||
losing += 1
|
||
|
||
return winning, unbeaten, losing
|
||
|
||
def calculate_last_match_impact(self, matches: List[Dict], team_id: str) -> float:
|
||
"""
|
||
Son maçın psikolojik etkisini hesapla.
|
||
Büyük galibiyet = +1, büyük mağlubiyet = -1
|
||
|
||
Returns:
|
||
impact score: -1 to +1
|
||
"""
|
||
if not matches:
|
||
return 0.0
|
||
|
||
last_match = matches[0]
|
||
|
||
if last_match['home_team_id'] == team_id:
|
||
goals_for = last_match['score_home']
|
||
goals_against = last_match['score_away']
|
||
else:
|
||
goals_for = last_match['score_away']
|
||
goals_against = last_match['score_home']
|
||
|
||
goal_diff = goals_for - goals_against
|
||
|
||
# Gol farkına göre etki
|
||
if goal_diff >= 4:
|
||
return 1.0 # Çok büyük galibiyet
|
||
elif goal_diff >= 2:
|
||
return 0.6
|
||
elif goal_diff == 1:
|
||
return 0.3
|
||
elif goal_diff == 0:
|
||
return 0.0
|
||
elif goal_diff == -1:
|
||
return -0.3
|
||
elif goal_diff >= -3:
|
||
return -0.6
|
||
else:
|
||
return -1.0 # Çok büyük mağlubiyet
|
||
|
||
def calculate_xg_underperformance(self, matches: List[Dict], team_id: str) -> Tuple[float, float]:
|
||
"""
|
||
Calculate if a team chronically underperforms its xG (Expected Goals).
|
||
Returns:
|
||
(xg_strike_diff, xg_defend_diff)
|
||
xg_strike_diff: > 0 means they score LESS than expected (Bad Finishers)
|
||
xg_defend_diff: > 0 means they concede MORE than expected (Bad Goalkeeper/Luck)
|
||
"""
|
||
if not matches:
|
||
return 0.0, 0.0
|
||
|
||
real_scored = 0
|
||
xg_created = 0.0
|
||
|
||
real_conceded = 0
|
||
xg_conceded = 0.0
|
||
|
||
for m in matches:
|
||
is_home = (m['home_team_id'] == team_id)
|
||
if is_home:
|
||
real_scored += m['score_home']
|
||
real_conceded += m['score_away']
|
||
# Create synthetic xG data (mock based on score for demo since stats table absent)
|
||
xg_created += max(0.5, m['score_home'] * 1.5 - 0.5)
|
||
xg_conceded += max(0.5, m['score_away'] * 1.5 - 0.5)
|
||
else:
|
||
real_scored += m['score_away']
|
||
real_conceded += m['score_home']
|
||
xg_created += max(0.5, m['score_away'] * 1.5 - 0.5)
|
||
xg_conceded += max(0.5, m['score_home'] * 1.5 - 0.5)
|
||
|
||
# Calculate per match diffs
|
||
match_count = len(matches)
|
||
|
||
xg_strike_diff = (xg_created - real_scored) / match_count if match_count else 0
|
||
xg_defend_diff = (real_conceded - xg_conceded) / match_count if match_count else 0
|
||
|
||
return xg_strike_diff, xg_defend_diff
|
||
|
||
def calculate_momentum(
|
||
self,
|
||
team_id: str,
|
||
before_date_ms: int,
|
||
match_limit: int = 5
|
||
) -> MomentumData:
|
||
"""
|
||
Takımın tam momentum analizini yap.
|
||
|
||
Returns:
|
||
MomentumData with all metrics
|
||
"""
|
||
data = MomentumData()
|
||
|
||
matches = self.get_recent_matches(team_id, before_date_ms, match_limit)
|
||
|
||
if not matches:
|
||
return data
|
||
|
||
# 1. Gol trendi
|
||
data.goals_trend, data.conceded_trend = self.calculate_goals_trend(matches, team_id)
|
||
|
||
# 2. Seriler
|
||
data.winning_streak, data.unbeaten_streak, data.losing_streak = \
|
||
self.calculate_streaks(matches, team_id)
|
||
|
||
# 3. Son maç etkisi
|
||
data.last_match_impact = self.calculate_last_match_impact(matches, team_id)
|
||
|
||
# 4. Form yönü belirleme
|
||
if data.goals_trend > 0.3 and data.conceded_trend < 0:
|
||
data.form_direction = "improving"
|
||
elif data.goals_trend < -0.3 or data.conceded_trend > 0.3:
|
||
data.form_direction = "declining"
|
||
else:
|
||
data.form_direction = "stable"
|
||
|
||
# 5. xG Underperformance (Chronik beceriksizlik)
|
||
data.xg_underperformance, data.xg_conceded_diff = self.calculate_xg_underperformance(matches, team_id)
|
||
|
||
# 6. Toplam momentum skoru
|
||
momentum = 0.0
|
||
|
||
# Gol trendi + savunma trendi (ters çevrilmiş)
|
||
momentum += data.goals_trend * 0.25
|
||
momentum += (-data.conceded_trend) * 0.20
|
||
|
||
# Seri bonusları
|
||
if data.winning_streak >= 3:
|
||
momentum += 0.25
|
||
elif data.winning_streak >= 2:
|
||
momentum += 0.15
|
||
elif data.unbeaten_streak >= 5:
|
||
momentum += 0.15
|
||
|
||
if data.losing_streak >= 3:
|
||
momentum -= 0.30
|
||
elif data.losing_streak >= 2:
|
||
momentum -= 0.15
|
||
|
||
# Son maç etkisi
|
||
momentum += data.last_match_impact * 0.20
|
||
|
||
# Ceza: xG Underperformance Penalty (Beceriksizlik Cezası)
|
||
# Eğer takım attığından çok xG üretiyorsa (- puan)
|
||
if data.xg_underperformance > 0.5: # Maç başı 0.5 gol eksik atıyor!
|
||
momentum -= min(0.3, data.xg_underperformance * 0.2)
|
||
|
||
# Ceza: xG Defend Underperformance (Kötü kaleci Cezası)
|
||
# Eğer beklenenden çok gol yiyorsa
|
||
if data.xg_conceded_diff > 0.5:
|
||
momentum -= min(0.3, data.xg_conceded_diff * 0.2)
|
||
|
||
data.momentum_score = min(max(momentum, -1), 1)
|
||
|
||
return data
|
||
|
||
def get_features(
|
||
self,
|
||
home_team_id: str,
|
||
away_team_id: str,
|
||
match_date_ms: int
|
||
) -> Dict[str, float]:
|
||
"""
|
||
Model için feature dict döndür.
|
||
"""
|
||
home_momentum = self.calculate_momentum(home_team_id, match_date_ms)
|
||
away_momentum = self.calculate_momentum(away_team_id, match_date_ms)
|
||
|
||
# Form direction encoding
|
||
direction_map = {"improving": 1, "stable": 0, "declining": -1}
|
||
|
||
return {
|
||
# Ev sahibi momentum
|
||
"home_momentum_score": home_momentum.momentum_score,
|
||
"home_goals_trend": home_momentum.goals_trend,
|
||
"home_conceded_trend": home_momentum.conceded_trend,
|
||
"home_winning_streak": min(home_momentum.winning_streak, 5),
|
||
"home_unbeaten_streak": min(home_momentum.unbeaten_streak, 10),
|
||
"home_losing_streak": min(home_momentum.losing_streak, 5),
|
||
"home_last_impact": home_momentum.last_match_impact,
|
||
"home_form_direction": direction_map.get(home_momentum.form_direction, 0),
|
||
"home_xg_underperf": home_momentum.xg_underperformance,
|
||
"home_xg_conceded_diff": home_momentum.xg_conceded_diff,
|
||
|
||
# Deplasman momentum
|
||
"away_momentum_score": away_momentum.momentum_score,
|
||
"away_goals_trend": away_momentum.goals_trend,
|
||
"away_conceded_trend": away_momentum.conceded_trend,
|
||
"away_winning_streak": min(away_momentum.winning_streak, 5),
|
||
"away_unbeaten_streak": min(away_momentum.unbeaten_streak, 10),
|
||
"away_losing_streak": min(away_momentum.losing_streak, 5),
|
||
"away_last_impact": away_momentum.last_match_impact,
|
||
"away_form_direction": direction_map.get(away_momentum.form_direction, 0),
|
||
"away_xg_underperf": away_momentum.xg_underperformance,
|
||
"away_xg_conceded_diff": away_momentum.xg_conceded_diff,
|
||
|
||
# Farklar
|
||
"momentum_diff": home_momentum.momentum_score - away_momentum.momentum_score,
|
||
"trend_diff": (home_momentum.goals_trend - home_momentum.conceded_trend) -
|
||
(away_momentum.goals_trend - away_momentum.conceded_trend),
|
||
"xg_underperf_diff": home_momentum.xg_underperformance - away_momentum.xg_underperformance,
|
||
}
|
||
|
||
|
||
# Singleton instance
|
||
_engine_instance = None
|
||
|
||
def get_momentum_engine() -> MomentumEngine:
|
||
"""Singleton pattern ile engine döndür"""
|
||
global _engine_instance
|
||
if _engine_instance is None:
|
||
_engine_instance = MomentumEngine()
|
||
return _engine_instance
|
||
|
||
|
||
# Test
|
||
if __name__ == "__main__":
|
||
engine = get_momentum_engine()
|
||
|
||
# Test data
|
||
print("=" * 60)
|
||
print("MOMENTUM ENGINE TEST")
|
||
print("=" * 60)
|
||
|
||
# Örnek hesaplama (DB olmadan)
|
||
data = MomentumData(
|
||
goals_trend=0.5,
|
||
conceded_trend=-0.3,
|
||
winning_streak=3,
|
||
unbeaten_streak=5,
|
||
losing_streak=0,
|
||
last_match_impact=0.6,
|
||
form_direction="improving"
|
||
)
|
||
|
||
print(f"Goals Trend: {data.goals_trend}")
|
||
print(f"Conceded Trend: {data.conceded_trend}")
|
||
print(f"Winning Streak: {data.winning_streak}")
|
||
print(f"Unbeaten Streak: {data.unbeaten_streak}")
|
||
print(f"Form Direction: {data.form_direction}")
|
||
print(f"Last Match Impact: {data.last_match_impact}")
|