import torch import torch.nn as nn import torch.nn.functional as F import numpy as np import matplotlib.pyplot as plt # ============================================================================= # THEORY OF THE PARADOX HIDDEN LAYER # ============================================================================= # A standard MLP hidden layer is memoryless: y_t = σ(W x_t + b). # It cannot distinguish a smooth DC signal from a rapidly oscillating one. # # The Paradox Layer introduces a "contradiction detector". It measures how # much the current state contradicts its recent past (i.e. oscillates). # # Definitions: # v_t = x_t - x_{t-1} # velocity (first derivative) # a_t = x_t - 2x_{t-1} + x_{t-2} # acceleration (second derivative) # z_t = ZCR(v_t) # zero-crossing rate of velocity # # Paradox Score: # p_t = σ( λ · ( ||a_t||² / (||v_t||² + ε) ) · z_t + β ) # # λ = learnable temperature (scales sensitivity) # β = learnable bias (shifts operating point) # ε = numerical stability # # The score p_t → 1 when the signal is oscillatory (high curvature + many # sign changes), and p_t → 0 when the signal is smooth or DC. # # The layer maintains two expert pathways: # Consensus : y_c = ReLU( W_c x_t + b_c ) -- stable regime # Paradox : y_p = tanh( W_p x_t + b_p ) -- oscillatory regime # # Final output: # y_t = (1 - p_t) ⊙ y_c + p_t ⊙ y_p # # When oscillation is detected, the network switches to the tanh pathway, # which preserves negative lobes and naturally supports oscillatory dynamics. # ============================================================================= class ParadoxLayer(nn.Module): """Differentiable hidden layer that detects temporal oscillation.""" def __init__(self, in_features: int, out_features: int, eps: float = 1e-5): super().__init__() self.in_features = in_features self.out_features = out_features self.eps = eps # Dual expert pathways self.consensus = nn.Linear(in_features, out_features) self.paradox = nn.Linear(in_features, out_features) # Causal temporal derivative kernels (fixed, per-channel) d1 = torch.tensor([1.0, -1.0]).view(1, 1, 2) self.register_buffer("d1_kernel", d1.repeat(in_features, 1, 1)) d2 = torch.tensor([1.0, -2.0, 1.0]).view(1, 1, 3) self.register_buffer("d2_kernel", d2.repeat(in_features, 1, 1)) # Learnable gating parameters self.temperature = nn.Parameter(torch.ones(1) * 2.0) self.bias = nn.Parameter(torch.zeros(1)) # ------------------------------------------------------------------------- # Internal: compute the scalar paradox score p_t from a short history # ------------------------------------------------------------------------- def _paradox_score(self, x_seq: torch.Tensor) -> torch.Tensor: """ x_seq : (B, T, D) -- batch, time, features returns p : (B, T, 1) in [0, 1] """ B, T, D = x_seq.shape x = x_seq.permute(0, 2, 1) # (B, D, T) # Causal first derivative (velocity) v = F.conv1d( F.pad(x, (1, 0), mode="replicate"), self.d1_kernel, groups=D ) # (B, D, T) # Causal second derivative (acceleration / curvature) a = F.conv1d( F.pad(x, (2, 0), mode="replicate"), self.d2_kernel, groups=D ) # (B, D, T) # Energy averaged over the feature dimension v2 = v.pow(2).mean(dim=1, keepdim=True) # (B, 1, T) a2 = a.pow(2).mean(dim=1, keepdim=True) # (B, 1, T) # Zero-crossing rate of velocity (robust oscillation indicator) v_sign = torch.sign(v) zcr = ( (v_sign[:, :, 1:] != v_sign[:, :, :-1]).float().mean(dim=2, keepdim=True) ) # (B, 1, T-1) zcr = F.pad(zcr, (1, 0), mode="replicate") # (B, 1, T) # Paradox score: high when curvature is large relative to velocity # AND zero-crossings are frequent. ratio = a2 / (v2 + self.eps) score = self.temperature * (ratio * zcr + self.bias) p = torch.sigmoid(score) # (B, 1, T) return p.permute(0, 2, 1) # (B, T, 1) # ------------------------------------------------------------------------- # Forward: accepts single step (B, D) or full sequence (B, T, D) # ------------------------------------------------------------------------- def forward(self, x: torch.Tensor) -> torch.Tensor: if x.dim() == 2: # Streaming / single-step mode x_seq = x.unsqueeze(1) # (B, 1, D) p = self._paradox_score(x_seq).squeeze(1) # (B, 1) else: # Batch sequence mode p = self._paradox_score(x) # (B, T, 1) y_c = F.relu(self.consensus(x)) y_p = torch.tanh(self.paradox(x)) y = (1 - p) * y_c + p * y_p return y # ============================================================================= # EXAMPLE MLP THAT STACKS PARADOX LAYERS # ============================================================================= class ParadoxMLP(nn.Module): def __init__( self, in_dim: int = 4, hidden_dim: int = 256, out_dim: int = 1, num_paradox: int = 2 ): super().__init__() dims = [in_dim] + [hidden_dim] * 3 + [out_dim] layers = [] for i in range(len(dims) - 1): if i < num_paradox: layers.append(ParadoxLayer(dims[i], dims[i + 1])) else: layers.append(nn.Linear(dims[i], dims[i + 1])) if i < len(dims) - 2: layers.append(nn.ReLU()) self.net = nn.Sequential(*layers) def forward(self, x: torch.Tensor) -> torch.Tensor: return self.net(x) # ============================================================================= # TEST SIGNAL & VISUALISATION # ============================================================================= def make_test_signal(T: int = 2800) -> torch.Tensor: """ Build a multi-feature signal that transitions from stable to oscillatory regimes, producing the stripe-like heatmaps shown in the reference. """ t = np.linspace(0, 1, T) signal = np.zeros((T, 4), dtype=np.float32) # Feature 0: continuous low-frequency sine signal[:, 0] = np.sin(2 * np.pi * 3 * t) # Feature 1: chirp that starts silent then sweeps upward # (creates diagonal interference stripes in the hidden layer) freq = np.where(t < 0.2, 0, 10 + 80 * (t - 0.2)) signal[:, 1] = np.where(t < 0.2, 0.0, np.sin(2 * np.pi * freq * t)) # Feature 2: smooth random walk (non-oscillatory) raw = np.random.randn(T).cumsum() signal[:, 2] = (raw - raw.mean()) / (raw.std() + 1e-6) * 0.3 # Feature 3: sudden high-frequency burst after t = 0.6 signal[:, 3] = np.where(t > 0.6, 0.5 * np.sin(2 * np.pi * 60 * t), 0.0) return torch.from_numpy(signal).unsqueeze(0) # (1, T, 4) def visualize(): torch.manual_seed(42) model = ParadoxMLP(in_dim=4, hidden_dim=256, out_dim=1, num_paradox=2) x = make_test_signal(T=2800) # (1, 2800, 4) # Capture hidden activations with forward hooks hidden_acts = {} def capture(name): def hook(module, inp, out): hidden_acts[name] = out.detach().squeeze(0).cpu().numpy() return hook model.net[0].register_forward_hook(capture("paradox_1")) model.net[1].register_forward_hook(capture("paradox_2")) with torch.no_grad(): _ = model(x) # Plot heatmaps matching the user's reference images fig, axes = plt.subplots(2, 1, figsize=(16, 6)) for ax, (name, act) in zip(axes, hidden_acts.items()): # act shape: (T, D) im = ax.imshow( act.T, aspect="auto", cmap="gray", interpolation="nearest", vmin=-1, vmax=1, extent=[0, act.shape[0], act.shape[1], 0], ) ax.set_xlabel("Time step") ax.set_ylabel("Hidden unit") ax.set_title(f"{name} ({act.shape[1]} units, {act.shape[0]} steps)") plt.colorbar(im, ax=ax, fraction=0.02) plt.tight_layout() plt.savefig("paradox_activations.png", dpi=150, bbox_inches="tight") plt.show() print("Saved: paradox_activations.png") print(" - Vertical stripes indicate oscillatory input regimes.") print(" - The Paradox Layer switches to the tanh pathway (preserving") print(" negative lobes) exactly when ZCR and curvature rise.") if __name__ == "__main__": visualize()