#!/usr/bin/env python3 """ Overfitting Filter Detector — Emergent Function Analysis ======================================================== Identifies neural network filters that exhibit memorization behavior: activating selectively on training examples rather than general features. Detects "emergent functions" that work for particular solutions, not general solutions. Key Metrics: - Activation Specificity: How selective a filter is to training patterns - Train-Test Activation Gap: Difference in activation between train and test - Activation Entropy: Diversity of activations across the dataset - Memorization Score: Composite overfitting indicator Usage: python overfitting_filter_detector.py """ import os import json import math import numpy as np import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader from torchvision import datasets, transforms from collections import defaultdict from scipy import stats # ─── Configuration ──────────────────────────────────────────────────── LR = 0.01 BATCH_SIZE = 128 EPOCHS = 10 DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") SEED = 42 LOG_DIR = "./overfit_filter_logs" MEASURE_INTERVAL = 50 # Measure filter activations every N batches # ─── Filter Activation Tracker ────────────────────────────────────── class FilterActivationTracker: """ Tracks filter activations across training to detect overfitting patterns. Identifies filters that exhibit memorization behavior. """ def __init__(self, model, layer_name="conv2"): """ Initialize tracker for specified model layer. Args: model: PyTorch model layer_name: Name of the convolutional layer to monitor """ self.model = model self.layer_name = layer_name self.hooks = [] # Activation storage self.train_activations = defaultdict(list) # {filter_idx: [activation_stats]} self.test_activations = defaultdict(list) # Pre-computed statistics self.activation_sums = defaultdict(float) self.activation_squared_sums = defaultdict(float) self.activation_counts = defaultdict(int) # Overfitting metrics self.train_test_gaps = defaultdict(float) self.activation_entropies = defaultdict(float) self.specificity_scores = defaultdict(float) self.memorization_scores = defaultdict(float) self._register_hooks() def _register_hooks(self): """Register forward hooks to capture activations.""" for name, module in self.model.named_modules(): if name == self.layer_name and isinstance(module, (nn.Conv2d, nn.Linear)): def hook_fn(module, input, output): self._capture_activations(output, is_train=True) self.hooks.append(module.register_forward_hook(hook_fn)) return def _capture_activations(self, activations, is_train=True): """Process captured activations for each filter.""" # activations shape: (batch_size, num_filters, ...) if len(activations.shape) == 4: # Conv2d: (N, C, H, W) # Global average pooling across spatial dimensions pooled = activations.mean(dim=(2, 3)) # (N, C) elif len(activations.shape) == 2: # Linear: (N, C) pooled = activations else: return for filter_idx in range(pooled.shape[1]): activations_per_filter = pooled[:, filter_idx].cpu().numpy() if is_train: # Track training activations self.train_activations[filter_idx].append(activations_per_filter) self.activation_sums[filter_idx] += np.sum(activations_per_filter) self.activation_squared_sums[filter_idx] += np.sum(activations_per_filter ** 2) self.activation_counts[filter_idx] += len(activations_per_filter) else: self.test_activations[filter_idx].append(activations_per_filter) def compute_overfitting_metrics(self): """ Compute overfitting metrics for each filter. Returns: dict: Overfitting metrics per filter """ metrics = {} for filter_idx in range(self.num_filters): train_acts = np.concatenate(self.train_activations.get(filter_idx, [])) test_acts = np.concatenate(self.test_activations.get(filter_idx, [])) if len(train_acts) == 0 or len(test_acts) == 0: continue # 1. Train-Test Activation Gap train_mean = np.mean(train_acts) test_mean = np.mean(test_acts) gap = abs(train_mean - test_mean) self.train_test_gaps[filter_idx] = gap # 2. Activation Entropy (measures diversity of activations) # Higher entropy = more general, Lower entropy = more specific (memorizing) train_dist = self._compute_activation_distribution(train_acts) test_dist = self._compute_activation_distribution(test_acts) train_entropy = self._shannon_entropy(train_dist) test_entropy = self._shannon_entropy(test_dist) avg_entropy = (train_entropy + test_entropy) / 2 self.activation_entropies[filter_idx] = avg_entropy # 3. Specificity Score (how selective the filter is) # High specificity = only activates on certain patterns specificity = self._compute_specificity(train_acts, test_acts) self.specificity_scores[filter_idx] = specificity # 4. Memorization Score (composite metric) # Combines gap, low entropy, and high specificity mem_score = self._compute_memorization_score(gap, avg_entropy, specificity) self.memorization_scores[filter_idx] = mem_score metrics[filter_idx] = { 'train_mean': float(train_mean), 'test_mean': float(test_mean), 'train_test_gap': float(gap), 'train_entropy': float(train_entropy), 'test_entropy': float(test_entropy), 'avg_entropy': float(avg_entropy), 'specificity': float(specificity), 'memorization_score': float(mem_score), } return metrics def _compute_activation_distribution(self, activations, n_bins=10): """Compute histogram distribution of activations.""" hist, _ = np.histogram(activations, bins=n_bins, range=(-1, 1)) return hist / np.sum(hist) def _shannon_entropy(self, distribution): """Compute Shannon entropy of distribution.""" return -np.sum(distribution * np.log(distribution + 1e-10)) def _compute_specificity(self, train_acts, test_acts, threshold=0.5): """ Compute how specific a filter is to training patterns. Specificity = fraction of test samples that DON'T activate the filter when the filter is active on training samples. Higher value = more specific (memorizing) """ # Find activation threshold from training train_threshold = np.percentile(train_acts, 75) # Count training activations above threshold train_active = np.sum(train_acts > train_threshold) train_active_fraction = train_active / len(train_acts) # Count test activations above same threshold test_active = np.sum(test_acts > train_threshold) test_active_fraction = test_active / len(test_acts) # Specificity: difference in activation rates if train_active_fraction > 0: specificity = 1 - (test_active_fraction / train_active_fraction) else: specificity = 0 return max(0, min(1, specificity)) def _compute_memorization_score(self, gap, entropy, specificity, gap_weight=0.3, entropy_weight=0.4, specificity_weight=0.3): """ Composite memorization score combining multiple metrics. Higher score = more likely to be memorizing (overfitting). """ # Normalize metrics to [0, 1] range # Gap: higher is more overfitting gap_normalized = min(gap / 0.5, 1.0) # Normalize by expected max gap # Entropy: lower is more overfitting (inverse) entropy_normalized = 1 - (entropy / 3.0) # Normalize by max entropy # Specificity: higher is more overfitting specificity_normalized = specificity # Weighted combination score = (gap_weight * gap_normalized + entropy_weight * entropy_normalized + specificity_weight * specificity_normalized) return float(score) def get_overfitting_filters(self, threshold=0.6): """ Get filters that show overfitting behavior. Args: threshold: Memorization score threshold Returns: list: Filter indices that are likely overfitting """ overfitting_filters = [] for filter_idx, score in self.memorization_scores.items(): if score > threshold: overfitting_filters.append(filter_idx) return overfitting_filters def get_filter_ranking(self): """ Rank filters by memorization score (descending). Returns: list: Sorted list of (filter_idx, score) tuples """ return sorted( [(idx, score) for idx, score in self.memorization_scores.items()], key=lambda x: x[1], reverse=True ) def cleanup(self): """Remove hooks and free memory.""" for hook in self.hooks: hook.remove() self.hooks = [] # ─── CNN Model for MNIST ──────────────────────────────────────────── class MNISTCNN(nn.Module): """Simple CNN for MNIST with identifiable convolutional layers.""" def __init__(self): super().__init__() self.conv1 = nn.Conv2d(1, 16, 3, padding=1) self.conv2 = nn.Conv2d(16, 32, 3, padding=1) self.pool = nn.MaxPool2d(2, 2) self.fc1 = nn.Linear(32 * 7 * 7, 128) self.fc2 = nn.Linear(128, 10) self.relu = nn.ReLU() def forward(self, x): x = self.pool(self.relu(self.conv1(x))) # (N, 16, 14, 14) x = self.pool(self.relu(self.conv2(x))) # (N, 32, 7, 7) x = x.view(-1, 32 * 7 * 7) x = self.relu(self.fc1(x)) x = self.fc2(x) return x # ─── Training Loop ──────────────────────────────────────────────────── def train_with_filter_monitoring(): """Train model while monitoring filter activations for overfitting.""" print(f"[Overfitting Filter Detector] Device: {DEVICE}") print(f"[Overfitting Filter Detector] LR={LR}, BS={BATCH_SIZE}, Epochs={EPOCHS}") print(f"[Overfitting Filter Detector] Log dir: {LOG_DIR}") print("=" * 80) os.makedirs(LOG_DIR, exist_ok=True) # Data transform = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)), ]) train_ds = datasets.MNIST("./data", train=True, download=True, transform=transform) test_ds = datasets.MNIST("./data", train=False, transform=transform) train_loader = DataLoader(train_ds, batch_size=BATCH_SIZE, shuffle=True, drop_last=True) test_loader = DataLoader(test_ds, batch_size=BATCH_SIZE, shuffle=False) # Model model = MNISTCNN().to(DEVICE) optimizer = optim.SGD(model.parameters(), lr=LR, momentum=0.9) criterion = nn.CrossEntropyLoss() # Filter activation tracker tracker = FilterActivationTracker(model, layer_name="conv2") print(f"[Overfitting Filter Detector] Tracking conv2 activations") # Logging overall_start = time.time() step_count = 0 for epoch in range(EPOCHS): model.train() epoch_loss = 0.0 correct = 0 total = 0 for batch_idx, (data, target) in enumerate(train_loader): data, target = data.to(DEVICE), target.to(DEVICE) optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() # Track activations periodically if step_count % MEASURE_INTERVAL == 0: tracker._capture_activations( model.conv2(model.relu(model.conv1(data))), is_train=True ) # Accumulate metrics epoch_loss += loss.item() _, pred = output.max(1) correct += pred.eq(target).sum().item() total += target.size(0) step_count += 1 # Test set evaluation model.eval() test_correct = 0 test_total = 0 test_loss = 0.0 with torch.no_grad(): for data, target in test_loader: data, target = data.to(DEVICE), target.to(DEVICE) output = model(data) test_loss += criterion(output, target).item() _, pred = output.max(1) test_correct += pred.eq(target).sum().item() test_total += target.size(0) # Track test activations if step_count % MEASURE_INTERVAL == 0: tracker._capture_activations( model.conv2(model.relu(model.conv1(data))), is_train=False ) # Compute training metrics train_acc = correct / total test_acc = test_correct / test_total avg_train_loss = epoch_loss / len(train_loader) avg_test_loss = test_loss / len(test_loader) print(f"\nEpoch {epoch+1}:") print(f" Train Loss: {avg_train_loss:.4f} | Train Acc: {train_acc:.4f}") print(f" Test Loss: {avg_test_loss:.4f} | Test Acc: {test_acc:.4f}") print(f" Gap: {avg_train_loss - avg_test_loss:.4f}") # Compute overfitting metrics every few epochs if epoch % 2 == 1 or epoch == EPOCHS - 1: metrics = tracker.compute_overfitting_metrics() if metrics: print(f"\n === Filter Overfitting Analysis (Conv2) ===") print(f" Total Filters: {len(tracker.memorization_scores)}") # Get overfitting filters overfitting = tracker.get_overfitting_filters(threshold=0.6) print(f" Overfitting Filters (score > 0.6): {len(overfitting)}") # Show top 5 most memorizing filters print(f"\n Top 5 Memorizing Filters:") for i, (filter_idx, score) in enumerate(tracker.get_filter_ranking()[:5]): m = metrics.get(filter_idx, {}) print(f" Filter {filter_idx:2d}: MemScore={score:.3f} " f"| Gap={m.get('train_test_gap', 0):.4f} " f"| Entropy={m.get('avg_entropy', 0):.3f} " f"| Specificity={m.get('specificity', 0):.3f}") print() total_time = time.time() - overall_start print(f"[Overfitting Filter Detector] Training complete in {total_time:.1f}s") # Final analysis print("\n=== Final Overfitting Filter Analysis ===") final_metrics = tracker.compute_overfitting_metrics() if final_metrics: # Save metrics metrics_file = os.path.join(LOG_DIR, "filter_overfitting_metrics.json") with open(metrics_file, "w") as f: json.dump(final_metrics, f, indent=2) print(f" Metrics saved to: {metrics_file}") # Identify overfitting filters overfitting = tracker.get_overfitting_filters(threshold=0.6) print(f" Overfitting Filters: {overfitting}") # Print all filter rankings print(f"\n Complete Filter Ranking:") for filter_idx, score in tracker.get_filter_ranking(): m = final_metrics.get(filter_idx, {}) status = "⚠️ OVERFIT" if score > 0.6 else "✓ OK" print(f" Filter {filter_idx:2d}: MemScore={score:.3f} " f"| Gap={m.get('train_test_gap', 0):.4f} " f"| Entropy={m.get('avg_entropy', 0):.3f} {status}") tracker.cleanup() return model, final_metrics # ─── Main ───────────────────────────────────────────────────────────── if __name__ == "__main__": model, metrics = train_with_filter_monitoring()