Upload scorer.py with huggingface_hub
Browse files
scorer.py
ADDED
|
@@ -0,0 +1,61 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""
|
| 2 |
+
Pure-Python port of the TS logistic scorer (`scoreFeatures` in
|
| 3 |
+
packages/intelligence/src/model.ts). Deliberately scores an ALREADY-EXTRACTED
|
| 4 |
+
feature vector — it does not walk a serialization graph. Graph feature extraction
|
| 5 |
+
stays single-sourced in the TS engine (see docs/huggingface-model-plan.md §0);
|
| 6 |
+
duplicating that traversal here would drift silently and undetectably.
|
| 7 |
+
|
| 8 |
+
This file has exactly one job, and hf/parity_test.py holds it to that: given the
|
| 9 |
+
same model.json and the same feature vector, produce the same
|
| 10 |
+
{score, uncertainty, explanation} the TS engine produces.
|
| 11 |
+
"""
|
| 12 |
+
|
| 13 |
+
from __future__ import annotations
|
| 14 |
+
import json
|
| 15 |
+
import math
|
| 16 |
+
from dataclasses import dataclass
|
| 17 |
+
|
| 18 |
+
|
| 19 |
+
@dataclass
|
| 20 |
+
class AnomalyOutput:
|
| 21 |
+
score: float
|
| 22 |
+
uncertainty: float
|
| 23 |
+
explanation: str
|
| 24 |
+
|
| 25 |
+
def to_dict(self) -> dict:
|
| 26 |
+
return {"score": self.score, "uncertainty": self.uncertainty, "explanation": self.explanation}
|
| 27 |
+
|
| 28 |
+
|
| 29 |
+
def load_model(path: str) -> dict:
|
| 30 |
+
with open(path, "r", encoding="utf-8") as fh:
|
| 31 |
+
return json.load(fh)
|
| 32 |
+
|
| 33 |
+
|
| 34 |
+
def score_features(model: dict, features: list[float]) -> AnomalyOutput:
|
| 35 |
+
"""Byte-for-byte port of scoreFeatures() in model.ts — same iteration order,
|
| 36 |
+
same tie-break (first-seen wins on equal |contrib|, matching JS `>` semantics)."""
|
| 37 |
+
mean = model["mean"]
|
| 38 |
+
std = model["std"]
|
| 39 |
+
weights = model["weights"]
|
| 40 |
+
feature_names = model.get("featureNames", [])
|
| 41 |
+
|
| 42 |
+
logit = float(model["bias"])
|
| 43 |
+
top_name = ""
|
| 44 |
+
top_abs = -1.0
|
| 45 |
+
for i, fi in enumerate(features):
|
| 46 |
+
denom = std[i] if std[i] else 1.0
|
| 47 |
+
z = (fi - mean[i]) / denom
|
| 48 |
+
contrib = weights[i] * z
|
| 49 |
+
logit += contrib
|
| 50 |
+
if abs(contrib) > top_abs:
|
| 51 |
+
top_abs = abs(contrib)
|
| 52 |
+
top_name = feature_names[i] if i < len(feature_names) else f"f{i}"
|
| 53 |
+
|
| 54 |
+
p = 1.0 / (1.0 + math.exp(-logit))
|
| 55 |
+
uncertainty = 1.0 - min(1.0, 2.0 * abs(p - 0.5))
|
| 56 |
+
explanation = (
|
| 57 |
+
f"learned anomaly {round(p * 100)}% (top: {top_name})"
|
| 58 |
+
if p > 0.5
|
| 59 |
+
else f"learned: consistent (top: {top_name})"
|
| 60 |
+
)
|
| 61 |
+
return AnomalyOutput(score=p, uncertainty=uncertainty, explanation=explanation)
|