updloads

app/core/engine.py

@@ -0,0 +1,73 @@
+import numpy as np
+from typing import List
+from app.api.schemas.telemetry import KeystrokeData
+from fastapi import HTTPException
+
+class DECI_Engine:
+    def __init__(self):
+        self.MIN_KEYSTROKES = 25  # [cite: 5, 52]
+        self.WEIGHTS = {"entropy": 0.35, "cv": 0.30, "mean": 0.15, "corrections": 0.20} # [cite: 5, 101]
+
+    def process_session(self, events: List[KeystrokeData]) -> dict:
+        if len(events) < self.MIN_KEYSTROKES:
+            return {"verdict": "INCONCLUSIVE", "reason": f"Need {self.MIN_KEYSTROKES} keys"}
+
+        # Extraer intervalos compatibles con v2.0 [cite: 5, 111]
+        timestamps = np.array([float(e.timestamp) for e in events])
+        intervals = np.diff(timestamps)
+        intervals = intervals[intervals > 0]
+
+        # 1. Shannon Entropy Normalizada (Fixed by Claude) [cite: 5, 111]
+        entropy = self._compute_shannon_entropy(intervals)
+        
+        # 2. Coeficiente de Variación (CV) [cite: 5, 111]
+        cv = np.std(intervals) / np.mean(intervals) if len(intervals) > 0 else 0
+        
+        # 3. Burst Correction Ratio [cite: 5, 59]
+        burst_ratio = self._compute_burst_ratio(events)
+
+        # ── Scoring Multi-señal (Vertex Elite Logic) ──
+        # Entropy Goldilocks Zone [cite: 5, 59]
+        s_entropy = 1.0 if 0.50 <= entropy <= 0.85 else (0.5 if 0.35 <= entropy <= 0.95 else 0.05)
+        s_cv = 1.0 if cv > 0.50 else (0.6 if cv > 0.30 else 0.1)
+        
+        # Mean IKL Scoring
+        mean_ms = np.mean(intervals)
+        s_mean = 1.0 if 60 <= mean_ms <= 500 else 0.2
+        
+        # Correction Scoring con Bonus de Ráfaga [cite: 5, 59]
+        corr_rate = sum(1 for e in events if e.key == 'Backspace') / len(events)
+        s_corrections = (0.5 + burst_ratio * 0.5) if 0.02 <= corr_rate <= 0.15 else 0.2
+
+        # Weighted Final Score
+        final_score = float(np.clip(
+            s_entropy * self.WEIGHTS["entropy"] +
+            s_cv * self.WEIGHTS["cv"] +
+            s_mean * self.WEIGHTS["mean"] +
+            s_corrections * self.WEIGHTS["corrections"], 0.0, 1.0
+        ))
+
+        return {
+            "score": round(final_score, 4),
+            "verdict": "HUMAN" if final_score >= 0.65 else ("SUSPECT" if final_score >= 0.40 else "BOT"), # 
+            "is_human": final_score >= 0.65,
+            "breakdown": {"entropy": round(entropy, 4), "cv": round(cv, 4), "burst": round(burst_ratio, 4)}
+        }
+
+    def _compute_shannon_entropy(self, intervals, bins=20):
+        hist, _ = np.histogram(intervals, bins=bins, density=True)
+        hist += 1e-10
+        raw_entropy = -np.sum(hist * np.log2(hist)) * (intervals.max() - intervals.min()) / bins
+        return float(np.clip(raw_entropy / np.log2(bins), 0.0, 1.0))
+
+    def _compute_burst_ratio(self, events):
+        # Detecta ráfagas de 3+ correcciones (comportamiento humano) [cite: 5, 59]
+        streak = burst = total = 0
+        for e in events:
+            if e.key == 'Backspace':
+                streak += 1
+                total += 1
+            else:
+                if streak >= 3: burst += streak
+                streak = 0
+        return float(burst / max(total, 1))