diff --git a/MALKHUT/malkhut/tests/test_asset_bridge.py b/MALKHUT/malkhut/tests/test_asset_bridge.py new file mode 100644 index 0000000..7abf335 --- /dev/null +++ b/MALKHUT/malkhut/tests/test_asset_bridge.py @@ -0,0 +1,419 @@ +""" +Tests for asset bridge (directory ↔ classification integration). + +Covers: +- Bridge sync (directory TRADING statuses → AssetProfile.exchanges) +- normalize_symbol correctness +- ExchangeListing validation +- AssetDirectory CRUD + persistence +- symbols_for_exchange filtering by status +- venue_symbol mapping +- Import idempotence +- Round-trip load/save +- Integration with ScenarioFactory +- Edge cases: unknown exchange, empty directory, re-sync +""" +import pytest +from pathlib import Path + +from malkhut.assets.directory import ( + KNOWN_EXCHANGES, AssetDirectory, AssetRecord, + ExchangeListing, ListingStatus, normalize_symbol, +) +from malkhut.training.asset_bridge import ( + sync_asset_to_profile, sync_exchanges_from_directory, get_universe_stats, +) +from malkhut.training.asset_classification import ( + ASSET_PROFILES, get_asset_profile, +) +from malkhut.training.cma_trainer import ScenarioFactory + + +# ============================================================================== +# normalize_symbol — canonical form +# ============================================================================== + +class TestNormalizeSymbol: + def test_dash_separator(self): + assert normalize_symbol("BTC-USDT") == "BTCUSDT" + + def test_underscore_separator(self): + assert normalize_symbol("eth_usdt") == "ETHUSDT" + + def test_slash_separator(self): + assert normalize_symbol("ETH/USDT") == "ETHUSDT" + + def test_whitespace_stripped(self): + assert normalize_symbol(" BTC-USDT ") == "BTCUSDT" + + def test_already_normalized(self): + assert normalize_symbol("BTCUSDT") == "BTCUSDT" + + def test_lowercase_normalized(self): + assert normalize_symbol("band-usdt") == "BANDUSDT" + + def test_mixed_separators(self): + assert normalize_symbol("BAND-USDT_USDC/BTC") == "BANDUSDTUSDCBTC" + + +# ============================================================================== +# ExchangeListing — validation +# ============================================================================== + +class TestExchangeListing: + def test_valid_listing(self): + l = ExchangeListing(venue_symbol="BTC-USDT", status=ListingStatus.TRADING) + assert l.venue_symbol == "BTC-USDT" + assert l.status == "TRADING" + + def test_invalid_status_rejected(self): + with pytest.raises(ValueError): + ExchangeListing(venue_symbol="X", status="INVALID") + + def test_unknown_status(self): + l = ExchangeListing(venue_symbol="X", status=ListingStatus.UNKNOWN) + assert l.status == "UNKNOWN" + + def test_offline_status(self): + l = ExchangeListing(venue_symbol="X", status=ListingStatus.OFFLINE) + assert l.status == "OFFLINE" + + +# ============================================================================== +# AssetRecord — listing queries +# ============================================================================== + +class TestAssetRecord: + def test_listed_on_trading(self): + r = AssetRecord( + symbol="BTCUSDT", + exchanges={"BINANCE": ExchangeListing("BTCUSDT", "TRADING")}, + ) + assert r.listed_on("BINANCE") is True + + def test_listed_on_offline(self): + r = AssetRecord( + symbol="BTCUSDT", + exchanges={"BINANCE": ExchangeListing("BTCUSDT", "OFFLINE")}, + ) + assert r.listed_on("BINANCE") is False + + def test_listed_on_unknown(self): + r = AssetRecord( + symbol="BTCUSDT", + exchanges={"BINANCE": ExchangeListing("BTCUSDT", "UNKNOWN")}, + ) + assert r.listed_on("BINANCE") is False + + def test_listed_on_missing(self): + r = AssetRecord(symbol="BTCUSDT") + assert r.listed_on("BINANCE") is False + + def test_base_quote_split(self): + r = AssetRecord(symbol="BTCUSDT", base="BTC", quote="USDT") + assert r.base == "BTC" + assert r.quote == "USDT" + + +# ============================================================================== +# AssetDirectory — CRUD + persistence +# ============================================================================== + +class TestAssetDirectory: + def test_empty_directory(self, tmp_path): + d = AssetDirectory(tmp_path / "empty.json") + assert len(d) == 0 + + def test_upsert_creates(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + rec = d.upsert("BTCUSDT") + assert rec.symbol == "BTCUSDT" + assert rec.base == "BTC" + assert rec.quote == "USDT" + assert len(d) == 1 + + def test_upsert_is_idempotent(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.upsert("BTCUSDT") + d.upsert("BTCUSDT") + assert len(d) == 1 + + def test_upsert_splits_usdt(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + rec = d.upsert("ETHUSDT") + assert rec.base == "ETH" + assert rec.quote == "USDT" + + def test_upsert_custom_base_quote(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + rec = d.upsert("XBTC", base="XBT", quote="USD") + assert rec.base == "XBT" + assert rec.quote == "USD" + + def test_get_returns_none_for_missing(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + assert d.get("BTCUSDT") is None + + def test_get_case_insensitive(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.upsert("BTCUSDT") + assert d.get("btcusdt") is not None + assert d.get("BTC-USDT") is not None + + def test_set_listing_creates_exchange(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.set_listing("BTCUSDT", "BINANCE", status=ListingStatus.TRADING) + rec = d.get("BTCUSDT") + assert rec.listed_on("BINANCE") + + def test_set_listing_rejects_unknown_exchange(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + with pytest.raises(ValueError): + d.set_listing("BTCUSDT", "KRAKEN") + + def test_set_listing_custom_venue_symbol(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.set_listing("ETHUSDT", "BINGX_VST", venue_symbol="ETH-USDT") + assert d.venue_symbol("ETHUSDT", "BINGX_VST") == "ETH-USDT" + + def test_import_symbols_bulk(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + n = d.import_symbols( + ["BTCUSDT", "ETHUSDT", "SOLUSDT"], "BINANCE", + status=ListingStatus.TRADING, + ) + assert n == 3 + assert len(d) == 3 + + def test_import_is_idempotent(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT"], "BINANCE") + d.import_symbols(["BTCUSDT"], "BINANCE") + assert len(d) == 1 + + def test_import_preserves_other_exchanges(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT"], "BINANCE") + d.set_listing("BTCUSDT", "BINGX_VST", venue_symbol="BTC-USDT", + status=ListingStatus.TRADING) + d.import_symbols(["BTCUSDT"], "BINANCE") # re-import + rec = d.get("BTCUSDT") + assert set(rec.exchanges) == {"BINANCE", "BINGX_VST"} + + def test_save_and_load(self, tmp_path): + p = tmp_path / "d.json" + d = AssetDirectory(p) + d.import_symbols(["BTCUSDT", "ETHUSDT"], "BINANCE") + d.set_listing("BTCUSDT", "BINGX_VST", venue_symbol="BTC-USDT", + status=ListingStatus.TRADING) + d.save() + + d2 = AssetDirectory(p) + assert len(d2) == 2 + assert d2.symbols_for_exchange("BINANCE") == ["BTCUSDT", "ETHUSDT"] + assert d2.symbols_for_exchange("BINGX_VST") == ["BTCUSDT"] + + def test_save_atomic(self, tmp_path): + """Save uses tmp-rename — no partial writes.""" + p = tmp_path / "d.json" + d = AssetDirectory(p) + d.import_symbols(["BTCUSDT"], "BINANCE") + d.save() + assert p.exists() + assert not (tmp_path / "d.json.tmp").exists() + + +# ============================================================================== +# symbols_for_exchange — filtering by status +# ============================================================================== + +class TestSymbolsForExchange: + def test_trading_only(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT", "ETHUSDT"], "BINANCE") + d.set_listing("BTCUSDT", "BINGX_VST", status=ListingStatus.TRADING) + d.set_listing("ETHUSDT", "BINGX_VST", status=ListingStatus.OFFLINE) + + assert d.symbols_for_exchange("BINGX_VST") == ["BTCUSDT"] + assert d.symbols_for_exchange("BINGX_VST", status=ListingStatus.OFFLINE) == ["ETHUSDT"] + + def test_sorted_output(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["SOLUSDT", "BTCUSDT", "ETHUSDT"], "BINANCE") + result = d.symbols_for_exchange("BINANCE") + assert result == ["BTCUSDT", "ETHUSDT", "SOLUSDT"] + + def test_unknown_exchange_raises(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + with pytest.raises(ValueError): + d.symbols_for_exchange("KRAKEN") + + def test_empty_exchange(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT"], "BINANCE") + assert d.symbols_for_exchange("BINGX_VST") == [] + + +# ============================================================================== +# venue_symbol — venue-local spelling +# ============================================================================== + +class TestVenueSymbol: + def test_venue_symbol_when_set(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.set_listing("ETHUSDT", "BINGX_VST", venue_symbol="ETH-USDT") + assert d.venue_symbol("ETHUSDT", "BINGX_VST") == "ETH-USDT" + + def test_venue_symbol_defaults_to_canonical(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.set_listing("ETHUSDT", "BINANCE") # no venue_symbol → uses canonical + assert d.venue_symbol("ETHUSDT", "BINANCE") == "ETHUSDT" + + def test_venue_symbol_unlisted(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + assert d.venue_symbol("BTCUSDT", "BINANCE") == "" + + +# ============================================================================== +# Bridge — directory → profile sync +# ============================================================================== + +class TestBridgeSync: + """Tests for directory → profile sync. Each test saves/restores profile state.""" + + @pytest.fixture(autouse=True) + def _save_restore_profiles(self): + """Snapshot and restore ASSET_PROFILES after each test.""" + snapshot = {k: v for k, v in ASSET_PROFILES.items()} + yield + ASSET_PROFILES.clear() + ASSET_PROFILES.update(snapshot) + + def test_sync_updates_profile_exchanges(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT", "ETHUSDT"], "BINANCE") + d.save() + updated = sync_exchanges_from_directory(d) + assert updated >= 2 + btc = get_asset_profile("BTCUSDT") + assert "BINANCE" in btc.exchanges + + def test_sync_adds_new_exchanges(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT"], "BINGX_VST") + d.save() + sync_exchanges_from_directory(d) + btc = get_asset_profile("BTCUSDT") + assert "binance" in btc.exchanges # pre-existing (lowercase default) + assert "BINGX_VST" in btc.exchanges # newly added + + def test_sync_does_not_add_offline(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.set_listing("BTCUSDT", "BINGX_VST", status=ListingStatus.OFFLINE) + d.save() + sync_exchanges_from_directory(d) + btc = get_asset_profile("BTCUSDT") + assert "BINGX_VST" not in btc.exchanges + + def test_sync_returns_update_count(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT"], "BINGX_VST") + d.save() + count = sync_exchanges_from_directory(d) + assert count >= 1 + + def test_sync_preserves_existing_fields(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT"], "BINGX_VST") + d.save() + btc_before = get_asset_profile("BTCUSDT") + sync_exchanges_from_directory(d) + btc_after = get_asset_profile("BTCUSDT") + assert btc_after.sector == btc_before.sector + assert btc_after.tick_size == btc_before.tick_size + assert btc_after.coingecko_id == btc_before.coingecko_id + + def test_sync_single_asset(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.set_listing("BTCUSDT", "BINGX_VST", status=ListingStatus.TRADING) + d.save() + result = sync_asset_to_profile(d, "BTCUSDT") + assert result is True + btc = get_asset_profile("BTCUSDT") + assert "BINGX_VST" in btc.exchanges + + def test_sync_unknown_asset_returns_false(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + result = sync_asset_to_profile(d, "NONEXISTENT") + assert result is False + + def test_universe_stats(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT", "ETHUSDT", "BANDUSDT"], "BINANCE") + d.save() + stats = get_universe_stats(d) + assert stats["directory_assets"] == 3 + assert stats["matched_profiles"] >= 2 + assert stats["unmatched_assets"] >= 1 + assert stats["profile_count"] == 13 + + +# ============================================================================== +# Integration with ScenarioFactory +# ============================================================================== + +class TestDirectoryIntegration: + @pytest.fixture(autouse=True) + def _save_restore_profiles(self): + snapshot = {k: v for k, v in ASSET_PROFILES.items()} + yield + ASSET_PROFILES.clear() + ASSET_PROFILES.update(snapshot) + + def test_scenario_factory_after_sync(self, tmp_path): + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT", "ETHUSDT"], "BINGX_VST") + d.save() + sync_exchanges_from_directory(d) + + btc = get_asset_profile("BTCUSDT") + assert "BINGX_VST" in btc.exchanges + + factory = ScenarioFactory() + suite = factory.build_suite(symbols=("BTCUSDT",), steps_per_scenario=3) + assert len(suite) >= 30 + + def test_query_by_new_exchange(self, tmp_path): + from malkhut.training.asset_classification import get_assets_on_exchange + + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT", "ETHUSDT"], "BINGX_VST") + d.save() + sync_exchanges_from_directory(d) + + vst_assets = get_assets_on_exchange("BINGX_VST") + symbols = {p.symbol for p in vst_assets} + assert "BTCUSDT" in symbols + assert "ETHUSDT" in symbols + + def test_full_roundtrip(self, tmp_path): + from malkhut.training.asset_classification import get_assets_on_exchange + + d = AssetDirectory(tmp_path / "d.json") + d.import_symbols(["BTCUSDT", "ETHUSDT", "SOLUSDT"], "BINGX_VST", + status=ListingStatus.TRADING, source="test") + d.save() + + updated = sync_exchanges_from_directory(d) + assert updated >= 3 + + vst = get_assets_on_exchange("BINGX_VST") + assert len(vst) >= 3 + + factory = ScenarioFactory() + suite = factory.build_suite(symbols=tuple(p.symbol for p in vst[:2]), + steps_per_scenario=3) + assert len(suite) >= 60 + + stats = get_universe_stats(d) + assert stats["matched_profiles"] >= 3 diff --git a/MALKHUT/malkhut/training/asset_bridge.py b/MALKHUT/malkhut/training/asset_bridge.py new file mode 100644 index 0000000..f6e5b89 --- /dev/null +++ b/MALKHUT/malkhut/training/asset_bridge.py @@ -0,0 +1,142 @@ +""" +Asset Directory ↔ Classification Bridge + +Connects the operational layer (Fable's AssetDirectory — runtime-mutable, +JSON-backed listing status) with the taxonomic layer (our AssetProfile — +frozen, in-code, invariant classification). + +Direction: directory → profiles (one-way). +A record in the directory says "this asset EXISTS and is TRADING on venue X". +The bridge propagates that to AssetProfile.exchanges so other consumers +(classification queries, behavior model, ScenarioFactory) can filter by venue. + +Usage: + from malkhut.training.asset_bridge import sync_exchanges_from_directory + + # After directory is loaded/populated: + sync_exchanges_from_directory() # updates all matching AssetProfile.exchanges + + # Or sync a specific asset: + sync_asset_to_profile("BTCUSDT") +""" +from __future__ import annotations + +from typing import List, Optional, Set + +from malkhut.assets.directory import ( + AssetDirectory, + ListingStatus, + normalize_symbol, +) +from malkhut.training.asset_classification import ( + ASSET_PROFILES, + AssetProfile, + get_asset_profile, +) + + +def _get_trading_exchanges(directory: AssetDirectory, symbol: str) -> List[str]: + """Get all exchanges where an asset has TRADING status.""" + rec = directory.get(symbol) + if rec is None: + return [] + return [ + ex for ex, listing in rec.exchanges.items() + if listing.status == ListingStatus.TRADING + ] + + +def sync_asset_to_profile( + directory: AssetDirectory, + symbol: str, +) -> bool: + """Sync a single asset's exchange list from directory to profile. + + Returns True if the profile was updated, False if not found. + """ + norm = normalize_symbol(symbol) + profile = get_asset_profile(norm) + if profile is None: + return False + + trading_exchanges = _get_trading_exchanges(directory, norm) + if not trading_exchanges: + return False + + new_exchanges = tuple(sorted(set(profile.exchanges + tuple(trading_exchanges)))) + + if new_exchanges != profile.exchanges: + # Frozen dataclass — must reconstruct + idx = list(ASSET_PROFILES.keys()).index(norm) + ASSET_PROFILES[norm] = AssetProfile( + symbol=profile.symbol, + base_asset=profile.base_asset, + name=profile.name, + unified_symbol=profile.unified_symbol, + quote_currency=profile.quote_currency, + coingecko_id=profile.coingecko_id, + cmc_id=profile.cmc_id, + blockchain=profile.blockchain, + contract_address=profile.contract_address, + sectors=profile.sectors, + token_roles=profile.token_roles, + supply_model=profile.supply_model, + consensus=profile.consensus, + smart_contracts=profile.smart_contracts, + market_cap_tier=profile.market_cap_tier, + volatility_profile=profile.volatility_profile, + liquidity_profile=profile.liquidity_profile, + derivative_access=profile.derivative_access, + tick_size=profile.tick_size, + lot_size=profile.lot_size, + price_decimals=profile.price_decimals, + maker_fee_bps=profile.maker_fee_bps, + taker_fee_bps=profile.taker_fee_bps, + typical_spread_bps=profile.typical_spread_bps, + typical_depth_usd=profile.typical_depth_usd, + typical_daily_volume_usd=profile.typical_daily_volume_usd, + has_funding=profile.has_funding, + has_options=profile.has_options, + exchanges=new_exchanges, + ) + return True + return False + + +def sync_exchanges_from_directory( + directory: Optional[AssetDirectory] = None, +) -> int: + """Sync ALL matching assets' exchange lists from directory to profiles. + + Returns the number of profiles that were updated. + """ + if directory is None: + directory = AssetDirectory() + + updated = 0 + for symbol in list(ASSET_PROFILES.keys()): + if sync_asset_to_profile(directory, symbol): + updated += 1 + return updated + + +def get_universe_stats(directory: Optional[AssetDirectory] = None) -> dict: + """Summary stats: how many directory assets match our profiles, how many don't.""" + if directory is None: + directory = AssetDirectory() + + matched = 0 + unmatched = 0 + for sym in directory.records: + norm = normalize_symbol(sym) + if get_asset_profile(norm) is not None: + matched += 1 + else: + unmatched += 1 + + return { + "directory_assets": len(directory), + "matched_profiles": matched, + "unmatched_assets": unmatched, + "profile_count": len(ASSET_PROFILES), + }