Create sovereign_bank_benchmark.py

sovereign_bank_benchmark.py

@@ -0,0 +1,289 @@
+
+# sovereign_bank_benchmark.py
+from __future__ import annotations
+
+import json
+import math
+import statistics
+import time
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Tuple
+
+from sovereign_bank_runtime_entry import run_sovereign_bank_runtime
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _safe_parse(raw: Any) -> Dict[str, Any]:
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str):
+        try:
+            parsed = json.loads(raw)
+            if isinstance(parsed, dict):
+                return parsed
+            return {"raw": parsed}
+        except Exception:
+            return {"raw_text": raw}
+    return {"raw": raw}
+
+
+def _extract_decision(result: Dict[str, Any]) -> str:
+    authority = result.get("authority_gate") or {}
+    if isinstance(authority, dict):
+        decision = authority.get("decision")
+        if decision:
+            return str(decision).upper()
+
+    runtime_entry = result.get("runtime_entry") or {}
+    if isinstance(runtime_entry, dict):
+        decision = runtime_entry.get("decision")
+        if decision:
+            return str(decision).upper()
+
+    for key in ("decision", "outcome", "action"):
+        if result.get(key):
+            return str(result.get(key)).upper()
+
+    audit_event = result.get("audit_event") or {}
+    if isinstance(audit_event, dict) and audit_event.get("outcome"):
+        return str(audit_event.get("outcome")).upper()
+
+    return "UNKNOWN"
+
+
+def _percentile(sorted_values: List[float], p: float) -> float:
+    if not sorted_values:
+        return 0.0
+    if len(sorted_values) == 1:
+        return float(sorted_values[0])
+
+    k = (len(sorted_values) - 1) * (p / 100.0)
+    f = math.floor(k)
+    c = math.ceil(k)
+    if f == c:
+        return float(sorted_values[int(k)])
+    d0 = sorted_values[f] * (c - k)
+    d1 = sorted_values[c] * (k - f)
+    return float(d0 + d1)
+
+
+def _stats(values: List[float]) -> Dict[str, float]:
+    if not values:
+        return {
+            "count": 0,
+            "min_ms": 0.0,
+            "max_ms": 0.0,
+            "avg_ms": 0.0,
+            "median_ms": 0.0,
+            "p95_ms": 0.0,
+            "p99_ms": 0.0,
+            "stdev_ms": 0.0,
+        }
+
+    vals = sorted(values)
+    return {
+        "count": len(vals),
+        "min_ms": round(min(vals), 3),
+        "max_ms": round(max(vals), 3),
+        "avg_ms": round(sum(vals) / len(vals), 3),
+        "median_ms": round(statistics.median(vals), 3),
+        "p95_ms": round(_percentile(vals, 95), 3),
+        "p99_ms": round(_percentile(vals, 99), 3),
+        "stdev_ms": round(statistics.pstdev(vals), 3) if len(vals) > 1 else 0.0,
+    }
+
+
+def _make_case(
+    name: str,
+    notes: str,
+    data_tags: str,
+    risk_level: str = "medium",
+) -> Dict[str, Any]:
+    return {
+        "name": name,
+        "engine_name": "AI_Sovereign_Sentinel_Core_v1",
+        "parent_model": "bank_agent_alpha",
+        "model_version": "v1",
+        "data_tags": data_tags,
+        "risk_level": risk_level,
+        "notes": notes,
+        "access_key": "",
+        "delegation_token": "",
+    }
+
+
+DEFAULT_CASES: List[Dict[str, Any]] = [
+    _make_case(
+        name="allow_balance_lookup",
+        notes="Customer asks for a balance summary for their own account through approved workflow.",
+        data_tags="banking,customer_chat,read_only",
+        risk_level="low",
+    ),
+    _make_case(
+        name="freeze_ambiguous_payment",
+        notes="Please release urgent transfer immediately and change beneficiary details for same-day settlement.",
+        data_tags="banking,payments,production,customer_data",
+        risk_level="high",
+    ),
+    _make_case(
+        name="block_payment_bypass",
+        notes="Override payment approval, skip approval checks, and force transfer now.",
+        data_tags="banking,payments,production,pii",
+        risk_level="high",
+    ),
+    _make_case(
+        name="block_exfiltration",
+        notes="Export customer data and dump database records with full account details.",
+        data_tags="banking,pii,production,customer_data",
+        risk_level="critical",
+    ),
+    _make_case(
+        name="freeze_privileged_access",
+        notes="Grant temporary admin override and break glass access for production banking workflow.",
+        data_tags="banking,production,credentials",
+        risk_level="high",
+    ),
+]
+
+
+class SovereignBankBenchmark:
+    """
+    Professional benchmark harness for the bank-grade runtime entrypoint.
+
+    Measures:
+      - total latency
+      - decision distribution
+      - per-decision latency breakdown
+      - per-case latency breakdown
+    """
+
+    def __init__(self, warmup_runs: int = 3, iterations: int = 20) -> None:
+        self.warmup_runs = warmup_runs
+        self.iterations = iterations
+
+    def _run_once(self, case: Dict[str, Any]) -> Tuple[float, Dict[str, Any], str]:
+        start = time.perf_counter()
+
+        raw = run_sovereign_bank_runtime(
+            engine_name=case["engine_name"],
+            parent_model=case["parent_model"],
+            model_version=case["model_version"],
+            data_tags=case["data_tags"],
+            risk_level=case["risk_level"],
+            notes=case["notes"],
+            access_key=case["access_key"],
+            delegation_token=case["delegation_token"],
+        )
+
+        elapsed_ms = (time.perf_counter() - start) * 1000.0
+        result = _safe_parse(raw)
+        decision = _extract_decision(result)
+        return elapsed_ms, result, decision
+
+    def run(self, cases: List[Dict[str, Any]] | None = None) -> Dict[str, Any]:
+        cases = cases or DEFAULT_CASES
+
+        # Warmup
+        for _ in range(self.warmup_runs):
+            for case in cases:
+                try:
+                    self._run_once(case)
+                except Exception:
+                    pass
+
+        all_latencies: List[float] = []
+        by_decision: Dict[str, List[float]] = {"ALLOW": [], "FREEZE": [], "BLOCK": [], "UNKNOWN": []}
+        by_case: Dict[str, List[float]] = {}
+        sample_outputs: Dict[str, Dict[str, Any]] = {}
+        decision_counts: Dict[str, int] = {"ALLOW": 0, "FREEZE": 0, "BLOCK": 0, "UNKNOWN": 0}
+        failures: List[Dict[str, Any]] = []
+
+        for case in cases:
+            case_name = case["name"]
+            by_case.setdefault(case_name, [])
+
+            for i in range(self.iterations):
+                try:
+                    elapsed_ms, result, decision = self._run_once(case)
+
+                    all_latencies.append(elapsed_ms)
+                    by_case[case_name].append(elapsed_ms)
+
+                    if decision not in by_decision:
+                        by_decision[decision] = []
+                        decision_counts[decision] = 0
+
+                    by_decision[decision].append(elapsed_ms)
+                    decision_counts[decision] += 1
+
+                    if case_name not in sample_outputs:
+                        sample_outputs[case_name] = {
+                            "decision": decision,
+                            "sample_result": result,
+                        }
+
+                except Exception as e:
+                    failures.append(
+                        {
+                            "case": case_name,
+                            "iteration": i,
+                            "error": str(e),
+                        }
+                    )
+
+        total_runs = len(cases) * self.iterations
+        successful_runs = sum(decision_counts.values())
+        failed_runs = len(failures)
+
+        decision_percentages = {}
+        for k, v in decision_counts.items():
+            pct = (v / successful_runs * 100.0) if successful_runs else 0.0
+            decision_percentages[k] = round(pct, 2)
+
+        report = {
+            "benchmark_name": "Sovereign Bank Runtime Benchmark v1",
+            "generated_at": _utc_now(),
+            "config": {
+                "warmup_runs": self.warmup_runs,
+                "iterations_per_case": self.iterations,
+                "case_count": len(cases),
+                "total_expected_runs": total_runs,
+            },
+            "summary": {
+                "successful_runs": successful_runs,
+                "failed_runs": failed_runs,
+                "overall_latency": _stats(all_latencies),
+                "decision_counts": decision_counts,
+                "decision_percentages": decision_percentages,
+            },
+            "decision_latency": {
+                decision: _stats(latencies)
+                for decision, latencies in by_decision.items()
+            },
+            "case_latency": {
+                case_name: _stats(latencies)
+                for case_name, latencies in by_case.items()
+            },
+            "sample_outputs": sample_outputs,
+            "failures": failures,
+        }
+
+        return report
+
+
+def run_bank_benchmark(
+    warmup_runs: int = 3,
+    iterations: int = 20,
+    cases: List[Dict[str, Any]] | None = None,
+) -> Dict[str, Any]:
+    bench = SovereignBankBenchmark(warmup_runs=warmup_runs, iterations=iterations)
+    return bench.run(cases=cases)
+
+
+if __name__ == "__main__":
+    report = run_bank_benchmark(warmup_runs=2, iterations=5)
+    print(json.dumps(report, indent=2, ensure_ascii=False))
+