""" 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}")