Spaces:
Build error
Build error
File size: 1,905 Bytes
ea6f215 |
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 |
import json
import os
import pandas as pd
from datetime import datetime
from src.core import setup_logger
logger = setup_logger(__name__)
class ExperimentTracker:
def __init__(self, log_path='logs/experiments.json'):
self.log_path = log_path
os.makedirs(os.path.dirname(self.log_path), exist_ok=True)
def log_experiment(self, name, params, metrics):
experiment = {
"timestamp": datetime.now().isoformat(),
"name": name,
"params": params,
"metrics": metrics
}
experiments = []
if os.path.exists(self.log_path):
with open(self.log_path, 'r') as f:
try:
experiments = json.load(f)
except json.JSONDecodeError:
experiments = []
experiments.append(experiment)
with open(self.log_path, 'w') as f:
json.dump(experiments, f, indent=4)
class DriftDetector:
def __init__(self, baseline_stats=None):
self.baseline_stats = baseline_stats
def calculate_stats(self, df, features):
return df[features].agg(['mean', 'std']).to_dict()
def check_drift(self, df, features, threshold=0.2):
if self.baseline_stats is None:
self.baseline_stats = self.calculate_stats(df, features)
return False, {}
current_stats = self.calculate_stats(df, features)
drifts = {}
drift_detected = False
for feature in features:
base_mean = self.baseline_stats[feature]['mean']
curr_mean = current_stats[feature]['mean']
if base_mean != 0:
change = abs(curr_mean - base_mean) / abs(base_mean)
if change > threshold:
drifts[feature] = change
drift_detected = True
return drift_detected, drifts
|