import torch import torch.nn as nn import torch.optim as optim from torchvision import datasets, transforms from torch.utils.data import DataLoader from collections import deque import math # --------------------------------------------------------- # 1. Custom Optimizer: Multi-Point Derivative Object Updates # --------------------------------------------------------- class GeometricDerivativeOptimizer(optim.Optimizer): """ An optimizer utilizing features from the 32 Derivative Objects text: - Hessian Spectrum Diamond: Adaptive scaling per dimension based on curvature variance. - Itô Strip: Balancing deterministic drift and random diffusion. - Caputo Fractional Memory: Power-law decay history matching fractional calculus. """ def __init__(self, params, lr=1e-3, alpha=0.75, drift_coef=0.9, diff_coef=0.01, memory_len=5): if not 0 < alpha <= 1: raise ValueError(f"Fractional order alpha must be in (0, 1], got {alpha}") defaults = dict(lr=lr, alpha=alpha, drift_coef=drift_coef, diff_coef=diff_coef, memory_len=memory_len) super(GeometricDerivativeOptimizer, self).__init__(params, defaults) # Initialize memory structures for fractional arcs for group in self.param_groups: for p in group['params']: state = self.state[p] state['step'] = 0 # Store update history for Caputo fractional memory kernel state['history'] = deque(maxlen=memory_len) @torch.no_grad() def step(self, closure=None): loss = None if closure != None: with torch.enable_grad(): loss = closure() for group in self.param_groups: lr = group['lr'] alpha = group['alpha'] drift_coef = group['drift_coef'] diff_coef = group['diff_coef'] memory_len = group['memory_len'] for p in group['params']: if p.grad is None: continue grad = p.grad state = self.state[p] state['step'] += 1 # --- CONCEPT 7: HESSIAN SPECTRUM DIAMOND --- # We track an online running estimate of directional variance/curvature # to scale updates differently across landscape dimensions. if 'curvature_buffer' not in state: state['curvature_buffer'] = torch.zeros_like(grad) # Update exponential tracking of gradient magnitude as local landscape shape state['curvature_buffer'].mul_(0.99).add_(grad.pow(2), alpha=0.01) # Diamond scaling factor: avoids division by zero, dampens high-frequency jitter diamond_scale = 1.0 / (torch.sqrt(state['curvature_buffer']) + 1e-8) # --- CONCEPT 21: ITÔ STRIP --- # Split the instantaneous derivative into deterministic drift and noise diffusion drift = grad * diamond_scale diffusion = torch.randn_like(grad) * diff_coef # Combine into a single stochastic geometric derivative step ito_update = (drift_coef * drift) + ((1.0 - drift_coef) * diffusion) # --- CONCEPT 26: CAPUTO FRACTIONAL ARC (MEMORY) --- # Append current step update to our memory queue state['history'].append(ito_update.clone()) # Compute Riemann-Liouville / Caputo fractional memory summation # Kernel weight scales with t^(1 - alpha) as power-law memory scaling fractional_step = torch.zeros_like(grad) history_size = len(state['history']) for i, past_update in enumerate(state['history']): # Calculate depth distance from current moment k = history_size - i # Power law coefficient modeling memory weight weight = 1.0 / math.pow(k, 1.0 - alpha) fractional_step.add_(past_update, alpha=weight) # Average the historical memory tensor effect fractional_step.div_(history_size) # Apply final multi-point calibrated geometric update to parameters p.add_(fractional_step, alpha=-lr) return loss # --------------------------------------------------------- # 2. Model Architecture, Data Pipelines, & Training # --------------------------------------------------------- class MNIST_MLP(nn.Module): def __init__(self): super(MNIST_MLP, self).__init__() self.pipeline = nn.Sequential( nn.Flatten(), nn.Linear(28 * 28, 256), nn.ReLU(), nn.Dropout(0.2), nn.Linear(256, 10) ) def forward(self, x): return self.pipeline(x) def main(): # Hardware Configuration device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"Training ongoing using device environment: {device}") # Standard MNIST Data Loading Pipelines transform = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ]) train_dataset = datasets.MNIST(root='../data', train=True, download=True, transform=transform) test_dataset = datasets.MNIST(root='../data', train=False, download=True, transform=transform) train_loader = DataLoader(train_dataset, batch_size=128, shuffle=True) test_loader = DataLoader(test_dataset, batch_size=1000, shuffle=False) # Initialize Neural Network & Custom Derivative Optimizer model = MNIST_MLP().to(device) criterion = nn.CrossEntropyLoss() optimizer = GeometricDerivativeOptimizer( model.parameters(), lr=0.002, alpha=0.75, # Fractional memory attenuation variable drift_coef=0.95, # High weight towards deterministic landscape shape diff_coef=0.02, # Multi-directional exploratory diffusion scaling memory_len=5 # Past system state history depth ) epochs = 3 print("Beginning Training Optimization via Custom Geometric Derivatives...\n") for epoch in range(1, epochs + 1): model.train() total_loss = 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() total_loss += loss.item() if batch_idx % 100 == 0: print(f"Epoch {epoch} [{batch_idx * len(data)}/{len(train_loader.dataset)}] " f"Current Loss: {loss.item():.4f}") # Test Performance Verification Loop model.eval() test_loss = 0 correct = 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.argmax(dim=1, keepdim=True) correct += pred.eq(target.view_as(pred)).sum().item() test_loss /= len(test_loader) accuracy = 100. * correct / len(test_loader.dataset) print(f"\n--> Epoch {epoch} Completed Summary Evaluation:") print(f" Average Test Set Loss: {test_loss:.4f}") print(f" Target Accuracy Metrics: {correct}/{len(test_loader.dataset)} ({accuracy:.2f}%)\n" + "-"*50) if __name__ == '__main__': main()