Instrument API timing and reduce detail fallbacks

2026-05-31 18:58:57 +08:00
parent 668f4d9bd3
commit 5083b7c433
18 changed files with 832 additions and 135 deletions
@@ -13,6 +13,7 @@ from fastapi.concurrency import run_in_threadpool
 from loguru import logger

 import web.routes as legacy_routes
+from web.services.request_timing import ServerTimingRecorder

 _RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None
 _RECENT_DEB_CACHE_TS = 0.0
@@ -368,16 +369,41 @@ async def get_city_detail_aggregate_payload(
    target_date: Optional[str] = None,
    resolution: Optional[str] = "10m",
 ) -> Dict[str, Any]:
-    legacy_routes._assert_entitlement(request)
-    city = legacy_routes._normalize_city_or_404(name)
-    data = await _get_city_full_data(city, force_refresh=force_refresh)
-
-    return await _build_city_detail_payload_cached(
-        data,
-        market_slug,
-        target_date,
-        resolution,
+    timer = ServerTimingRecorder(
+        request,
+        log_name="city_detail_timing",
+        prefix="city_detail",
+        state_attr="city_detail_server_timing",
    )
+    outcome = "ok"
+    status_code = 200
+    try:
+        timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
+        city = timer.measure("normalize_city", lambda: legacy_routes._normalize_city_or_404(name))
+        data = await timer.measure_async(
+            "full_data",
+            lambda: _get_city_full_data(city, force_refresh=force_refresh),
+        )
+
+        return await timer.measure_async(
+            "detail_payload",
+            lambda: _build_city_detail_payload_cached(
+                data,
+                market_slug,
+                target_date,
+                resolution,
+            ),
+        )
+    except HTTPException as exc:
+        outcome = f"http_{exc.status_code}"
+        status_code = exc.status_code
+        raise
+    except Exception:
+        outcome = "exception"
+        status_code = 500
+        raise
+    finally:
+        timer.finish(outcome=outcome, status_code=status_code)


 def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
@@ -404,14 +430,30 @@ async def _build_city_detail_batch_item_async(
    market_slug: Optional[str],
    target_date: Optional[str],
    resolution: Optional[str],
+    timing_recorder: Optional[ServerTimingRecorder] = None,
 ) -> Tuple[str, Dict[str, Any]]:
-    data = await _get_city_full_data(city, force_refresh=force_refresh)
-    detail = await _build_city_detail_payload_cached(
-        data,
-        market_slug,
-        target_date,
-        resolution,
-    )
+    if timing_recorder is not None:
+        data = await timing_recorder.measure_async(
+            f"full_data_{city}",
+            lambda: _get_city_full_data(city, force_refresh=force_refresh),
+        )
+        detail = await timing_recorder.measure_async(
+            f"detail_payload_{city}",
+            lambda: _build_city_detail_payload_cached(
+                data,
+                market_slug,
+                target_date,
+                resolution,
+            ),
+        )
+    else:
+        data = await _get_city_full_data(city, force_refresh=force_refresh)
+        detail = await _build_city_detail_payload_cached(
+            data,
+            market_slug,
+            target_date,
+            resolution,
+        )
    return city, detail


@@ -433,39 +475,68 @@ async def get_city_detail_batch_payload(
    resolution: Optional[str] = "10m",
    limit: int = 12,
 ) -> Dict[str, Any]:
-    legacy_routes._assert_entitlement(request)
-    city_names = _parse_batch_city_names(cities, limit=max(1, min(24, int(limit or 12))))
-    if not city_names:
-        return {"cities": [], "details": {}, "errors": {}}
+    timer = ServerTimingRecorder(
+        request,
+        log_name="city_detail_batch_timing",
+        prefix="city_detail_batch",
+        state_attr="city_detail_batch_server_timing",
+    )
+    outcome = "ok"
+    status_code = 200
+    try:
+        timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
+        city_names = timer.measure(
+            "parse_cities",
+            lambda: _parse_batch_city_names(
+                cities,
+                limit=max(1, min(24, int(limit or 12))),
+            ),
+        )
+        if not city_names:
+            return {"cities": [], "details": {}, "errors": {}}

-    semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
+        semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())

-    async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
-        async with semaphore:
-            return await _build_city_detail_batch_item_async(
-                city,
-                force_refresh=force_refresh,
-                market_slug=market_slug,
-                target_date=target_date,
-                resolution=resolution,
-            )
+        async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
+            async with semaphore:
+                return await _build_city_detail_batch_item_async(
+                    city,
+                    force_refresh=force_refresh,
+                    market_slug=market_slug,
+                    target_date=target_date,
+                    resolution=resolution,
+                    timing_recorder=timer,
+                )

-    tasks = [
-        _build_with_limit(city)
-        for city in city_names
-    ]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-    details: Dict[str, Any] = {}
-    errors: Dict[str, str] = {}
-    for city, result in zip(city_names, results):
-        if isinstance(result, Exception):
-            errors[city] = str(result)
-            continue
-        result_city, payload = result
-        details[result_city] = payload
+        tasks = [
+            _build_with_limit(city)
+            for city in city_names
+        ]
+        results = await timer.measure_async(
+            "build_details",
+            lambda: asyncio.gather(*tasks, return_exceptions=True),
+        )
+        details: Dict[str, Any] = {}
+        errors: Dict[str, str] = {}
+        for city, result in zip(city_names, results):
+            if isinstance(result, Exception):
+                errors[city] = str(result)
+                continue
+            result_city, payload = result
+            details[result_city] = payload

-    return {
-        "cities": city_names,
-        "details": details,
-        "errors": errors,
-    }
+        return {
+            "cities": city_names,
+            "details": details,
+            "errors": errors,
+        }
+    except HTTPException as exc:
+        outcome = f"http_{exc.status_code}"
+        status_code = exc.status_code
+        raise
+    except Exception:
+        outcome = "exception"
+        status_code = 500
+        raise
+    finally:
+        timer.finish(outcome=outcome, status_code=status_code)
@@ -0,0 +1,87 @@
+"""Small helpers for exposing request stage timings via Server-Timing."""
+
+from __future__ import annotations
+
+import re
+import threading
+import time
+from typing import Awaitable, Callable, Dict, Optional, TypeVar
+
+from fastapi import Request, Response
+from loguru import logger
+
+
+T = TypeVar("T")
+
+
+class ServerTimingRecorder:
+    def __init__(
+        self,
+        request: Optional[Request],
+        *,
+        log_name: str,
+        prefix: str,
+        state_attr: str,
+    ) -> None:
+        self.request = request
+        self.log_name = log_name
+        self.prefix = prefix
+        self.state_attr = state_attr
+        self.started = time.perf_counter()
+        self.timings_ms: Dict[str, float] = {}
+        self._lock = threading.Lock()
+
+    def _record(self, stage: str, started: float) -> None:
+        elapsed_ms = round((time.perf_counter() - started) * 1000.0, 1)
+        with self._lock:
+            self.timings_ms[stage] = elapsed_ms
+
+    def measure(self, stage: str, action: Callable[[], T]) -> T:
+        started = time.perf_counter()
+        try:
+            return action()
+        finally:
+            self._record(stage, started)
+
+    async def measure_async(self, stage: str, action: Callable[[], Awaitable[T]]) -> T:
+        started = time.perf_counter()
+        try:
+            return await action()
+        finally:
+            self._record(stage, started)
+
+    def server_timing_value(self) -> str:
+        with self._lock:
+            items = list(self.timings_ms.items())
+        return ", ".join(
+            f"{self._metric_name(stage)};dur={max(0.0, duration):.1f}"
+            for stage, duration in items
+        )
+
+    def finish(self, *, outcome: str, status_code: int) -> None:
+        self._record("total", self.started)
+        value = self.server_timing_value()
+        state = getattr(self.request, "state", None)
+        if state is not None:
+            setattr(state, self.state_attr, value)
+        logger.info(
+            "{} outcome={} status_code={} timings_ms={}",
+            self.log_name,
+            outcome,
+            status_code,
+            dict(self.timings_ms),
+        )
+
+    def _metric_name(self, stage: str) -> str:
+        raw = f"{self.prefix}_{stage}"
+        return re.sub(r"[^A-Za-z0-9_-]", "_", raw)
+
+
+def attach_server_timing_header(
+    response: Response,
+    request: Request,
+    state_attr: str,
+) -> None:
+    value = str(getattr(request.state, state_attr, "") or "").strip()
+    if value:
+        response.headers["Server-Timing"] = value
@@ -2,12 +2,25 @@

 from __future__ import annotations

+from inspect import Parameter, signature
 from typing import Any, Dict

-from fastapi import Request
+from fastapi import HTTPException, Request
 from fastapi.concurrency import run_in_threadpool

 import web.routes as legacy_routes
+from web.services.request_timing import ServerTimingRecorder
+
+
+def _supports_timing_recorder(func: Any) -> bool:
+    try:
+        params = signature(func).parameters.values()
+    except (TypeError, ValueError):
+        return True
+    return any(
+        param.name == "timing_recorder" or param.kind == Parameter.VAR_KEYWORD
+        for param in params
+    )


 async def get_scan_terminal_payload(
@@ -26,27 +39,49 @@ async def get_scan_terminal_payload(
    region: str = "",
    timezone_offset_seconds: int | None = None,
 ) -> Dict[str, Any]:
-    legacy_routes._assert_entitlement(request)
-    filters: Dict[str, Any] = {
-        "scan_mode": scan_mode,
-        "min_price": min_price,
-        "max_price": max_price,
-        "min_edge_pct": min_edge_pct,
-        "min_liquidity": min_liquidity,
-        "high_liquidity_only": high_liquidity_only,
-        "market_type": market_type,
-        "time_range": time_range,
-        "limit": limit,
-    }
-    if timezone_offset_seconds is not None:
-        filters["timezone_offset_seconds"] = timezone_offset_seconds
-    if region:
-        filters["trading_region"] = region
-    return await run_in_threadpool(
-        legacy_routes.build_scan_terminal_payload,
-        filters,
-        force_refresh=force_refresh,
+    timer = ServerTimingRecorder(
+        request,
+        log_name="scan_terminal_timing",
+        prefix="scan_terminal",
+        state_attr="scan_terminal_server_timing",
    )
+    outcome = "ok"
+    status_code = 200
+    try:
+        timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
+        filters: Dict[str, Any] = {
+            "scan_mode": scan_mode,
+            "min_price": min_price,
+            "max_price": max_price,
+            "min_edge_pct": min_edge_pct,
+            "min_liquidity": min_liquidity,
+            "high_liquidity_only": high_liquidity_only,
+            "market_type": market_type,
+            "time_range": time_range,
+            "limit": limit,
+        }
+        if timezone_offset_seconds is not None:
+            filters["timezone_offset_seconds"] = timezone_offset_seconds
+        if region:
+            filters["trading_region"] = region
+        async def build_payload():
+            builder = legacy_routes.build_scan_terminal_payload
+            kwargs: Dict[str, Any] = {"force_refresh": force_refresh}
+            if _supports_timing_recorder(builder):
+                kwargs["timing_recorder"] = timer
+            return await run_in_threadpool(builder, filters, **kwargs)
+
+        return await timer.measure_async("build_payload", build_payload)
+    except HTTPException as exc:
+        outcome = f"http_{exc.status_code}"
+        status_code = exc.status_code
+        raise
+    except Exception:
+        outcome = "exception"
+        status_code = 500
+        raise
+    finally:
+        timer.finish(outcome=outcome, status_code=status_code)


 async def get_scan_terminal_overview_payload(request: Request) -> Dict[str, Any]: