scripts/feature_ablation.py

import os
import sys
import numpy as np
import pandas as pd
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from sklearn.calibration import CalibratedClassifierCV
import xgboost as xgb

sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
from train import gather_data
from features import get_feature_names

def run_ablation():
    feature_groups = {
        'brightness_color': ['brightness', 'contrast', 'saturation'],
        'blur_sharpness': ['laplacian_var', 'sobel_mean', 'edge_density'],
        'fft_global_freq': ['fft_hf_ratio', 'h_freq_peak', 'v_freq_peak', 'diag_freq_peak'],
        'local_patch_fft': ['local_fft_hf'],
        'moire_banding': ['moire_score', 'banding_score'],
        'jpeg_compression': ['compression_diff', 'blockiness'],
        'bezel_border': ['bezel_score', 'rect_contour_score'],
        'perspective_rect': ['perspective_score'],
        'glare_overexposure': ['glare_ratio', 'glare_patch_size'],
        'printout_paper': ['paper_texture']
    }
    
    data_dir = os.path.join(os.path.dirname(__file__), '..', 'dataset')
    phone_dir = os.path.join(os.path.dirname(__file__), '..', 'dataset', 'my_photos')
    
    print("Loading datasets...")
    X_tr_g, y_tr_g, _ = gather_data(data_dir)
    X_phone, y_phone, _ = gather_data(phone_dir)
    
    f_names = get_feature_names()
    
    results = []
    
    def eval_model(X_train, y_train, X_test, y_test, disabled_indices=[]):
        X_train_sub = np.delete(X_train, disabled_indices, axis=1)
        X_test_sub = np.delete(X_test, disabled_indices, axis=1)
        
        ratio = float(np.sum(y_train == 0)) / np.sum(y_train == 1) if np.sum(y_train == 1) > 0 else 1.0
        base = xgb.XGBClassifier(n_estimators=100, max_depth=4, reg_lambda=10, random_state=42, scale_pos_weight=ratio, tree_method='hist', device='cpu')
        model = CalibratedClassifierCV(base, method='sigmoid', cv=3)
        model.fit(X_train_sub, y_train)
        
        probs = model.predict_proba(X_test_sub)[:, 1]
        
        best_th, best_f1 = 0.5, 0
        for th in np.arange(0.1, 0.9, 0.05):
            f1 = f1_score(y_test, (probs >= th).astype(int), zero_division=0)
            if f1 > best_f1:
                best_f1, best_th = f1, th
                
        preds = (probs >= best_th).astype(int)
        acc = accuracy_score(y_test, preds)
        prec = precision_score(y_test, preds, zero_division=0)
        rec = recall_score(y_test, preds, zero_division=0)
        
        fp = np.sum((preds == 1) & (y_test == 0))
        fn = np.sum((preds == 0) & (y_test == 1))
        
        return acc, prec, rec, best_f1, fp, fn
        
    def cv_eval(disabled_indices):
        skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
        metrics = []
        for tr_idx, te_idx in skf.split(X_phone, y_phone):
            X_comb = np.vstack([X_tr_g, X_phone[tr_idx]])
            y_comb = np.concatenate([y_tr_g, y_phone[tr_idx]])
            
            acc, prec, rec, f1, fp, fn = eval_model(X_comb, y_comb, X_phone[te_idx], y_phone[te_idx], disabled_indices)
            metrics.append([acc, prec, rec, f1, fp, fn])
        
        return np.mean(metrics, axis=0)
        
    print("Evaluating Baseline...")
    base_metrics = cv_eval([])
    results.append({
        'Ablated Group': 'None (Baseline)',
        'Accuracy': base_metrics[0],
        'Precision': base_metrics[1],
        'Recall': base_metrics[2],
        'F1': base_metrics[3],
        'False Positives': base_metrics[4] * 5, 
        'False Negatives': base_metrics[5] * 5
    })
    
    for group, features in feature_groups.items():
        print(f"Evaluating without {group}...")
        indices = [f_names.index(f) for f in features if f in f_names]
        metrics = cv_eval(indices)
        results.append({
            'Ablated Group': group,
            'Accuracy': metrics[0],
            'Precision': metrics[1],
            'Recall': metrics[2],
            'F1': metrics[3],
            'False Positives': metrics[4] * 5,
            'False Negatives': metrics[5] * 5
        })
        
    df = pd.DataFrame(results)
    
    os.makedirs(os.path.join(os.path.dirname(__file__), '..', 'reports'), exist_ok=True)
    df.to_csv(os.path.join(os.path.dirname(__file__), '..', 'reports', 'feature_ablation.csv'), index=False)
    
    with open(os.path.join(os.path.dirname(__file__), '..', 'reports', 'feature_ablation_summary.md'), 'w') as f:
        f.write("# Feature Ablation Summary\n\n")
        f.write(df.to_markdown(index=False))
        f.write("\n")

if __name__ == "__main__":
    run_ablation()