Machine Learning con Claude Sonnet 3.5

Dev: Ho un dataset di transazioni finanziarie. Come posso iniziare l'analisi?

Claude: Prima di tutto, carichiamo e analizziamo il dataset:

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

def analyze_transactions(file_path):
    # Caricamento dati
    df = pd.read_csv(file_path)
    
    # Statistiche di base
    print("Dimensione dataset:", df.shape)
    print("\nStatistiche principali:")
    print(df.describe())
    
    # Valori mancanti
    print("\nValori mancanti:")
    print(df.isnull().sum())
    
    return df

df = analyze_transactions('transactions.csv')

Dev: Ho notato molti valori anomali nelle colonne amount e timestamp

Claude: Ecco una funzione per identificare e analizzare le anomalie:

def analyze_outliers(df):
    # Calcolo IQR per amount
    Q1 = df['amount'].quantile(0.25)
    Q3 = df['amount'].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - 1.5 * IQR
    upper = Q3 + 1.5 * IQR
    
    outliers = df[
        (df['amount'] < lower) | 
        (df['amount'] > upper)
    ]
    
    # Analisi timestamp
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    time_diffs = df['timestamp'].diff()
    suspicious_times = df[
        time_diffs > pd.Timedelta(hours=24)
    ]
    
    return outliers, suspicious_times

# Visualizzazione risultati
def plot_anomalies(df, outliers):
    plt.figure(figsize=(12, 6))
    
    plt.subplot(1, 2, 1)
    sns.boxplot(y=df['amount'])
    plt.title('Distribuzione Importi')
    
    plt.subplot(1, 2, 2)
    sns.scatterplot(
        data=df, 
        x='timestamp', 
        y='amount',
        alpha=0.5
    )
    sns.scatterplot(
        data=outliers,
        x='timestamp',
        y='amount',
        color='red',
        alpha=0.7
    )
    plt.title('Anomalie Identificate')
    
    plt.tight_layout()
    plt.show()

Dev: Come posso preparare questi dati per un modello di fraud detection?

Claude: Ecco una pipeline di feature engineering per fraud detection:

def engineer_features(df):
    # Features temporali
    df['hour'] = df['timestamp'].dt.hour
    df['day_of_week'] = df['timestamp'].dt.dayofweek
    df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
    
    # Aggregazioni per utente
    user_stats = df.groupby('user_id').agg({
        'amount': ['mean', 'std', 'min', 'max', 'count'],
        'merchant_id': 'nunique'
    }).reset_index()
    
    user_stats.columns = [
        'user_id', 
        'avg_amount',
        'std_amount', 
        'min_amount',
        'max_amount',
        'transaction_count',
        'unique_merchants'
    ]
    
    # Velocity features
    df['time_since_last'] = df.groupby('user_id')['timestamp'].diff()
    df['amount_ratio'] = df['amount'] / df.groupby('user_id')['amount'].shift()
    
    # One-hot encoding per categorie
    cat_features = pd.get_dummies(
        df[['merchant_category', 'transaction_type']],
        prefix=['cat', 'type']
    )
    
    return pd.concat([df, cat_features], axis=1)

Dev: Come gestisco lo sbilanciamento delle classi nel dataset?

Claude: Ecco alcune tecniche per gestire dataset sbilanciati:

from imblearn.over_sampling import SMOTE
from imblearn.under_sampling import RandomUnderSampler
from imblearn.pipeline import Pipeline

def balance_dataset(X, y):
    # Combina over e under sampling
    steps = [
        ('over', SMOTE(sampling_strategy=0.1)),
        ('under', RandomUnderSampler(sampling_strategy=0.5))
    ]
    
    pipeline = Pipeline(steps=steps)
    
    # Fit resample
    X_resampled, y_resampled = pipeline.fit_resample(X, y)
    
    print("Distribuzione originale:", np.bincount(y))
    print("Distribuzione bilanciata:", np.bincount(y_resampled))
    
    return X_resampled, y_resampled

Dev: Quale modello mi suggerisci per fraud detection?

Claude: Per fraud detection, un approccio ensemble spesso funziona bene:

from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report
from sklearn.model_selection import cross_val_score

def train_fraud_model(X, y):
    # Split train/test
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, stratify=y
    )
    
    # Train model
    model = RandomForestClassifier(
        n_estimators=100,
        max_depth=10,
        class_weight='balanced'
    )
    
    # Cross validation
    scores = cross_val_score(
        model, X_train, y_train, 
        cv=5, scoring='f1'
    )
    print(f"Cross-val scores: {scores.mean():.3f} (+/- {scores.std() * 2:.3f})")
    
    # Final training
    model.fit(X_train, y_train)
    
    # Evaluation
    y_pred = model.predict(X_test)
    print("\nTest Set Performance:")
    print(classification_report(y_test, y_pred))
    
    return model

Dev: Come interpreto i risultati del modello?

Claude: Ecco alcune tecniche per l'interpretabilità del modello:

from sklearn.inspection import permutation_importance
import shap

def interpret_model(model, X, y):
    # Feature importance
    importances = pd.DataFrame(
        model.feature_importances_,
        index=X.columns,
        columns=['importance']
    ).sort_values('importance', ascending=False)
    
    # Permutation importance
    perm_importance = permutation_importance(
        model, X, y, 
        n_repeats=10
    )
    
    # SHAP values
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X)
    
    # Plot results
    plt.figure(figsize=(15, 5))
    
    plt.subplot(1, 3, 1)
    importances.head(10).plot(kind='bar')
    plt.title('Feature Importance')
    
    plt.subplot(1, 3, 2)
    shap.summary_plot(shap_values, X, plot_type='bar')
    plt.title('SHAP Values')
    
    plt.tight_layout()
    plt.show()

Dev: Come posso deployare il modello in produzione?

Claude: Ecco una struttura per il deployment con FastAPI e monitoring:

from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import prometheus_client as prom

# Metriche
PREDICTION_LATENCY = prom.Histogram(
    'model_prediction_latency_seconds',
    'Time spent processing prediction'
)

PREDICTION_COUNT = prom.Counter(
    'model_predictions_total',
    'Number of predictions served'
)

class Transaction(BaseModel):
    amount: float
    merchant_id: str
    user_id: str
    timestamp: str
    
class FraudDetectionService:
    def __init__(self):
        self.model = joblib.load('fraud_model.pkl')
        self.feature_engineer = joblib.load('feature_engineer.pkl')
        
    @PREDICTION_LATENCY.time()
    def predict(self, transaction: Transaction):
        # Preprocess
        features = self.feature_engineer.transform(
            transaction.dict()
        )
        
        # Predict
        prediction = self.model.predict_proba(features)[0]
        PREDICTION_COUNT.inc()
        
        return {
            'fraud_probability': float(prediction[1]),
            'threshold': 0.8,
            'is_fraud': prediction[1] > 0.8
        }

app = FastAPI()
service = FraudDetectionService()

@app.post('/predict')
async def predict(transaction: Transaction):
    return service.predict(transaction)

@app.get('/metrics')
async def metrics():
    return prom.generate_latest()