malkhut: three-layer identifier architecture for cross-system asset identification

Layer 1 (canonical identity): symbol, base_asset, name, unified_symbol (CCXT format), quote_currency Layer 2 (cross-system): coingecko_id, cmc_id, blockchain, contract_address Layer 3 (exchange mapping): exchanges tuple All 13 pre-defined assets migrated with accurate CoinGecko IDs, CMC IDs, blockchains, and contract addresses (ERC-20 tokens). 7 new query functions: get_asset_by_coingecko_id, get_asset_by_cmc_id, get_assets_by_base_asset, get_assets_by_blockchain, get_assets_by_unified_symbol. 33 new tests covering: Layer 1 identity, Layer 2 cross-system identifiers, identifier query functions, identifier consistency (uniqueness, derivation), exchange registry. Total: 1189 tests, 47 files, all green. Based on research: CCXT BASE/QUOTE is de facto standard, CoinGecko ID most widely used in crypto-native, ISO 24165 DTI emerging, FIGI for institutional.
2026-07-12 00:40:14 +02:00
parent da61c8f6c2
commit 1f709af6d1
4 changed files with 341 additions and 13 deletions
--- a/MALKHUT/malkhut/tests/test_asset_classification.py
+++ b/MALKHUT/malkhut/tests/test_asset_classification.py
@@ -2,19 +2,24 @@
 Tests for asset classification — multi-label, invariant characteristics, exhaustive.
 Covers: per-asset profiles (all 13), multi-label overlap, cross-dimensional
 consistency, enum coverage, query edge cases, orderbook fingerprint ordering,
-predictive properties, and ScenarioFactory integration.
+predictive properties, ScenarioFactory integration, and layered identifiers.
 """
 import pytest
 from malkhut.training.asset_classification import (
    Sector, TokenRole, SupplyModel, ConsensusFamily, SmartContractCapability,
    MarketCapTier, DerivativeAccess, VolatilityProfile, LiquidityProfile,
    AssetProfile, ASSET_PROFILES,
+    ExchangeProfile, EXCHANGE_PROFILES, get_exchange, list_exchanges,
    get_asset_profile, list_assets,
    get_assets_by_sector, get_assets_by_token_role, get_assets_by_supply,
    get_assets_by_consensus, get_assets_by_market_cap, get_assets_by_volatility,
    get_assets_by_liquidity, get_assets_by_derivatives, get_gas_tokens,
    get_pov_assets, get_shortable_assets,
    get_multi_sector_assets, get_multi_role_assets,
+    get_assets_on_exchange, get_common_assets, get_exchange_for_asset,
+    get_asset_by_coingecko_id, get_asset_by_cmc_id,
+    get_assets_by_base_asset, get_assets_by_blockchain,
+    get_assets_by_unified_symbol,
 )
 from malkhut.training.cma_trainer import ScenarioFactory

@@ -996,3 +1001,200 @@ class TestMultiAssetScenarios:
        suite = factory.build_suite(symbols=("ETHUSDT",), steps_per_scenario=10)
        for s in suite:
            assert s.max_steps >= 10
+
+
+# ==============================================================================
+# Layered Identifier Architecture Tests
+# ==============================================================================
+
+class TestLayeredIdentifiers:
+    """Layer 1: Canonical identity."""
+
+    def test_all_have_base_asset(self):
+        for p in ASSET_PROFILES.values():
+            assert p.base_asset, f'{p.symbol}: empty base_asset'
+
+    def test_all_have_name(self):
+        for p in ASSET_PROFILES.values():
+            assert p.name, f'{p.symbol}: empty name'
+
+    def test_all_have_unified_symbol(self):
+        for p in ASSET_PROFILES.values():
+            assert '/' in p.unified_symbol, f'{p.symbol}: unified missing /'
+            base, quote = p.unified_symbol.split('/')
+            assert base == p.base_asset
+            assert quote == p.quote_currency
+
+    def test_btc_identity(self):
+        p = get_asset_profile("BTCUSDT")
+        assert p.base_asset == "BTC"
+        assert p.name == "BTC"
+        assert p.unified_symbol == "BTC/USDT"
+        assert p.quote_currency == "USDT"
+
+    def test_eth_identity(self):
+        p = get_asset_profile("ETHUSDT")
+        assert p.base_asset == "ETH"
+        assert p.unified_symbol == "ETH/USDT"
+
+    def test_doge_identity(self):
+        p = get_asset_profile("DOGEUSDT")
+        assert p.base_asset == "DOGE"
+        assert p.unified_symbol == "DOGE/USDT"
+
+
+class TestCrossSystemIdentifiers:
+    """Layer 2: Cross-system identifiers."""
+
+    def test_all_have_coingecko_id(self):
+        for p in ASSET_PROFILES.values():
+            assert p.coingecko_id, f'{p.symbol}: empty coingecko_id'
+
+    def test_all_have_cmc_id(self):
+        for p in ASSET_PROFILES.values():
+            assert p.cmc_id > 0, f'{p.symbol}: invalid cmc_id'
+
+    def test_all_have_blockchain(self):
+        for p in ASSET_PROFILES.values():
+            assert p.blockchain, f'{p.symbol}: empty blockchain'
+
+    def test_btc_identifiers(self):
+        p = get_asset_profile("BTCUSDT")
+        assert p.coingecko_id == "bitcoin"
+        assert p.cmc_id == 1
+        assert p.blockchain == "bitcoin"
+        assert p.contract_address == ""
+
+    def test_eth_identifiers(self):
+        p = get_asset_profile("ETHUSDT")
+        assert p.coingecko_id == "ethereum"
+        assert p.cmc_id == 1027
+        assert p.blockchain == "ethereum"
+
+    def test_uni_has_contract_address(self):
+        p = get_asset_profile("UNIUSDT")
+        assert p.contract_address.startswith("0x")
+        assert p.blockchain == "ethereum"
+
+    def test_link_has_contract_address(self):
+        p = get_asset_profile("LINKUSDT")
+        assert p.contract_address.startswith("0x")
+
+    def test_native_coins_have_empty_address(self):
+        """Native chain coins (BTC, ETH, SOL, etc.) have no contract address."""
+        for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT", "ADAUSDT",
+                    "AVAXUSDT", "DOTUSDT", "ATOMUSDT"):
+            p = get_asset_profile(sym)
+            assert p.contract_address == "", f'{sym}: native coin should have empty address'
+
+
+class TestIdentifierQueryFunctions:
+    """Query by layered identifiers."""
+
+    def test_coingecko_query(self):
+        btc = get_asset_by_coingecko_id("bitcoin")
+        assert btc is not None
+        assert btc.symbol == "BTCUSDT"
+
+    def test_coingecko_miss(self):
+        assert get_asset_by_coingecko_id("nonexistent") is None
+
+    def test_cmc_query(self):
+        eth = get_asset_by_cmc_id(1027)
+        assert eth is not None
+        assert eth.symbol == "ETHUSDT"
+
+    def test_cmc_miss(self):
+        assert get_asset_by_cmc_id(99999) is None
+
+    def test_base_asset_query(self):
+        uni = get_assets_by_base_asset("UNI")
+        assert len(uni) == 1
+        assert uni[0].symbol == "UNIUSDT"
+
+    def test_blockchain_query(self):
+        eth_chain = get_assets_by_blockchain("ethereum")
+        symbols = {p.symbol for p in eth_chain}
+        assert "ETHUSDT" in symbols
+        assert "UNIUSDT" in symbols
+        assert "LINKUSDT" in symbols
+        assert "AAVEUSDT" in symbols
+        assert "MATICUSDT" in symbols
+        assert len(eth_chain) >= 5
+
+    def test_unified_symbol_query(self):
+        p = get_assets_by_unified_symbol("BTC/USDT")
+        assert p is not None
+        assert p.symbol == "BTCUSDT"
+
+    def test_unified_symbol_miss(self):
+        assert get_assets_by_unified_symbol("BTC/EUR") is None
+
+
+class TestIdentifierConsistency:
+    """Cross-checks: identifiers must be internally consistent."""
+
+    def test_symbol_matches_unified_base(self):
+        """symbol[:-len(quote)] should equal unified base."""
+        for p in ASSET_PROFILES.values():
+            expected_base = p.symbol.replace(p.quote_currency, "")
+            assert p.base_asset == expected_base, \
+                f'{p.symbol}: base_asset={p.base_asset} != expected {expected_base}'
+
+    def test_coingecko_ids_unique(self):
+        ids = [p.coingecko_id for p in ASSET_PROFILES.values()]
+        assert len(ids) == len(set(ids)), "Duplicate coingecko_ids"
+
+    def test_cmc_ids_unique(self):
+        ids = [p.cmc_id for p in ASSET_PROFILES.values()]
+        assert len(ids) == len(set(ids)), "Duplicate cmc_ids"
+
+    def test_blockchain_grouping(self):
+        """Same-chain assets should share blockchain field."""
+        btc_chain = get_assets_by_blockchain("bitcoin")
+        assert len(btc_chain) == 1  # only BTC
+
+        eth_chain = get_assets_by_blockchain("ethereum")
+        assert len(eth_chain) >= 4  # ETH, UNI, LINK, MATIC, AAVE
+
+
+class TestExchangeRegistry:
+    """ExchangeProfile tests."""
+
+    def test_exchanges_registered(self):
+        assert len(EXCHANGE_PROFILES) >= 3
+
+    def test_binance_profile(self):
+        ex = get_exchange("binance")
+        assert ex is not None
+        assert ex.display_name == "Binance"
+        assert ex.has_spot
+        assert ex.has_perps
+        assert ex.has_options
+        assert ex.default_taker_fee_bps > 0
+        assert ex.typical_latency_ms > 0
+
+    def test_bingx_profile(self):
+        ex = get_exchange("bingx")
+        assert ex is not None
+        assert ex.display_name == "BingX"
+        assert not ex.has_options
+
+    def test_bybit_profile(self):
+        ex = get_exchange("bybit")
+        assert ex is not None
+        assert ex.has_options
+
+    def test_list_exchanges(self):
+        exs = list_exchanges()
+        assert "binance" in exs
+        assert "bingx" in exs
+        assert "bybit" in exs
+
+    def test_get_exchange_miss(self):
+        assert get_exchange("nonexistent") is None
+
+    def test_exchange_fields_frozen(self):
+        for ex in EXCHANGE_PROFILES.values():
+            with pytest.raises(AttributeError):
+                ex.display_name = "X"