Startup model check: 2025-08-03 13:02:33

src/predict/config.py

@@ -0,0 +1,40 @@
+"""Configuration module for UFC prediction models."""
+
+# Model settings
+DEFAULT_ELO = 1500
+N_FIGHTS_HISTORY = 5
+DEFAULT_ROUNDS_DURATION = 5 * 60  # 5 minutes per round
+
+# Date formats
+DATE_FORMAT_EVENT = '%B %d, %Y'
+DATE_FORMAT_DOB = '%b %d, %Y'
+
+# Feature settings
+FEATURE_COLUMNS = [
+    'height_cm', 
+    'reach_in', 
+    'elo', 
+    'stance', 
+    'dob'
+]
+
+# Model hyperparameters
+MODEL_DEFAULTS = {
+    'LogisticRegression': {},
+    'XGBClassifier': {
+        'use_label_encoder': False,
+        'eval_metric': 'logloss',
+        'random_state': 42
+    },
+    'SVC': {
+        'probability': True,
+        'random_state': 42
+    },
+    'RandomForestClassifier': {
+        'random_state': 42
+    },
+    'BernoulliNB': {},
+    'LGBMClassifier': {
+        'random_state': 42
+    }
+}

src/predict/main.py

@@ -68,18 +68,7 @@ def main():
     elif use_existing_models:
         print("Using existing models if available and no new data detected.")
 
-    # --- Define Models to Run ---
-    # Instantiate all the models you want to evaluate here.
-    models_to_run = [
-        EloBaselineModel(),
-        LogisticRegressionModel(),
-        XGBoostModel(),
-        SVCModel(),
-        RandomForestModel(),
-        BernoulliNBModel(),
-        LGBMModel(),
-    ]
-    # --- End of Model Definition ---
+    # Use the already defined MODELS_TO_RUN from the top of the file
 
     pipeline = PredictionPipeline(
         models=MODELS_TO_RUN, 

src/predict/models.py

@@ -2,6 +2,7 @@ from abc import ABC, abstractmethod
 import sys
 import os
 import pandas as pd
+from typing import Dict, Any, Optional
 from sklearn.linear_model import LogisticRegression
 from sklearn.svm import SVC
 from sklearn.naive_bayes import BernoulliNB
@@ -10,7 +11,8 @@ from xgboost import XGBClassifier
 from lightgbm import LGBMClassifier
 from ..analysis.elo import process_fights_for_elo, INITIAL_ELO
 from ..config import FIGHTERS_CSV_PATH
-from .preprocess import preprocess_for_ml, _get_fighter_history_stats, _calculate_age
+from .preprocess import preprocess_for_ml, _get_fighter_history_stats
+from .utils import calculate_age, prepare_fighters_data, DEFAULT_ELO
 
 class BaseModel(ABC):
     """
@@ -53,7 +55,7 @@ class EloBaselineModel(BaseModel):
         self.fighters_df['full_name'] = self.fighters_df['first_name'] + ' ' + self.fighters_df['last_name']
         self.fighters_df = self.fighters_df.drop_duplicates(subset=['full_name']).set_index('full_name')
 
-    def predict(self, fight):
+    def predict(self, fight: Dict[str, Any]) -> Dict[str, Optional[float]]:
         """Predicts the winner based on ELO and calculates win probability."""
         f1_name, f2_name = fight['fighter_1'], fight['fighter_2']
         
@@ -85,7 +87,7 @@ class BaseMLModel(BaseModel):
         self.fighters_df = None
         self.fighter_histories = {}
 
-    def train(self, train_fights):
+    def train(self, train_fights: List[Dict[str, Any]]) -> None:
         """
         Trains the machine learning model. This involves loading fighter data,
         pre-calculating histories, and fitting the model on the preprocessed data.
@@ -93,12 +95,7 @@ class BaseMLModel(BaseModel):
         print(f"--- Training {self.model.__class__.__name__} ---")
         
         # 1. Prepare data for prediction-time feature generation
-        self.fighters_df = pd.read_csv(FIGHTERS_CSV_PATH)
-        self.fighters_df['full_name'] = self.fighters_df['first_name'] + ' ' + self.fighters_df['last_name']
-        self.fighters_df = self.fighters_df.drop_duplicates(subset=['full_name']).set_index('full_name')
-        for col in ['height_cm', 'reach_in', 'elo']:
-            if col in self.fighters_df.columns:
-                self.fighters_df[col] = pd.to_numeric(self.fighters_df[col], errors='coerce')
+        self.fighters_df = prepare_fighters_data(pd.read_csv(FIGHTERS_CSV_PATH))
 
         # 2. Pre-calculate fighter histories
         train_fights_with_dates = []
@@ -136,8 +133,8 @@ class BaseMLModel(BaseModel):
         f1_hist_stats = _get_fighter_history_stats(f1_name, fight_date, f1_hist, self.fighters_df)
         f2_hist_stats = _get_fighter_history_stats(f2_name, fight_date, f2_hist, self.fighters_df)
         
-        f1_age = _calculate_age(f1_stats.get('dob'), fight['event_date'])
-        f2_age = _calculate_age(f2_stats.get('dob'), fight['event_date'])
+        f1_age = calculate_age(f1_stats.get('dob'), fight['event_date'])
+        f2_age = calculate_age(f2_stats.get('dob'), fight['event_date'])
 
         features = {
             'elo_diff': f1_stats.get('elo', 1500) - f2_stats.get('elo', 1500),
@@ -194,4 +191,4 @@ class BernoulliNBModel(BaseMLModel):
 class LGBMModel(BaseMLModel):
     """A thin wrapper for LightGBM's LGBMClassifier."""
     def __init__(self):
-        super().__init__(model=LGBMClassifier(random_state=42)) 
+        super().__init__(model=LGBMClassifier(random_state=42))

src/predict/pipeline.py

@@ -149,16 +149,13 @@ class PredictionPipeline:
         print("No new data detected and all model files exist. Using existing models.")
         return False
 
-    def _load_and_split_data(self, num_test_events=1):
+    def _load_and_split_data(self, num_test_events: int = 1) -> None:
         """Loads and splits the data into chronological training and testing sets."""
         print("\n--- Loading and Splitting Data ---")
         if not os.path.exists(FIGHTS_CSV_PATH):
             raise FileNotFoundError(f"Fights data not found at '{FIGHTS_CSV_PATH}'.")
 
-        with open(FIGHTS_CSV_PATH, 'r', encoding='utf-8') as f:
-            fights = list(csv.DictReader(f))
-        
-        fights.sort(key=lambda x: datetime.strptime(x['event_date'], '%B %d, %Y'))
+        fights = self._load_fights()
         
         all_events = list(OrderedDict.fromkeys(f['event_name'] for f in fights))
         if len(all_events) < num_test_events:
@@ -171,7 +168,15 @@ class PredictionPipeline:
         print(f"Data loaded. {len(self.train_fights)} training fights, {len(self.test_fights)} testing fights.")
         print(f"Testing on the last {num_test_events} event(s): {', '.join(test_event_names)}")
 
-    def run(self, detailed_report=True):
+    def _load_fights(self) -> list:
+        """Helper method to load and sort fights from CSV."""
+        with open(FIGHTS_CSV_PATH, 'r', encoding='utf-8') as f:
+            fights = list(csv.DictReader(f))
+        
+        fights.sort(key=lambda x: datetime.strptime(x['event_date'], '%B %d, %Y'))
+        return fights
+
+    def run(self, detailed_report: bool = True) -> None:
         """Executes the full pipeline: load, train, evaluate, report and save models."""
         self._load_and_split_data()
         
@@ -349,4 +354,4 @@ class PredictionPipeline:
         # A summary is printed to the console for convenience.
         self._report_summary()
         # The detailed report is now saved to a JSON file.
-        self._save_report_to_json() 
+        self._save_report_to_json()

src/predict/preprocess.py

@@ -1,53 +1,22 @@
 import pandas as pd
 import os
 from datetime import datetime
+from typing import Dict, List, Tuple, Any, Optional
 from ..config import FIGHTERS_CSV_PATH
+from .utils import (
+    parse_round_time_to_seconds, parse_striking_stats, to_int_safe, 
+    calculate_age, prepare_fighters_data, DEFAULT_ELO, N_FIGHTS_HISTORY
+)
 
-def _clean_numeric_column(series):
-    """A helper to clean string columns into numbers, handling errors."""
-    series_str = series.astype(str)
-    return pd.to_numeric(series_str.str.replace(r'[^0-9.]', '', regex=True), errors='coerce')
-
-def _calculate_age(dob_str, fight_date_str):
-    """Calculates age in years from a date of birth string and fight date string."""
-    if pd.isna(dob_str) or not dob_str:
-        return None
-    try:
-        dob = datetime.strptime(dob_str, '%b %d, %Y')
-        fight_date = datetime.strptime(fight_date_str, '%B %d, %Y')
-        return (fight_date - dob).days / 365.25
-    except (ValueError, TypeError):
-        return None
-
-def _parse_round_time_to_seconds(round_str, time_str):
-    """Converts fight duration from round and time to total seconds."""
-    try:
-        rounds = int(round_str)
-        minutes, seconds = map(int, time_str.split(':'))
-        # Assuming 5-minute rounds for calculation simplicity
-        return ((rounds - 1) * 5 * 60) + (minutes * 60) + seconds
-    except (ValueError, TypeError, AttributeError):
-        return 0
-
-def _parse_striking_stats(stat_str):
-    """Parses striking stats string like '10 of 20' into (landed, attempted)."""
-    try:
-        landed, attempted = map(int, stat_str.split(' of '))
-        return landed, attempted
-    except (ValueError, TypeError, AttributeError):
-        return 0, 0
-
-def _to_int_safe(val):
-    """Safely converts a value to an integer, returning 0 if it's invalid or empty."""
-    if pd.isna(val):
-        return 0
-    try:
-        # handle strings with whitespace or empty strings
-        return int(str(val).strip() or 0)
-    except (ValueError, TypeError):
-        return 0
-
-def _get_fighter_history_stats(fighter_name, current_fight_date, fighter_history, fighters_df, n=5):
+
+
+def _get_fighter_history_stats(
+    fighter_name: str, 
+    current_fight_date: datetime, 
+    fighter_history: List[Dict[str, Any]], 
+    fighters_df: pd.DataFrame, 
+    n: int = N_FIGHTS_HISTORY
+) -> Dict[str, float]:
     """
     Calculates performance statistics for a fighter based on their last n fights.
     """
@@ -58,7 +27,7 @@ def _get_fighter_history_stats(fighter_name, current_fight_date, fighter_history
         # Return a default dictionary with the correct keys for a fighter with no history
         return {
             'wins_last_n': 0,
-            'avg_opp_elo_last_n': 1500, # Assume average ELO for first opponent
+            'avg_opp_elo_last_n': DEFAULT_ELO,
             'ko_percent_last_n': 0,
             'sig_str_landed_per_min_last_n': 0,
             'takedown_accuracy_last_n': 0,
@@ -84,20 +53,20 @@ def _get_fighter_history_stats(fighter_name, current_fight_date, fighter_history
 
         if opponent_name in fighters_df.index:
             opp_elo = fighters_df.loc[opponent_name, 'elo']
-            stats['opponent_elos'].append(opp_elo if pd.notna(opp_elo) else 1500)
+            stats['opponent_elos'].append(opp_elo if pd.notna(opp_elo) else DEFAULT_ELO)
         
-        stats['total_time_secs'] += _parse_round_time_to_seconds(fight['round'], fight['time'])
+        stats['total_time_secs'] += parse_round_time_to_seconds(fight['round'], fight['time'])
         
         sig_str_stat = fight.get(f'{f_prefix}_sig_str', '0 of 0')
-        landed, _ = _parse_striking_stats(sig_str_stat)
+        landed, _ = parse_striking_stats(sig_str_stat)
         stats['sig_str_landed'] += landed
 
         td_stat = fight.get(f'{f_prefix}_td', '0 of 0')
-        td_landed, td_attempted = _parse_striking_stats(td_stat) # Can reuse this parser
+        td_landed, td_attempted = parse_striking_stats(td_stat)
         stats['td_landed'] += td_landed
         stats['td_attempted'] += td_attempted
         
-        stats['sub_attempts'] += _to_int_safe(fight.get(f'{f_prefix}_sub_att'))
+        stats['sub_attempts'] += to_int_safe(fight.get(f'{f_prefix}_sub_att'))
 
     # Final calculations
     avg_opp_elo = sum(stats['opponent_elos']) / len(stats['opponent_elos']) if stats['opponent_elos'] else 1500
@@ -112,36 +81,26 @@ def _get_fighter_history_stats(fighter_name, current_fight_date, fighter_history
         'sub_attempts_per_min_last_n': (stats['sub_attempts'] / total_minutes) if total_minutes > 0 else 0,
     }
 
-def preprocess_for_ml(fights_to_process, fighters_csv_path):
+def preprocess_for_ml(
+    fights_to_process: List[Dict[str, Any]], 
+    fighters_csv_path: str
+) -> Tuple[pd.DataFrame, pd.Series, pd.DataFrame]:
     """
     Transforms raw fight and fighter data into a feature matrix (X) and target vector (y)
     suitable for a binary classification machine learning model.
 
     Args:
-        fights_to_process (list of dict): The list of fights to process.
-        fighters_csv_path (str): Path to the CSV file with all fighter stats.
+        fights_to_process: The list of fights to process.
+        fighters_csv_path: Path to the CSV file with all fighter stats.
 
     Returns:
-        pd.DataFrame: Feature matrix X.
-        pd.Series: Target vector y.
-        pd.DataFrame: Metadata DataFrame.
+        Feature matrix X, target vector y, and metadata DataFrame.
     """
     if not os.path.exists(fighters_csv_path):
         raise FileNotFoundError(f"Fighters data not found at '{fighters_csv_path}'.")
 
     fighters_df = pd.read_csv(fighters_csv_path)
-    
-    # 1. Prepare fighters data for merging
-    fighters_prepared = fighters_df.copy()
-    fighters_prepared['full_name'] = fighters_prepared['first_name'] + ' ' + fighters_prepared['last_name']
-    
-    # Handle duplicate fighter names by keeping the first entry
-    fighters_prepared = fighters_prepared.drop_duplicates(subset=['full_name'], keep='first')
-    fighters_prepared = fighters_prepared.set_index('full_name')
-
-    for col in ['height_cm', 'reach_in', 'elo']:
-        if col in fighters_prepared.columns:
-            fighters_prepared[col] = _clean_numeric_column(fighters_prepared[col])
+    fighters_prepared = prepare_fighters_data(fighters_df)
 
     # 2. Pre-calculate fighter histories to speed up lookups
     # And convert date strings to datetime objects once

src/predict/utils.py

@@ -0,0 +1,70 @@
+import pandas as pd
+import os
+from datetime import datetime
+from typing import Optional, Dict, Any
+
+# Constants
+DEFAULT_ELO = 1500
+DEFAULT_AGE = 0
+DEFAULT_FIGHT_TIME = 0
+DEFAULT_ROUNDS_DURATION = 5 * 60  # 5 minutes per round
+N_FIGHTS_HISTORY = 5
+
+def clean_numeric_column(series: pd.Series) -> pd.Series:
+    """A helper to clean string columns into numbers, handling errors."""
+    series_str = series.astype(str)
+    return pd.to_numeric(series_str.str.replace(r'[^0-9.]', '', regex=True), errors='coerce')
+
+def calculate_age(dob_str: str, fight_date_str: str) -> Optional[float]:
+    """Calculates age in years from a date of birth string and fight date string."""
+    if pd.isna(dob_str) or not dob_str:
+        return None
+    try:
+        dob = datetime.strptime(dob_str, '%b %d, %Y')
+        fight_date = datetime.strptime(fight_date_str, '%B %d, %Y')
+        return (fight_date - dob).days / 365.25
+    except (ValueError, TypeError):
+        return None
+
+def parse_round_time_to_seconds(round_str: str, time_str: str) -> int:
+    """Converts fight duration from round and time to total seconds."""
+    try:
+        rounds = int(round_str)
+        minutes, seconds = map(int, time_str.split(':'))
+        # Assuming 5-minute rounds for calculation simplicity
+        return ((rounds - 1) * DEFAULT_ROUNDS_DURATION) + (minutes * 60) + seconds
+    except (ValueError, TypeError, AttributeError):
+        return 0
+
+def parse_striking_stats(stat_str: str) -> tuple[int, int]:
+    """Parses striking stats string like '10 of 20' into (landed, attempted)."""
+    try:
+        landed, attempted = map(int, stat_str.split(' of '))
+        return landed, attempted
+    except (ValueError, TypeError, AttributeError):
+        return 0, 0
+
+def to_int_safe(val: Any) -> int:
+    """Safely converts a value to an integer, returning 0 if it's invalid or empty."""
+    if pd.isna(val):
+        return 0
+    try:
+        # handle strings with whitespace or empty strings
+        return int(str(val).strip() or 0)
+    except (ValueError, TypeError):
+        return 0
+
+def prepare_fighters_data(fighters_df: pd.DataFrame) -> pd.DataFrame:
+    """Prepares fighter data for analysis by cleaning and standardizing."""
+    fighters_prepared = fighters_df.copy()
+    fighters_prepared['full_name'] = fighters_prepared['first_name'] + ' ' + fighters_prepared['last_name']
+    
+    # Handle duplicate fighter names by keeping the first entry
+    fighters_prepared = fighters_prepared.drop_duplicates(subset=['full_name'], keep='first')
+    fighters_prepared = fighters_prepared.set_index('full_name')
+
+    for col in ['height_cm', 'reach_in', 'elo']:
+        if col in fighters_prepared.columns:
+            fighters_prepared[col] = clean_numeric_column(fighters_prepared[col])
+    
+    return fighters_prepared