Transfer Learning Experiments

1# Cross-dataset pairs to test
2CROSS_DATASET_PAIRS = [
3    # (train_dataset, test_dataset, description)
4    ("FD002", "FD004", "6 conditions → 6 conditions + new fault mode"),
5    ("FD004", "FD002", "6 conditions + 2 faults → 6 conditions + 1 fault"),
6    ("FD001", "FD003", "1 condition → 1 condition + new fault mode"),
7    ("FD003", "FD001", "1 condition + 2 faults → 1 condition + 1 fault"),
8]
9
10# Seeds for statistical validity
11SEEDS = [42, 123, 456]
12
13# AMNL configuration (0.5/0.5 - our best)
14AMNL_CONFIG = {
15    'amnl_weight_rul': 0.5,
16    'amnl_weight_health': 0.5,
17    'use_attention': True,
18    'use_weighted_mse': True,
19    'use_warmup': True,
20    'warmup_epochs': 10,
21    'use_ema': True,
22    'use_adaptive_weight_decay': True,
23    'initial_weight_decay': 1e-4,
24}

1def run_cross_dataset_experiment(train_dataset, test_dataset, seed, output_dir):
2    """Run a single cross-dataset experiment."""
3    set_seed(seed)
4    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
5
6    print(f"  Training on {train_dataset}, Testing on {test_dataset} (seed={seed})")
7
8    # Load TRAINING dataset
9    train_dataset_obj = EnhancedNASACMAPSSDataset(
10        dataset_name=train_dataset,
11        train=True,
12        sequence_length=30,
13        max_rul=125,
14        random_seed=seed,
15        per_condition_norm=False
16    )
17
18    # Source test set (same dataset as training)
19    source_test_dataset = EnhancedNASACMAPSSDataset(
20        dataset_name=train_dataset,
21        train=False,
22        sequence_length=30,
23        max_rul=125,
24        scaler_params=train_dataset_obj.get_scaler_params(),
25        random_seed=seed,
26        per_condition_norm=False
27    )
28
29    # Load TARGET dataset (different dataset for cross-dataset evaluation)
30    target_test_dataset = EnhancedNASACMAPSSDataset(
31        dataset_name=test_dataset,
32        train=False,
33        sequence_length=30,
34        max_rul=125,
35        scaler_params=train_dataset_obj.get_scaler_params(),  # Use source scaler!
36        random_seed=seed,
37        per_condition_norm=False
38    )

1def evaluate_model(model, test_loader, device):
2    """Evaluate model and return metrics."""
3    model.eval()
4    predictions = []
5    targets = []
6
7    with torch.no_grad():
8        for batch_x, batch_y in test_loader:
9            batch_x = batch_x.to(device)
10            rul_pred, _ = model(batch_x)
11            predictions.extend(rul_pred.squeeze().cpu().numpy())
12            targets.extend(batch_y.numpy())
13
14    predictions = np.array(predictions)
15    targets = np.array(targets)
16
17    # Calculate metrics
18    rmse = np.sqrt(np.mean((predictions - targets) ** 2))
19    mae = np.mean(np.abs(predictions - targets))
20
21    # R² score
22    ss_res = np.sum((targets - predictions) ** 2)
23    ss_tot = np.sum((targets - np.mean(targets)) ** 2)
24    r2 = 1 - (ss_res / ss_tot) if ss_tot > 0 else 0
25
26    # NASA Score
27    errors = predictions - targets
28    nasa_score = np.sum(np.where(errors < 0,
29                                  np.exp(-errors / 13) - 1,
30                                  np.exp(errors / 10) - 1))
31
32    return {
33        'rmse': float(rmse),
34        'mae': float(mae),
35        'r2': float(r2),
36        'nasa_score': float(nasa_score),
37        'predictions': predictions,
38        'targets': targets
39    }

1def main():
2    """Main function to run all cross-dataset experiments."""
3    print("=" * 70)
4    print("AMNL Cross-Dataset Generalization Experiment")
5    print("=" * 70)
6
7    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
8
9    # Run experiments
10    all_results = []
11    total_experiments = len(CROSS_DATASET_PAIRS) * len(SEEDS)
12
13    for train_dataset, test_dataset, description in CROSS_DATASET_PAIRS:
14        print(f"\n[{train_dataset} → {test_dataset}] {description}")
15
16        for seed in SEEDS:
17            results = run_cross_dataset_experiment(
18                train_dataset, test_dataset, seed, OUTPUT_DIR
19            )
20            all_results.append(results)
21
22    # Generate summary
23    results_df = pd.DataFrame(all_results)
24
25    # Calculate generalization gap
26    for (train, test), group in results_df.groupby(['train_dataset', 'test_dataset']):
27        src = group['source_rmse'].mean()
28        tgt = group['target_rmse'].mean()
29        gap = group['generalization_gap'].mean()
30        print(f"  {train} → {test}: Source={src:.2f}, Target={tgt:.2f}, Gap={gap:+.2f}")
31
32    avg_gap = results_df['generalization_gap'].mean()
33    print(f"\n🎯 Average Generalization Gap: {avg_gap:.2f} RMSE")
34
35    if avg_gap < 3:
36        print("   ✅ Excellent generalization! Strong evidence for robust feature learning.")
37    elif avg_gap < 5:
38        print("   ✅ Good generalization. Model transfers well to new conditions.")
39    else:
40        print("   ⚠️ Moderate generalization. Some performance drop on unseen data.")