Machine Learning with Claude Sonnet 3.5

Dev: I have a dataset of financial transactions. How can I start the analysis?

Claude: First of all, let's load and analyze the dataset:

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

def analyze_transactions(file_path):
    # Load data
    df = pd.read_csv(file_path)
    
    # Basic statistics
    print("Dataset size:", df.shape)
    print("\nMain statistics:")
    print(df.describe())
    
    # Missing values
    print("\nMissing values:")
    print(df.isnull().sum())
    
    return df

df = analyze_transactions('transactions.csv')

Dev: I noticed a lot of anomalous values in the amount and timestamp columns

Claude: Here's a function to identify and analyze the anomalies:

def analyze_outliers(df):
    # Compute IQR for 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)
    ]
    
    # Timestamp analysis
    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

# Plot the results
def plot_anomalies(df, outliers):
    plt.figure(figsize=(12, 6))
    
    plt.subplot(1, 2, 1)
    sns.boxplot(y=df['amount'])
    plt.title('Amount Distribution')
    
    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('Detected Anomalies')
    
    plt.tight_layout()
    plt.show()

Dev: How can I prepare this data for a fraud detection model?

Claude: Here's a feature engineering pipeline for fraud detection:

def engineer_features(df):
    # Time-based features
    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)
    
    # Per-user aggregations
    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 for categories
    cat_features = pd.get_dummies(
        df[['merchant_category', 'transaction_type']],
        prefix=['cat', 'type']
    )
    
    return pd.concat([df, cat_features], axis=1)

Dev: How do I handle class imbalance in the dataset?

Claude: Here are a few techniques for handling imbalanced datasets:

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

def balance_dataset(X, y):
    # Combine over- and 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("Original distribution:", np.bincount(y))
    print("Balanced distribution:", np.bincount(y_resampled))
    
    return X_resampled, y_resampled

Dev: Which model would you suggest for fraud detection?

Claude: For fraud detection, an ensemble approach often works well:

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: How do I interpret the model's results?

Claude: Here are a few techniques for model interpretability:

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: How can I deploy the model to production?

Claude: Here's a structure for deployment with FastAPI and monitoring:

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

# Metrics
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()