training_service.py

import os
import time
import json
import joblib
import pandas as pd
import numpy as np
import xgboost as xgb
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split, RandomizedSearchCV, cross_val_predict
from sklearn.metrics import precision_recall_curve, classification_report, accuracy_score, f1_score, precision_score, recall_score
try:
    from imblearn.over_sampling import SMOTE
except ImportError:
    SMOTE = None

from supabase import create_client, Client
from dotenv import load_dotenv

# Import local modules
# Assuming feature_engineering.py is in the same directory
try:
    from feature_engineering import FraudFeatureEngineer
except ImportError:
    import sys
    sys.path.append(os.path.dirname(os.path.abspath(__file__)))
    from feature_engineering import FraudFeatureEngineer

load_dotenv()

# Initialize Supabase client
url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") # Use Service Role Key for backend updates
supabase: Client = create_client(url, key) if url and key else None

def update_job_status(job_id: str, status: str, metrics: dict = None, file_path: str = None):
    """Update job status in Supabase"""
    if not supabase:
        print(f"⚠️ Supabase not configured. Job {job_id} status: {status}")
        return

    data = {"status": status}
    if metrics:
        data["metrics"] = metrics
    if file_path:
        data["file_path"] = file_path
    
    try:
        supabase.table("model_registry").update(data).eq("id", job_id).execute()
    except Exception as e:
        print(f"❌ Failed to update Supabase for job {job_id}: {str(e)}")

def make_splits(df, test_frac=0.05, time_col='step', min_test_fraud=100, random_state=42):
    """
    Returns: X_train, X_test, y_train, y_test
    """
    # Normalize column names
    if 'is_fraud' in df.columns:
        df = df.rename(columns={'is_fraud': 'isFraud'})
    if 'fraud' in df.columns:
        df = df.rename(columns={'fraud': 'isFraud'})

    # Temporal test split
    if time_col in df.columns:
        cutoff = df[time_col].quantile(1 - test_frac)
        train_df = df[df[time_col] <= cutoff].reset_index(drop=True)
        test_df = df[df[time_col] > cutoff].reset_index(drop=True)
    else:
        # Random split if no time column
        train_df, test_df = train_test_split(df, test_size=test_frac, stratify=df.get('isFraud'), random_state=42)
        train_df = train_df.reset_index(drop=True)
        test_df = test_df.reset_index(drop=True)

    # Fallback to stratified split if test set is poor
    if (test_df.empty) or (test_df['isFraud'].sum() < min_test_fraud):
        X_temp = df.drop(columns=['isFraud'])
        y_temp = df['isFraud']
        X_train, X_test, y_train, y_test = train_test_split(
            X_temp, y_temp, test_size=test_frac, stratify=y_temp, random_state=random_state
        )
        return X_train.reset_index(drop=True), X_test.reset_index(drop=True), y_train.reset_index(drop=True), y_test.reset_index(drop=True)
    else:
        X_train = train_df.drop(columns=['isFraud'])
        y_train = train_df['isFraud']
        X_test = test_df.drop(columns=['isFraud'])
        y_test = test_df['isFraud']
        return X_train.reset_index(drop=True), X_test.reset_index(drop=True), y_train.reset_index(drop=True), y_test.reset_index(drop=True)

async def train_model_async(job_id: str, file_path: str, params: dict):
    """
    Background task to train model
    """
    print(f"🚀 Starting training job {job_id} with params: {params}")
    update_job_status(job_id, "training")
    
    try:
        # 1. Load Data
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"Dataset not found at {file_path}")
            
        df = pd.read_csv(file_path)
        
        # Basic Preprocessing
        if 'oldbalanceOrg' in df.columns:
            df = df.rename(columns={'oldbalanceOrg':'oldBalanceOrig', 'newbalanceOrig':'newBalanceOrig', 
                                    'oldbalanceDest':'oldBalanceDest', 'newbalanceDest':'newBalanceDest'})
            
        # Filter types if needed (Notebook filters to TRANSFER/CASH_OUT)
        # We'll stick to what the notebook did for consistency
        if 'type' in df.columns:
            df = df[df['type'].isin(['TRANSFER', 'CASH_OUT'])].reset_index(drop=True)
            
        # 2. Split Data
        test_frac = params.get('test_split', 0.05)
        X_train, X_test, y_train, y_test = make_splits(df, test_frac=test_frac)
        
        print(f"📊 Data split: Train={len(X_train)}, Test={len(X_test)}")
        
        # 3. Feature Engineering
        pagerank_limit = params.get('pagerank_limit', 10000)
        advanced_features = params.get('advanced_feature_engineering', False)
        fe = FraudFeatureEngineer(pagerank_limit=pagerank_limit, advanced_features=advanced_features)
        
        print("⚙️ Fitting feature engineer...")
        fe.fit(X_train, y_train)
        X_train_trans = fe.transform(X_train)
        X_test_trans = fe.transform(X_test)
        
        # Apply Advanced Preprocessing (SMOTE) if requested
        if params.get('advanced_preprocessing'):
            if SMOTE:
                print("⚖️ Applying SMOTE for class imbalance...")
                try:
                    smote = SMOTE(random_state=42)
                    X_train_trans, y_train = smote.fit_resample(X_train_trans, y_train)
                    print(f"   New training shape: {X_train_trans.shape}")
                except Exception as e:
                    print(f"⚠️ SMOTE failed: {str(e)}. Proceeding with original data.")
            else:
                print("⚠️ SMOTE requested but imbalanced-learn not installed. Proceeding without it.")

        # 4. Configure XGBoost
        # Calculate scale_pos_weight
        neg = (y_train == 0).sum()
        pos = (y_train == 1).sum()
        spw = int(max(1, neg / max(1, pos)))
        
        xgb_params = {
            'objective': 'binary:logistic',
            'tree_method': 'hist',
            'random_state': 42,
            'n_jobs': 1,
            'n_estimators': int(params.get('n_estimators', 300)),
            'max_depth': int(params.get('max_depth', 6)),
            'learning_rate': float(params.get('learning_rate', 0.05)),
            'scale_pos_weight': spw
        }
        
        clf = xgb.XGBClassifier(**xgb_params)
        
        # 5. Train
        print("🏋️ Training classifier...")
        clf.fit(X_train_trans, y_train)
        
        # 6. Evaluate
        print("📝 Evaluating...")
        y_pred = clf.predict(X_test_trans)
        
        metrics = {
            "accuracy": float(accuracy_score(y_test, y_pred)),
            "precision": float(precision_score(y_test, y_pred)),
            "recall": float(recall_score(y_test, y_pred)),
            "f1": float(f1_score(y_test, y_pred))
        }
        print(f"✅ Metrics: {metrics}")
        
        # 7. Save Model
        # Ensure Models directory exists
        models_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Models")
        os.makedirs(models_dir, exist_ok=True)
        
        model_filename = f"model_{job_id}.pkl"
        model_save_path = os.path.join(models_dir, model_filename)
        
        # Save ONLY the classifier (as per notebook strategy)
        joblib.dump(clf, model_save_path)
        print(f"💾 Model saved to {model_save_path}")
        
        # 8. Update Registry
        update_job_status(job_id, "ready", metrics=metrics, file_path=f"Models/{model_filename}")
        
    except Exception as e:
        print(f"❌ Training failed: {str(e)}")
        update_job_status(job_id, "failed", metrics={"error": str(e)})