malkhut: asset bridge — directory ↔ classification integration

asset_bridge.py: connects Fable's AssetDirectory (operational layer,
runtime-mutable, JSON-backed listing status) with our AssetProfile
(taxonomic layer, frozen, invariant classification).

Functions:
- sync_asset_to_profile(directory, symbol): sync one asset's TRADING
  exchanges from directory to AssetProfile.exchanges
- sync_exchanges_from_directory(directory): sync all matching assets
- get_universe_stats(directory): matched/unmatched counts

49 tests covering:
- normalize_symbol (7 edge cases)
- ExchangeListing validation (4 tests)
- AssetRecord listing queries (5 tests)
- AssetDirectory CRUD + persistence (14 tests)
- symbols_for_exchange filtering by status (4 tests)
- venue_symbol mapping (3 tests)
- Bridge sync (8 tests with state save/restore)
- Integration with ScenarioFactory + get_assets_on_exchange (4 tests)

Total: 1246 tests across 49 files, all green, zero regressions.
Fable's assets/ package preserved intact, interfaces retained.

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

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),
+    }