"""Feature engineering""" from typing import List, Dict, Optional from datetime import datetime, timedelta import pandas as pd from sqlalchemy.orm import Session from app.core.db.models.match import Match, MatchStatus from app.core.db.models.team import Team from app.core.db.models.feature import TeamFormDaily, Feature from app.core.db.repositories.match_repository import MatchRepository from app.ml.elo import EloRating from app.core.config import settings import logging logger = logging.getLogger(__name__) class FeatureBuilder: """Build features for matches""" def __init__(self, db: Session): self.db = db self.match_repo = MatchRepository(db) self.rolling_window = settings.rolling_window_matches def calculate_team_form( self, team_id: int, date: datetime, is_home: bool ) -> Dict[str, float]: """Calculate team form statistics up to a given date""" # Get recent matches for the team matches = self.match_repo.get_finished( from_date=date - timedelta(days=365), to_date=date - timedelta(days=1), limit=1000 ) # Filter by team and home/away team_matches = [] for match in matches: if match.home_team_id == team_id and is_home: team_matches.append((match, True)) elif match.away_team_id == team_id and not is_home: team_matches.append((match, False)) # Sort by date descending and take last N team_matches.sort(key=lambda x: x[0].match_date, reverse=True) team_matches = team_matches[:self.rolling_window] if not team_matches: return { "goals_for_avg": 0.0, "goals_against_avg": 0.0, "points_avg": 0.0, "form_last_n": 0.0, "days_rest": 7.0 } # Calculate statistics goals_for = [] goals_against = [] points = [] form_points = [] # 3 for win, 1 for draw, 0 for loss last_match_date = None for match, is_home_team in team_matches: if is_home_team: goals_for.append(match.home_score or 0) goals_against.append(match.away_score or 0) if match.home_score > match.away_score: points.append(3) form_points.append(3) elif match.home_score == match.away_score: points.append(1) form_points.append(1) else: points.append(0) form_points.append(0) else: goals_for.append(match.away_score or 0) goals_against.append(match.home_score or 0) if match.away_score > match.home_score: points.append(3) form_points.append(3) elif match.away_score == match.home_score: points.append(1) form_points.append(1) else: points.append(0) form_points.append(0) if not last_match_date: last_match_date = match.match_date # Calculate days rest days_rest = 7.0 if last_match_date: days_rest = (date - last_match_date).days return { "goals_for_avg": sum(goals_for) / len(goals_for) if goals_for else 0.0, "goals_against_avg": sum(goals_against) / len(goals_against) if goals_against else 0.0, "points_avg": sum(points) / len(points) if points else 0.0, "form_last_n": sum(form_points) / (len(form_points) * 3) if form_points else 0.0, "days_rest": days_rest } def calculate_elo_ratings( self, matches: List[Match], date: datetime ) -> Dict[int, float]: """Calculate Elo ratings for all teams up to a date""" team_ratings: Dict[int, float] = {} # Process matches chronologically historical_matches = [m for m in matches if m.match_date < date] historical_matches.sort(key=lambda x: x.match_date) for match in historical_matches: if match.status != MatchStatus.FINISHED.value: continue if match.home_score is None or match.away_score is None: continue EloRating.calculate_ratings( team_ratings, match.home_team_id, match.away_team_id, match.home_score, match.away_score ) return team_ratings def calculate_h2h( self, home_team_id: int, away_team_id: int, date: datetime, limit: int = 10 ) -> Dict[str, int]: """Calculate head-to-head statistics""" matches = self.match_repo.get_finished( from_date=date - timedelta(days=365 * 2), to_date=date - timedelta(days=1), limit=1000 ) h2h_matches = [] for match in matches: if ((match.home_team_id == home_team_id and match.away_team_id == away_team_id) or (match.home_team_id == away_team_id and match.away_team_id == home_team_id)): h2h_matches.append(match) h2h_matches.sort(key=lambda x: x.match_date, reverse=True) h2h_matches = h2h_matches[:limit] home_wins = 0 draws = 0 away_wins = 0 for match in h2h_matches: if match.home_score is None or match.away_score is None: continue if match.home_team_id == home_team_id: if match.home_score > match.away_score: home_wins += 1 elif match.home_score == match.away_score: draws += 1 else: away_wins += 1 else: if match.away_score > match.home_score: home_wins += 1 elif match.away_score == match.home_score: draws += 1 else: away_wins += 1 return { "h2h_home_wins": home_wins, "h2h_draws": draws, "h2h_away_wins": away_wins } def build_features_for_match(self, match: Match) -> Feature: """Build features for a single match""" # Calculate team forms home_form = self.calculate_team_form(match.home_team_id, match.match_date, is_home=True) away_form = self.calculate_team_form(match.away_team_id, match.match_date, is_home=False) # Calculate Elo ratings all_matches = self.match_repo.get_finished(limit=5000) elo_ratings = self.calculate_elo_ratings(all_matches, match.match_date) home_elo = elo_ratings.get(match.home_team_id, EloRating.INITIAL_RATING) away_elo = elo_ratings.get(match.away_team_id, EloRating.INITIAL_RATING) # Calculate H2H h2h = self.calculate_h2h(match.home_team_id, match.away_team_id, match.match_date) # Create feature record feature = Feature( match_id=match.id, calculated_at=datetime.utcnow(), home_goals_for_avg=home_form["goals_for_avg"], home_goals_against_avg=home_form["goals_against_avg"], home_points_avg=home_form["points_avg"], home_form_last_n=home_form["form_last_n"], home_elo=home_elo, home_days_rest=home_form["days_rest"], away_goals_for_avg=away_form["goals_for_avg"], away_goals_against_avg=away_form["goals_against_avg"], away_points_avg=away_form["points_avg"], away_form_last_n=away_form["form_last_n"], away_elo=away_elo, away_days_rest=away_form["days_rest"], h2h_home_wins=h2h["h2h_home_wins"], h2h_draws=h2h["h2h_draws"], h2h_away_wins=h2h["h2h_away_wins"] ) return feature def build_feature_vector(self, feature: Feature) -> List[float]: """Convert feature to vector for ML model""" return [ feature.home_goals_for_avg or 0.0, feature.home_goals_against_avg or 0.0, feature.home_points_avg or 0.0, feature.home_form_last_n or 0.0, feature.home_elo or 1500.0, feature.home_days_rest or 7.0, feature.away_goals_for_avg or 0.0, feature.away_goals_against_avg or 0.0, feature.away_points_avg or 0.0, feature.away_form_last_n or 0.0, feature.away_elo or 1500.0, feature.away_days_rest or 7.0, float(feature.h2h_home_wins or 0), float(feature.h2h_draws or 0), float(feature.h2h_away_wins or 0), ]