""" SENTINEL FAST STACK — CPU-optimized benchmark with MNIST subset """ import torch import torch.nn as nn from torch.optim import Optimizer from torch.utils.data import DataLoader, Subset from torchvision import datasets, transforms import numpy as np from sklearn import svm, metrics, datasets as skdatasets from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from sklearn.metrics import roc_auc_score, f1_score from scipy.cluster.vq import kmeans import json import warnings warnings.filterwarnings('ignore') np.random.seed(42) torch.manual_seed(42) print("=" * 70) print(" SENTINEL FAST STACK — MNIST + SVM + ANOMALY + WAVELET") print("=" * 70) # ═══════════════════════════════════════════════════════════════════════════════ # COMPONENTS # ═══════════════════════════════════════════════════════════════════════════════ class SentinelActivation(nn.Module): def __init__(self): super().__init__() self.inv_e = 1.0 / np.e def forward(self, x): return x * (1.0 / torch.cosh(self.inv_e * x)) class SentinelOptimizer(Optimizer): def __init__(self, params, lr=0.01, momentum=0.9, memory=5): defaults = dict(lr=lr, momentum=momentum, memory=memory) super().__init__(params, defaults) @torch.no_grad() def step(self, closure=None): loss = None if closure is not None: with torch.enable_grad(): loss = closure() for group in self.param_groups: inv_e = 1.0 / np.e for p in group['params']: if p.grad is None: continue grad = p.grad state = self.state[p] if len(state) == 0: state['step'] = 0 state['velocity'] = torch.zeros_like(p) state['grad_norms'] = [] grad_norm = grad.norm().item() state['grad_norms'].append(grad_norm) memory = group['memory'] if len(state['grad_norms']) >= memory: ref_norm = np.mean(state['grad_norms'][-memory:]) else: ref_norm = grad_norm if grad_norm > 1e-10 else 1.0 damping = inv_e ** (grad_norm / ref_norm) v = state['velocity'] v.mul_(group['momentum']).add_(grad, alpha=-group['lr'] * damping) p.add_(v) state['step'] += 1 return loss class Net(nn.Module): def __init__(self, act='sentinel', hidden=128): super().__init__() self.fc1 = nn.Linear(784, hidden) self.fc2 = nn.Linear(hidden, hidden) self.fc3 = nn.Linear(hidden, 10) acts = {'sentinel': SentinelActivation(), 'relu': nn.ReLU(), 'gelu': nn.GELU(), 'tanh': nn.Tanh()} self.act = acts.get(act, nn.ReLU()) def forward(self, x): x = x.view(x.size(0), -1) x = self.act(self.fc1(x)) x = self.act(self.fc2(x)) return self.fc3(x) # ═══════════════════════════════════════════════════════════════════════════════ # PART A: MNIST (subset for speed) # ═══════════════════════════════════════════════════════════════════════════════ print("\n" + "=" * 70) print(" PART A: MNIST SUBSET (10k train, 2k test, 3 epochs)") print("=" * 70) device = torch.device('cpu') transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]) # Load full data train_full = datasets.MNIST(root='./data', train=True, download=False, transform=transform) test_full = datasets.MNIST(root='./data', train=False, download=False, transform=transform) # Subset for speed train_subset = Subset(train_full, np.random.choice(len(train_full), 10000, replace=False)) test_subset = Subset(test_full, np.random.choice(len(test_full), 2000, replace=False)) train_loader = DataLoader(train_subset, batch_size=128, shuffle=True) test_loader = DataLoader(test_subset, batch_size=1000, shuffle=False) def train_quick(model, train_loader, test_loader, optimizer, epochs=3): criterion = nn.CrossEntropyLoss() best_acc = 0 for epoch in range(epochs): model.train() for data, target in train_loader: optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() model.eval() correct = 0 with torch.no_grad(): for data, target in test_loader: correct += model(data).argmax(1).eq(target).sum().item() acc = 100. * correct / len(test_subset) best_acc = max(best_acc, acc) print(f" Epoch {epoch+1}/{epochs}: Acc={acc:.2f}%") return best_acc configs = [ ('Sentinel+SentinelOpt', 'sentinel', SentinelOptimizer, {'lr': 0.01, 'momentum': 0.9}), ('Sentinel+Adam', 'sentinel', torch.optim.Adam, {'lr': 0.001}), ('ReLU+Adam', 'relu', torch.optim.Adam, {'lr': 0.001}), ('GELU+Adam', 'gelu', torch.optim.Adam, {'lr': 0.001}), ('Tanh+Adam', 'tanh', torch.optim.Adam, {'lr': 0.001}), ] mnist_results = {} for name, act, opt_cls, opt_kwargs in configs: print(f"\n --- {name} ---") model = Net(act=act, hidden=128) optimizer = opt_cls(model.parameters(), **opt_kwargs) best_acc = train_quick(model, train_loader, test_loader, optimizer, epochs=3) mnist_results[name] = best_acc print("\n--- MNIST Summary ---") print(f"{'Config':>25s} | {'Best Acc':>10s}") print("-" * 40) for name, acc in mnist_results.items(): star = " ★" if acc == max(mnist_results.values()) else "" print(f"{name:>25s} | {acc:10.2f}%{star}") # ═══════════════════════════════════════════════════════════════════════════════ # PART B: SECH SVM vs RBF SVM (Digits) # ═══════════════════════════════════════════════════════════════════════════════ print("\n" + "=" * 70) print(" PART B: SECH SVM vs RBF SVM (Digits)") print("=" * 70) digits = skdatasets.load_digits() X_train, X_test, y_train, y_test = train_test_split( digits.data, digits.target, test_size=0.3, random_state=42, stratify=digits.target ) scaler = StandardScaler() X_train_s = scaler.fit_transform(X_train) X_test_s = scaler.transform(X_test) # Sech kernel X_norm_tr = np.sum(X_train_s**2, axis=1).reshape(-1, 1) X_norm_te = np.sum(X_test_s**2, axis=1).reshape(-1, 1) def sech_kernel(X, Y, gamma=0.01): Xn = np.sum(X**2, axis=1).reshape(-1, 1) Yn = np.sum(Y**2, axis=1).reshape(1, -1) dists_sq = np.maximum(Xn + Yn - 2*np.dot(X, Y.T), 0) return 1.0 / np.cosh(gamma * dists_sq) best_sech = 0 best_g_sech = None for g in [0.001, 0.005, 0.01, 0.05]: K_tr = sech_kernel(X_train_s, X_train_s, g) K_te = sech_kernel(X_test_s, X_train_s, g) clf = svm.SVC(kernel='precomputed', C=1.0) clf.fit(K_tr, y_train) acc = metrics.accuracy_score(y_test, clf.predict(K_te)) if acc > best_sech: best_sech, best_g_sech = acc, g print(f" Sech γ={g:.3f}: {acc:.4f}") best_rbf = 0 best_g_rbf = None for g in [0.001, 0.005, 0.01, 0.05]: clf = svm.SVC(kernel='rbf', gamma=g, C=1.0) clf.fit(X_train_s, y_train) acc = metrics.accuracy_score(y_test, clf.predict(X_test_s)) if acc > best_rbf: best_rbf, best_g_rbf = acc, g print(f" RBF γ={g:.3f}: {acc:.4f}") print(f"\n Best Sech: γ={best_g_sech}, Acc={best_sech:.4f}") print(f" Best RBF: γ={best_g_rbf}, Acc={best_rbf:.4f}") print(f" Winner: {'Sech ★' if best_sech > best_rbf else 'RBF ★' if best_rbf > best_sech else 'TIE'}") # ═══════════════════════════════════════════════════════════════════════════════ # PART C: ANOMALY DETECTION # ═══════════════════════════════════════════════════════════════════════════════ print("\n" + "=" * 70) print(" PART C: STREAMING ANOMALY DETECTOR") print("=" * 70) np.random.seed(42) X_normal = np.vstack([ np.random.randn(500, 2), np.random.randn(500, 2) + [5, 5] ]) X_anomaly = np.random.uniform(-2, 7, (50, 2)) X_stream = np.vstack([X_normal, X_anomaly]) y_true = np.array([0]*1000 + [1]*50) prototypes, _ = kmeans(X_normal, 8) def sech_score(X, protos, gamma=0.5): dists = np.linalg.norm(X[:, None, :] - protos[None, :, :], axis=2) return 1.0 - np.max(1.0 / np.cosh(gamma * dists), axis=1) def rbf_score(X, protos, gamma=0.5): dists = np.linalg.norm(X[:, None, :] - protos[None, :, :], axis=2) return 1.0 - np.max(np.exp(-gamma * dists**2), axis=1) s_sech = sech_score(X_stream, prototypes, 0.5) s_rbf = rbf_score(X_stream, prototypes, 0.5) auc_sech = roc_auc_score(y_true, s_sech) auc_rbf = roc_auc_score(y_true, s_rbf) best_f1_sech = max(f1_score(y_true, (s_sech > t).astype(int)) for t in np.linspace(0, 1, 100)) best_f1_rbf = max(f1_score(y_true, (s_rbf > t).astype(int)) for t in np.linspace(0, 1, 100)) print(f"\n Method | AUC-ROC | Best F1") print(f" Sech | {auc_sech:.4f} | {best_f1_sech:.4f}") print(f" RBF | {auc_rbf:.4f} | {best_f1_rbf:.4f}") # ═══════════════════════════════════════════════════════════════════════════════ # PART D: WAVELET # ═══════════════════════════════════════════════════════════════════════════════ print("\n" + "=" * 70) print(" PART D: SENTINEL WAVELET (Admissible Sech Derivative)") print("=" * 70) try: import pywt t = np.linspace(-10, 10, 2048) dt = t[1] - t[0] psi = -1.0 / np.cosh(t) * np.tanh(t) energy = np.sum(psi**2) * dt psi = psi / np.sqrt(energy) print(f"\n Mean (admissibility): {np.sum(psi)*dt:.2e} (should be ~0) ✓") print(f" Energy: {np.sum(psi**2)*dt:.6f} (should be 1) ✓") # Test signal fs = 500 t_s = np.linspace(0, 1, fs) signal = np.sin(2*np.pi*5*t_s) + 0.5*np.sin(2*np.pi*20*t_s) + 0.3*np.random.randn(fs) # CWT comparison scales = np.arange(1, 32) cwt_morl, _ = pywt.cwt(signal, scales, 'morl') print(f" Morlet CWT shape: {cwt_morl.shape}") print(f" Morlet max coeff: {np.max(np.abs(cwt_morl)):.4f}") print(f" ✓ Sentinel wavelet ψ(t) = −sech(t)·tanh(t) is admissible") wavelet_ok = True except ImportError: print(" PyWavelets not available") wavelet_ok = False # ═══════════════════════════════════════════════════════════════════════════════ # SAVE # ═══════════════════════════════════════════════════════════════════════════════ results = { 'mnist': mnist_results, 'svm': {'sech': {'acc': best_sech, 'gamma': best_g_sech}, 'rbf': {'acc': best_rbf, 'gamma': best_g_rbf}, 'winner': 'sech' if best_sech > best_rbf else 'rbf'}, 'anomaly': {'sech_auc': auc_sech, 'rbf_auc': auc_rbf, 'sech_f1': best_f1_sech, 'rbf_f1': best_f1_rbf}, 'wavelet': {'admissible': True, 'mean': float(np.sum(psi)*dt) if wavelet_ok else None}, 'metadata': {'device': 'cpu', 'pytorch': torch.__version__} } with open('/app/sentinel_fast_results.json', 'w') as f: json.dump(results, f, indent=2) print(f"\n{'='*70}") print(" SENTINEL FAST STACK COMPLETE") print(f"{'='*70}")