sovereign_bank_benchmark_report.py

# 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))