File size: 4,487 Bytes
2532605 77f1d1a 2532605 | 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 | """
KRONECTOR - Drift Detection Pipeline
Monitors incoming F1 data for Concept and Data Drift using Evidently AI.
"""
import logging
import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset
from evidently import ColumnMapping
from ml.feature_engineering import prepare_model_data
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(name)s | %(levelname)s | %(message)s")
def detect_drift(
data_path: str = "data_output/fastf1_races.parquet",
current_season: int = 2026,
report_output_path: str = "data_output/drift_report.html"
) -> bool:
"""
Detects drift between the reference dataset (historical) and the current dataset.
Returns True if significant drift is detected.
"""
try:
df = pd.read_parquet(data_path)
except FileNotFoundError:
logger.error(f"Data not found at {data_path}")
return False
# To detect drift accurately, we should evaluate the model's engineered features.
# We use prepare_model_data to impute missing values and encode categoricals.
logger.info("Preparing features for drift detection...")
bundle, _ = prepare_model_data(df)
# Re-attach season for splitting
eval_df = bundle.X.copy()
eval_df["season"] = bundle.metadata["season"]
eval_df["win_probability"] = bundle.y
# Define reference (e.g., up to 2025) and current (2026)
reference = eval_df[eval_df["season"] < current_season].copy()
current = eval_df[eval_df["season"] >= current_season].copy()
# Drop the season column as it's not a model feature we want to check for drift
reference = reference.drop(columns=["season"])
current = current.drop(columns=["season"])
if current.empty:
logger.warning(f"No data available for current season {current_season}. Skipping drift detection.")
return False
logger.info(f"Running drift detection: Reference ({len(reference)} rows) vs Current ({len(current)} rows)")
# Define column mapping
target = "win_probability"
prediction = None
column_mapping = ColumnMapping()
column_mapping.target = target
column_mapping.prediction = prediction
# Since we passed the data through prepare_model_data, all categories are now numerically encoded.
# We treat them as numerical features for Evidently so it can detect distributional shifts in the encoded space.
numerical_features = [col for col in reference.columns if col != target]
column_mapping.numerical_features = numerical_features
column_mapping.categorical_features = []
# Create Evidently Report
report = Report(metrics=[
DataDriftPreset(),
TargetDriftPreset(),
])
logger.info("Generating Evidently AI Drift Report...")
report.run(reference_data=reference, current_data=current, column_mapping=column_mapping)
# Save HTML report
import os
os.makedirs(os.path.dirname(report_output_path), exist_ok=True)
report.save_html(report_output_path)
logger.info(f"Drift report saved to {report_output_path}")
# Parse JSON results to determine if drift was detected
result = report.as_dict()
# Check data drift
data_drift = False
target_drift = False
try:
# Evidently structure navigation
for metric in result["metrics"]:
if metric["metric"] == "DataDriftPreset":
if metric["result"]["dataset_drift"]:
data_drift = True
logger.warning("🚨 DATA DRIFT DETECTED!")
elif metric["metric"] == "TargetDriftPreset":
if metric["result"]["drift_detected"]:
target_drift = True
logger.warning("🚨 TARGET DRIFT DETECTED!")
except KeyError as e:
logger.warning(f"Could not parse drift results exactly: {e}")
return data_drift or target_drift
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Run Drift Detection")
parser.add_argument("--data-path", default="data_output/fastf1_races.parquet")
parser.add_argument("--current-season", type=int, default=2026)
args = parser.parse_args()
drift_detected = detect_drift(
data_path=args.data_path,
current_season=args.current_season
)
print(f"\nFinal Result -> Drift Detected: {drift_detected}")
|