19787274411

Committed 29 Nov 2025 05:49PM UTC coverage: 97.087% (-2.9%) from 100.0%

Build # 19787274411

Build Type

Pull #32

github

Committed by

web-flow

Commit Message

Merge dbbde655c into 66502aaa8

Pull Request Pull Request #32: add ml_experiments directory

Coverage Stats

53 of 56 new or added lines in 1 file covered. (94.64%)

100 of 103 relevant lines covered (97.09%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.09

/src/utils.py

from datetime import datetime
import logging
from typing import Any, Dict

from joblib import load
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sqlalchemy.engine.base import Connection, Engine
from jyablonski_common_modules.sql import write_to_sql_upsert

from ml_experiments.training_pipeline import TrainingPipeline

# Set up module-level logger
logger = logging.getLogger(__name__)

# Constants
INPUT_ML_TABLE = "ml_game_features"
OUTPUT_ML_TABLE = "ml_game_predictions"

# Columns to drop for V1 ML prediction
ML_EXCLUDE_COLUMNS = [
    "home_team",
    "home_moneyline",
    "away_team",
    "away_moneyline",
    "game_date",
    "outcome",
]


def get_feature_flags(
    connection: Connection | Engine, schema: str = "gold"
) -> pd.DataFrame:
    """Retrieve feature flags from the database."""
    flags = pd.read_sql_query(
        sql=f"select * from {schema}.feature_flags", con=connection
    )
    logger.info(f"Retrieved {len(flags)} feature flags")
    return flags


def check_feature_flag(flag: str, flags_df: pd.DataFrame) -> bool:
    """Check if a specific feature flag is enabled."""
    flag_data = flags_df.query(f"flag == '{flag}'")

    if flag_data.empty:
        logger.info(f"Feature flag '{flag}' not found")
        return False

    is_enabled = bool(flag_data["is_enabled"].iloc[0])

    if not is_enabled:
        logger.info(f"Feature flag '{flag}' is disabled")

    return is_enabled


def pull_tonights_games(
    connection: Connection | Engine,
    schema: str,
    table: str = INPUT_ML_TABLE,
) -> pd.DataFrame:
    """Pull tonight's game features from the database."""
    games = pd.read_sql_query(sql=f"SELECT * FROM {schema}.{table}", con=connection)

    # Sort for consistency (handle missing column gracefully)
    if "home_team_avg_pts_scored" in games.columns:
        games = games.sort_values("home_team_avg_pts_scored")

    logger.info(f"Retrieved {len(games)} games from {schema}.{table}")
    return games


def load_ml_model(model_path: str) -> LogisticRegression:
    """Load a trained V1 ML model from disk (simple joblib)."""
    model = load(model_path)
    logger.info(f"Loaded V1 model from {model_path}")
    return model


def load_v2_artifacts(model_path: str) -> Dict[str, Any]:
    """
    Load V2 Production Artifacts.
    Returns dict with keys: 'model', 'feature_engineer', 'feature_names', etc.
    """
    try:
        artifacts = TrainingPipeline.load_artifacts(model_path)
        logger.info(f"Loaded V2 artifacts from {model_path}")
        return artifacts
    except FileNotFoundError:
        logger.error(f"V2 Model Artifact not found at {model_path}")
        return {}


def generate_win_predictions(
    games_df: pd.DataFrame,
    ml_model: LogisticRegression,
) -> pd.DataFrame:
    """Generate V1 win predictions."""
    if games_df.empty:
        logger.error("No data available for predictions")
        return pd.DataFrame()

    current_date = datetime.now().date()
    latest_date = pd.to_datetime(games_df["game_date"]).iloc[0].date()

    if latest_date != current_date:
        logger.error(
            f"Date mismatch: data is from {latest_date}, expected {current_date}"
        )
        return pd.DataFrame()

    # Prepare features for prediction
    ml_features = games_df.drop(columns=ML_EXCLUDE_COLUMNS)

    # Generate predictions
    predictions = ml_model.predict_proba(ml_features)
    prediction_df = pd.DataFrame(
        predictions,
        columns=["away_team_predicted_win_pct", "home_team_predicted_win_pct"],
    ).round(3)

    # Combine with original data
    result_df = games_df.reset_index(drop=True).drop(
        columns=["outcome"], errors="ignore"
    )
    result_df = pd.concat([result_df, prediction_df], axis=1)

    logger.info(f"Generated V1 predictions for {len(result_df)} games")
    return result_df


def generate_win_predictions_v2(
    games_df: pd.DataFrame,
    artifacts: Dict[str, Any],
) -> pd.DataFrame:
    """
    Generate V2 win predictions using the FeatureEngineer pipeline.
    Returns the original DataFrame with two new prediction columns appended.
    """
    if games_df.empty:
        logger.warning("No V2 data available for predictions")
        return pd.DataFrame()

    # 1. Date Validation
    current_date = datetime.now().date()
    game_dates = pd.to_datetime(games_df["game_date"])
    latest_date = game_dates.iloc[0].date()

    if latest_date != current_date:
        logger.warning(
            f"V2 Date mismatch: data is from {latest_date}, expected {current_date}. Proceeding anyway."
        )

    # 2. Extract Artifacts
    model = artifacts["model"]
    engineer = artifacts["feature_engineer"]
    feature_names = artifacts["feature_names"]

    # 3. Preprocess Data (Inference Mode)
    try:
        # is_training=False ensures we use saved medians/scalers
        X_processed = engineer.preprocess_data(games_df, is_training=False)

        # Ensure exact column alignment with training data (Robustness)
        for col in feature_names:
            if col not in X_processed.columns:
                X_processed[col] = 0

        X_final = X_processed[feature_names]

    except Exception as e:
        logger.error(f"V2 Feature Engineering failed: {e}")
        return pd.DataFrame()

    # 4. Predict
    try:
        # model.predict_proba returns [Prob_Class_0, Prob_Class_1] (Loss, Win)
        # We grab index 1 for Home Win Probability
        home_probs = model.predict_proba(X_final)[:, 1]
    except Exception as e:
        logger.error(f"V2 Model Prediction failed: {e}")
        return pd.DataFrame()

    # 5. Format Output (Similar to V1 Process)
    # We clone the original DF so we retain all the feature columns (Rank, VORP, etc.)
    # needed for the destination DDL.
    result_df = games_df.reset_index(drop=True)

    # Drop 'outcome' if it exists in raw data (usually NULL in inference, but safe to drop)
    if "outcome" in result_df.columns:
        result_df = result_df.drop(columns=["outcome"])

    # Append the 2 new columns
    result_df["home_team_predicted_win_pct"] = np.round(home_probs, 3)
    result_df["away_team_predicted_win_pct"] = np.round(1.0 - home_probs, 3)

    logger.info(f"Generated V2 predictions for {len(result_df)} games")
    return result_df


def write_predictions_to_database(
    connection: Connection | Engine,
    predictions_df: pd.DataFrame,
    schema: str,
    table: str = OUTPUT_ML_TABLE,
    primary_keys: list[str] = None,
) -> None:
    """Write predictions to the database using upsert logic."""
    if primary_keys is None:
        primary_keys = ["home_team", "game_date"]

    if predictions_df.empty:
        logger.warning(f"Attempted to write empty dataframe to {table}")
        return

    write_to_sql_upsert(
        conn=connection,
        table=table,
        schema=schema,
        df=predictions_df,
        primary_keys=primary_keys,
    )

    logger.info(
        f"Successfully wrote {len(predictions_df)} predictions to {schema}.{table}"
    )

    return None

1	from datetime import datetime	1✔
2	import logging	1✔
3	from typing import Any, Dict	1✔
4
5	from joblib import load	1✔
6	import pandas as pd	1✔
7	import numpy as np	1✔
8	from sklearn.linear_model import LogisticRegression	1✔
9	from sqlalchemy.engine.base import Connection, Engine	1✔
10	from jyablonski_common_modules.sql import write_to_sql_upsert	1✔
11
12	from ml_experiments.training_pipeline import TrainingPipeline	1✔
13
14	# Set up module-level logger
15	logger = logging.getLogger(__name__)	1✔
16
17	# Constants
18	INPUT_ML_TABLE = "ml_game_features"	1✔
19	OUTPUT_ML_TABLE = "ml_game_predictions"	1✔
20
21	# Columns to drop for V1 ML prediction
22	ML_EXCLUDE_COLUMNS = [	1✔
23	"home_team",
24	"home_moneyline",
25	"away_team",
26	"away_moneyline",
27	"game_date",
28	"outcome",
29	]
30
31
32	def get_feature_flags(	1✔
33	connection: Connection \| Engine, schema: str = "gold"
34	) -> pd.DataFrame:
35	"""Retrieve feature flags from the database."""
36	flags = pd.read_sql_query(	1✔
37	sql=f"select * from {schema}.feature_flags", con=connection
38	)
39	logger.info(f"Retrieved {len(flags)} feature flags")	1✔
40	return flags	1✔
41
42
43	def check_feature_flag(flag: str, flags_df: pd.DataFrame) -> bool:	1✔
44	"""Check if a specific feature flag is enabled."""
45	flag_data = flags_df.query(f"flag == '{flag}'")	1✔
46
47	if flag_data.empty:	1✔
48	logger.info(f"Feature flag '{flag}' not found")	1✔
49	return False	1✔
50
51	is_enabled = bool(flag_data["is_enabled"].iloc[0])	1✔
52
53	if not is_enabled:	1✔
54	logger.info(f"Feature flag '{flag}' is disabled")	1✔
55
56	return is_enabled	1✔
57
58
59	def pull_tonights_games(	1✔
60	connection: Connection \| Engine,
61	schema: str,
62	table: str = INPUT_ML_TABLE,
63	) -> pd.DataFrame:
64	"""Pull tonight's game features from the database."""
65	games = pd.read_sql_query(sql=f"SELECT * FROM {schema}.{table}", con=connection)	1✔
66
67	# Sort for consistency (handle missing column gracefully)
68	if "home_team_avg_pts_scored" in games.columns:	1✔
69	games = games.sort_values("home_team_avg_pts_scored")	1✔
70
71	logger.info(f"Retrieved {len(games)} games from {schema}.{table}")	1✔
72	return games	1✔
73
74
75	def load_ml_model(model_path: str) -> LogisticRegression:	1✔
76	"""Load a trained V1 ML model from disk (simple joblib)."""
77	model = load(model_path)	1✔
78	logger.info(f"Loaded V1 model from {model_path}")	1✔
79	return model	1✔
80
81
82	def load_v2_artifacts(model_path: str) -> Dict[str, Any]:	1✔
83	"""
84	Load V2 Production Artifacts.
85	Returns dict with keys: 'model', 'feature_engineer', 'feature_names', etc.
86	"""
87	try:	1✔
88	artifacts = TrainingPipeline.load_artifacts(model_path)	1✔
89	logger.info(f"Loaded V2 artifacts from {model_path}")	1✔
90	return artifacts	1✔
91	except FileNotFoundError:	1✔
92	logger.error(f"V2 Model Artifact not found at {model_path}")	1✔
93	return {}	1✔
94
95
96	def generate_win_predictions(	1✔
97	games_df: pd.DataFrame,
98	ml_model: LogisticRegression,
99	) -> pd.DataFrame:
100	"""Generate V1 win predictions."""
101	if games_df.empty:	1✔
102	logger.error("No data available for predictions")	1✔
103	return pd.DataFrame()	1✔
104
105	current_date = datetime.now().date()	1✔
106	latest_date = pd.to_datetime(games_df["game_date"]).iloc[0].date()	1✔
107
108	if latest_date != current_date:	1✔
109	logger.error(	1✔
110	f"Date mismatch: data is from {latest_date}, expected {current_date}"
111	)
112	return pd.DataFrame()	1✔
113
114	# Prepare features for prediction
115	ml_features = games_df.drop(columns=ML_EXCLUDE_COLUMNS)	1✔
116
117	# Generate predictions
118	predictions = ml_model.predict_proba(ml_features)	1✔
119	prediction_df = pd.DataFrame(	1✔
120	predictions,
121	columns=["away_team_predicted_win_pct", "home_team_predicted_win_pct"],
122	).round(3)
123
124	# Combine with original data
125	result_df = games_df.reset_index(drop=True).drop(	1✔
126	columns=["outcome"], errors="ignore"
127	)
128	result_df = pd.concat([result_df, prediction_df], axis=1)	1✔
129
130	logger.info(f"Generated V1 predictions for {len(result_df)} games")	1✔
131	return result_df	1✔
132
133
134	def generate_win_predictions_v2(	1✔
135	games_df: pd.DataFrame,
136	artifacts: Dict[str, Any],
137	) -> pd.DataFrame:
138	"""
139	Generate V2 win predictions using the FeatureEngineer pipeline.
140	Returns the original DataFrame with two new prediction columns appended.
141	"""
142	if games_df.empty:	1✔
NEW 143	logger.warning("No V2 data available for predictions")	×
NEW 144	return pd.DataFrame()	×
145
146	# 1. Date Validation
147	current_date = datetime.now().date()	1✔
148	game_dates = pd.to_datetime(games_df["game_date"])	1✔
149	latest_date = game_dates.iloc[0].date()	1✔
150
151	if latest_date != current_date:	1✔
152	logger.warning(	1✔
153	f"V2 Date mismatch: data is from {latest_date}, expected {current_date}. Proceeding anyway."
154	)
155
156	# 2. Extract Artifacts
157	model = artifacts["model"]	1✔
158	engineer = artifacts["feature_engineer"]	1✔
159	feature_names = artifacts["feature_names"]	1✔
160
161	# 3. Preprocess Data (Inference Mode)
162	try:	1✔
163	# is_training=False ensures we use saved medians/scalers
164	X_processed = engineer.preprocess_data(games_df, is_training=False)	1✔
165
166	# Ensure exact column alignment with training data (Robustness)
167	for col in feature_names:	1✔
168	if col not in X_processed.columns:	1✔
NEW 169	X_processed[col] = 0	×
170
171	X_final = X_processed[feature_names]	1✔
172
173	except Exception as e:	1✔
174	logger.error(f"V2 Feature Engineering failed: {e}")	1✔
175	return pd.DataFrame()	1✔
176
177	# 4. Predict
178	try:	1✔
179	# model.predict_proba returns [Prob_Class_0, Prob_Class_1] (Loss, Win)
180	# We grab index 1 for Home Win Probability
181	home_probs = model.predict_proba(X_final)[:, 1]	1✔
182	except Exception as e:	1✔
183	logger.error(f"V2 Model Prediction failed: {e}")	1✔
184	return pd.DataFrame()	1✔
185
186	# 5. Format Output (Similar to V1 Process)
187	# We clone the original DF so we retain all the feature columns (Rank, VORP, etc.)
188	# needed for the destination DDL.
189	result_df = games_df.reset_index(drop=True)	1✔
190
191	# Drop 'outcome' if it exists in raw data (usually NULL in inference, but safe to drop)
192	if "outcome" in result_df.columns:	1✔
193	result_df = result_df.drop(columns=["outcome"])	1✔
194
195	# Append the 2 new columns
196	result_df["home_team_predicted_win_pct"] = np.round(home_probs, 3)	1✔
197	result_df["away_team_predicted_win_pct"] = np.round(1.0 - home_probs, 3)	1✔
198
199	logger.info(f"Generated V2 predictions for {len(result_df)} games")	1✔
200	return result_df	1✔
201
202
203	def write_predictions_to_database(	1✔
204	connection: Connection \| Engine,
205	predictions_df: pd.DataFrame,
206	schema: str,
207	table: str = OUTPUT_ML_TABLE,
208	primary_keys: list[str] = None,
209	) -> None:
210	"""Write predictions to the database using upsert logic."""
211	if primary_keys is None:	1✔
212	primary_keys = ["home_team", "game_date"]	1✔
213
214	if predictions_df.empty:	1✔
215	logger.warning(f"Attempted to write empty dataframe to {table}")	1✔
216	return	1✔
217
218	write_to_sql_upsert(	1✔
219	conn=connection,
220	table=table,
221	schema=schema,
222	df=predictions_df,
223	primary_keys=primary_keys,
224	)
225
226	logger.info(	1✔
227	f"Successfully wrote {len(predictions_df)} predictions to {schema}.{table}"
228	)
229
230	return None	1✔

jyablonski / nba_elt_mlflow / 19787274411

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous