Spaces:
Running
Running
| #!/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()) | |