170 lines
5.1 KiB
Python
170 lines
5.1 KiB
Python
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,
|
|
]
|
|
) |