Spaces:
Sleeping
Sleeping
File size: 6,067 Bytes
a3d10df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | import pandas as pd
import numpy as np
class FeatureEngineer:
def __init__(self, rankings_df, form_df, teams_df, manual_features_df=None):
self.rankings = rankings_df
self.form = form_df
self.teams = teams_df
self.manual_features = manual_features_df if manual_features_df is not None else pd.DataFrame()
self.team_momentum = self._calculate_momentum()
def _calculate_momentum(self):
"""
Calculate a form momentum score for each team based on their last 10 matches.
A win against a strong opponent (high Elo) gives more momentum than against a weak one.
"""
momentum = {}
for team_id, group in self.form.groupby('team_id'):
# Take up to 10 most recent matches
recent = group.head(10).copy()
if len(recent) == 0:
momentum[team_id] = 0
continue
# Points: 3 for win, 1 for draw, 0 for loss
recent['points'] = np.where(recent['goals_for'] > recent['goals_against'], 3,
np.where(recent['goals_for'] == recent['goals_against'], 1, 0))
# Weight points by opponent Elo (normalized roughly around 1500)
# E.g., beating an 1800 Elo team gives weight 1.2, a 1200 Elo team gives 0.8
recent['weight'] = recent['opponent_elo'] / 1500.0
# Weighted average points
score = (recent['points'] * recent['weight']).sum() / recent['weight'].sum()
momentum[team_id] = score
return pd.Series(momentum, name='momentum')
def engineer_match_features(self, match_id, home_team_id, away_team_id):
"""
Given match_id and home/away team IDs, return a feature dictionary.
"""
# Elo rating
if home_team_id not in self.rankings.index:
raise ValueError(f"Missing ranking for home team: {home_team_id}")
if away_team_id not in self.rankings.index:
raise ValueError(f"Missing ranking for away team: {away_team_id}")
home_rating = self.rankings.loc[home_team_id, 'rating']
away_rating = self.rankings.loc[away_team_id, 'rating']
if pd.isna(home_rating) or pd.isna(away_rating):
raise ValueError(f"NaN rating for {home_team_id} or {away_team_id}")
# Momentum
home_mom = self.team_momentum.get(home_team_id, 1.0)
away_mom = self.team_momentum.get(away_team_id, 1.0)
# Host advantage
home_host = 1 if (home_team_id in self.teams.index and self.teams.loc[home_team_id, 'is_host']) else 0
away_host = 1 if (away_team_id in self.teams.index and self.teams.loc[away_team_id, 'is_host']) else 0
# In World Cup, if home team is host, they have home advantage.
# Often neither is host, meaning neutral venue.
home_advantage = 1 if home_host else 0
# Manual Features Extraction
injury_impact_home = 0.0
injury_impact_away = 0.0
lineup_strength_home = 1.0
lineup_strength_away = 1.0
odds_implied_home_prob = 0.0
odds_implied_away_prob = 0.0
if match_id in self.manual_features.index:
m_feat = self.manual_features.loc[match_id]
injury_impact_home = m_feat.get('injury_impact_home') or 0.0
injury_impact_away = m_feat.get('injury_impact_away') or 0.0
lineup_strength_home = m_feat.get('lineup_strength_home') or 1.0
lineup_strength_away = m_feat.get('lineup_strength_away') or 1.0
odds_h = m_feat.get('odds_1x2_home')
odds_d = m_feat.get('odds_1x2_draw')
odds_a = m_feat.get('odds_1x2_away')
# Simple implied probability from odds (1/odds)
if pd.notna(odds_h) and pd.notna(odds_d) and pd.notna(odds_a):
margin = (1/odds_h) + (1/odds_d) + (1/odds_a)
odds_implied_home_prob = (1/odds_h) / margin
odds_implied_away_prob = (1/odds_a) / margin
return {
'elo_diff': home_rating - away_rating,
'momentum_diff': home_mom - away_mom,
'home_advantage': home_advantage,
'injury_impact_home': float(injury_impact_home),
'injury_impact_away': float(injury_impact_away),
'lineup_strength_home': float(lineup_strength_home),
'lineup_strength_away': float(lineup_strength_away),
'odds_implied_home_prob': float(odds_implied_home_prob),
'odds_implied_away_prob': float(odds_implied_away_prob)
}
def build_dataset(self, matches_df):
"""
Build a training dataset from a matches dataframe.
"""
features = []
labels = []
for _, row in matches_df.iterrows():
if pd.isna(row['home_team_id']) or pd.isna(row['away_team_id']):
continue
try:
# Extract features
f = self.engineer_match_features(row['id'], row['home_team_id'], row['away_team_id'])
# Track the original index to match back easily
f['match_id'] = row['id']
features.append(f)
# Determine label (0: Away Win, 1: Draw, 2: Home Win)
if not pd.isna(row['home_score_90']) and not pd.isna(row['away_score_90']):
if row['home_score_90'] > row['away_score_90']:
labels.append(2)
elif row['home_score_90'] == row['away_score_90']:
labels.append(1)
else:
labels.append(0)
else:
labels.append(None) # Unplayed
except ValueError as e:
print(f"Skipping match {row['id']}: {e}")
return pd.DataFrame(features), pd.Series(labels)
|