110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from llm_trade_bot.binance_client import (
|
|
SPOT_REST_URL,
|
|
BinanceCredentials,
|
|
BinanceRestClient,
|
|
encode_parameters,
|
|
sign_parameters,
|
|
)
|
|
|
|
|
|
def test_hmac_signature_uses_sorted_urlencoded_parameters():
|
|
parameters = {
|
|
"symbol": "BTCUSDT",
|
|
"side": "BUY",
|
|
"quantity": "0.01000000",
|
|
"recvWindow": 100,
|
|
"timestamp": 1645423376532,
|
|
}
|
|
payload = encode_parameters(parameters)
|
|
assert payload == "quantity=0.01000000&recvWindow=100&side=BUY&symbol=BTCUSDT×tamp=1645423376532"
|
|
assert sign_parameters(parameters, "secret") == hmac.new(
|
|
b"secret", payload.encode(), hashlib.sha256
|
|
).hexdigest()
|
|
|
|
|
|
async def test_signed_request_resynchronizes_time_and_resigns():
|
|
seen: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.append(request)
|
|
if request.url.path == "/api/v3/time":
|
|
return httpx.Response(200, json={"serverTime": 1_000_100}, request=request)
|
|
if len([item for item in seen if item.url.path == "/api/v3/account"]) == 1:
|
|
return httpx.Response(400, json={"code": -1021, "msg": "outside recvWindow"}, request=request)
|
|
query = dict(request.url.params)
|
|
unsigned = {key: value for key, value in query.items() if key != "signature"}
|
|
assert query["signature"] == sign_parameters(unsigned, "secret")
|
|
assert query["recvWindow"] == "5000"
|
|
return httpx.Response(200, json={"ok": True}, request=request)
|
|
|
|
client = BinanceRestClient(
|
|
SPOT_REST_URL,
|
|
BinanceCredentials("key", "secret"),
|
|
http_client=httpx.AsyncClient(base_url=SPOT_REST_URL, transport=httpx.MockTransport(handler)),
|
|
clock=lambda: 1000.0,
|
|
)
|
|
assert await client.request("GET", "/api/v3/account", signed=True) == {"ok": True}
|
|
assert [request.url.path for request in seen] == ["/api/v3/account", "/api/v3/time", "/api/v3/account"]
|
|
|
|
|
|
async def test_rate_limit_honors_retry_after():
|
|
attempts = 0
|
|
pauses: list[float] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal attempts
|
|
attempts += 1
|
|
if attempts == 1:
|
|
return httpx.Response(429, headers={"Retry-After": "0.25"}, json={"code": -1003}, request=request)
|
|
return httpx.Response(200, json={"value": "ok"}, request=request)
|
|
|
|
async def sleep(delay: float) -> None:
|
|
pauses.append(delay)
|
|
|
|
client = BinanceRestClient(
|
|
SPOT_REST_URL,
|
|
BinanceCredentials("key", "secret"),
|
|
http_client=httpx.AsyncClient(base_url=SPOT_REST_URL, transport=httpx.MockTransport(handler)),
|
|
sleep=sleep,
|
|
)
|
|
assert await client.request("GET", "/api/v3/ping") == {"value": "ok"}
|
|
assert pauses == [0.25]
|
|
|
|
|
|
async def test_ambiguous_mutation_queries_client_id_instead_of_retrying_post():
|
|
methods: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
methods.append(request.method)
|
|
if request.method == "POST":
|
|
return httpx.Response(500, json={"msg": "backend timeout"}, request=request)
|
|
return httpx.Response(200, json={"orderId": 42, "clientOrderId": "ltb-1"}, request=request)
|
|
|
|
client = BinanceRestClient(
|
|
SPOT_REST_URL,
|
|
BinanceCredentials("key", "secret"),
|
|
http_client=httpx.AsyncClient(base_url=SPOT_REST_URL, transport=httpx.MockTransport(handler)),
|
|
)
|
|
result = await client.mutate_with_recovery(
|
|
"POST",
|
|
"/api/v3/order",
|
|
{"symbol": "BTCUSDT", "newClientOrderId": "ltb-1"},
|
|
lookup_path="/api/v3/order",
|
|
lookup_parameters={"symbol": "BTCUSDT", "origClientOrderId": "ltb-1"},
|
|
)
|
|
assert result["orderId"] == 42
|
|
assert methods == ["POST", "GET"]
|
|
|
|
|
|
def test_production_url_is_not_accepted():
|
|
with pytest.raises(ValueError, match="Testnet"):
|
|
BinanceRestClient("https://api.binance.com", BinanceCredentials("key", "secret"))
|