135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
# app/tools/dzengi_probe/ticker24hr_update_probe.py
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import csv
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
from app.tools.dzengi_probe.config import DzengiProbeConfig
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Ticker24hrSample:
|
|
sample: int
|
|
local_time_ms: int
|
|
ask_price: str | None
|
|
bid_price: str | None
|
|
last_price: str | None
|
|
close_time: int | None
|
|
|
|
|
|
class Ticker24hrUpdateProbe:
|
|
def __init__(self, config: DzengiProbeConfig) -> None:
|
|
self.config = config
|
|
|
|
async def run(
|
|
self,
|
|
*,
|
|
samples: int = 300,
|
|
interval_seconds: float = 0.2,
|
|
) -> Path:
|
|
output_path = (
|
|
self.config.output_dir
|
|
/ "reports"
|
|
/ f"ticker24hr_update_{self.config.symbol_file_name}.csv"
|
|
)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
rows: list[Ticker24hrSample] = []
|
|
|
|
for index in range(1, samples + 1):
|
|
payload = self._get_ticker24hr()
|
|
|
|
rows.append(
|
|
Ticker24hrSample(
|
|
sample=index,
|
|
local_time_ms=int(time.time() * 1000),
|
|
ask_price=self._decimal_str(payload.get("askPrice")),
|
|
bid_price=self._decimal_str(payload.get("bidPrice")),
|
|
last_price=self._decimal_str(payload.get("lastPrice")),
|
|
close_time=self._int_or_none(payload.get("closeTime")),
|
|
)
|
|
)
|
|
|
|
await asyncio.sleep(interval_seconds)
|
|
|
|
self._save_csv(output_path, rows)
|
|
return output_path
|
|
|
|
def _get_ticker24hr(self) -> dict:
|
|
query = urlencode({"symbol": self.config.symbol})
|
|
url = (
|
|
f"{self.config.base_url}"
|
|
f"/api/{self.config.api_version}/ticker/24hr"
|
|
f"?{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 _decimal_str(self, value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return str(Decimal(str(value)))
|
|
|
|
def _int_or_none(self, value: object) -> int | None:
|
|
if value is None:
|
|
return None
|
|
|
|
return int(str(value))
|
|
|
|
def _save_csv(self, path: Path, rows: list[Ticker24hrSample]) -> None:
|
|
with path.open("w", encoding="utf-8", newline="") as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(
|
|
[
|
|
"sample",
|
|
"local_time_ms",
|
|
"close_time",
|
|
"ask_price",
|
|
"bid_price",
|
|
"last_price",
|
|
"ask_changed",
|
|
"bid_changed",
|
|
"last_changed",
|
|
"close_time_changed",
|
|
"last_equals_bid",
|
|
]
|
|
)
|
|
|
|
previous: Ticker24hrSample | None = None
|
|
|
|
for row in rows:
|
|
writer.writerow(
|
|
[
|
|
row.sample,
|
|
row.local_time_ms,
|
|
row.close_time,
|
|
row.ask_price,
|
|
row.bid_price,
|
|
row.last_price,
|
|
previous is not None and row.ask_price != previous.ask_price,
|
|
previous is not None and row.bid_price != previous.bid_price,
|
|
previous is not None and row.last_price != previous.last_price,
|
|
previous is not None and row.close_time != previous.close_time,
|
|
row.last_price == row.bid_price,
|
|
]
|
|
)
|
|
|
|
previous = row |