Create sovereign_capacity_report.py

sovereign_capacity_report.py

@@ -0,0 +1,242 @@
+
+# sovereign_capacity_report.py
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from sovereign_capacity_probe import run_capacity_probe
+
+
+CAPACITY_OUTPUT_DIR = "sovereign_capacity_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 _rps(v: Any) -> str:
+    try:
+        return f"{float(v):.3f} rps"
+    except Exception:
+        return "0.000 rps"
+
+
+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 Capacity Report
+
+Generated at: {_safe(report.get("generated_at"), _utc_now())}
+
+## 1. Probe Identity
+- Probe Name: {_safe(report.get("probe_name"), "Sovereign Capacity Probe")}
+- 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 Requests: {_safe(cfg.get("total_expected_requests"), 0)}
+
+## 2. Executive Summary
+- Successful Requests: {_safe(summary.get("successful_requests"), 0)}
+- Failed Requests: {_safe(summary.get("failed_requests"), 0)}
+- Total Wall Time: {_safe(summary.get("total_wall_seconds"), 0)} sec
+- Approx Throughput: {_rps(summary.get("throughput_rps", 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 Under Load
+- 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_case_latency(report: Dict[str, Any]) -> str:
+    blocks = ["## 4. Per-Scenario Latency Under Load"]
+    per_case = report.get("per_case_latency", {})
+
+    if not per_case:
+        blocks.append("- No per-case data available.")
+        return "\n".join(blocks)
+
+    for case_name, stats in per_case.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 = ["## 5. Failures"]
+
+    if not failures:
+        lines.append("- No capacity-probe 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 = ["## 6. Sample Outputs Under Load"]
+
+    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_verdict(report: Dict[str, Any]) -> str:
+    summary = report.get("summary", {})
+    overall = summary.get("overall_latency", {})
+    failures = int(summary.get("failed_requests", 0) or 0)
+    p95 = float(overall.get("p95_ms", 0.0) or 0.0)
+    p99 = float(overall.get("p99_ms", 0.0) or 0.0)
+    throughput = float(summary.get("throughput_rps", 0.0) or 0.0)
+
+    if failures == 0 and p95 <= 100 and p99 <= 200:
+        verdict = "Capacity posture is stable for current probe volume, with acceptable runtime behavior under repeated load."
+    elif failures == 0:
+        verdict = "Capacity posture is operationally viable, but latency under load should be further optimized before strong bank-facing performance claims."
+    else:
+        verdict = "Capacity probe indicates additional hardening or stability work is required before bank-facing performance commitments."
+
+    return f"""## 7. Preliminary Capacity Verdict
+{verdict}
+
+## 8. Notes
+- This report reflects current repeated-request behavior, not final production-scale distributed load testing.
+- Throughput in this report is an approximate runtime throughput for the current execution path.
+- Bank-facing performance claims should be based on repeated probes, tuning iterations, and later higher-intensity validation.
+- This report is suitable as a technical capacity artifact, not yet as a final production SLA statement.
+"""
+
+
+def build_capacity_markdown(report: Dict[str, Any]) -> str:
+    parts = [
+        _render_summary(report),
+        _render_case_latency(report),
+        _render_failures(report),
+        _render_samples(report),
+        _render_verdict(report),
+    ]
+    return "\n\n".join(parts).strip() + "\n"
+
+
+def save_capacity_report(
+    report: Dict[str, Any],
+    output_dir: str = CAPACITY_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_capacity_report_{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_capacity_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_capacity_report(
+    iterations_per_case: int = 25,
+    warmup_runs: int = 2,
+    output_dir: str = CAPACITY_OUTPUT_DIR,
+    base_name: Optional[str] = None,
+) -> Dict[str, Any]:
+    report = run_capacity_probe(
+        iterations_per_case=iterations_per_case,
+        warmup_runs=warmup_runs,
+    )
+
+    files = save_capacity_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_capacity_report(iterations_per_case=10, warmup_runs=1)
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+