build 039: complete Quotes Feed migration foundation

app/tools/dzengi_probe/ask_price_equivalence_probe.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import asyncio
+import csv
+import json
+from dataclasses import dataclass
+from decimal import Decimal
+from pathlib import Path
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+import websockets
+from websockets.typing import Subprotocol
+
+from app.tools.dzengi_probe.config import DzengiProbeConfig
+
+
+@dataclass(frozen=True)
+class AskPriceSample:
+    sample: int
+    depth_ask: str | None
+    stream_ofr: str | None
+    ticker_ask_price: str | None
+
+
+class AskPriceEquivalenceProbe:
+    def __init__(self, config: DzengiProbeConfig) -> None:
+        self.config = config
+
+    async def run(self, *, samples: int = 30) -> Path:
+        output_path = (
+            self.config.output_dir
+            / "reports"
+            / f"ask_price_equivalence_{self.config.symbol_file_name}.csv"
+        )
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        rows: list[AskPriceSample] = []
+
+        async with websockets.connect(
+            self.config.ws_url,
+            extra_headers={
+                "Origin": self.config.base_url,
+                "Content-Type": "application/json",
+            },
+            subprotocols=[Subprotocol("json")],
+            ping_interval=20,
+            ping_timeout=20,
+            close_timeout=5,
+        ) as websocket:
+            await websocket.send(
+                json.dumps(
+                    {
+                        "correlationId": "probe-market-data-subscribe",
+                        "destination": "marketData.subscribe",
+                        "payload": {"symbols": [self.config.symbol]},
+                    }
+                )
+            )
+
+            for index in range(1, samples + 1):
+                stream_ofr = await self._read_next_ofr(websocket)
+                depth_ask = self._get_depth_ask()
+                ticker_ask_price = self._get_ticker_ask_price()
+
+                rows.append(
+                    AskPriceSample(
+                        sample=index,
+                        depth_ask=depth_ask,
+                        stream_ofr=stream_ofr,
+                        ticker_ask_price=ticker_ask_price,
+                    )
+                )
+
+                await asyncio.sleep(0.2)
+
+        self._save_csv(output_path, rows)
+        return output_path
+
+    async def _read_next_ofr(self, websocket) -> str | None:
+        while True:
+            raw_message = await asyncio.wait_for(
+                websocket.recv(),
+                timeout=self.config.timeout_seconds,
+            )
+            payload = json.loads(raw_message)
+
+            if payload.get("destination") != "internal.quote":
+                continue
+
+            data = payload.get("payload")
+            if not isinstance(data, dict):
+                continue
+
+            value = data.get("ofr")
+            if value is None:
+                continue
+
+            return str(Decimal(str(value)))
+
+    def _get_depth_ask(self) -> str | None:
+        payload = self._get_json(
+            f"/api/{self.config.api_version}/depth",
+            {
+                "symbol": self.config.symbol,
+                "limit": self.config.depth_limit,
+            },
+        )
+
+        asks = payload.get("asks")
+        if not asks:
+            return None
+
+        return str(Decimal(str(asks[0][0])))
+
+    def _get_ticker_ask_price(self) -> str | None:
+        payload = self._get_json(
+            f"/api/{self.config.api_version}/ticker/24hr",
+            {"symbol": self.config.symbol},
+        )
+
+        value = payload.get("askPrice")
+        if value is None:
+            return None
+
+        return str(Decimal(str(value)))
+
+    def _get_json(self, path: str, params: dict[str, str]) -> dict:
+        query = urlencode(params)
+        url = f"{self.config.base_url}{path}?{query}"
+
+        request = Request(
+            url=url,
+            method="GET",
+            headers={
+                "Accept": "application/json",
+                "User-Agent": "dzentra-dzengi-probe/1.0",
+            },
+        )
+
+        with urlopen(request, timeout=self.config.timeout_seconds) as response:
+            return json.loads(response.read().decode("utf-8"))
+
+    def _save_csv(self, path: Path, rows: list[AskPriceSample]) -> None:
+        with path.open("w", encoding="utf-8", newline="") as file:
+            writer = csv.writer(file)
+            writer.writerow(
+                [
+                    "sample",
+                    "depth_ask",
+                    "stream_ofr",
+                    "ticker_ask_price",
+                    "depth_equals_stream",
+                    "ticker_equals_depth",
+                    "ticker_equals_stream",
+                ]
+            )
+
+            for row in rows:
+                writer.writerow(
+                    [
+                        row.sample,
+                        row.depth_ask,
+                        row.stream_ofr,
+                        row.ticker_ask_price,
+                        row.depth_ask == row.stream_ofr,
+                        row.ticker_ask_price == row.depth_ask,
+                        row.ticker_ask_price == row.stream_ofr,
+                    ]
+                )