feat(hyperliquid): add methods for info endpoint

Author: smypmsa

api/read/hyperliquid.py

@@ -9,6 +9,10 @@
     HTTPGetLogsLatencyMetric,
     HTTPTxReceiptLatencyMetric,
 )
+from metrics.hyperliquid_info import (
+    HTTPClearinghouseStateLatencyMetric,
+    HTTPOpenOrdersLatencyMetric,
+)
 
 METRIC_NAME = f"{MetricsServiceConfig.METRIC_PREFIX}response_latency_seconds"
 ALLOWED_REGIONS: list[str] = [
@@ -26,6 +30,8 @@
         (HTTPAccBalanceLatencyMetric, METRIC_NAME),
         (HTTPTxReceiptLatencyMetric, METRIC_NAME),
         (HTTPGetLogsLatencyMetric, METRIC_NAME),
+        (HTTPClearinghouseStateLatencyMetric, METRIC_NAME),
+        (HTTPOpenOrdersLatencyMetric, METRIC_NAME),
     ]
     if os.getenv("VERCEL_REGION") in ALLOWED_REGIONS  # System env var, standard name
     else []

common/http_timing.py

@@ -0,0 +1,103 @@
+"""Shared HTTP request timing utilities for metric collection."""
+
+import asyncio
+import time
+from typing import Any, Dict, Optional
+
+import aiohttp
+
+MAX_RETRIES = 2
+
+
+class HttpTimingCollector:
+    """Utility class for measuring HTTP request timing with detailed breakdown."""
+
+    def __init__(self):
+        self.timing: Dict[str, float] = {}
+
+    def create_trace_config(self) -> aiohttp.TraceConfig:
+        """Create aiohttp trace configuration for detailed timing measurement."""
+        trace_config = aiohttp.TraceConfig()
+
+        async def on_request_start(session, context, params):
+            self.timing["start"] = time.monotonic()
+
+        async def on_dns_resolvehost_start(session, context, params):
+            self.timing["dns_start"] = time.monotonic()
+
+        async def on_dns_resolvehost_end(session, context, params):
+            self.timing["dns_end"] = time.monotonic()
+
+        async def on_connection_create_start(session, context, params):
+            self.timing["conn_start"] = time.monotonic()
+
+        async def on_connection_create_end(session, context, params):
+            self.timing["conn_end"] = time.monotonic()
+
+        async def on_request_end(session, context, params):
+            self.timing["end"] = time.monotonic()
+
+        trace_config.on_request_start.append(on_request_start)
+        trace_config.on_dns_resolvehost_start.append(on_dns_resolvehost_start)
+        trace_config.on_dns_resolvehost_end.append(on_dns_resolvehost_end)
+        trace_config.on_connection_create_start.append(on_connection_create_start)
+        trace_config.on_connection_create_end.append(on_connection_create_end)
+        trace_config.on_request_end.append(on_request_end)
+
+        return trace_config
+
+    def get_connection_time(self) -> float:
+        """Get connection establishment time in seconds."""
+        if "conn_start" in self.timing and "conn_end" in self.timing:
+            return self.timing["conn_end"] - self.timing["conn_start"]
+        return 0.0
+
+    def get_dns_time(self) -> float:
+        """Get DNS resolution time in seconds."""
+        if "dns_start" in self.timing and "dns_end" in self.timing:
+            return self.timing["dns_end"] - self.timing["dns_start"]
+        return 0.0
+
+
+async def measure_http_request_timing(
+    session: aiohttp.ClientSession,
+    method: str,
+    url: str,
+    headers: Optional[Dict[str, str]] = None,
+    json_data: Optional[Dict[str, Any]] = None,
+    exclude_connection_time: bool = True,
+) -> tuple[float, aiohttp.ClientResponse]:
+    """Measure HTTP request timing with retry logic and detailed breakdown.
+
+    Returns:
+        tuple: (response_time_seconds, response)
+    """
+    response_time = 0.0
+    response = None
+
+    for retry_count in range(MAX_RETRIES):
+        start_time = time.monotonic()
+
+        # Send request
+        if method.upper() == "POST":
+            response = await session.post(
+                url, headers=headers, json=json_data
+            )
+        else:
+            response = await session.get(url, headers=headers)
+
+        response_time = time.monotonic() - start_time
+
+        # Handle rate limiting
+        if response.status == 429 and retry_count < MAX_RETRIES - 1:
+            wait_time = int(response.headers.get("Retry-After", 3))
+            await response.release()
+            await asyncio.sleep(wait_time)
+            continue
+
+        break
+
+    if not response:
+        raise ValueError("No response received")
+
+    return response_time, response

common/hyperliquid_info_base.py

@@ -0,0 +1,118 @@
+"""Base class for Hyperliquid /info endpoint metrics."""
+
+from abc import abstractmethod
+from typing import Any
+
+import aiohttp
+
+from common.http_timing import measure_http_request_timing
+from common.metric_config import MetricConfig, MetricLabelKey, MetricLabels
+from common.metric_types import HttpMetric
+from common.metrics_handler import MetricsHandler
+
+
+class HyperliquidInfoMetricBase(HttpMetric):
+    """Base class for Hyperliquid /info endpoint latency metrics.
+
+    Handles request configuration, state validation, and response time
+    measurement for Hyperliquid info API endpoints.
+    """
+
+    @property
+    @abstractmethod
+    def method(self) -> str:
+        """Info API method to be implemented by subclasses."""
+        pass
+
+    def __init__(
+        self,
+        handler: "MetricsHandler",
+        metric_name: str,
+        labels: MetricLabels,
+        config: MetricConfig,
+        **kwargs: Any,
+    ) -> None:
+        """Initialize Hyperliquid info metric with state-based parameters."""
+        state_data = kwargs.get("state_data", {})
+        if not self.validate_state(state_data):
+            raise ValueError(f"Invalid state data for {self.method}")
+
+        super().__init__(
+            handler=handler,
+            metric_name=metric_name,
+            labels=labels,
+            config=config,
+        )
+
+        params = self.get_params_from_state(state_data)
+        self.user_address = params["user"]
+        self.labels.update_label(MetricLabelKey.API_METHOD, self.method)
+        self.request_payload = self._build_request_payload()
+
+    def _build_request_payload(self) -> dict[str, Any]:
+        """Build the Hyperliquid info API request payload."""
+        return {"type": self.method, "user": self.user_address}
+
+    @staticmethod
+    def validate_state(state_data: dict[str, Any]) -> bool:
+        """Validate state data. Override in subclasses if needed."""
+        return True
+
+    @staticmethod
+    def get_params_from_state(state_data: dict[str, Any]) -> dict[str, str]:
+        """Get parameters from state data. Override in subclasses if needed."""
+        return {}
+
+    def get_info_endpoint(self) -> str:
+        """Transform EVM endpoint to info endpoint."""
+        base_endpoint = self.get_endpoint()
+
+        if base_endpoint.endswith("/evm"):
+            return base_endpoint.replace("/evm", "/info")
+        else:
+            # Handle cases where endpoint doesn't end with /evm
+            if base_endpoint.endswith("/"):
+                return base_endpoint + "info"
+            else:
+                return base_endpoint + "/info"
+
+    async def fetch_data(self) -> float:
+        """Measure single request latency for Hyperliquid info API."""
+        endpoint = self.get_info_endpoint()
+
+        headers = {
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+
+        async with aiohttp.ClientSession() as session:
+            response_time, response = await measure_http_request_timing(
+                session=session,
+                method="POST",
+                url=endpoint,
+                headers=headers,
+                json_data=self.request_payload,
+                exclude_connection_time=True,
+            )
+
+            try:
+                # Validate response
+                if response.status != 200:
+                    raise aiohttp.ClientResponseError(
+                        request_info=response.request_info,
+                        history=(),
+                        status=response.status,
+                        message=f"Status code: {response.status}",
+                        headers=response.headers,
+                    )
+
+                response_data = await response.json()
+
+                return response_time
+
+            finally:
+                await response.release()
+
+    def process_data(self, value: float) -> float:
+        """Process raw latency measurement."""
+        return value