Codex
94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from crypto_spot_bot.bybit import BybitClient, websocket_subscribe_message, _looks_like_leveraged_token, _looks_like_stablecoin
|
|
|
|
|
|
def test_leveraged_token_filter() -> None:
|
|
assert _looks_like_leveraged_token("BTC3L")
|
|
assert _looks_like_leveraged_token("ETHDOWN")
|
|
assert not _looks_like_leveraged_token("BTC")
|
|
|
|
|
|
def test_stablecoin_filter() -> None:
|
|
assert _looks_like_stablecoin("USDC")
|
|
assert _looks_like_stablecoin("FDUSD")
|
|
assert not _looks_like_stablecoin("BTC")
|
|
|
|
|
|
def test_spot_instrument_uses_min_order_amt_and_base_precision(make_settings, tmp_path) -> None:
|
|
client = BybitClient(make_settings(tmp_path))
|
|
client.public_get = lambda *_args, **_kwargs: {
|
|
"list": [
|
|
{
|
|
"symbol": "BTCUSDT",
|
|
"baseCoin": "BTC",
|
|
"quoteCoin": "USDT",
|
|
"status": "Trading",
|
|
"priceFilter": {"tickSize": "0.01"},
|
|
"lotSizeFilter": {
|
|
"basePrecision": "0.000001",
|
|
"minOrderQty": "0.000001",
|
|
"minOrderAmt": "5",
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
instrument = client.instruments()["BTCUSDT"]
|
|
|
|
assert instrument.qty_step == 0.000001
|
|
assert instrument.min_notional_value == 5.0
|
|
|
|
|
|
def test_live_spot_order_explicitly_disables_leverage(make_settings, tmp_path) -> None:
|
|
client = BybitClient(make_settings(tmp_path))
|
|
captured = {}
|
|
|
|
def fake_private_post(path, payload):
|
|
captured["path"] = path
|
|
captured["payload"] = payload
|
|
return {"orderId": "test"}
|
|
|
|
client.private_post = fake_private_post
|
|
client.place_spot_market_order("BTCUSDT", "Buy", 10, "quoteCoin", "order-1")
|
|
|
|
assert captured["path"] == "/v5/order/create"
|
|
assert captured["payload"]["category"] == "spot"
|
|
assert captured["payload"]["isLeverage"] == 0
|
|
assert captured["payload"]["orderFilter"] == "Order"
|
|
|
|
|
|
def test_private_get_signs_the_same_query_it_sends(make_settings, tmp_path) -> None:
|
|
client = BybitClient(make_settings(tmp_path))
|
|
captured = {}
|
|
|
|
class Response:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"retCode": 0, "result": {"ok": True}}
|
|
|
|
class Session:
|
|
def get(self, url, params, headers, timeout):
|
|
captured["url"] = url
|
|
captured["params"] = params
|
|
captured["headers"] = headers
|
|
captured["timeout"] = timeout
|
|
return Response()
|
|
|
|
client.session = Session()
|
|
|
|
assert client.wallet_balance(coin=None) == {"ok": True}
|
|
|
|
assert captured["params"] == [("accountType", "UNIFIED")]
|
|
assert "coin" not in dict(captured["params"])
|
|
assert captured["headers"]["X-BAPI-SIGN"]
|
|
|
|
|
|
def test_websocket_subscribe_uses_configured_kline_interval() -> None:
|
|
payload = websocket_subscribe_message(["BTCUSDT"], interval="60")
|
|
|
|
assert "kline.60.BTCUSDT" in payload
|
|
assert "kline.1.BTCUSDT" not in payload
|