Chittrarasu/Cricket-match-prediction-FastAPI
0
1import pandas as pd2import numpy as np3from fastapi import HTTPException4from models.train_model import (5 load_and_preprocess_data, train_team_performance_model, train_player_score_model,6 predict_player_score, predict_team_performance7)8from groq import Groq9 10# Global variables to store models and data11TEAM_WIN_MODEL = None12TEAM_SCORE_MODEL = None13TEAM_DATA = None14TEAM_SCALER = None15PLAYER_SCORE_MODEL = None16PLAYER_SCALER = None17PLAYER_DATA = None18MATCH_DF = None19BALL_DF = None20 21# Initialize Groq client22GROQ_API_KEY = "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"23client = Groq(api_key=GROQ_API_KEY)24 25# Load data and train models at startup26def initialize_models():27 global TEAM_WIN_MODEL, TEAM_SCORE_MODEL, TEAM_DATA, TEAM_SCALER28 global PLAYER_SCORE_MODEL, PLAYER_SCALER, PLAYER_DATA, MATCH_DF, BALL_DF29 30 MATCH_DF, BALL_DF = load_and_preprocess_data()31 TEAM_WIN_MODEL, TEAM_SCORE_MODEL, TEAM_DATA, TEAM_SCALER = train_team_performance_model(MATCH_DF)32 PLAYER_SCORE_MODEL, PLAYER_SCALER, PLAYER_DATA = train_player_score_model(MATCH_DF, BALL_DF)33 print("Models trained and loaded into memory.")34 35# Call this at app startup (see main.py below)36initialize_models()37 38# Player-team mapping39player_team_mapping = BALL_DF.groupby('striker')['batting_team'].agg(lambda x: x.mode()[0] if len(x.mode()) > 0 else None).to_dict()40 41# Clean JSON data (unchanged)42def clean_json(data):43 if isinstance(data, dict):44 return {k: clean_json(v) for k, v in data.items()}45 elif isinstance(data, list):46 return [clean_json(v) for v in data]47 elif isinstance(data, float):48 return 0.0 if pd.isna(data) or np.isinf(data) else data49 elif pd.isna(data):50 return None51 elif isinstance(data, pd.Timestamp):52 return data.strftime('%Y-%m-%d') if pd.notna(data) else None53 elif isinstance(data, (int, bool)):54 return data55 return str(data)56 57# Summary generation (unchanged)58def generate_summary(data, context_type):59 prompt = ""60 if context_type == "player_stats":61 prompt = f"Summarize this player data in one sentence: {data}"62 elif context_type == "team_stats":63 prompt = f"Summarize this team data in one sentence: {data}"64 elif context_type == "match_history":65 prompt = f"Summarize this match history between {data['team1']} and {data['team2']} in one sentence: {data['matches']}"66 elif context_type == "prediction_score":67 prompt = f"Summarize this prediction in one sentence: {data}"68 elif context_type == "prediction_team":69 prompt = f"Summarize this team prediction in one sentence: {data}"70 71 try:72 chat_completion = client.chat.completions.create(73 model="mixtral-8x7b-32768",74 messages=[75 {"role": "system", "content": "You are a concise cricket analyst."},76 {"role": "user", "content": prompt}77 ],78 max_tokens=50,79 temperature=0.780 )81 return chat_completion.choices[0].message.content.strip()82 except Exception as e:83 return f"Summary unavailable due to error: {str(e)}"84 85# Player stats (unchanged except using global BALL_DF)86def get_player_stats(player_name: str, season: str = None, role: str = "Batting"):87 player_name = player_name.strip().title()88 name_variations = [player_name, player_name.replace(" ", ""), " ".join(reversed(player_name.split()))]89 player_data = BALL_DF[BALL_DF['striker'].isin(name_variations) | BALL_DF['bowler'].isin(name_variations)]90 if season and 'season' in BALL_DF.columns:91 player_data = player_data[player_data['season'] == season]92 if player_data.empty:93 raise HTTPException(status_code=404, detail=f"Player '{player_name}' not found. Variations tried: {name_variations}")94 95 if role == "Batting":96 batting_data = player_data[player_data['striker'].isin(name_variations)]97 total_runs = int(batting_data['runs_off_bat'].sum())98 balls_faced = int(batting_data.shape[0])99 strike_rate = float((total_runs / balls_faced * 100) if balls_faced > 0 else 0)100 matches_played = int(len(batting_data['match_id'].unique()))101 102 stats = {103 "player_name": player_name,104 "role": role,105 "total_runs": total_runs,106 "balls_faced": balls_faced,107 "strike_rate": strike_rate,108 "matches_played": matches_played,109 "season": season if season else "All Seasons"110 }111 stats["summary"] = generate_summary(stats, "player_stats")112 return clean_json(stats)113 114 elif role == "Bowling":115 bowling_data = player_data[player_data['bowler'].isin(name_variations)]116 bowler_wicket_types = ["caught", "bowled", "lbw", "caught and bowled", "hit wicket"]117 wickets_data = bowling_data[bowling_data['player_dismissed'].notna() & 118 bowling_data['wicket_type'].isin(bowler_wicket_types)]119 total_wickets = int(wickets_data.shape[0])120 total_runs_conceded = int(bowling_data['total_runs'].sum())121 total_balls_bowled = int(bowling_data.shape[0])122 total_overs_bowled = float(total_balls_bowled / 6)123 bowling_average = float(total_runs_conceded / total_wickets) if total_wickets > 0 else float('inf')124 economy_rate = float(total_runs_conceded / total_overs_bowled) if total_overs_bowled > 0 else 0125 bowling_strike_rate = float(total_balls_bowled / total_wickets) if total_wickets > 0 else float('inf')126 bowling_matches = int(len(bowling_data['match_id'].unique()))127 128 stats = {129 "player_name": player_name,130 "role": role,131 "total_wickets": total_wickets,132 "bowling_average": 0.0 if np.isinf(bowling_average) else round(bowling_average, 2),133 "economy_rate": round(economy_rate, 2),134 "bowling_strike_rate": 0.0 if np.isinf(bowling_strike_rate) else round(bowling_strike_rate, 2),135 "overs_bowled": round(total_overs_bowled, 1),136 "bowling_matches": bowling_matches,137 "season": season if season else "All Seasons"138 }139 stats["summary"] = generate_summary(stats, "player_stats")140 return clean_json(stats)141 142# Team stats (unchanged except using global MATCH_DF)143def get_team_stats(team_name: str, season: str = None):144 team_name = team_name.strip().title()145 team_matches = MATCH_DF[(MATCH_DF['team1'] == team_name) | (MATCH_DF['team2'] == team_name)]146 if season and 'season' in MATCH_DF.columns:147 team_matches = team_matches[team_matches['season'] == season]148 if team_matches.empty:149 raise HTTPException(status_code=404, detail="Team not found")150 151 wins = int(team_matches[team_matches['winner'] == team_name].shape[0])152 total_matches = int(team_matches.shape[0])153 154 stats = {155 "total_matches": total_matches,156 "wins": wins,157 "losses": total_matches - wins,158 "win_percentage": float((wins / total_matches * 100) if total_matches > 0 else 0),159 "season": season if season else "All Seasons"160 }161 stats["summary"] = generate_summary(stats, "team_stats")162 return clean_json(stats)163 164# Match history (unchanged except using global MATCH_DF)165def get_match_history(team1: str, team2: str, season: str = None):166 team1 = team1.strip().title()167 team2 = team2.strip().title()168 available_teams = set(MATCH_DF['team1'].unique().tolist() + MATCH_DF['team2'].unique().tolist())169 if team1 not in available_teams or team2 not in available_teams:170 raise HTTPException(status_code=404, detail=f"Team {team1 if team1 not in available_teams else team2} not found.")171 172 team_matches = MATCH_DF[173 ((MATCH_DF['team1'] == team1) & (MATCH_DF['team2'] == team2)) |174 ((MATCH_DF['team1'] == team2) & (MATCH_DF['team2'] == team1))175 ].copy()176 if season and 'season' in MATCH_DF.columns:177 team_matches = team_matches[team_matches['season'] == season]178 if team_matches.empty:179 raise HTTPException(status_code=404, detail=f"No match history found between {team1} and {team2}.")180 181 team_matches['date'] = team_matches['date'].apply(lambda x: x.strftime('%Y-%m-%d') if pd.notna(x) else None)182 team_matches['winner'] = team_matches['winner'].fillna("Draw")183 for column in ['team1', 'team2', 'winner']:184 team_matches[column] = team_matches[column].apply(lambda x: str(x) if pd.notna(x) else None)185 history = team_matches[['date', 'team1', 'team2', 'winner']].to_dict(orient='records')186 187 response = {188 "team1": team1,189 "team2": team2,190 "season": season if season else "All Seasons",191 "matches": history192 }193 response["summary"] = generate_summary(response, "match_history")194 return clean_json(response)195 196# Prediction functions using in-memory models197def predict_score(player_name: str, opposition_team: str):198 try:199 player_name = player_name.strip().replace("+", " ").title()200 name_variations = [player_name, player_name.replace(" ", ""), " ".join(reversed(player_name.split()))]201 player_team = None202 for name in name_variations:203 if name in player_team_mapping:204 player_team = player_team_mapping[name]205 player_name = name206 break207 if not player_team:208 raise ValueError(f"Player {player_name} not found in historical data")209 210 predicted_runs = predict_player_score(211 player=player_name,212 team=player_team,213 opponent=opposition_team,214 venue=None,215 city=None,216 toss_winner=None,217 toss_decision=None,218 score_model=PLAYER_SCORE_MODEL,219 scaler=PLAYER_SCALER,220 player_data=PLAYER_DATA221 )222 stats = {223 "player": player_name,224 "team": player_team,225 "opposition": opposition_team,226 "predicted_runs": predicted_runs["expected_score"]227 }228 stats["summary"] = generate_summary(stats, "prediction_score")229 return clean_json(stats)230 except Exception as e:231 raise HTTPException(status_code=500, detail=f"Error predicting score for {player_name} against {opposition_team}: {str(e)}")232 233def predict_team_outcome(team1: str, team2: str):234 prediction = predict_team_performance(235 team1=team1,236 team2=team2,237 venue=None,238 city=None,239 toss_winner=None,240 toss_decision=None,241 win_model=TEAM_WIN_MODEL,242 score_model=TEAM_SCORE_MODEL,243 data=TEAM_DATA,244 scaler=TEAM_SCALER245 )246 prediction["summary"] = generate_summary(prediction, "prediction_team")247 return clean_json(prediction)248 249# Utility functions (unchanged except using global dataframes)250def get_teams():251 return clean_json({"teams": sorted(set(MATCH_DF['team1'].unique().tolist() + MATCH_DF['team2'].unique().tolist()))})252 253def get_players():254 unique_players = sorted(set(BALL_DF['striker'].dropna().unique().tolist()))255 return clean_json({"players": unique_players})256 257def get_seasons():258 return clean_json({"seasons": ["All Seasons"] + sorted(MATCH_DF['season'].dropna().unique().tolist())})259 260# Team trends (unchanged except using global MATCH_DF)261def get_team_trends(team_name: str):262 team_name = team_name.strip().title()263 team_matches = MATCH_DF[(MATCH_DF['team1'] == team_name) | (MATCH_DF['team2'] == team_name)]264 if team_matches.empty:265 raise HTTPException(status_code=404, detail="Team not found")266 267 trends = []268 for season in MATCH_DF['season'].unique():269 season_matches = team_matches[team_matches['season'] == season]270 if not season_matches.empty:271 wins = season_matches[season_matches['winner'] == team_name].shape[0]272 total_matches = season_matches.shape[0]273 win_percentage = (wins / total_matches * 100) if total_matches > 0 else 0274 trends.append({275 "season": season,276 "wins": wins,277 "total_matches": total_matches,278 "win_percentage": win_percentage279 })280 281 return {"team_name": team_name, "trends": trends}282 283# Player trends (unchanged except using global BALL_DF)284def get_player_trends(player_name: str, role: str = "Batting"):285 player_name = player_name.strip().title()286 name_variations = [player_name, player_name.replace(" ", ""), " ".join(reversed(player_name.split()))]287 player_data = BALL_DF[BALL_DF['striker'].isin(name_variations) | BALL_DF['bowler'].isin(name_variations)]288 if player_data.empty:289 raise HTTPException(status_code=404, detail=f"Player '{player_name}' not found")290 291 trends = []292 for season in BALL_DF['season'].unique():293 season_data = player_data[player_data['season'] == season]294 if not season_data.empty:295 if role == "Batting":296 total_runs = int(season_data['runs_off_bat'].sum())297 balls_faced = int(season_data.shape[0])298 strike_rate = float((total_runs / balls_faced * 100) if balls_faced > 0 else 0)299 matches_played = int(len(season_data['match_id'].unique()))300 trends.append({301 "season": season,302 "total_runs": total_runs,303 "strike_rate": strike_rate,304 "matches_played": matches_played305 })306 elif role == "Bowling":307 total_wickets = int(season_data[season_data['wicket_type'].notna()].shape[0])308 total_runs_conceded = int(season_data['total_runs'].sum())309 total_overs_bowled = float(season_data.shape[0] / 6)310 bowling_average = float(total_runs_conceded / total_wickets) if total_wickets > 0 else float('inf')311 economy_rate = float(total_runs_conceded / total_overs_bowled) if total_overs_bowled > 0 else 0312 matches_played = int(len(season_data['match_id'].unique()))313 trends.append({314 "season": season,315 "total_wickets": total_wickets,316 "bowling_average": bowling_average,317 "economy economy_rate": economy_rate,318 "matches_played": matches_played319 })320 321 return {"player_name": player_name, "role": role, "trends": trends}