build 058: validate websocket OHLC market data
This commit is contained in:
328
app/scripts/check_ohlc_websocket.py
Normal file
328
app/scripts/check_ohlc_websocket.py
Normal file
@@ -0,0 +1,328 @@
|
||||
# app/scripts/check_ohlc_websocket.py
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
import websockets
|
||||
from websockets.typing import Subprotocol
|
||||
|
||||
from src.core.config import load_settings
|
||||
|
||||
|
||||
SYMBOLS = (
|
||||
"BTC/USD_LEVERAGE",
|
||||
)
|
||||
|
||||
INTERVALS = (
|
||||
"1m",
|
||||
"5m",
|
||||
)
|
||||
|
||||
CANDLE_TYPE = "classic"
|
||||
|
||||
MAX_MESSAGES = 30
|
||||
TOTAL_TIMEOUT_SECONDS = 300.0
|
||||
RECEIVE_TIMEOUT_SECONDS = 15.0
|
||||
|
||||
|
||||
def build_ws_url() -> str:
|
||||
settings = load_settings()
|
||||
|
||||
raw_url = settings.exchange_ws_url or settings.exchange_base_url
|
||||
|
||||
if raw_url.startswith("https://"):
|
||||
raw_url = raw_url.replace("https://", "wss://", 1)
|
||||
elif raw_url.startswith("http://"):
|
||||
raw_url = raw_url.replace("http://", "ws://", 1)
|
||||
|
||||
raw_url = raw_url.rstrip("/")
|
||||
|
||||
if raw_url.endswith("/connect"):
|
||||
return raw_url
|
||||
|
||||
return f"{raw_url}/connect"
|
||||
|
||||
|
||||
def build_headers() -> dict[str, str]:
|
||||
settings = load_settings()
|
||||
|
||||
headers = {
|
||||
"Origin": settings.exchange_base_url.rstrip("/"),
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
if settings.exchange_api_key:
|
||||
headers["X-MBX-APIKEY"] = settings.exchange_api_key
|
||||
|
||||
return headers
|
||||
|
||||
|
||||
def build_subscribe_request(
|
||||
symbols: tuple[str, ...],
|
||||
intervals: tuple[str, ...],
|
||||
candle_type: str,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"correlationId": str(uuid4()),
|
||||
"destination": "OHLCMarketData.subscribe",
|
||||
"payload": {
|
||||
"intervals": list(intervals),
|
||||
"symbols": list(symbols),
|
||||
"type": candle_type,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def message_to_text(
|
||||
raw_message: str | bytes | bytearray | memoryview,
|
||||
) -> str:
|
||||
if isinstance(raw_message, str):
|
||||
return raw_message
|
||||
|
||||
if isinstance(raw_message, memoryview):
|
||||
return raw_message.tobytes().decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
return bytes(raw_message).decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
|
||||
def decode_document(
|
||||
raw_message: str | bytes | bytearray | memoryview,
|
||||
) -> object | None:
|
||||
text = message_to_text(raw_message)
|
||||
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
|
||||
def format_json(
|
||||
raw_message: str | bytes | bytearray | memoryview,
|
||||
) -> str:
|
||||
text = message_to_text(raw_message)
|
||||
|
||||
try:
|
||||
document = json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return text
|
||||
|
||||
return json.dumps(
|
||||
document,
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
sort_keys=True,
|
||||
)
|
||||
|
||||
|
||||
def destination_from_document(document: object) -> str | None:
|
||||
if not isinstance(document, dict):
|
||||
return None
|
||||
|
||||
direct_destination = document.get("destination")
|
||||
|
||||
if isinstance(direct_destination, str):
|
||||
return direct_destination
|
||||
|
||||
upper_destination = document.get("Destination")
|
||||
|
||||
if isinstance(upper_destination, str):
|
||||
return upper_destination
|
||||
|
||||
payload = document.get("payload")
|
||||
|
||||
if isinstance(payload, dict):
|
||||
nested_destination = payload.get("destination")
|
||||
|
||||
if isinstance(nested_destination, str):
|
||||
return nested_destination
|
||||
|
||||
nested_upper_destination = payload.get("Destination")
|
||||
|
||||
if isinstance(nested_upper_destination, str):
|
||||
return nested_upper_destination
|
||||
|
||||
upper_payload = document.get("Payload")
|
||||
|
||||
if isinstance(upper_payload, dict):
|
||||
nested_destination = upper_payload.get("destination")
|
||||
|
||||
if isinstance(nested_destination, str):
|
||||
return nested_destination
|
||||
|
||||
nested_upper_destination = upper_payload.get("Destination")
|
||||
|
||||
if isinstance(nested_upper_destination, str):
|
||||
return nested_upper_destination
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def is_ohlc_event(document: object) -> bool:
|
||||
destination = destination_from_document(document)
|
||||
|
||||
if destination is None:
|
||||
return False
|
||||
|
||||
return destination.lower() == "ohlc.event"
|
||||
|
||||
|
||||
def elapsed_seconds(
|
||||
loop: asyncio.AbstractEventLoop,
|
||||
started_at: float,
|
||||
) -> float:
|
||||
return max(0.0, loop.time() - started_at)
|
||||
|
||||
|
||||
async def receive_ohlc() -> None:
|
||||
settings = load_settings()
|
||||
|
||||
ws_url = build_ws_url()
|
||||
headers = build_headers()
|
||||
|
||||
request = build_subscribe_request(
|
||||
SYMBOLS,
|
||||
INTERVALS,
|
||||
CANDLE_TYPE,
|
||||
)
|
||||
|
||||
print(f"WebSocket URL: {ws_url}")
|
||||
print(f"Symbols: {', '.join(SYMBOLS)}")
|
||||
print(f"Intervals: {', '.join(INTERVALS)}")
|
||||
print(f"Candle type: {CANDLE_TYPE}")
|
||||
print(f"Maximum messages: {MAX_MESSAGES}")
|
||||
print(f"Total timeout: {TOTAL_TIMEOUT_SECONDS:.0f} seconds")
|
||||
print()
|
||||
|
||||
print("Subscription request:")
|
||||
print(
|
||||
json.dumps(
|
||||
request,
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
)
|
||||
print()
|
||||
|
||||
async with websockets.connect(
|
||||
ws_url,
|
||||
extra_headers=headers,
|
||||
subprotocols=[Subprotocol("json")],
|
||||
ping_interval=20,
|
||||
ping_timeout=float(settings.exchange_timeout_sec),
|
||||
open_timeout=float(settings.exchange_timeout_sec),
|
||||
close_timeout=float(settings.exchange_timeout_sec),
|
||||
) as websocket:
|
||||
await websocket.send(
|
||||
json.dumps(
|
||||
request,
|
||||
ensure_ascii=False,
|
||||
)
|
||||
)
|
||||
|
||||
print("Subscription request sent.")
|
||||
print("Waiting for acknowledgement and OHLC events...")
|
||||
print()
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
started_at = loop.time()
|
||||
deadline = started_at + TOTAL_TIMEOUT_SECONDS
|
||||
|
||||
received_count = 0
|
||||
ohlc_event_count = 0
|
||||
timeout_count = 0
|
||||
|
||||
while received_count < MAX_MESSAGES:
|
||||
remaining_seconds = deadline - loop.time()
|
||||
|
||||
if remaining_seconds <= 0:
|
||||
print(
|
||||
"Total timeout reached after "
|
||||
f"{elapsed_seconds(loop, started_at):.1f} seconds."
|
||||
)
|
||||
break
|
||||
|
||||
receive_timeout = min(
|
||||
RECEIVE_TIMEOUT_SECONDS,
|
||||
remaining_seconds,
|
||||
)
|
||||
|
||||
try:
|
||||
raw_message = await asyncio.wait_for(
|
||||
websocket.recv(),
|
||||
timeout=receive_timeout,
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
timeout_count += 1
|
||||
|
||||
print(
|
||||
"No OHLC event yet; connection is alive. "
|
||||
f"Elapsed: {elapsed_seconds(loop, started_at):.1f}s, "
|
||||
f"remaining: {max(0.0, remaining_seconds):.1f}s."
|
||||
)
|
||||
continue
|
||||
|
||||
if not isinstance(
|
||||
raw_message,
|
||||
(str, bytes, bytearray, memoryview),
|
||||
):
|
||||
print(
|
||||
"Skipped unsupported WebSocket message type: "
|
||||
f"{type(raw_message).__name__}"
|
||||
)
|
||||
continue
|
||||
|
||||
received_count += 1
|
||||
document = decode_document(raw_message)
|
||||
|
||||
if is_ohlc_event(document):
|
||||
ohlc_event_count += 1
|
||||
message_kind = "OHLC EVENT"
|
||||
else:
|
||||
message_kind = "CONTROL / ACKNOWLEDGEMENT"
|
||||
|
||||
print("=" * 80)
|
||||
print(
|
||||
f"Message #{received_count} — {message_kind} — "
|
||||
f"elapsed {elapsed_seconds(loop, started_at):.1f}s"
|
||||
)
|
||||
print(format_json(raw_message))
|
||||
print()
|
||||
|
||||
print("=" * 80)
|
||||
print("Diagnostic summary")
|
||||
print(f"Received messages: {received_count}")
|
||||
print(f"OHLC events: {ohlc_event_count}")
|
||||
print(f"Receive timeouts: {timeout_count}")
|
||||
print(
|
||||
"Elapsed time: "
|
||||
f"{elapsed_seconds(loop, started_at):.1f} seconds"
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
try:
|
||||
asyncio.run(receive_ohlc())
|
||||
except KeyboardInterrupt:
|
||||
print()
|
||||
print("Stopped by user.")
|
||||
except Exception as exc:
|
||||
print()
|
||||
print(
|
||||
"WebSocket diagnostic failed: "
|
||||
f"{type(exc).__name__}: {exc}"
|
||||
)
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user