mirror of
https://github.com/OpenByteInc/QuantDinger.git
synced 2026-09-28 15:22:56 +08:00
313 lines
11 KiB
Python
313 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from types import SimpleNamespace
|
|
from urllib.parse import urlencode
|
|
|
|
import pytest
|
|
|
|
from app.services.execution_streams.adapters import (
|
|
ADAPTERS,
|
|
AlpacaExecutionAdapter,
|
|
BinanceExecutionAdapter,
|
|
BitgetExecutionAdapter,
|
|
BybitExecutionAdapter,
|
|
GateExecutionAdapter,
|
|
HtxExecutionAdapter,
|
|
OkxExecutionAdapter,
|
|
)
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self) -> None:
|
|
self.messages: list[object] = []
|
|
self.closed = False
|
|
|
|
def send(self, raw: str) -> None:
|
|
try:
|
|
self.messages.append(json.loads(raw))
|
|
except (TypeError, json.JSONDecodeError):
|
|
self.messages.append(raw)
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class StopAfterOneHeartbeat:
|
|
def __init__(self) -> None:
|
|
self.waits = 0
|
|
|
|
def wait(self, _timeout: float) -> bool:
|
|
self.waits += 1
|
|
return self.waits > 1
|
|
|
|
|
|
def _adapter(adapter_cls, *, market_type="swap", symbols=()):
|
|
states: list[str] = []
|
|
adapter = adapter_cls(
|
|
credential_id=9,
|
|
user_id=3,
|
|
exchange_id="test",
|
|
market_type=market_type,
|
|
config={
|
|
"api_key": "key",
|
|
"secret_key": "secret",
|
|
"passphrase": "pass",
|
|
"paper": True,
|
|
},
|
|
symbols=symbols,
|
|
on_event=lambda _event: None,
|
|
on_state=lambda state, _error, _reconnect: states.append(state),
|
|
)
|
|
return adapter, states
|
|
|
|
|
|
def test_adapter_registry_covers_six_exchanges_and_two_brokers():
|
|
assert set(ADAPTERS) == {
|
|
"binance",
|
|
"okx",
|
|
"bitget",
|
|
"bybit",
|
|
"gate",
|
|
"htx",
|
|
"alpaca",
|
|
"ibkr",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"adapter_cls",
|
|
(OkxExecutionAdapter, BybitExecutionAdapter, BitgetExecutionAdapter, HtxExecutionAdapter, AlpacaExecutionAdapter),
|
|
)
|
|
def test_authenticated_adapters_are_not_healthy_before_auth_ack(adapter_cls):
|
|
adapter, _states = _adapter(adapter_cls)
|
|
assert adapter.ready_on_open() is False
|
|
assert adapter.connected is False
|
|
|
|
|
|
def test_okx_subscribes_only_after_successful_login():
|
|
adapter, states = _adapter(OkxExecutionAdapter)
|
|
assert adapter.on_open_messages()[0]["op"] == "login"
|
|
ws = FakeSocket()
|
|
assert adapter.handle_control(ws, {"event": "login", "code": "0"})
|
|
assert ws.messages == [{"op": "subscribe", "args": [{"channel": "orders", "instType": "ANY"}]}]
|
|
assert adapter.connected
|
|
assert states == ["connected"]
|
|
|
|
|
|
def test_okx_and_bitget_use_literal_ping_and_bybit_uses_json_ping():
|
|
okx, _states = _adapter(OkxExecutionAdapter)
|
|
bitget, _states = _adapter(BitgetExecutionAdapter)
|
|
bybit, _states = _adapter(BybitExecutionAdapter)
|
|
|
|
assert 0 < okx.application_heartbeat_interval() < 30
|
|
assert okx.application_heartbeat_payload() == "ping"
|
|
assert 0 < bitget.application_heartbeat_interval() < 30
|
|
assert bitget.application_heartbeat_payload() == "ping"
|
|
assert bybit.application_heartbeat_interval() == 20
|
|
assert bybit.application_heartbeat_payload() == {"op": "ping"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("adapter_cls", "expected"),
|
|
(
|
|
(OkxExecutionAdapter, "ping"),
|
|
(BitgetExecutionAdapter, "ping"),
|
|
(BybitExecutionAdapter, {"op": "ping"}),
|
|
),
|
|
)
|
|
def test_application_heartbeat_sends_venue_payload(adapter_cls, expected):
|
|
adapter, _states = _adapter(adapter_cls)
|
|
adapter._last_message_at = -100.0
|
|
ws = FakeSocket()
|
|
|
|
adapter._application_heartbeat_loop(ws, StopAfterOneHeartbeat())
|
|
|
|
assert ws.messages == [expected]
|
|
|
|
|
|
@pytest.mark.parametrize(("environment", "expected_url"), (
|
|
("live", "wss://ws-api.binance.com:443/ws-api/v3"),
|
|
("demo", "wss://demo-ws-api.binance.com/ws-api/v3"),
|
|
("testnet", "wss://ws-api.testnet.binance.vision/ws-api/v3"),
|
|
))
|
|
def test_binance_spot_uses_signed_websocket_api_subscription(monkeypatch, environment, expected_url):
|
|
monkeypatch.setattr("app.services.execution_streams.adapters.time.time", lambda: 1_700_000_000.125)
|
|
monkeypatch.setattr("app.services.execution_streams.adapters.requests.post", lambda *a, **kw: pytest.fail("Spot must not request a listenKey"))
|
|
adapter, states = _adapter(BinanceExecutionAdapter, market_type="spot")
|
|
adapter.config["environment"] = environment
|
|
|
|
assert adapter.ready_on_open() is False
|
|
assert adapter.url() == expected_url
|
|
adapter.prepare() # Must not call the retired REST listenKey endpoint.
|
|
request = adapter.on_open_messages()[0]
|
|
assert request["method"] == "userDataStream.subscribe.signature"
|
|
params = request["params"]
|
|
expected_payload = urlencode(sorted({
|
|
"apiKey": "key",
|
|
"timestamp": 1_700_000_000_125,
|
|
}.items()))
|
|
assert params["signature"] == hmac.new(
|
|
b"secret", expected_payload.encode(), hashlib.sha256
|
|
).hexdigest()
|
|
|
|
ws = FakeSocket()
|
|
assert adapter.handle_control(
|
|
ws,
|
|
{"id": request["id"], "status": 200, "result": {"subscriptionId": 0}},
|
|
)
|
|
assert adapter.connected
|
|
assert states == ["connected"]
|
|
|
|
|
|
@pytest.mark.parametrize(("environment", "rest_host", "ws_host"), (
|
|
("live", "https://fapi.binance.com", "wss://fstream.binance.com/ws"),
|
|
("demo", "https://demo-fapi.binance.com", "wss://fstream.binancefuture.com/ws"),
|
|
))
|
|
def test_binance_futures_retains_listen_key_subscription(monkeypatch, environment, rest_host, ws_host):
|
|
adapter, _states = _adapter(BinanceExecutionAdapter)
|
|
adapter.config["environment"] = environment
|
|
calls = []
|
|
response = SimpleNamespace(raise_for_status=lambda: None, json=lambda: {"listenKey": "stream-key"})
|
|
monkeypatch.setattr("app.services.execution_streams.adapters.requests.post", lambda url, **kw: calls.append((url, kw)) or response)
|
|
adapter._keepalive_thread = SimpleNamespace(is_alive=lambda: True)
|
|
|
|
adapter.prepare()
|
|
|
|
assert calls[0][0] == rest_host + "/fapi/v1/listenKey"
|
|
assert calls[0][1]["headers"]["X-MBX-APIKEY"] == "key"
|
|
assert adapter.url() == ws_host + "/stream-key"
|
|
assert adapter.ready_on_open()
|
|
assert adapter.on_open_messages() == []
|
|
|
|
|
|
def test_binance_websocket_api_event_envelope_is_unwrapped():
|
|
adapter, _states = _adapter(BinanceExecutionAdapter, market_type="spot")
|
|
events = adapter.parse({
|
|
"subscriptionId": 0,
|
|
"event": {
|
|
"e": "executionReport",
|
|
"E": 1_700_000_000_000,
|
|
"s": "BTCUSDT",
|
|
"i": 12,
|
|
"t": 34,
|
|
"S": "BUY",
|
|
"X": "FILLED",
|
|
"L": "100",
|
|
"l": "0.5",
|
|
"z": "0.5",
|
|
},
|
|
})
|
|
|
|
assert len(events) == 1
|
|
assert events[0].symbol == "BTC/USDT"
|
|
assert events[0].exchange_order_id == "12"
|
|
|
|
|
|
def test_bitget_demo_and_live_use_matching_private_websocket_hosts():
|
|
demo, _states = _adapter(BitgetExecutionAdapter)
|
|
live, _states = _adapter(BitgetExecutionAdapter)
|
|
live.config["environment"] = "live"
|
|
|
|
assert demo.url() == "wss://wspap.bitget.com/v2/ws/private"
|
|
assert live.url() == "wss://ws.bitget.com/v2/ws/private"
|
|
|
|
|
|
def test_bybit_subscribes_only_after_successful_authentication():
|
|
adapter, _states = _adapter(BybitExecutionAdapter)
|
|
ws = FakeSocket()
|
|
adapter.handle_control(ws, {"op": "auth", "success": True})
|
|
assert ws.messages == [{"op": "subscribe", "args": ["execution"]}]
|
|
assert adapter.connected
|
|
|
|
|
|
def test_bybit_heartbeat_response_does_not_trigger_an_extra_pong():
|
|
adapter, _states = _adapter(BybitExecutionAdapter)
|
|
ws = FakeSocket()
|
|
|
|
assert adapter.handle_control(ws, {"success": True, "ret_msg": "pong", "op": "ping"})
|
|
assert ws.messages == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("market_type", "inst_type"),
|
|
(("spot", "SPOT"), ("swap", "USDT-FUTURES")),
|
|
)
|
|
def test_bitget_uses_market_specific_fill_subscription(market_type, inst_type):
|
|
adapter, _states = _adapter(BitgetExecutionAdapter, market_type=market_type)
|
|
assert adapter.on_open_messages()[0]["op"] == "login"
|
|
ws = FakeSocket()
|
|
adapter.handle_control(ws, {"event": "login", "code": "0"})
|
|
assert ws.messages[0]["args"][0] == {
|
|
"instType": inst_type,
|
|
"channel": "fill",
|
|
"instId": "default",
|
|
}
|
|
|
|
|
|
def test_gate_becomes_healthy_only_after_authenticated_subscription_ack():
|
|
adapter, _states = _adapter(GateExecutionAdapter, market_type="spot")
|
|
assert adapter.ready_on_open() is False
|
|
request = adapter.on_open_messages()[0]
|
|
assert request["channel"] == "spot.usertrades"
|
|
assert request["auth"]["KEY"] == "key"
|
|
adapter.handle_control(FakeSocket(), {"event": "subscribe", "result": {"status": "success"}})
|
|
assert adapter.connected
|
|
|
|
|
|
def test_gate_uses_current_official_testnet_websocket_paths():
|
|
spot, _states = _adapter(GateExecutionAdapter, market_type="spot")
|
|
swap, _states = _adapter(GateExecutionAdapter, market_type="swap")
|
|
|
|
assert spot.url() == "wss://ws-testnet.gate.com/v4/ws/spot"
|
|
assert swap.url() == "wss://ws-testnet.gate.com/v4/ws/futures/usdt"
|
|
|
|
|
|
def test_gate_futures_logs_in_for_uid_before_subscribing_to_all_contracts():
|
|
adapter, _states = _adapter(GateExecutionAdapter, market_type="swap")
|
|
login = adapter.on_open_messages()[0]
|
|
assert login["channel"] == "futures.login"
|
|
assert login["event"] == "api"
|
|
assert login["payload"]["api_key"] == "key"
|
|
assert login["payload"]["signature"]
|
|
|
|
ws = FakeSocket()
|
|
adapter.handle_control(
|
|
ws,
|
|
{
|
|
"header": {"channel": "futures.login", "event": "api", "status": "200"},
|
|
"data": {"result": {"api_key": "key", "uid": "110284739"}},
|
|
},
|
|
)
|
|
assert ws.messages[0]["channel"] == "futures.usertrades"
|
|
assert ws.messages[0]["payload"] == ["110284739", "!all"]
|
|
assert adapter.connected is False
|
|
|
|
adapter.handle_control(ws, {"event": "subscribe", "result": {"status": "success"}})
|
|
assert adapter.connected
|
|
|
|
|
|
def test_htx_spot_subscribes_known_symbols_after_authentication():
|
|
adapter, _states = _adapter(HtxExecutionAdapter, market_type="spot", symbols=("BTC/USDT", "ETH/USDT"))
|
|
assert adapter.on_open_messages()[0]["ch"] == "auth"
|
|
ws = FakeSocket()
|
|
adapter.handle_control(ws, {"action": "req", "ch": "auth", "code": 200})
|
|
assert {message["ch"] for message in ws.messages} == {
|
|
"trade.clearing#btcusdt",
|
|
"trade.clearing#ethusdt",
|
|
}
|
|
|
|
|
|
def test_alpaca_listens_for_trade_updates_after_authorization():
|
|
adapter, _states = _adapter(AlpacaExecutionAdapter, market_type="usstock")
|
|
assert adapter.on_open_messages()[0]["action"] == "auth"
|
|
ws = FakeSocket()
|
|
adapter.handle_control(
|
|
ws,
|
|
{"stream": "authorization", "data": {"status": "authorized"}},
|
|
)
|
|
assert ws.messages == [{"action": "listen", "data": {"streams": ["trade_updates"]}}]
|
|
assert adapter.connected
|