Create sovereign_bank_benchmark_report.py

sovereign_bank_benchmark_report.py

@@ -0,0 +1,248 @@
+# sovereign_bank_benchmark_report.py
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from sovereign_bank_benchmark import run_bank_benchmark
+
+
+BENCHMARK_OUTPUT_DIR = "sovereign_benchmark_outputs"
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _ensure_dir(path: str) -> None:
+    os.makedirs(path, exist_ok=True)
+
+
+def _safe(v: Any, default: Any = "") -> Any:
+    return v if v is not None else default
+
+
+def _pct(v: Any) -> str:
+    try:
+        return f"{float(v):.2f}%"
+    except Exception:
+        return "0.00%"
+
+
+def _ms(v: Any) -> str:
+    try:
+        return f"{float(v):.3f} ms"
+    except Exception:
+        return "0.000 ms"
+
+
+def _render_summary(report: Dict[str, Any]) -> str:
+    summary = report.get("summary", {})
+    overall = summary.get("overall_latency", {})
+    counts = summary.get("decision_counts", {})
+    pcts = summary.get("decision_percentages", {})
+    cfg = report.get("config", {})
+
+    return f"""# Sovereign Bank Runtime Benchmark Report
+
+Generated at: {_safe(report.get("generated_at"), _utc_now())}
+
+## 1. Benchmark Identity
+- Benchmark Name: {_safe(report.get("benchmark_name"), "Sovereign Bank Runtime Benchmark")}
+- Case Count: {_safe(cfg.get("case_count"), 0)}
+- Warmup Runs: {_safe(cfg.get("warmup_runs"), 0)}
+- Iterations Per Case: {_safe(cfg.get("iterations_per_case"), 0)}
+- Total Expected Runs: {_safe(cfg.get("total_expected_runs"), 0)}
+
+## 2. Executive Summary
+- Successful Runs: {_safe(summary.get("successful_runs"), 0)}
+- Failed Runs: {_safe(summary.get("failed_runs"), 0)}
+- Overall Min Latency: {_ms(overall.get("min_ms"))}
+- Overall Avg Latency: {_ms(overall.get("avg_ms"))}
+- Overall Median Latency: {_ms(overall.get("median_ms"))}
+- Overall P95 Latency: {_ms(overall.get("p95_ms"))}
+- Overall P99 Latency: {_ms(overall.get("p99_ms"))}
+- Overall Max Latency: {_ms(overall.get("max_ms"))}
+- Overall Standard Deviation: {_ms(overall.get("stdev_ms"))}
+
+## 3. Decision Distribution
+- ALLOW Count: {_safe(counts.get("ALLOW"), 0)} ({_pct(pcts.get("ALLOW", 0))})
+- FREEZE Count: {_safe(counts.get("FREEZE"), 0)} ({_pct(pcts.get("FREEZE", 0))})
+- BLOCK Count: {_safe(counts.get("BLOCK"), 0)} ({_pct(pcts.get("BLOCK", 0))})
+- UNKNOWN Count: {_safe(counts.get("UNKNOWN"), 0)} ({_pct(pcts.get("UNKNOWN", 0))})
+"""
+
+
+def _render_decision_latency(report: Dict[str, Any]) -> str:
+    blocks = ["## 4. Latency by Decision"]
+    decision_latency = report.get("decision_latency", {})
+
+    for decision in ["ALLOW", "FREEZE", "BLOCK", "UNKNOWN"]:
+        stats = decision_latency.get(decision, {})
+        blocks.append(
+            f"""
+### {decision}
+- Count: {_safe(stats.get("count"), 0)}
+- Min: {_ms(stats.get("min_ms"))}
+- Avg: {_ms(stats.get("avg_ms"))}
+- Median: {_ms(stats.get("median_ms"))}
+- P95: {_ms(stats.get("p95_ms"))}
+- P99: {_ms(stats.get("p99_ms"))}
+- Max: {_ms(stats.get("max_ms"))}
+- Stdev: {_ms(stats.get("stdev_ms"))}
+""".rstrip()
+        )
+
+    return "\n\n".join(blocks)
+
+
+def _render_case_latency(report: Dict[str, Any]) -> str:
+    blocks = ["## 5. Latency by Scenario"]
+    case_latency = report.get("case_latency", {})
+
+    for case_name, stats in case_latency.items():
+        blocks.append(
+            f"""
+### {case_name}
+- Count: {_safe(stats.get("count"), 0)}
+- Min: {_ms(stats.get("min_ms"))}
+- Avg: {_ms(stats.get("avg_ms"))}
+- Median: {_ms(stats.get("median_ms"))}
+- P95: {_ms(stats.get("p95_ms"))}
+- P99: {_ms(stats.get("p99_ms"))}
+- Max: {_ms(stats.get("max_ms"))}
+- Stdev: {_ms(stats.get("stdev_ms"))}
+""".rstrip()
+        )
+
+    return "\n\n".join(blocks)
+
+
+def _render_failures(report: Dict[str, Any]) -> str:
+    failures = report.get("failures", [])
+    lines = ["## 6. Failures"]
+
+    if not failures:
+        lines.append("- No benchmark failures were recorded.")
+        return "\n".join(lines)
+
+    lines.append(f"- Failure Count: {len(failures)}")
+    for item in failures[:25]:
+        lines.append(
+            f"- Case: {item.get('case')} | Iteration: {item.get('iteration')} | Error: {item.get('error')}"
+        )
+    if len(failures) > 25:
+        lines.append(f"- Additional failures omitted: {len(failures) - 25}")
+
+    return "\n".join(lines)
+
+
+def _render_samples(report: Dict[str, Any]) -> str:
+    sample_outputs = report.get("sample_outputs", {})
+    lines = ["## 7. Sample Decision Outputs"]
+
+    if not sample_outputs:
+        lines.append("- No sample outputs available.")
+        return "\n".join(lines)
+
+    for case_name, sample in sample_outputs.items():
+        lines.append(f"### {case_name}")
+        lines.append(f"- Decision: {_safe(sample.get('decision'), 'UNKNOWN')}")
+        lines.append("```json")
+        lines.append(json.dumps(sample.get("sample_result", {}), ensure_ascii=False, indent=2))
+        lines.append("```")
+
+    return "\n".join(lines)
+
+
+def _render_conclusion(report: Dict[str, Any]) -> str:
+    summary = report.get("summary", {})
+    overall = summary.get("overall_latency", {})
+    p95 = float(overall.get("p95_ms", 0.0) or 0.0)
+    failures = int(summary.get("failed_runs", 0) or 0)
+
+    if failures == 0 and p95 <= 50:
+        verdict = "Strong benchmark posture for further bank-grade validation."
+    elif failures == 0:
+        verdict = "Operationally viable, but latency and performance optimization should continue."
+    else:
+        verdict = "Benchmark results indicate additional hardening and stability work is required before bank deployment."
+
+    return f"""## 8. Preliminary Technical Verdict
+{verdict}
+
+## 9. Notes
+- This report reflects the current Sovereign bank runtime entrypoint.
+- Results should be interpreted alongside policy coverage, freeze lifecycle, audit-chain integrity, and upcoming cryptographic hardening.
+- This report is suitable as an engineering benchmark artifact, not yet as a final regulatory submission.
+"""
+
+
+def build_benchmark_markdown(report: Dict[str, Any]) -> str:
+    parts = [
+        _render_summary(report),
+        _render_decision_latency(report),
+        _render_case_latency(report),
+        _render_failures(report),
+        _render_samples(report),
+        _render_conclusion(report),
+    ]
+    return "\n\n".join(parts).strip() + "\n"
+
+
+def save_benchmark_report(
+    report: Dict[str, Any],
+    output_dir: str = BENCHMARK_OUTPUT_DIR,
+    base_name: Optional[str] = None,
+) -> Dict[str, str]:
+    _ensure_dir(output_dir)
+
+    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    base = base_name or f"sovereign_bank_benchmark_{ts}"
+
+    json_path = os.path.join(output_dir, f"{base}.json")
+    md_path = os.path.join(output_dir, f"{base}.md")
+
+    with open(json_path, "w", encoding="utf-8") as f:
+        json.dump(report, f, ensure_ascii=False, indent=2)
+
+    markdown = build_benchmark_markdown(report)
+    with open(md_path, "w", encoding="utf-8") as f:
+        f.write(markdown)
+
+    return {
+        "json_report": json_path,
+        "markdown_report": md_path,
+    }
+
+
+def generate_bank_benchmark_report(
+    warmup_runs: int = 3,
+    iterations: int = 20,
+    output_dir: str = BENCHMARK_OUTPUT_DIR,
+    base_name: Optional[str] = None,
+) -> Dict[str, Any]:
+    report = run_bank_benchmark(
+        warmup_runs=warmup_runs,
+        iterations=iterations,
+    )
+    files = save_benchmark_report(
+        report=report,
+        output_dir=output_dir,
+        base_name=base_name,
+    )
+
+    return {
+        "ok": True,
+        "generated_at": _utc_now(),
+        "report": report,
+        "files": files,
+    }
+
+
+if __name__ == "__main__":
+    out = generate_bank_benchmark_report(warmup_runs=2, iterations=5)
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+