Upload leviathan.py with huggingface_hub

leviathan.py

@@ -1,51 +1,35 @@
 """
-🔱 LEVIATHAN v2 — EVM Exploit Topology Classifier
+🔱 Leviathan v2 — EVM Exploit Topology Classifier
 Pure NumPy inference. Zero PyTorch dependency.
 
-Takes 2-channel 256x256 EVM execution manifolds and classifies
-THREAT (1.0) vs CLEAN (0.0) via CNN + ZKAEDI PRIME bistable refinement.
-
 Architecture:
-    256x256 manifold -> downsample 20x20 -> Conv2d(2,16,3) -> ReLU
-    -> Conv2d(16,16,3) -> ReLU -> Flatten(4096) -> FC(64) -> ReLU -> FC(1)
-    -> PRIME refinement (eta=3.50, gamma=0.30, beta=0.10, sigma=0.05, T=256)
-
-Channel semantics:
-    Channel 0: Opcode energy density (H activator field)
-    Channel 1: Stack depth / state mutation intensity (V inhibitor field)
-
-Manifold encoding:
-    EVM traces are Hilbert-curve-encoded into 256x256 spatial manifolds
-    where spatial locality = execution locality.
-
-Validated results:
-    Gnosis Multisig       -> 0.0000 (CLEAN)
-    SWC-107 reentrancy    -> 1.0000 (THREAT)
-    SWC-112 delegatecall  -> 1.0000 (THREAT)
-    SWC-101 overflow      -> 1.0000 (THREAT)
-    Cross-fn reentrancy   -> 1.0000 (THREAT)
-    Flash loan manip      -> 1.0000 (THREAT)
+    EVM Trace → Hilbert Encoder → 256×256 Manifold
+      → Downsample 20×20 → Conv2d(2→16) → Conv2d(16→16)
+      → FC(4096→64) → FC(64→1) → Raw Score
+      → PRIME Bistable Attractor Refinement → THREAT/CLEAN
+
+Channels:
+    0: Opcode energy density (H activator field)
+    1: Stack depth / state mutation intensity (V inhibitor field)
 
 Usage:
     from leviathan import Leviathan
     model = Leviathan.from_safetensors("leviathan_v2_session_trained.safetensors")
-    result = model.audit(H_256x256, V_256x256)
-    # -> {"verdict": "THREAT", "committed": True, "refined_score": 0.9998, ...}
+    score = model.predict_manifold(H_256, V_256)
+    verdict = model.prime_refine(score)
 """
 from __future__ import annotations
 import numpy as np
 from pathlib import Path
 
-CNN_INPUT_SPATIAL = 20
-MANIFOLD_SIZE = 256
-PRIME_ETA = 3.50
+CNN_SPATIAL = 20
+PRIME_ETA   = 3.50
 PRIME_GAMMA = 0.30
-PRIME_BETA = 0.10
+PRIME_BETA  = 0.10
 PRIME_SIGMA = 0.05
-PRIME_ITERATIONS = 256
+PRIME_T     = 256
 THREAT_THRESHOLD = 0.90
 BENIGN_THRESHOLD = 0.10
-NEGATIVE_FIXED_POINT = -3.054
 
 
 def _conv2d(x, weight, bias):
@@ -76,106 +60,25 @@ def _downsample(field, target):
     if H == target and W == target:
         return field
     bh, bw = H // target, W // target
-    cropped = field[:bh * target, :bw * target]
+    if bh < 1 or bw < 1:
+        xi = np.round(np.linspace(0, H-1, target)).astype(int)
+        yi = np.round(np.linspace(0, W-1, target)).astype(int)
+        return field[np.ix_(xi, yi)]
+    cropped = field[:bh*target, :bw*target]
     return cropped.reshape(target, bh, target, bw).mean(axis=(1, 3))
 
 
-def _hilbert_d2xy(n, d):
-    x = y = 0
-    s = 1
-    while s < n:
-        rx = 1 if (d & 2) else 0
-        ry = 1 if ((d & 1) ^ rx) else 0
-        if ry == 0:
-            if rx == 1:
-                x = s - 1 - x
-                y = s - 1 - y
-            x, y = y, x
-        x += s * rx
-        y += s * ry
-        d >>= 2
-        s <<= 1
-    return x, y
-
-
-_HILBERT_LUT_256 = None
-
-def encode_trace_to_manifold(opcode_energies, stack_depths, size=256):
-    """Encode EVM trace -> 2-channel 256x256 manifold via Hilbert curve."""
-    global _HILBERT_LUT_256
-    if _HILBERT_LUT_256 is None:
-        _HILBERT_LUT_256 = np.array([_hilbert_d2xy(size, d) for d in range(size * size)])
-
-    N = len(opcode_energies)
-    total_cells = size * size
-    H = np.zeros((size, size), dtype=np.float32)
-    V = np.zeros((size, size), dtype=np.float32)
-
-    for i in range(min(N, total_cells)):
-        x, y = _HILBERT_LUT_256[i]
-        H[x, y] += opcode_energies[i]
-        V[x, y] += stack_depths[i]
-
-    if N > total_cells:
-        for i in range(total_cells, N):
-            x, y = _HILBERT_LUT_256[i % total_cells]
-            H[x, y] += opcode_energies[i]
-            V[x, y] += stack_depths[i]
-
-    if H.max() > 0:
-        H /= H.max()
-    if V.max() > 0:
-        V /= V.max()
-
-    try:
-        from scipy.ndimage import gaussian_filter
-        H = gaussian_filter(H, sigma=1.5).astype(np.float32)
-        V = gaussian_filter(V, sigma=1.5).astype(np.float32)
-    except ImportError:
-        pass  # scipy optional — skip smoothing
-
-    return H, V
-
-
-def prime_refine(raw_score, iterations=PRIME_ITERATIONS, eta=PRIME_ETA,
-                 gamma=PRIME_GAMMA, beta=PRIME_BETA, sigma=PRIME_SIGMA, seed=42):
-    """
-    ZKAEDI PRIME bistable attractor refinement.
-    Maps CNN raw score to committed THREAT or CLEAN via Hamiltonian evolution.
-    Fixed points: H* = -3.054 (BENIGN) and H* = +3.054 (THREAT).
-    """
-    rng = np.random.default_rng(seed)
-    H = (raw_score - 0.5) * 6.0
-    for _ in range(iterations):
-        sig = 1.0 / (1.0 + np.exp(-gamma * H))
-        noise = rng.normal(0, 1 + beta * abs(H)) * sigma
-        H = H + eta * H * sig + noise
-        H = np.clip(H, -10.0, 10.0)
-
-    refined = 1.0 / (1.0 + np.exp(-H))
-    if refined > THREAT_THRESHOLD:
-        verdict, committed = "THREAT", True
-    elif refined < BENIGN_THRESHOLD:
-        verdict, committed = "CLEAN", True
-    else:
-        verdict, committed = "UNCERTAIN", False
-
-    return {
-        "raw_score": float(raw_score),
-        "prime_H": float(H),
-        "refined_score": float(refined),
-        "verdict": verdict,
-        "committed": committed,
-        "iterations": iterations,
-    }
+def _sigmoid(x, gamma):
+    clamped = max(-500.0, min(500.0, -gamma * x))
+    return 1.0 / (1.0 + np.exp(clamped))
 
 
 class Leviathan:
     """
     Leviathan v2 — EVM Exploit Topology Classifier.
 
-    Pipeline:
-        256x256 manifold -> downsample 20x20 -> CNN -> raw score -> PRIME refine -> verdict
+    Two-channel CNN on Hamiltonian energy manifolds from EVM traces.
+    256x256 manifold -> downsample 20x20 -> CNN -> score -> PRIME refine.
     """
 
     def __init__(self, weights):
@@ -194,43 +97,104 @@ class Leviathan:
         return cls(load_file(str(path)))
 
     @classmethod
-    def from_huggingface(cls, repo_id="zkaedi/leviathan-v2"):
-        from huggingface_hub import hf_hub_download
-        path = hf_hub_download(repo_id, "leviathan_v2_session_trained.safetensors")
-        return cls.from_safetensors(path)
+    def from_dict(cls, weights):
+        return cls(weights)
 
-    def _forward_cnn(self, x):
+    def forward(self, x):
+        if x.ndim == 3:
+            x = x[None]
+        B = x.shape[0]
         h = _relu(_conv2d(x, self.conv0_w, self.conv0_b))
         h = _relu(_conv2d(h, self.conv1_w, self.conv1_b))
-        h = h.reshape(1, -1)
+        h = h.reshape(B, -1)
         h = _relu(_linear(h, self.fc1_w, self.fc1_b))
         h = _linear(h, self.fc2_w, self.fc2_b)
-        return float(h[0, 0])
-
-    def classify_manifold(self, H_field, V_field):
-        """Classify a 2-channel manifold (any spatial size, auto-downsamples to 20x20)."""
-        H_ds = _downsample(H_field.astype(np.float32), CNN_INPUT_SPATIAL)
-        V_ds = _downsample(V_field.astype(np.float32), CNN_INPUT_SPATIAL)
-        x = np.stack([H_ds, V_ds])[None]
-        return self._forward_cnn(x)
-
-    def audit(self, H_field, V_field, use_prime=True, seed=42):
-        """Full audit: CNN + PRIME bistable refinement -> verdict."""
-        raw = self.classify_manifold(H_field, V_field)
-        if use_prime:
-            return prime_refine(raw, seed=seed)
-        verdict = "THREAT" if raw > THREAT_THRESHOLD else ("CLEAN" if raw < BENIGN_THRESHOLD else "UNCERTAIN")
-        return {"raw_score": raw, "prime_H": 0.0, "refined_score": raw,
-                "verdict": verdict, "committed": verdict != "UNCERTAIN", "iterations": 0}
-
-    def audit_trace(self, opcode_energies, stack_depths, **kwargs):
-        """Full audit from raw EVM trace arrays."""
-        H, V = encode_trace_to_manifold(opcode_energies, stack_depths)
-        return self.audit(H, V, **kwargs)
+        return h
+
+    def predict(self, H_field, V_field):
+        """Score from 20x20 field patches (CNN native resolution)."""
+        x = np.stack([H_field, V_field])[None].astype(np.float32)
+        return float(self.forward(x)[0, 0])
+
+    def predict_manifold(self, H_256, V_256):
+        """Score from full 256x256 EVM execution manifold. Auto-downsamples."""
+        H_down = _downsample(H_256.astype(np.float32), CNN_SPATIAL)
+        V_down = _downsample(V_256.astype(np.float32), CNN_SPATIAL)
+        return self.predict(H_down, V_down)
+
+    def predict_region(self, H_field, V_field, region=None):
+        """Score a region of any-sized energy landscape."""
+        s = CNN_SPATIAL
+        if region is not None:
+            rx, ry, rw, rh = region
+            H_crop = H_field[rx:rx+rw, ry:ry+rh]
+            V_crop = V_field[rx:rx+rw, ry:ry+rh]
+        else:
+            cx = max(0, (H_field.shape[0] - s) // 2)
+            cy = max(0, (H_field.shape[1] - s) // 2)
+            H_crop = H_field[cx:cx+s, cy:cy+s]
+            V_crop = V_field[cx:cx+s, cy:cy+s]
+        H_down = _downsample(H_crop.astype(np.float32), s)
+        V_down = _downsample(V_crop.astype(np.float32), s)
+        return self.predict(H_down, V_down)
+
+    @staticmethod
+    def prime_refine(raw_score, seed=None):
+        """
+        PRIME bistable attractor refinement.
+        Evolves raw CNN score through recursive Hamiltonian dynamics.
+        H* = -3.054 (BENIGN attractor), positive (THREAT attractor).
+        """
+        rng = np.random.default_rng(seed)
+        H = (raw_score - 0.5) * 4.0
+        trajectory = [H]
+
+        for t in range(PRIME_T):
+            sig = _sigmoid(H, PRIME_GAMMA)
+            noise = rng.normal(0, 1 + PRIME_BETA * abs(H))
+            H = H + PRIME_ETA * H * sig + PRIME_SIGMA * noise
+            H = max(-10.0, min(10.0, H))
+            trajectory.append(H)
+
+        refined = 1.0 / (1.0 + np.exp(-H))
+
+        if refined > THREAT_THRESHOLD:
+            verdict = "THREAT"
+            confidence = refined
+        elif refined < BENIGN_THRESHOLD:
+            verdict = "BENIGN"
+            confidence = 1.0 - refined
+        else:
+            verdict = "UNCERTAIN"
+            confidence = 1.0 - 2.0 * abs(refined - 0.5)
+
+        return {
+            "verdict": verdict,
+            "confidence": round(float(confidence), 4),
+            "refined_score": round(float(refined), 6),
+            "raw_score": round(float(raw_score), 6),
+            "prime_iterations": PRIME_T,
+            "prime_params": {"eta": PRIME_ETA, "gamma": PRIME_GAMMA,
+                             "beta": PRIME_BETA, "sigma": PRIME_SIGMA},
+            "trajectory_final_10": [round(float(x), 4) for x in trajectory[-10:]],
+        }
+
+    def audit(self, H_manifold, V_manifold, seed=42):
+        """Full pipeline: manifold -> CNN -> PRIME refinement -> verdict."""
+        if H_manifold.shape[0] > CNN_SPATIAL:
+            raw_score = self.predict_manifold(H_manifold, V_manifold)
+        else:
+            raw_score = self.predict(H_manifold.astype(np.float32),
+                                     V_manifold.astype(np.float32))
+        result = self.prime_refine(raw_score, seed=seed)
+        result["manifold_size"] = list(H_manifold.shape)
+        result["cnn_input_size"] = [CNN_SPATIAL, CNN_SPATIAL]
+        return result
 
     def __repr__(self):
-        return (f"Leviathan(input=256x256->20x20, conv=[2->16->16], fc=[4096->64->1], "
-                f"params=264,897, PRIME=eta{PRIME_ETA}/gamma{PRIME_GAMMA})")
+        return (f"Leviathan(channels=2, conv=[2->16->16], fc=[4096->64->1], "
+                f"params=264,897, manifold=256x256->{CNN_SPATIAL}x{CNN_SPATIAL}, "
+                f"PRIME=[eta={PRIME_ETA},gamma={PRIME_GAMMA},T={PRIME_T}])")
 
 
 if __name__ == "__main__":
@@ -239,21 +203,22 @@ if __name__ == "__main__":
     model = Leviathan.from_safetensors(path)
     print(model)
 
-    print("\n--- 256x256 manifold (standard pipeline) ---")
-    H = np.random.randn(256, 256).astype(np.float32)
-    V = np.random.randn(256, 256).astype(np.float32)
-    result = model.audit(H, V)
-    print(f"  raw={result['raw_score']:.4f}  refined={result['refined_score']:.4f}  "
-          f"verdict={result['verdict']}  committed={result['committed']}  "
-          f"prime_H={result['prime_H']:.3f}")
-
-    print("\n--- 20x20 direct (no downsampling) ---")
-    raw = model.classify_manifold(np.random.randn(20, 20).astype(np.float32),
-                                   np.random.randn(20, 20).astype(np.float32))
-    print(f"  raw_score={raw:.4f}")
-
-    print("\n--- PRIME refinement sweep ---")
-    for s in [0.0, 0.25, 0.5, 0.75, 1.0]:
-        r = prime_refine(s)
-        print(f"  input={s:.2f} -> {r['verdict']:<10} refined={r['refined_score']:.4f} "
-              f"H*={r['prime_H']:.3f} committed={r['committed']}")
+    print("\n=== 256x256 Manifold Audit ===")
+    H_256 = np.random.randn(256, 256).astype(np.float32)
+    V_256 = np.random.randn(256, 256).astype(np.float32)
+    result = model.audit(H_256, V_256)
+    print(f"  Verdict:    {result['verdict']}")
+    print(f"  Raw:        {result['raw_score']}")
+    print(f"  Refined:    {result['refined_score']}")
+    print(f"  Confidence: {result['confidence']}")
+
+    print("\n=== 20x20 Direct ===")
+    score = model.predict(np.random.randn(20, 20).astype(np.float32),
+                          np.zeros((20, 20), dtype=np.float32))
+    print(f"  Score: {score:.6f}")
+
+    print("\n=== PRIME Refinement Sweep ===")
+    for s in [0.0, 0.3, 0.5, 0.7, 1.0]:
+        r = Leviathan.prime_refine(s, seed=42)
+        print(f"  input={s:.1f} -> {r['verdict']:>9} "
+              f"(refined={r['refined_score']:.4f}, conf={r['confidence']:.4f})")