""" Feature Engineering Pipeline Loads scraped CSVs, builds comprehensive features for UFC fight prediction. Outputs: data/training_data.csv, data/feature_names.json """ import os import re import json import warnings from pathlib import Path import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler, LabelEncoder import joblib warnings.filterwarnings("ignore") DATA_DIR = Path(__file__).parent.parent / "data" MODELS_DIR = Path(__file__).parent.parent / "models" FIGHTS_CSV = DATA_DIR / "ufc_fight_stats.csv" PROFILES_CSV = DATA_DIR / "fighter_profiles.csv" EXPERT_PICKS_CSV = DATA_DIR / "expert_picks.csv" EXPERT_HISTORY_CSV = DATA_DIR / "expert_history.csv" SENTIMENT_CSV = DATA_DIR / "fighter_news_sentiment.csv" TRAINING_CSV = DATA_DIR / "training_data.csv" SCALER_PATH = MODELS_DIR / "scaler.pkl" FEATURE_NAMES_PATH = MODELS_DIR / "feature_names.pkl" def parse_height(height_str): """Convert height string like 5' 11\" to inches.""" if pd.isna(height_str) or not height_str: return np.nan try: match = re.match(r"(\d+)[\'′].*?(\d+)[\"″]?", str(height_str)) if match: feet = int(match.group(1)) inches = int(match.group(2)) return feet * 12 + inches except Exception: pass return np.nan def parse_reach(reach_str): """Convert reach string to float inches.""" if pd.isna(reach_str) or not reach_str: return np.nan try: return float(str(reach_str).replace('"', "").replace("′", "").strip()) except Exception: return np.nan def parse_percent(pct_str): """Convert percentage string to float.""" if pd.isna(pct_str) or not pct_str: return np.nan try: return float(str(pct_str).replace("%", "").strip()) / 100 except Exception: return np.nan def parse_float(val): """Safely parse a value to float.""" if pd.isna(val) or val == "" or val == "--": return np.nan try: return float(str(val).replace(",", "").strip()) except Exception: return np.nan def load_and_clean_fight_stats(): """Load and clean UFC fight stats CSV.""" print("[1/5] Loading fight statistics...") if not FIGHTS_CSV.exists(): print(f" WARNING: {FIGHTS_CSV} not found. Generating synthetic data for demonstration.") return generate_synthetic_fight_data() df = pd.read_csv(FIGHTS_CSV, low_memory=False) print(f" Loaded {len(df)} fight records") numeric_cols = ["a_sig_str", "a_sig_str_pct", "a_total_str", "a_td", "a_td_pct", "a_sub_att", "a_rev", "a_ctrl", "a_head", "a_body", "a_leg", "a_distance", "a_clinch", "a_ground", "b_sig_str", "b_sig_str_pct", "b_total_str", "b_td", "b_td_pct", "b_sub_att", "b_rev", "b_ctrl", "b_head", "b_body", "b_leg", "b_distance", "b_clinch", "b_ground", "a_sig_str_landed", "a_sig_str_attempted", "a_td_landed", "a_td_attempted", "b_sig_str_landed", "b_sig_str_attempted", "b_td_landed", "b_td_attempted"] for col in numeric_cols: if col in df.columns: df[col] = df[col].apply(parse_float) return df def load_profiles(): """Load fighter profiles.""" print("[2/5] Loading fighter profiles...") if not PROFILES_CSV.exists(): print(f" WARNING: {PROFILES_CSV} not found. Using default profile data.") return pd.DataFrame() df = pd.read_csv(PROFILES_CSV) print(f" Loaded {len(df)} fighter profiles") if "height" in df.columns: df["height_inches"] = df["height"].apply(parse_height) if "reach" in df.columns: df["reach_inches"] = df["reach"].apply(parse_reach) if "sig_strike_accuracy" in df.columns: df["strike_acc"] = df["sig_strike_accuracy"].apply(parse_percent) if "sig_strike_defense" in df.columns: df["strike_def"] = df["sig_strike_defense"].apply(parse_percent) if "takedown_accuracy" in df.columns: df["td_acc"] = df["takedown_accuracy"].apply(parse_percent) if "takedown_defense" in df.columns: df["td_def"] = df["takedown_defense"].apply(parse_percent) if "sig_strikes_landed_per_min" in df.columns: df["slpm"] = df["sig_strikes_landed_per_min"].apply(parse_float) if "sig_strikes_absorbed_per_min" in df.columns: df["sapm"] = df["sig_strikes_absorbed_per_min"].apply(parse_float) if "takedown_avg" in df.columns: df["td_avg"] = df["takedown_avg"].apply(parse_float) if "submission_avg" in df.columns: df["sub_avg"] = df["submission_avg"].apply(parse_float) if "wins" in df.columns: df["career_wins"] = df["wins"].apply(parse_float) else: df["career_wins"] = 15.0 if "losses" in df.columns: df["career_losses"] = df["losses"].apply(parse_float) else: df["career_losses"] = 5.0 if "full_name" not in df.columns: df["full_name"] = (df.get("first_name", "").fillna("") + " " + df.get("last_name", "").fillna("")).str.strip() df["win_rate"] = np.where( (df["career_wins"].notna()) & (df["career_losses"].notna()), df["career_wins"] / (df["career_wins"] + df["career_losses"] + 0.001), np.nan, ) weight_map = {"Strawweight": 1, "Flyweight": 2, "Bantamweight": 3, "Featherweight": 4, "Lightweight": 5, "Welterweight": 6, "Middleweight": 7, "Light Heavyweight": 8, "Heavyweight": 9} if "weight" in df.columns: df["weight_class"] = df["weight"].map(weight_map).fillna(5.0) return df def build_fight_level_features(fight_df): """Aggregate round-by-round data into fight-level features.""" print(" Building fight-level features...") fight_df = fight_df.copy() if "winner" not in fight_df.columns: fight_df["winner"] = "A" fight_features = [] for (fighter_a, fighter_b), group in fight_df.groupby(["fighter_a", "fighter_b"], sort=False): group = group.sort_values("round") rounds = len(group) features = { "fighter_a": fighter_a, "fighter_b": fighter_b, "winner": group["winner"].iloc[0], "num_rounds": rounds, "method": group["method_type"].iloc[0] if "method_type" in group.columns else "", } strike_cols_a = [c for c in group.columns if c.startswith("a_sig_str") and "pct" not in c] strike_cols_b = [c for c in group.columns if c.startswith("b_sig_str") and "pct" not in c] for col_a, col_b in zip( ["a_sig_str", "a_total_str", "a_td", "a_sub_att", "a_ctrl", "a_head", "a_body", "a_leg", "a_distance", "a_clinch", "a_ground"], ["b_sig_str", "b_total_str", "b_td", "b_sub_att", "b_ctrl", "b_head", "b_body", "b_leg", "b_distance", "b_clinch", "b_ground"], ): name = col_a.replace("a_", "") if col_a in group.columns: features[f"{name}_a_total"] = group[col_a].sum() if col_b in group.columns: features[f"{name}_b_total"] = group[col_b].sum() for col_a, col_b in [("a_sig_str_pct", "b_sig_str_pct"), ("a_td_pct", "b_td_pct")]: name = col_a.replace("a_", "").replace("_pct", "") if col_a in group.columns: features[f"{name}_a_avg_pct"] = group[col_a].mean() if col_b in group.columns: features[f"{name}_b_avg_pct"] = group[col_b].mean() if rounds >= 2: for col_a, col_b in [("a_sig_str", "b_sig_str"), ("a_td", "b_td")]: name = col_a.replace("a_", "") if col_a in group.columns: r1_val = group[col_a].iloc[0] if len(group) > 0 else 0 r_last_val = group[col_a].iloc[-1] if len(group) > 0 else 0 features[f"{name}_a_rd1"] = r1_val features[f"{name}_a_rd_last"] = r_last_val features[f"{name}_a_dropoff"] = r1_val - r_last_val if r1_val > 0 else 0 if col_b in group.columns: r1_val = group[col_b].iloc[0] if len(group) > 0 else 0 r_last_val = group[col_b].iloc[-1] if len(group) > 0 else 0 features[f"{name}_b_rd1"] = r1_val features[f"{name}_b_rd_last"] = r_last_val features[f"{name}_b_dropoff"] = r1_val - r_last_val if r1_val > 0 else 0 if "a_sig_str" in group.columns and "b_sig_str" in group.columns: features["a_sig_str_per_round"] = group["a_sig_str"].sum() / max(rounds, 1) features["b_sig_str_per_round"] = group["b_sig_str"].sum() / max(rounds, 1) if "a_str_def_pct" in group.columns and "b_str_def_pct" in group.columns: features["a_striking_def"] = group["a_str_def_pct"].mean() features["b_striking_def"] = group["b_str_def_pct"].mean() fight_features.append(features) return pd.DataFrame(fight_features) def build_style_matchup_features(fight_df, profiles_df): """Create style matchup metrics: striking differential, grappling differential.""" print("[3/5] Building style matchup features...") features = [] unique_fighters = set(fight_df["fighter_a"].unique()) | set(fight_df["fighter_b"].unique()) fighter_stats = {} for fighter in unique_fighters: stats = {} fights_as_a = fight_df[fight_df["fighter_a"] == fighter] fights_as_b = fight_df[fight_df["fighter_b"] == fighter] for prefix, fights in [("a_", fights_as_a), ("b_", fights_as_b)]: for metric in ["sig_str", "total_str", "td", "sub_att", "ctrl", "head", "body", "leg"]: col = f"{prefix}{metric}" if col in fight_df.columns and len(fights) > 0: val = fights[col].sum() stats[f"{metric}_career"] = stats.get(f"{metric}_career", 0) + val num_fights = len(fights_as_a) + len(fights_as_b) stats["num_fights"] = num_fights profile_row = profiles_df[profiles_df["full_name"].str.lower() == fighter.lower()] if not profiles_df.empty else pd.DataFrame() if not profile_row.empty: for col in ["height_inches", "reach_inches", "strike_acc", "strike_def", "td_acc", "td_def", "slpm", "sapm", "td_avg", "sub_avg", "win_rate", "weight_class", "career_wins", "career_losses"]: if col in profile_row.columns: stats[f"profile_{col}"] = profile_row[col].values[0] fighter_stats[fighter] = stats for _, fight in fight_df.iterrows(): fa_name = fight["fighter_a"] fb_name = fight["fighter_b"] fa = fighter_stats.get(fa_name, {}) fb = fighter_stats.get(fb_name, {}) row = { "fighter_a": fa_name, "fighter_b": fb_name, "winner": fight.get("winner", "A"), "method": fight.get("method_type", fight.get("method", "")), } for metric in ["sig_str", "total_str", "td", "sub_att", "ctrl", "head", "body", "leg"]: a_val = fa.get(f"{metric}_career", 0) b_val = fb.get(f"{metric}_career", 0) total = a_val + b_val + 0.001 row[f"diff_{metric}"] = a_val - b_val row[f"ratio_{metric}"] = a_val / total for prof_metric in ["height_inches", "reach_inches", "strike_acc", "strike_def", "td_acc", "td_def", "slpm", "sapm", "td_avg", "sub_avg", "win_rate", "weight_class"]: a_val = fa.get(f"profile_{prof_metric}", np.nan) b_val = fb.get(f"profile_{prof_metric}", np.nan) if not pd.isna(a_val) and not pd.isna(b_val): row[f"diff_{prof_metric}"] = a_val - b_val row[f"ratio_{prof_metric}"] = a_val / (a_val + b_val + 0.001) a_fights = fa.get("num_fights", 0) b_fights = fb.get("num_fights", 0) row["a_experience"] = a_fights row["b_experience"] = b_fights row["experience_diff"] = a_fights - b_fights a_wc = fa.get("profile_weight_class", np.nan) b_wc = fb.get("profile_weight_class", np.nan) if not pd.isna(a_wc) and not pd.isna(b_wc): row["same_weight_class"] = 1.0 if a_wc == b_wc else 0.0 else: row["same_weight_class"] = 1.0 features.append(row) return pd.DataFrame(features) def build_sentiment_features(fight_df): """Integrate NLP sentiment scores.""" print("[4/5] Building sentiment features...") if not SENTIMENT_CSV.exists(): print(f" WARNING: {SENTIMENT_CSV} not found. Using neutral sentiment (0.0).") return fight_df sentiment_df = pd.read_csv(SENTIMENT_CSV) sentiment_summary = sentiment_df.groupby("fighter_name").agg( avg_sentiment=("sentiment_score", "mean"), article_count=("sentiment_score", "count"), sentiment_std=("sentiment_score", "std"), max_positive=("sentiment_score", "max"), min_negative=("sentiment_score", "min"), ).reset_index() sentiment_summary["momentum_score"] = ( sentiment_summary["avg_sentiment"].fillna(0) * 0.4 + (sentiment_summary["article_count"] / max(1, int(sentiment_summary["article_count"].max()))) * 0.2 + (1 - sentiment_summary["sentiment_std"].fillna(1).clip(0, 1)) * 0.4 ) result = fight_df.copy() for side in ["a", "b"]: col_name = f"fighter_{side}" result = result.merge( sentiment_summary[["fighter_name", "avg_sentiment", "article_count", "momentum_score"]], left_on=col_name, right_on="fighter_name", how="left", suffixes=("", f"_{side}"), ) result.rename(columns={ "avg_sentiment": f"{side}_sentiment", "article_count": f"{side}_news_articles", "momentum_score": f"{side}_momentum", }, inplace=True) if "fighter_name" in result.columns: result.drop(columns=["fighter_name"], inplace=True) for col in ["a_sentiment", "b_sentiment", "a_momentum", "b_momentum"]: if col in result.columns: result[col] = result[col].fillna(0.0) if "a_momentum" in result.columns and "b_momentum" in result.columns: result["sentiment_diff"] = result["a_momentum"] - result["b_momentum"] return result def build_expert_consensus_features(fight_df): """Build weighted expert consensus features.""" print("[5/5] Building expert consensus features...") result = fight_df.copy() if not EXPERT_PICKS_CSV.exists() or not EXPERT_HISTORY_CSV.exists(): print(f" WARNING: Expert data not found. Using neutral consensus (0.5).") result["expert_consensus_a"] = 0.5 result["expert_consensus_b"] = 0.5 result["consensus_diff"] = 0.0 result["expert_agreement"] = 0.5 return result picks_df = pd.read_csv(EXPERT_PICKS_CSV) history_df = pd.read_csv(EXPERT_HISTORY_CSV) if "expert_name" not in picks_df.columns or "reliability_weight" not in history_df.columns: result["expert_consensus_a"] = 0.5 result["expert_consensus_b"] = 0.5 result["consensus_diff"] = 0.0 result["expert_agreement"] = 0.5 return result if "expert_name" in picks_df.columns and "reliability_weight" in history_df.columns and "expert_name" in history_df.columns: picks_weighted = picks_df.merge( history_df[["expert_name", "reliability_weight"]], on="expert_name", how="left", ) picks_weighted["reliability_weight"] = picks_weighted["reliability_weight"].fillna(0.5) consensus_list = [] for (fa, fb), group in picks_weighted.groupby(["fighter_a", "fighter_b"]): total_weight = group["reliability_weight"].sum() if total_weight > 0: a_picks = group[group["predicted_winner"] == fa]["reliability_weight"].sum() b_picks = group[group["predicted_winner"] == fb]["reliability_weight"].sum() consensus_list.append({ "fighter_a": fa, "fighter_b": fb, "expert_consensus_a": a_picks / total_weight, "expert_consensus_b": b_picks / total_weight, "expert_agreement": max(a_picks, b_picks) / total_weight, }) if consensus_list: consensus_df = pd.DataFrame(consensus_list) result = result.merge(consensus_df, on=["fighter_a", "fighter_b"], how="left") for col in ["expert_consensus_a", "expert_consensus_b", "expert_agreement"]: if col in result.columns: result[col] = result[col].fillna(0.5) else: result[col] = 0.5 result["consensus_diff"] = result["expert_consensus_a"] - result["expert_consensus_b"] return result def generate_synthetic_fight_data(n_fights=500): """Generate realistic synthetic UFC fight data for testing.""" import random random.seed(42) np.random.seed(42) fighters = [ "Jon Jones", "Islam Makhachev", "Alexander Volkanovski", "Israel Adesanya", "Alex Pereira", "Charles Oliveira", "Leon Edwards", "Max Holloway", "Dustin Poirier", "Justin Gaethje", "Sean O'Malley", "Ilia Topuria", "Sean Strickland", "Jiri Prochazka", "Khamzat Chimaev", "Tom Aspinall", "Ciryl Gane", "Brandon Moreno", "Colby Covington", "Robert Whittaker", "Petr Yan", "Cory Sandhagen", "Marlon Vera", "Gilbert Burns", "Belal Muhammad", "Shavkat Rakhmonov", "Arman Tsarukyan", "Beneil Dariush", ] data = [] for i in range(n_fights): fa = random.choice(fighters) fb = random.choice([f for f in fighters if f != fa]) striker_types = ["Jon Jones", "Israel Adesanya", "Max Holloway", "Sean O'Malley", "Israel Adesanya"] grappler_types = ["Islam Makhachev", "Khamzat Chimaev", "Charles Oliveira", "Beneil Dariush"] fa_sig_str = np.random.gamma(25, 1.5) if fa in striker_types else np.random.gamma(15, 1.2) fb_sig_str = np.random.gamma(25, 1.5) if fb in striker_types else np.random.gamma(15, 1.2) fa_td = np.random.gamma(3, 0.8) if fa in grappler_types else np.random.gamma(1, 0.5) fb_td = np.random.gamma(3, 0.8) if fb in grappler_types else np.random.gamma(1, 0.5) fa_ctrl = np.random.gamma(120, 1) if fa in grappler_types else np.random.gamma(30, 1) fb_ctrl = np.random.gamma(120, 1) if fb in grappler_types else np.random.gamma(30, 1) fa_strength = fa_sig_str * 0.4 + fa_td * 1.5 + fa_ctrl * 0.02 + np.random.normal(0, 2) fb_strength = fb_sig_str * 0.4 + fb_td * 1.5 + fb_ctrl * 0.02 + np.random.normal(0, 2) winner = "A" if fa_strength > fb_strength else "B" rounds = np.random.choice([1, 3, 5], p=[0.15, 0.60, 0.25]) method = np.random.choice(["KO/TKO", "Submission", "Decision"], p=[0.35, 0.20, 0.45]) for r in range(1, rounds + 1): fatigue_factor = max(0.7, 1.0 - (r - 1) * 0.1 - np.random.uniform(0, 0.05)) round_data = { "fighter_a": fa, "fighter_b": fb, "winner": winner, "round": r, "method_type": method, "a_sig_str": round(fa_sig_str * fatigue_factor * np.random.uniform(0.8, 1.2)), "b_sig_str": round(fb_sig_str * fatigue_factor * np.random.uniform(0.8, 1.2)), "a_total_str": round(fa_sig_str * fatigue_factor * 1.5 * np.random.uniform(0.8, 1.2)), "b_total_str": round(fb_sig_str * fatigue_factor * 1.5 * np.random.uniform(0.8, 1.2)), "a_td": round(fa_td * fatigue_factor * np.random.uniform(0, 1.5)), "b_td": round(fb_td * fatigue_factor * np.random.uniform(0, 1.5)), "a_sub_att": round(np.random.uniform(0, 1.5) if fa in grappler_types else np.random.uniform(0, 0.5)), "b_sub_att": round(np.random.uniform(0, 1.5) if fb in grappler_types else np.random.uniform(0, 0.5)), "a_ctrl": round(fa_ctrl * fatigue_factor * np.random.uniform(0.5, 1.5)), "b_ctrl": round(fb_ctrl * fatigue_factor * np.random.uniform(0.5, 1.5)), "a_head": round(fa_sig_str * fatigue_factor * 0.7), "b_head": round(fb_sig_str * fatigue_factor * 0.7), "a_body": round(fa_sig_str * fatigue_factor * 0.2), "b_body": round(fb_sig_str * fatigue_factor * 0.2), "a_leg": round(fa_sig_str * fatigue_factor * 0.1), "b_leg": round(fb_sig_str * fatigue_factor * 0.1), "a_sig_str_pct": round(np.random.uniform(0.35, 0.65), 2), "b_sig_str_pct": round(np.random.uniform(0.35, 0.65), 2), "a_td_pct": round(np.random.uniform(0.2, 0.6), 2), "b_td_pct": round(np.random.uniform(0.2, 0.6), 2), } data.append(round_data) df = pd.DataFrame(data) print(f" Generated {len(df)} synthetic fight records") return df def encode_and_export(features_df, scaler=None): """Encode categoricals, normalize numeric features, and export.""" print("\nEncoding features and preparing for training...") df = features_df.copy() if "winner" in df.columns: df["target"] = (df["winner"] == "A").astype(int) if "method" in df.columns: method_dummies = pd.get_dummies(df["method"], prefix="method") df = pd.concat([df, method_dummies], axis=1) drop_cols = ["fighter_a", "fighter_b", "winner", "method", "target"] feature_cols = [c for c in df.columns if c not in drop_cols] feats = df[feature_cols].copy() for col in feats.columns: if feats[col].dtype == "object": feats[col] = pd.to_numeric(feats[col], errors="coerce") feats = feats.fillna(feats.median(numeric_only=True)) feats = feats.replace([np.inf, -np.inf], np.nan) feats = feats.fillna(0.0) if scaler is None: scaler = StandardScaler() numeric_cols = feats.select_dtypes(include=[np.number]).columns if not hasattr(scaler, "mean_"): scaled_array = scaler.fit_transform(feats[numeric_cols]) else: scaled_array = scaler.transform(feats[numeric_cols]) scaled_df = pd.DataFrame(scaled_array, columns=numeric_cols, index=feats.index) scaled_df["target"] = df["target"].values os.makedirs(MODELS_DIR, exist_ok=True) joblib.dump(scaler, SCALER_PATH) joblib.dump(list(numeric_cols), FEATURE_NAMES_PATH) scaled_df.to_csv(TRAINING_CSV, index=False) print(f" Saved {len(scaled_df)} training samples to {TRAINING_CSV}") print(f" Features: {len(numeric_cols)}") print(f" Targets: A wins = {scaled_df['target'].sum():.0f}, B wins = {(1 - scaled_df['target']).sum():.0f}") return scaled_df, scaler, list(numeric_cols) def main(): os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(MODELS_DIR, exist_ok=True) fight_df = load_and_clean_fight_stats() profiles_df = load_profiles() styled_df = build_style_matchup_features(fight_df, profiles_df) if "fighter_a" in fight_df.columns and "fighter_b" in fight_df.columns: styled_df["method"] = fight_df["method_type"].values if "method_type" in fight_df.columns else "" sentiment_df = build_sentiment_features(styled_df) consensus_df = build_expert_consensus_features(sentiment_df) training_df, scaler, feature_names = encode_and_export(consensus_df) print(f"\nFeature engineering complete!") print(f" Training data: {TRAINING_CSV}") print(f" Scaler: {SCALER_PATH}") print(f" Next step: python scripts/model_training.py") return training_df if __name__ == "__main__": main()