SOFAM/cross_validation.py at main · cwccie/SOFAM

261 lines (221 loc) · 9.55 KB
Block 3, Step 3: 5-Fold Stratified Cross-Validation.
Compares SOFAM against LR, RF, GB on 30K subset.
Reports mean +/- std for all metrics.
Dr. Sethi requested cross-validation detail.
import json
import time
import numpy as np
import torch
os.chdir(os.path.dirname(os.path.abspath(__file__)))
from dataset_cache import load_cached
from sklearn.model_selection import StratifiedKFold
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.metrics import (accuracy_score, precision_score, recall_score,
                             f1_score, matthews_corrcoef, cohen_kappa_score)
import gqfam
from gqfam import (Config, FAM, MapField, run_fam_with_parameters)
# Config for CV runs — keep it manageable
Config.DATASET_MAX_ROWS = None
Config.GENERATIONS = 3
Config.POPULATION_SIZE = 3
Config.HEURISTIC_GENERATION = 3
Config.HEURISTIC_POPULATION = 3
Config.MAX_NODES = 800
Config.BATCH_SIZE = 128
device = Config.device
N_FOLDS = 5
SUBSET_SIZE = 10000
def compute_metrics(y_true, y_pred):
    """Compute all metrics for a fold."""
    return {
        "Accuracy": float(accuracy_score(y_true, y_pred)),
        "Precision": float(precision_score(y_true, y_pred, average='weighted', zero_division=0)),
        "Recall": float(recall_score(y_true, y_pred, average='weighted', zero_division=0)),
        "F1": float(f1_score(y_true, y_pred, average='weighted', zero_division=0)),
        "MCC": float(matthews_corrcoef(y_true, y_pred)),
        "Kappa": float(cohen_kappa_score(y_true, y_pred)),
def evaluate_fam(fam_model, map_field_model, X_data, device):
    """Get predictions from a trained FAM model."""
    predictions = []
    with torch.no_grad():
        for i in range(len(X_data)):
            sample = X_data[i].to(device) if isinstance(X_data[i], torch.Tensor) else torch.tensor(X_data[i], dtype=torch.float32, device=device)
            J, _ = fam_model.find_matching_category(sample, 0.0)
            if J is not None:
                pred_output = map_field_model.predict(J)
                pred_label = np.argmax(pred_output) if isinstance(pred_output, np.ndarray) else pred_output
                predictions.append(pred_label)
            else:
                predictions.append(0)
    return np.array(predictions)
def run_sofam_fold(X_train_cc, y_train, X_test_cc, y_test, le, fold_num):
    """Run SOFAM (baseline + optimized) on one fold."""
    num_features = X_train_cc.shape[1]
    num_categories = len(np.unique(y_train))
    # Use 20% of train as validation for tuning
    n_val = max(100, int(len(y_train) * 0.2))
    perm = np.random.permutation(len(y_train))
    val_idx = perm[:n_val]
    train_idx = perm[n_val:]
    X_tr = torch.tensor(X_train_cc[train_idx], dtype=torch.float32, device=device)
    y_tr = torch.tensor(y_train[train_idx], dtype=torch.long, device=device)
    X_vl = torch.tensor(X_train_cc[val_idx], dtype=torch.float32, device=device)
    y_vl = torch.tensor(y_train[val_idx], dtype=torch.long)
    X_te = torch.tensor(X_test_cc, dtype=torch.float32, device=device)
    # Baseline
    bl_metrics, bl_fam, bl_mf = run_fam_with_parameters(
        X_train=X_tr, y_train=y_tr,
        X_validation=X_vl, y_validation=y_vl,
        num_features=num_features, num_categories=num_categories,
        learning_rate=Config.BASELINE_LR, vigilance=Config.BASELINE_VIG,
        label_encoder=le, device=device
    bl_preds = evaluate_fam(bl_fam, bl_mf, X_te, device)
    bl_test_metrics = compute_metrics(y_test, bl_preds)
    # Quick HA (3x3 grid)
    best_ha = {"accuracy": 0, "lr": Config.BASELINE_LR, "vig": Config.BASELINE_VIG}
    lr_values = np.linspace(Config.MIN_LEARNING_RATE, Config.MAX_LEARNING_RATE, Config.HEURISTIC_GENERATION)
    vig_values = np.linspace(Config.MIN_VIGILANCE, Config.MAX_VIGILANCE, Config.HEURISTIC_POPULATION)
    for lr in lr_values:
        for vig in vig_values:
            metrics, fam, mf = run_fam_with_parameters(
                X_train=X_tr, y_train=y_tr,
                X_validation=X_vl, y_validation=y_vl,
                num_features=num_features, num_categories=num_categories,
                learning_rate=lr, vigilance=vig,
                label_encoder=le, device=device
            if metrics and metrics['Accuracy'] > best_ha['accuracy']:
                best_ha = {"accuracy": metrics['Accuracy'], "lr": lr, "vig": vig, "fam": fam, "mf": mf}
    # Optimized test
    if best_ha.get('fam'):
        opt_preds = evaluate_fam(best_ha['fam'], best_ha['mf'], X_te, device)
        opt_preds = bl_preds
    opt_test_metrics = compute_metrics(y_test, opt_preds)
    return {
        "Baseline": bl_test_metrics,
        "Optimized": opt_test_metrics,
        "best_lr": float(best_ha['lr']),
        "best_vig": float(best_ha['vig']),
def main():
    print("=" * 60)
    print(f"5-FOLD STRATIFIED CROSS-VALIDATION (10K subset)")
    print(f"  Models: SOFAM, LR, RF, GB")
    print(f"  Folds: {N_FOLDS}, Seed: {SEED}")
    print("=" * 60)
    # Load cached dataset
    X_scaled, X_complement, y, le = load_cached()
    # Take a 30K stratified subset for CV
    np.random.seed(SEED)
    idx_all = np.arange(len(y))
    # Stratified subsample
    from sklearn.model_selection import train_test_split
    if SUBSET_SIZE < len(y):
        idx_subset, _ = train_test_split(idx_all, train_size=SUBSET_SIZE, random_state=SEED, stratify=y)
        idx_subset = idx_all
    X_cc_sub = X_complement[idx_subset]
    X_raw_sub = X_scaled[idx_subset]
    y_sub = y[idx_subset]
    print(f"Subset: {len(y_sub)} samples")
    print(f"  Class dist: {dict(zip(*np.unique(y_sub, return_counts=True)))}")
    skf = StratifiedKFold(n_splits=N_FOLDS, shuffle=True, random_state=SEED)
    # Collectors
    all_fold_results = {
        "SOFAM_Baseline": [],
        "SOFAM_Optimized": [],
        "LogisticRegression": [],
        "RandomForest": [],
        "GradientBoosting": [],
    for fold_num, (train_idx, test_idx) in enumerate(skf.split(X_cc_sub, y_sub), 1):
        print(f"\n{'='*60}")
        print(f"FOLD {fold_num}/{N_FOLDS}")
        print(f"  Train: {len(train_idx)}, Test: {len(test_idx)}")
        print(f"{'='*60}")
        X_train_cc = X_cc_sub[train_idx]
        X_test_cc = X_cc_sub[test_idx]
        X_train_raw = X_raw_sub[train_idx]
        X_test_raw = X_raw_sub[test_idx]
        y_train = y_sub[train_idx]
        y_test = y_sub[test_idx]
        # --- SOFAM ---
        print(f"  Running SOFAM...")
        t0 = time.time()
        sofam_results = run_sofam_fold(X_train_cc, y_train, X_test_cc, y_test, le, fold_num)
        t1 = time.time()
        all_fold_results["SOFAM_Baseline"].append(sofam_results["Baseline"])
        all_fold_results["SOFAM_Optimized"].append(sofam_results["Optimized"])
        print(f"  SOFAM Fold {fold_num}: Baseline Acc={sofam_results['Baseline']['Accuracy']:.4f}, "
              f"Optimized Acc={sofam_results['Optimized']['Accuracy']:.4f} ({t1-t0:.1f}s)")
        # --- Sklearn baselines (on raw features, not complement coded) ---
        for name, clf in [
            ("LogisticRegression", LogisticRegression(max_iter=1000, random_state=SEED)),
            ("RandomForest", RandomForestClassifier(n_estimators=100, random_state=SEED, n_jobs=-1)),
            ("GradientBoosting", GradientBoostingClassifier(n_estimators=100, random_state=SEED)),
            print(f"  Running {name}...")
            t0 = time.time()
            clf.fit(X_train_raw, y_train)
            y_pred = clf.predict(X_test_raw)
            fold_metrics = compute_metrics(y_test, y_pred)
            t1 = time.time()
            all_fold_results[name].append(fold_metrics)
            print(f"  {name} Fold {fold_num}: Acc={fold_metrics['Accuracy']:.4f} MCC={fold_metrics['MCC']:.4f} ({t1-t0:.1f}s)")
    # --- Aggregate results ---
    print("\n" + "=" * 70)
    print("CROSS-VALIDATION RESULTS (Mean +/- Std)")
    print("=" * 70)
    summary = {}
    metric_names = ["Accuracy", "Precision", "Recall", "F1", "MCC", "Kappa"]
    print(f"{'Model':<22} {'Accuracy':>14} {'MCC':>14} {'F1':>14} {'Kappa':>14}")
    print("-" * 78)
    for model_name, fold_list in all_fold_results.items():
        model_summary = {}
        for metric in metric_names:
            values = [f[metric] for f in fold_list]
            model_summary[metric] = {
                "mean": float(np.mean(values)),
                "std": float(np.std(values)),
                "values": [float(v) for v in values],
        summary[model_name] = model_summary
        acc = model_summary["Accuracy"]
        mcc = model_summary["MCC"]
        f1 = model_summary["F1"]
        kap = model_summary["Kappa"]
        print(f"{model_name:<22} {acc['mean']:.4f}+/-{acc['std']:.4f} "
              f"{mcc['mean']:.4f}+/-{mcc['std']:.4f} "
              f"{f1['mean']:.4f}+/-{f1['std']:.4f} "
              f"{kap['mean']:.4f}+/-{kap['std']:.4f}")
    output = {
        "meta": {
            "n_folds": N_FOLDS,
            "subset_size": SUBSET_SIZE,
            "actual_size": len(y_sub),
            "seed": SEED,
            "sofam_config": {
                "generations": Config.GENERATIONS,
                "population": Config.POPULATION_SIZE,
                "max_nodes": Config.MAX_NODES,
        "summary": summary,
        "per_fold": {k: v for k, v in all_fold_results.items()},
    outfile = f"CrossValidation_Results_{int(time.time())}.json"
    with open(outfile, 'w') as f:
        json.dump(output, f, indent=2)
    print(f"\nResults saved to {outfile}")
if __name__ == "__main__":
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

cross_validation.py

Latest commit

History

cross_validation.py

File metadata and controls
