#!/usr/bin/env python3
"""Benchmark dashboard API latency and write JSON/HTML reports."""
from __future__ import annotations
import argparse
import html
import json
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
DEFAULT_ENDPOINTS = [
("health", "/health", {}),
("stock_history_2330_6m", "/api/stock/2330/history", {"months": 6}),
("stock_predict_2330", "/api/stock/2330/predict", {}),
("stock_backtest_2330_6m", "/api/stock/2330/backtest", {"months": 6}),
("recommend_all_top20", "/api/stock/recommend", {"limit": 20, "signal_filter": "ALL"}),
(
"hotspots_top20",
"/api/stock/hotspots",
{"limit": 20, "watchlist_limit": 20, "candidate_limit": 20},
),
(
"paper_fast_summary",
"/api/paper-trades",
{"mark_prices": "false", "price_timeout_seconds": 1.0},
),
(
"paper_mark_prices",
"/api/paper-trades",
{"mark_prices": "true", "price_timeout_seconds": 1.0},
),
("oil_brent_history", "/api/oil/BZ=F/history", {"months": 6, "timeout_seconds": 4}),
("oil_brent_predict", "/api/oil/BZ=F/predict", {"timeout_seconds": 4}),
("oil_brent_backtest", "/api/oil/BZ=F/backtest", {"months": 12, "timeout_seconds": 4}),
]
def _url(base_url: str, path: str, params: dict[str, Any]) -> str:
query = urlencode(params)
return f"{base_url.rstrip('/')}{path}{'?' + query if query else ''}"
def _request(url: str, timeout: float) -> dict[str, Any]:
started = time.perf_counter()
status = None
size = 0
error = None
try:
with urlopen(Request(url, headers={"Accept": "application/json"}), timeout=timeout) as resp:
body = resp.read()
status = resp.status
size = len(body)
except HTTPError as exc:
status = exc.code
try:
size = len(exc.read())
except Exception:
size = 0
error = str(exc)
except (TimeoutError, URLError, OSError) as exc:
error = str(exc)
elapsed_ms = round((time.perf_counter() - started) * 1000, 2)
return {
"status": status,
"elapsed_ms": elapsed_ms,
"size_bytes": size,
"ok": bool(status and 200 <= int(status) < 400),
"error": error,
}
def _render_html(report: dict[str, Any]) -> str:
rows = []
for row in report["results"]:
cls = "ok" if row["ok"] else "fail"
rows.append(
""
f"| {html.escape(row['name'])} | "
f"{html.escape(row['path'])} | "
f"{html.escape(str(row['status']))} | "
f"{row['elapsed_ms']:.2f} | "
f"{row['size_bytes']} | "
f"{html.escape(str(row.get('error') or ''))} | "
"
"
)
return f"""
Dashboard Latency Report
Dashboard Latency Report
Generated: {html.escape(report['generated_at'])} · Base URL: {html.escape(report['base_url'])}
OK: {report['summary']['ok_count']} / {report['summary']['count']} · p50: {report['summary']['p50_ms']:.2f} ms · p95: {report['summary']['p95_ms']:.2f} ms
| Name | Path | Status | Latency ms | Bytes | Error |
|---|
{''.join(rows)}
"""
def _percentile(values: list[float], pct: float) -> float:
if not values:
return 0.0
ordered = sorted(values)
index = min(len(ordered) - 1, max(0, round((pct / 100) * (len(ordered) - 1))))
return float(ordered[index])
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-url", default="http://127.0.0.1:7861")
parser.add_argument("--timeout-seconds", type=float, default=15.0)
parser.add_argument("--output-json", type=Path, default=Path("docs/validation_runs/dashboard_latency_report.json"))
parser.add_argument("--output-html", type=Path, default=Path("docs/validation_runs/dashboard_latency_report.html"))
args = parser.parse_args()
results = []
for name, path, params in DEFAULT_ENDPOINTS:
target = _url(args.base_url, path, params)
result = _request(target, timeout=args.timeout_seconds)
results.append({"name": name, "path": path, "params": params, "url": target, **result})
latencies = [float(row["elapsed_ms"]) for row in results]
report = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"base_url": args.base_url,
"timeout_seconds": args.timeout_seconds,
"summary": {
"count": len(results),
"ok_count": sum(1 for row in results if row["ok"]),
"p50_ms": _percentile(latencies, 50),
"p95_ms": _percentile(latencies, 95),
"max_ms": max(latencies) if latencies else 0.0,
},
"results": results,
}
args.output_json.parent.mkdir(parents=True, exist_ok=True)
args.output_html.parent.mkdir(parents=True, exist_ok=True)
args.output_json.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
args.output_html.write_text(_render_html(report), encoding="utf-8")
print(json.dumps(report["summary"], ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())