a062507656
MALKHUT asset directory (malkhut/assets/): normalized canonical symbols, KNOWN_EXCHANGES aux table (BINANCE/BINGX/BINGX_VST), per-exchange listing status (TRADING/OFFLINE/UNKNOWN), JSON-backed, built for full-Binance-500 scale. Seeded with BLUE's NG7 feed universe (50 symbols) + live VST contracts probe: 35 TRADING / 15 OFFLINE on VST (BAND, CELR, COS, CVC, DENT, FUN, HOT, ICX, TFUEL, TUSD, USDC, WAN, WIN, XTZ, ZIL). uv_asset_universe.init_asset_universe() = runtime tradable set for the execution exchange; --onboard runs MALKHUT AssetCompiler for full taxonomy. 12 new tests, mutation-RED verified (status-filter + tradability guard).
91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""
|
|
Tests for malkhut.assets.directory — the system-wide asset universe store.
|
|
|
|
Mutation-sensitivity: breaking normalize_symbol's uppercasing, listed_on's
|
|
status comparison, or symbols_for_exchange's filter goes RED here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from malkhut.assets.directory import (
|
|
KNOWN_EXCHANGES,
|
|
AssetDirectory,
|
|
ExchangeListing,
|
|
ListingStatus,
|
|
normalize_symbol,
|
|
)
|
|
|
|
|
|
def test_normalize_symbol_variants():
|
|
assert normalize_symbol("BAND-USDT") == "BANDUSDT"
|
|
assert normalize_symbol("band_usdt") == "BANDUSDT"
|
|
assert normalize_symbol(" eth/usdt ") == "ETHUSDT"
|
|
assert normalize_symbol("BANDUSDT") == "BANDUSDT"
|
|
|
|
|
|
def test_known_exchanges_table_has_the_three_venues():
|
|
assert {"BINANCE", "BINGX", "BINGX_VST"} <= set(KNOWN_EXCHANGES)
|
|
|
|
|
|
def test_listing_status_validated():
|
|
with pytest.raises(ValueError):
|
|
ExchangeListing(venue_symbol="X-USDT", status="LISTED") # not a valid status
|
|
|
|
|
|
def test_set_listing_rejects_unknown_exchange(tmp_path):
|
|
d = AssetDirectory(tmp_path / "d.json")
|
|
with pytest.raises(ValueError):
|
|
d.set_listing("BTCUSDT", "KRAKEN", status=ListingStatus.TRADING)
|
|
|
|
|
|
def test_upsert_splits_base_quote_for_usdt():
|
|
d = AssetDirectory.__new__(AssetDirectory)
|
|
d.records = {}
|
|
rec = d.upsert("BANDUSDT")
|
|
assert rec.base == "BAND" and rec.quote == "USDT"
|
|
|
|
|
|
def test_import_and_query_roundtrip(tmp_path):
|
|
p = tmp_path / "dir.json"
|
|
d = AssetDirectory(p)
|
|
n = d.import_symbols(
|
|
["ETHUSDT", "BANDUSDT", "CELRUSDT"], "BINANCE",
|
|
status=ListingStatus.TRADING, checked_at="2026-07-11T20:00:00Z",
|
|
source="dolphin_ng7_scan_feed",
|
|
)
|
|
assert n == 3
|
|
d.set_listing("BANDUSDT", "BINGX_VST", venue_symbol="BAND-USDT",
|
|
status=ListingStatus.OFFLINE, source="vst_contracts_probe")
|
|
d.set_listing("ETHUSDT", "BINGX_VST", venue_symbol="ETH-USDT",
|
|
status=ListingStatus.TRADING, source="vst_contracts_probe")
|
|
d.save()
|
|
|
|
d2 = AssetDirectory(p) # reload from disk
|
|
assert len(d2) == 3
|
|
assert d2.symbols_for_exchange("BINANCE") == ["BANDUSDT", "CELRUSDT", "ETHUSDT"]
|
|
# BAND is OFFLINE on VST → excluded from the tradable set
|
|
assert d2.symbols_for_exchange("BINGX_VST") == ["ETHUSDT"]
|
|
assert d2.symbols_for_exchange("BINGX_VST", status=ListingStatus.OFFLINE) == ["BANDUSDT"]
|
|
assert d2.venue_symbol("ETHUSDT", "BINGX_VST") == "ETH-USDT"
|
|
assert d2.venue_symbol("BANDUSDT", "BINGX") == "" # never probed on live BingX
|
|
assert d2.get("band-usdt").listed_on("BINANCE") is True
|
|
assert d2.get("band-usdt").listed_on("BINGX_VST") is False
|
|
|
|
|
|
def test_import_is_idempotent(tmp_path):
|
|
d = AssetDirectory(tmp_path / "d.json")
|
|
d.import_symbols(["ETHUSDT"], "BINANCE")
|
|
d.import_symbols(["ETHUSDT"], "BINANCE")
|
|
assert len(d) == 1
|
|
assert d.symbols_for_exchange("BINANCE") == ["ETHUSDT"]
|
|
|
|
|
|
def test_listing_on_second_exchange_preserves_first(tmp_path):
|
|
d = AssetDirectory(tmp_path / "d.json")
|
|
d.import_symbols(["ETHUSDT"], "BINANCE")
|
|
d.set_listing("ETHUSDT", "BINGX_VST", venue_symbol="ETH-USDT",
|
|
status=ListingStatus.TRADING)
|
|
rec = d.get("ETHUSDT")
|
|
assert set(rec.exchanges) == {"BINANCE", "BINGX_VST"}
|