Spaces:
Running
Running
File size: 6,143 Bytes
d2d8dd9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | #!/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(
"<tr>"
f"<td>{html.escape(row['name'])}</td>"
f"<td>{html.escape(row['path'])}</td>"
f"<td class='{cls}'>{html.escape(str(row['status']))}</td>"
f"<td>{row['elapsed_ms']:.2f}</td>"
f"<td>{row['size_bytes']}</td>"
f"<td>{html.escape(str(row.get('error') or ''))}</td>"
"</tr>"
)
return f"""<!doctype html>
<html lang="zh-Hant">
<head>
<meta charset="utf-8">
<title>Dashboard Latency Report</title>
<style>
body {{ font-family: -apple-system, BlinkMacSystemFont, "Noto Sans TC", sans-serif; margin: 24px; color: #17202a; }}
table {{ border-collapse: collapse; width: 100%; }}
th, td {{ border: 1px solid #d5d8dc; padding: 8px; text-align: left; font-size: 13px; }}
th {{ background: #f4f6f7; }}
.ok {{ color: #117a65; font-weight: 700; }}
.fail {{ color: #b03a2e; font-weight: 700; }}
code {{ background: #f4f6f7; padding: 2px 4px; border-radius: 4px; }}
</style>
</head>
<body>
<h1>Dashboard Latency Report</h1>
<p>Generated: <code>{html.escape(report['generated_at'])}</code> 路 Base URL: <code>{html.escape(report['base_url'])}</code></p>
<p>OK: {report['summary']['ok_count']} / {report['summary']['count']} 路 p50: {report['summary']['p50_ms']:.2f} ms 路 p95: {report['summary']['p95_ms']:.2f} ms</p>
<table>
<thead><tr><th>Name</th><th>Path</th><th>Status</th><th>Latency ms</th><th>Bytes</th><th>Error</th></tr></thead>
<tbody>{''.join(rows)}</tbody>
</table>
</body>
</html>
"""
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())
|