from decimal import Decimal from ktrade.kraken.checksum import _enc, book_checksum, fmt_fixed from ktrade.kraken.ws import LocalBook def test_encoding_rules_from_docs(): assert _enc("45285.2") == "452852" assert _enc("0.00100000") == "100000" assert fmt_fixed(Decimal("45285.2"), 1) == "45285.2" assert fmt_fixed(Decimal("0.001"), 8) == "0.00100000" def test_checksum_is_order_independent_and_top10_only(): asks = [(f"{100 + i}.0", "1.00000000") for i in range(15)] bids = [(f"{99 - i}.0", "1.00000000") for i in range(15)] a = book_checksum(asks, bids) b = book_checksum(list(reversed(asks)), list(reversed(bids))) assert a == b # levels beyond the top 10 do not matter assert book_checksum(asks[:10], bids[:10]) == a assert book_checksum(asks[:9], bids[:10]) != a def test_local_book_apply_and_checksum(): b = LocalBook("BTC/USD", depth=10, price_decimals=1, qty_decimals=8) snap = {"bids": [{"price": 100.0, "qty": 1.0}, {"price": 99.5, "qty": 2.0}], "asks": [{"price": 100.5, "qty": 1.5}, {"price": 101.0, "qty": 3.0}], "timestamp": "t"} expected = book_checksum([("100.5", "1.50000000"), ("101.0", "3.00000000")], [("100.0", "1.00000000"), ("99.5", "2.00000000")]) assert b.apply("snapshot", {**snap, "checksum": expected}) # update: remove a level, change another upd = {"bids": [{"price": 99.5, "qty": 0}], "asks": [{"price": 100.5, "qty": 0.5}]} expected2 = book_checksum([("100.5", "0.50000000"), ("101.0", "3.00000000")], [("100.0", "1.00000000")]) assert b.apply("update", {**upd, "checksum": expected2}) assert not b.apply("update", {"bids": [], "asks": [], "checksum": 12345}) top = b.top() assert top["best_bid"] == 100.0 and top["best_ask"] == 100.5 and top["spread"] == 0.5