malkhut: asset-faithful book generation with composable toggles

Three independently toggleable features:
  1. Asset-faithful depth/spread: levels sized by OB study power-law per asset
  2. Intraday volume clock: depth scales by time-of-day (peak/trough)
  3. Realistic spread: per-asset spread from OB study + Flight7

Composable via BookGenerationConfig toggles:
  use_asset_faithful_depth, use_asset_faithful_spread, use_intraday_clock,
  use_weekend_mode, use_stress_mode, use_fragility, worst_case_mode

worst_case_mode overrides everything for max adversarial learning:
  spread * stress_mult, depth * fragility, no intraday/weekend.

DuckDB registry for online updates:
  AssetRegistry: upsert/get/list/delete/query
  RuntimeProfileCache: hot-reload during CWM runs
  upsert_from_csv/export_csv: pipeline support
  upsert_all_from_asset_behaviors(): seed from OB study

Results (8 assets):
  BTC: spread 0.031 bps, depth $350M (normal) / $4.9M (worst)
  DOGE: spread 2.86 bps, depth $2M (normal) / $132K (worst)
  ADA: spread 11.8 bps, depth $10M (normal) / $511K (worst)
  Intraday: BTC peak/trough = 2.8x depth ratio

All 99 tests green (31 new + 68 existing).

MALKHUT/malkhut/cwm/hft_cwm.py

@@ -106,6 +106,8 @@ class HftBacktestCWM:
         queue_model_n: int = 3,
         use_dynamic_book: bool = False,
         book_refresh_volatility: float = 0.1,
+        book_profile=None,
+        book_config=None,
     ) -> None:
         self.feature_extractor = feature_extractor or DefaultFeatureExtractor()
         self._tick_ns = tick_ns
@@ -113,6 +115,11 @@ class HftBacktestCWM:
         self._queue_model_n = queue_model_n
         self._use_dynamic_book = use_dynamic_book
         self._book_refresh_vol = book_refresh_volatility
+        self._book_generator = None
+        if use_dynamic_book and book_profile is not None:
+            from malkhut.training.asset_book_profile import BookGenerator, BookGenerationConfig
+            cfg = book_config or BookGenerationConfig()
+            self._book_generator = BookGenerator(book_profile, cfg)
 
         # Pre-compute fill probabilities for each level distance
         if self._use_queue_model:
@@ -382,8 +389,10 @@ class HftBacktestCWM:
                                 last_trade_side=Side.SELL,
                             )
 
-        # 4b. Dynamic book refresh (when use_dynamic_book=True)
+        # 4b. Dynamic book refresh
-        if self._use_dynamic_book and book.bids and book.asks:
+        if self._use_dynamic_book and self._book_generator:
+            book = self._book_generator.refresh_book(book, tick, rng)
+        elif self._use_dynamic_book and book.bids and book.asks:
             import numpy as np
             rng = np.random.RandomState(state.ts_ns % (2**31))
 

MALKHUT/malkhut/tests/test_asset_book_profile.py

@@ -0,0 +1,318 @@
+"""Tests for asset-faithful book generation."""
+from __future__ import annotations
+
+import os
+import math
+import random
+import tempfile
+
+from malkhut.training.asset_book_profile import (
+    AssetBookProfile, BookGenerationConfig, BookGenerator,
+    build_profile_from_behavior, _intraday_multiplier,
+)
+from malkhut.training.asset_registry import AssetRegistry, RuntimeProfileCache
+from malkhut.state import PriceLevel
+
+
+class TestIntradayMultiplier:
+    def test_peak_is_max(self):
+        m = _intraday_multiplier(15, 15, 19, 7.4)
+        assert m == 7.4
+
+    def test_trough_is_min(self):
+        m = _intraday_multiplier(19, 15, 19, 7.4)
+        assert m == 1.0
+
+    def test_midpoint_between_peak_and_trough(self):
+        m = _intraday_multiplier(17, 15, 19, 4.0)
+        assert 1.0 < m < 4.0
+
+    def test_all_hours_bounded(self):
+        for h in range(24):
+            m = _intraday_multiplier(h, 15, 19, 7.4)
+            assert 1.0 <= m <= 7.4, f"hour={h} mult={m}"
+
+
+class TestAssetBookProfile:
+    def test_build_from_btc(self):
+        p = build_profile_from_behavior("BTCUSDT")
+        assert p.symbol == "BTCUSDT"
+        assert p.depth_amplitude_usd == 750_000
+        assert p.depth_alpha == 0.70
+        assert p.depth_fragility == 0.10
+        assert p.spread_normal_bps == 0.01
+        assert p.spread_stress_mult == 50.0
+        assert p.typical_num_levels > 0
+        assert p.avg_level_size_usd > 0
+
+    def test_build_from_doge(self):
+        p = build_profile_from_behavior("DOGEUSDT")
+        assert p.symbol == "DOGEUSDT"
+        assert p.depth_amplitude_usd == 22_000
+        assert p.spread_normal_bps == 1.35
+        assert p.depth_alpha == 1.00
+
+    def test_build_from_unknown_raises(self):
+        try:
+            build_profile_from_behavior("FAKEUSDT")
+            assert False, "Should have raised ValueError"
+        except ValueError:
+            pass
+
+    def test_roundtrip_dict(self):
+        p = build_profile_from_behavior("ETHUSDT")
+        d = p.to_dict()
+        p2 = AssetBookProfile.from_dict(d)
+        assert p2.symbol == p.symbol
+        assert p2.depth_amplitude_usd == p.depth_amplitude_usd
+        assert p2.spread_normal_bps == p.spread_normal_bps
+
+
+class TestBookGenerationConfig:
+    def test_defaults(self):
+        c = BookGenerationConfig()
+        assert c.use_asset_faithful_depth is True
+        assert c.use_asset_faithful_spread is True
+        assert c.use_intraday_clock is True
+        assert c.use_weekend_mode is True
+        assert c.worst_case_mode is False
+
+    def test_worst_case_overrides(self):
+        c = BookGenerationConfig(worst_case_mode=True)
+        assert c.worst_case_mode is True
+
+    def test_independent_toggles(self):
+        c = BookGenerationConfig(
+            use_asset_faithful_depth=True,
+            use_intraday_clock=False,
+            use_weekend_mode=False,
+            use_stress_mode=True,
+        )
+        assert c.use_asset_faithful_depth is True
+        assert c.use_intraday_clock is False
+        assert c.use_weekend_mode is False
+        assert c.use_stress_mode is True
+
+
+class TestBookGenerator:
+    def test_generate_btc_book(self):
+        p = build_profile_from_behavior("BTCUSDT")
+        gen = BookGenerator(p, BookGenerationConfig())
+        book = gen.generate_initial_book(64000.0, 0.1, ts_ns=1_000_000)
+        assert len(book.bids) > 0
+        assert len(book.asks) > 0
+        assert book.bids[0].price < book.asks[0].price
+        assert book.mid > 0
+
+    def test_generate_doge_book(self):
+        p = build_profile_from_behavior("DOGEUSDT")
+        gen = BookGenerator(p, BookGenerationConfig())
+        book = gen.generate_initial_book(0.07, 0.00001, ts_ns=1_000_000)
+        assert len(book.bids) > 0
+        assert len(book.asks) > 0
+        spread = book.asks[0].price - book.bids[0].price
+        spread_bps = spread / book.mid * 10_000
+        assert spread_bps > 0.5
+
+    def test_worst_case_wider_spread(self):
+        p = build_profile_from_behavior("BTCUSDT")
+        normal = BookGenerator(p, BookGenerationConfig())
+        worst = BookGenerator(p, BookGenerationConfig(worst_case_mode=True))
+        b1 = normal.generate_initial_book(64000.0, 0.1)
+        b2 = worst.generate_initial_book(64000.0, 0.1)
+        s1 = (b1.asks[0].price - b1.bids[0].price) / b1.mid * 10_000
+        s2 = (b2.asks[0].price - b2.bids[0].price) / b2.mid * 10_000
+        assert s2 >= s1 * 10
+
+    def test_worst_case_thinner_book(self):
+        p = build_profile_from_behavior("BTCUSDT")
+        normal = BookGenerator(p, BookGenerationConfig())
+        worst = BookGenerator(p, BookGenerationConfig(worst_case_mode=True))
+        b1 = normal.generate_initial_book(64000.0, 0.1)
+        b2 = worst.generate_initial_book(64000.0, 0.1)
+        assert b2.bids[0].qty < b1.bids[0].qty * 0.2
+
+    def test_refresh_preserves_structure(self):
+        p = build_profile_from_behavior("BTCUSDT")
+        gen = BookGenerator(p, BookGenerationConfig())
+        book = gen.generate_initial_book(64000.0, 0.1)
+        rng = random.Random(42)
+        refreshed = gen.refresh_book(book, 0.1, rng)
+        assert len(refreshed.bids) > 0
+        assert len(refreshed.asks) > 0
+        assert refreshed.bids[0].price < refreshed.asks[0].price
+
+    def test_refresh_multiple_steps(self):
+        p = build_profile_from_behavior("ETHUSDT")
+        gen = BookGenerator(p, BookGenerationConfig())
+        book = gen.generate_initial_book(1800.0, 0.01)
+        rng = random.Random(42)
+        for _ in range(50):
+            book = gen.refresh_book(book, 0.01, rng)
+        assert len(book.bids) > 0
+        assert book.mid > 0
+
+    def test_worst_case_refresh_even_thinner(self):
+        p = build_profile_from_behavior("DOGEUSDT")
+        normal = BookGenerator(p, BookGenerationConfig())
+        worst = BookGenerator(p, BookGenerationConfig(worst_case_mode=True))
+        b1 = normal.generate_initial_book(0.07, 0.00001)
+        b2 = worst.generate_initial_book(0.07, 0.00001)
+        rng1 = random.Random(42)
+        rng2 = random.Random(42)
+        for _ in range(10):
+            b1 = normal.refresh_book(b1, 0.00001, rng1)
+            b2 = worst.refresh_book(b2, 0.00001, rng2)
+        avg_qty1 = sum(l.qty for l in b1.bids) / len(b1.bids)
+        avg_qty2 = sum(l.qty for l in b2.bids) / len(b2.bids)
+        assert avg_qty2 < avg_qty1 * 0.5
+
+    def test_no_cross_after_refresh(self):
+        for sym in ["BTCUSDT", "DOGEUSDT", "SOLUSDT", "ADAUSDT"]:
+            p = build_profile_from_behavior(sym)
+            gen = BookGenerator(p, BookGenerationConfig())
+            ref_p = p.reference_price if p.reference_price > 0 else 100.0
+            book = gen.generate_initial_book(ref_p, ref_p * 0.0001)
+            rng = random.Random(42)
+            for _ in range(20):
+                book = gen.refresh_book(book, ref_p * 0.0001, rng)
+            assert book.bids[0].price < book.asks[0].price, f"{sym} crossed"
+
+    def test_different_assets_different_books(self):
+        btc = BookGenerator(build_profile_from_behavior("BTCUSDT"), BookGenerationConfig())
+        doge = BookGenerator(build_profile_from_behavior("DOGEUSDT"), BookGenerationConfig())
+        b1 = btc.generate_initial_book(64000.0, 0.1)
+        b2 = doge.generate_initial_book(0.07, 0.00001)
+        s1 = (b1.asks[0].price - b1.bids[0].price) / b1.mid * 10_000
+        s2 = (b2.asks[0].price - b2.bids[0].price) / b2.mid * 10_000
+        assert s2 > s1 * 5
+
+
+class TestAssetRegistry:
+    def test_upsert_and_get(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = os.path.join(tmp, "test.db")
+            reg = AssetRegistry(db)
+            p = build_profile_from_behavior("BTCUSDT")
+            reg.upsert_profile(p)
+            got = reg.get_profile("BTCUSDT")
+            assert got is not None
+            assert got.symbol == "BTCUSDT"
+            assert got.depth_amplitude_usd == 750_000
+            reg.close()
+
+    def test_upsert_all_from_behaviors(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = os.path.join(tmp, "test.db")
+            reg = AssetRegistry(db)
+            count = reg.upsert_all_from_asset_behaviors()
+            assert count >= 8
+            syms = reg.list_symbols()
+            assert "BTCUSDT" in syms
+            assert "ETHUSDT" in syms
+            reg.close()
+
+    def test_upsert_overwrites(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = os.path.join(tmp, "test.db")
+            reg = AssetRegistry(db)
+            p = build_profile_from_behavior("BTCUSDT")
+            reg.upsert_profile(p)
+            reg.upsert_profile(p)
+            profiles = reg.list_profiles()
+            assert len(profiles) == 1
+            reg.close()
+
+    def test_delete_profile(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = os.path.join(tmp, "test.db")
+            reg = AssetRegistry(db)
+            p = build_profile_from_behavior("BTCUSDT")
+            reg.upsert_profile(p)
+            reg.delete_profile("BTCUSDT")
+            assert reg.get_profile("BTCUSDT") is None
+            reg.close()
+
+    def test_csv_roundtrip(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = os.path.join(tmp, "test.db")
+            csv_out = os.path.join(tmp, "export.csv")
+            reg = AssetRegistry(db)
+            reg.upsert_all_from_asset_behaviors()
+            n = reg.export_csv(csv_out)
+            assert n >= 8
+            assert os.path.exists(csv_out)
+            reg.close()
+
+            reg2 = AssetRegistry(os.path.join(tmp, "test2.db"))
+            n2 = reg2.upsert_from_csv(csv_out)
+            assert n2 >= 8
+            assert reg2.get_profile("BTCUSDT") is not None
+            reg2.close()
+
+
+class TestRuntimeProfileCache:
+    def test_put_and_get(self):
+        cache = RuntimeProfileCache()
+        p = build_profile_from_behavior("BTCUSDT")
+        cache.put(p)
+        assert cache.has("BTCUSDT")
+        assert cache.get("BTCUSDT").symbol == "BTCUSDT"
+
+    def test_load_from_registry(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = os.path.join(tmp, "test.db")
+            reg = AssetRegistry(db)
+            reg.upsert_all_from_asset_behaviors()
+            cache = RuntimeProfileCache()
+            n = cache.load_from_registry(reg)
+            assert n >= 8
+            assert cache.has("BTCUSDT")
+            assert cache.has("ETHUSDT")
+            reg.close()
+
+
+class TestHftCwmWithProfile:
+    def test_cwm_accepts_profile(self):
+        from malkhut.cwm.hft_cwm import HftBacktestCWM
+        p = build_profile_from_behavior("BTCUSDT")
+        cfg = BookGenerationConfig()
+        cwm = HftBacktestCWM(
+            use_queue_model=True,
+            use_dynamic_book=True,
+            book_profile=p,
+            book_config=cfg,
+        )
+        assert cwm._book_generator is not None
+
+    def test_cwm_without_profile_fallback(self):
+        from malkhut.cwm.hft_cwm import HftBacktestCWM
+        cwm = HftBacktestCWM(use_queue_model=True, use_dynamic_book=True)
+        assert cwm._book_generator is None
+
+    def test_cwm_default_backward_compat(self):
+        from malkhut.cwm.hft_cwm import HftBacktestCWM
+        cwm = HftBacktestCWM()
+        assert cwm._use_dynamic_book is False
+        assert cwm._book_generator is None
+
+
+class TestAllAssetsHaveProfiles:
+    def test_all_13_assets(self):
+        symbols = [
+            "BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT", "ADAUSDT",
+            "AVAXUSDT", "UNIUSDT", "LINKUSDT", "BNBUSDT", "MATICUSDT",
+            "AAVEUSDT", "DOTUSDT", "ATOMUSDT",
+        ]
+        for sym in symbols:
+            p = build_profile_from_behavior(sym)
+            assert p.symbol == sym
+            assert p.depth_amplitude_usd > 0
+            assert p.spread_normal_bps > 0
+            assert p.typical_num_levels > 0
+            gen = BookGenerator(p, BookGenerationConfig())
+            ref_p = p.reference_price if p.reference_price > 0 else 100.0
+            book = gen.generate_initial_book(ref_p, ref_p * 0.0001)
+            assert len(book.bids) > 0, f"{sym} no bids"
+            assert len(book.asks) > 0, f"{sym} no asks"
+            assert book.mid > 0, f"{sym} no mid"

MALKHUT/malkhut/training/asset_book_profile.py

@@ -0,0 +1,290 @@
+"""
+Asset-Faithful Book Generation — composable, per-asset order book simulation.
+
+Three independently toggleable features:
+  1. Asset-faithful depth/spread: levels sized by OB study power-law per asset
+  2. Intraday volume clock: depth scales by time-of-day volume profile
+  3. Realistic spread: per-asset spread from Flight7 calibration
+
+All features composed via BookGenerationConfig toggles.
+worst_case_mode overrides everything for max adversarial learning.
+
+Usage:
+    from malkhut.training.asset_book_profile import (
+        AssetBookProfile, BookGenerationConfig, BookGenerator,
+        build_profile_from_behavior,
+    )
+    profile = build_profile_from_behavior("BTCUSDT")
+    config = BookGenerationConfig(use_intraday_clock=True, intraday_hour=14)
+    gen = BookGenerator(profile, config)
+    book = gen.generate_initial_book(mid_price=64000.0, tick_size=0.1)
+"""
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass, field, asdict
+from typing import Dict, List, Optional
+
+from malkhut.state import OrderBookState, PriceLevel
+
+
+@dataclass(slots=True)
+class AssetBookProfile:
+    symbol: str
+    depth_amplitude_usd: float
+    depth_alpha: float
+    depth_fragility: float
+    depth_at_10bps_usd: float
+    depth_at_100bps_usd: float
+    spread_normal_bps: float
+    spread_stress_mult: float
+    flow_orders_per_sec: float
+    flow_cancel_fill_ratio: float
+    flow_median_order_usd: float
+    flow_avg_trade_usd: float
+    vol_annualized_normal: float
+    vol_annualized_crisis: float
+    vol_garch_alpha: float
+    vol_garch_beta: float
+    vol_half_life_hours: float
+    intraday_peak_hour_utc: int
+    intraday_trough_hour_utc: int
+    intraday_ratio: float
+    weekend_vol_mult: float
+    weekend_volume_mult: float
+    weekend_spread_mult: float
+    mm_max_inventory_usd: float
+    mm_pull_speed_ms: float
+    mm_margin_bps: float
+    avg_level_size_usd: float
+    typical_num_levels: int
+    reference_price: float
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, d: dict) -> AssetBookProfile:
+        return cls(**{k: v for k, v in d.items() if k in cls.__slots__})
+
+
+@dataclass(frozen=True, slots=True)
+class BookGenerationConfig:
+    use_asset_faithful_depth: bool = True
+    use_asset_faithful_spread: bool = True
+    use_intraday_clock: bool = True
+    use_weekend_mode: bool = True
+    use_stress_mode: bool = False
+    use_fragility: bool = False
+    use_asset_faithful_flow: bool = True
+    intraday_hour: int = 15
+    is_weekend: bool = False
+    stress_depth_mult: float = 1.0
+    worst_case_mode: bool = False
+
+
+def _intraday_multiplier(hour_utc: int, peak_hour: int, trough_hour: int, ratio: float) -> float:
+    """Sinusoidal intraday volume profile. Returns multiplier in [1/ratio, ratio]."""
+    hours = list(range(24))
+    trough_dist = [min(abs(h - trough_hour), 24 - abs(h - trough_hour)) for h in hours]
+    peak_dist = [min(abs(h - peak_hour), 24 - abs(h - peak_hour)) for h in hours]
+    max_dist = max(max(trough_dist), max(peak_dist), 1)
+    if hour_utc == peak_hour:
+        return ratio
+    if hour_utc == trough_hour:
+        return 1.0
+    t = 1.0 - trough_dist[hour_utc] / max_dist
+    return 1.0 + (ratio - 1.0) * t
+
+
+def build_profile_from_behavior(symbol: str) -> AssetBookProfile:
+    """Build AssetBookProfile from existing AssetBehavior data."""
+    from malkhut.training.asset_behavior import get_behavior
+    b = get_behavior(symbol)
+    if b is None:
+        raise ValueError(f"No AssetBehavior for {symbol}")
+
+    ref_price = b.reference_price if b.reference_price > 0 else 1.0
+    typical_levels = 50 if b.depth.amplitude_usd > 200_000 else 30 if b.depth.amplitude_usd > 50_000 else 20
+    avg_level = b.depth.amplitude_usd / typical_levels
+
+    return AssetBookProfile(
+        symbol=symbol,
+        depth_amplitude_usd=b.depth.amplitude_usd,
+        depth_alpha=b.depth.alpha,
+        depth_fragility=b.depth.fragility_factor,
+        depth_at_10bps_usd=b.depth.depth_at_10bps_usd,
+        depth_at_100bps_usd=b.depth.depth_at_100bps_usd,
+        spread_normal_bps=b.spread.normal_bps,
+        spread_stress_mult=b.spread.stress_multiplier,
+        flow_orders_per_sec=b.flow.orders_per_sec_normal,
+        flow_cancel_fill_ratio=b.flow.cancel_fill_ratio,
+        flow_median_order_usd=b.flow.median_order_usd,
+        flow_avg_trade_usd=b.flow.avg_trade_usd,
+        vol_annualized_normal=b.vol.annualized_normal,
+        vol_annualized_crisis=b.vol.annualized_crisis,
+        vol_garch_alpha=b.vol.garch_alpha,
+        vol_garch_beta=b.vol.garch_beta,
+        vol_half_life_hours=b.vol.half_life_hours,
+        intraday_peak_hour_utc=b.intraday.peak_hour_utc,
+        intraday_trough_hour_utc=b.intraday.trough_hour_utc,
+        intraday_ratio=b.intraday.ratio,
+        weekend_vol_mult=b.weekend.vol_mult,
+        weekend_volume_mult=b.weekend.volume_mult,
+        weekend_spread_mult=b.weekend.spread_mult,
+        mm_max_inventory_usd=b.market_maker.max_inventory_usd,
+        mm_pull_speed_ms=b.market_maker.pull_speed_ms,
+        mm_margin_bps=b.market_maker.margin_bps,
+        avg_level_size_usd=avg_level,
+        typical_num_levels=typical_levels,
+        reference_price=ref_price,
+    )
+
+
+class BookGenerator:
+    """Generates and refreshes order books faithful to a specific asset's characteristics."""
+
+    def __init__(self, profile: AssetBookProfile, config: BookGenerationConfig) -> None:
+        self._p = profile
+        self._c = config
+
+    def _effective_spread_bps(self) -> float:
+        s = self._p.spread_normal_bps
+        if self._c.worst_case_mode:
+            return s * self._p.spread_stress_mult
+        if self._c.use_stress_mode:
+            s *= self._p.spread_stress_mult
+        if self._c.use_weekend_mode and self._c.is_weekend:
+            s *= self._p.weekend_spread_mult
+        return s
+
+    def _effective_depth_multiplier(self) -> float:
+        m = 1.0
+        if self._c.worst_case_mode:
+            return self._p.depth_fragility
+        if self._c.use_intraday_clock:
+            m *= _intraday_multiplier(
+                self._c.intraday_hour,
+                self._p.intraday_peak_hour_utc,
+                self._p.intraday_trough_hour_utc,
+                self._p.intraday_ratio,
+            )
+        if self._c.use_weekend_mode and self._c.is_weekend:
+            m *= self._p.weekend_volume_mult
+        if self._c.use_stress_mode:
+            m *= self._c.stress_depth_mult
+        return m
+
+    def _level_qty_usd(self, distance_bps: float) -> float:
+        A = self._p.depth_amplitude_usd
+        alpha = self._p.depth_alpha
+        depth_usd = A * (distance_bps ** (1.0 - alpha))
+        return depth_usd
+
+    def generate_initial_book(self, mid_price: float, tick_size: float, ts_ns: int = 0) -> OrderBookState:
+        if mid_price <= 0:
+            return OrderBookState(ts_ns=ts_ns, symbol=self._p.symbol, bids=(), asks=(),
+                                  last_trade_price=0.0, last_trade_qty=0.0, last_trade_side=None)
+
+        spread_bps = self._effective_spread_bps()
+        depth_mult = self._effective_depth_multiplier()
+        half_spread = mid_price * spread_bps / 20_000.0
+        half_spread = max(half_spread, tick_size)
+
+        best_bid = mid_price - half_spread
+        best_ask = mid_price + half_spread
+
+        ref_price = self._p.reference_price if self._p.reference_price > 0 else mid_price
+        n_levels = self._p.typical_num_levels
+        flow_mult = self._p.flow_avg_trade_usd / max(ref_price, 1e-12)
+
+        bids = []
+        asks = []
+        for i in range(n_levels):
+            dist_bps = spread_bps / 2 + (i + 1) * 0.1
+            level_usd = self._level_qty_usd(dist_bps) * depth_mult
+            level_qty = level_usd / max(mid_price, 1e-12)
+            level_qty = max(level_qty, 1e-8)
+            bid_price = best_bid - i * tick_size
+            ask_price = best_ask + i * tick_size
+            if bid_price > 0:
+                bids.append(PriceLevel(round(bid_price, 10), level_qty))
+            asks.append(PriceLevel(round(ask_price, 10), level_qty))
+
+        return OrderBookState(
+            ts_ns=ts_ns, symbol=self._p.symbol,
+            bids=tuple(bids), asks=tuple(asks),
+            last_trade_price=mid_price, last_trade_qty=flow_mult,
+            last_trade_side=None,
+        )
+
+    def refresh_book(self, book: OrderBookState, tick_size: float, rng) -> OrderBookState:
+        if not book.bids or not book.asks:
+            return book
+
+        mid = book.mid
+        if mid <= 0:
+            return book
+
+        vol_ann = self._p.vol_annualized_normal
+        vol_per_step = vol_ann / math.sqrt(252 * 6.5 * 3600) * 0.1
+        if self._c.worst_case_mode:
+            vol_per_step *= 2.0
+        elif self._c.use_stress_mode:
+            vol_per_step *= math.sqrt(self._p.vol_annualized_crisis / max(self._p.vol_annualized_normal, 1e-12))
+
+        drift_bps = rng.gauss(0, vol_per_step * 100)
+        drift_price = mid * drift_bps / 10_000.0
+
+        cancel_ratio = self._p.flow_cancel_fill_ratio
+        qty_noise_frac = min(0.15, 1.0 / max(cancel_ratio, 1.0))
+
+        fragility = 1.0
+        if self._c.use_fragility and not self._c.worst_case_mode:
+            if rng.random() < 0.01:
+                fragility = self._p.depth_fragility
+
+        new_bids = []
+        for level in book.bids:
+            new_qty = level.qty * fragility
+            noise = rng.gauss(0, new_qty * qty_noise_frac)
+            new_qty = max(1e-8, new_qty + noise)
+            new_price = level.price + drift_price
+            if new_price > 0:
+                new_bids.append(PriceLevel(round(new_price, 10), new_qty))
+
+        new_asks = []
+        for level in book.asks:
+            new_qty = level.qty * fragility
+            noise = rng.gauss(0, new_qty * qty_noise_frac)
+            new_qty = max(1e-8, new_qty + noise)
+            new_price = level.price + drift_price
+            if new_price > 0:
+                new_asks.append(PriceLevel(round(new_price, 10), new_qty))
+
+        if not new_bids or not new_asks:
+            return book
+
+        if new_bids[0].price >= new_asks[0].price:
+            spread_bps = self._effective_spread_bps()
+            half_spread = mid * spread_bps / 20_000.0
+            half_spread = max(half_spread, tick_size)
+            new_bids = [PriceLevel(round(mid - half_spread, 10), new_bids[0].qty)]
+            new_asks = [PriceLevel(round(mid + half_spread, 10), new_asks[0].qty)]
+            for i in range(1, min(len(book.bids), self._p.typical_num_levels)):
+                dist_bps = spread_bps / 2 + (i + 1) * 0.1
+                lq = self._level_qty_usd(dist_bps) * self._effective_depth_multiplier() / max(mid, 1e-12)
+                new_bids.append(PriceLevel(round(mid - half_spread - i * tick_size, 10), max(lq, 1e-8)))
+            for i in range(1, min(len(book.asks), self._p.typical_num_levels)):
+                dist_bps = spread_bps / 2 + (i + 1) * 0.1
+                lq = self._level_qty_usd(dist_bps) * self._effective_depth_multiplier() / max(mid, 1e-12)
+                new_asks.append(PriceLevel(round(mid + half_spread + i * tick_size, 10), max(lq, 1e-8)))
+
+        n = min(len(new_bids), len(new_asks))
+        return OrderBookState(
+            ts_ns=book.ts_ns, symbol=book.symbol,
+            bids=tuple(new_bids[:n]), asks=tuple(new_asks[:n]),
+            last_trade_price=book.last_trade_price,
+            last_trade_qty=book.last_trade_qty,
+            last_trade_side=book.last_trade_side,
+        )

MALKHUT/malkhut/training/asset_registry.py

@@ -0,0 +1,168 @@
+"""
+Asset Book Profile Registry — DuckDB persistence + online update tooling.
+
+Provides upsert/query for per-asset book generation profiles.
+Profiles can be updated:
+  1. One-shot: upsert_all_from_asset_behaviors() seeds all 13 assets
+  2. Online: upsert_profile(symbol, ...) updates a single asset
+  3. Pipeline: upsert_from_csv(path) bulk-loads from a CSV
+  4. Runtime override: RuntimeProfileCache for hot-reload during CWM runs
+
+Usage:
+    from malkhut.training.asset_registry import AssetRegistry
+    reg = AssetRegistry()
+    reg.upsert_all_from_asset_behaviors()
+    profile = reg.get_profile("BTCUSDT")
+"""
+from __future__ import annotations
+
+import csv
+import os
+from typing import Dict, List, Optional
+
+from malkhut.training.asset_book_profile import AssetBookProfile
+
+_DEFAULT_DB = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                           "data", "asset_registry.db")
+
+_CREATE_SQL = """
+CREATE TABLE IF NOT EXISTS asset_book_profiles (
+    symbol TEXT PRIMARY KEY,
+    depth_amplitude_usd DOUBLE, depth_alpha DOUBLE, depth_fragility DOUBLE,
+    depth_at_10bps_usd DOUBLE, depth_at_100bps_usd DOUBLE,
+    spread_normal_bps DOUBLE, spread_stress_mult DOUBLE,
+    flow_orders_per_sec DOUBLE, flow_cancel_fill_ratio DOUBLE,
+    flow_median_order_usd DOUBLE, flow_avg_trade_usd DOUBLE,
+    vol_annualized_normal DOUBLE, vol_annualized_crisis DOUBLE,
+    vol_garch_alpha DOUBLE, vol_garch_beta DOUBLE, vol_half_life_hours DOUBLE,
+    intraday_peak_hour_utc INTEGER, intraday_trough_hour_utc INTEGER,
+    intraday_ratio DOUBLE,
+    weekend_vol_mult DOUBLE, weekend_volume_mult DOUBLE, weekend_spread_mult DOUBLE,
+    mm_max_inventory_usd DOUBLE, mm_pull_speed_ms DOUBLE, mm_margin_bps DOUBLE,
+    avg_level_size_usd DOUBLE, typical_num_levels INTEGER, reference_price DOUBLE,
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+)
+"""
+
+
+class AssetRegistry:
+    """DuckDB-backed asset profile registry with online update support."""
+
+    def __init__(self, db_path: str = _DEFAULT_DB) -> None:
+        os.makedirs(os.path.dirname(db_path), exist_ok=True)
+        import duckdb
+        self._db_path = db_path
+        self._conn = duckdb.connect(db_path)
+        self._conn.execute(_CREATE_SQL)
+
+    def upsert_profile(self, profile: AssetBookProfile) -> None:
+        d = profile.to_dict()
+        cols = list(d.keys())
+        placeholders = ", ".join(["?"] * len(cols))
+        col_str = ", ".join(cols)
+        self._conn.execute(
+            f"INSERT INTO asset_book_profiles ({col_str}) VALUES ({placeholders}) "
+            f"ON CONFLICT (symbol) DO UPDATE SET {', '.join(f'{c}=excluded.{c}' for c in cols)}",
+            list(d.values()),
+        )
+
+    def get_profile(self, symbol: str) -> Optional[AssetBookProfile]:
+        rows = self._conn.execute(
+            "SELECT * FROM asset_book_profiles WHERE symbol = ?", [symbol]
+        ).fetchall()
+        if not rows:
+            return None
+        cols = [desc[0] for desc in self._conn.description]
+        return AssetBookProfile.from_dict(dict(zip(cols, rows[0])))
+
+    def list_profiles(self) -> List[AssetBookProfile]:
+        rows = self._conn.execute("SELECT * FROM asset_book_profiles").fetchall()
+        cols = [desc[0] for desc in self._conn.description]
+        return [AssetBookProfile.from_dict(dict(zip(cols, r))) for r in rows]
+
+    def list_symbols(self) -> List[str]:
+        rows = self._conn.execute("SELECT symbol FROM asset_book_profiles").fetchall()
+        return [r[0] for r in rows]
+
+    def delete_profile(self, symbol: str) -> None:
+        self._conn.execute("DELETE FROM asset_book_profiles WHERE symbol = ?", [symbol])
+
+    def upsert_all_from_asset_behaviors(self) -> int:
+        from malkhut.training.asset_book_profile import build_profile_from_behavior
+        from malkhut.training.asset_behavior import list_behavior_symbols
+        count = 0
+        for sym in list_behavior_symbols():
+            try:
+                profile = build_profile_from_behavior(sym)
+                self.upsert_profile(profile)
+                count += 1
+            except Exception:
+                continue
+        return count
+
+    def upsert_from_csv(self, csv_path: str) -> int:
+        count = 0
+        with open(csv_path, "r") as f:
+            reader = csv.DictReader(f)
+            for row in reader:
+                try:
+                    profile = AssetBookProfile.from_dict(
+                        {k: float(v) if k not in ("symbol",) else v
+                         for k, v in row.items() if hasattr(AssetBookProfile, k)}
+                    )
+                    self.upsert_profile(profile)
+                    count += 1
+                except Exception:
+                    continue
+        return count
+
+    def export_csv(self, csv_path: str) -> int:
+        profiles = self.list_profiles()
+        if not profiles:
+            return 0
+        cols = list(profiles[0].to_dict().keys())
+        with open(csv_path, "w", newline="") as f:
+            writer = csv.DictWriter(f, fieldnames=cols)
+            writer.writeheader()
+            for p in profiles:
+                writer.writerow(p.to_dict())
+        return len(profiles)
+
+    def close(self) -> None:
+        self._conn.close()
+
+
+class RuntimeProfileCache:
+    """Hot-reloadable in-memory cache of AssetBookProfiles.
+
+    CWM uses this to pick up profile updates mid-run without restart.
+    Supports polling (check for updates) and push (explicit update).
+    """
+
+    def __init__(self) -> None:
+        self._cache: Dict[str, AssetBookProfile] = {}
+
+    def get(self, symbol: str) -> Optional[AssetBookProfile]:
+        return self._cache.get(symbol)
+
+    def put(self, profile: AssetBookProfile) -> None:
+        self._cache[profile.symbol] = profile
+
+    def put_all(self, profiles: List[AssetBookProfile]) -> None:
+        for p in profiles:
+            self._cache[p.symbol] = p
+
+    def load_from_registry(self, registry: AssetRegistry, symbols: Optional[List[str]] = None) -> int:
+        if symbols is None:
+            profiles = registry.list_profiles()
+        else:
+            profiles = [registry.get_profile(s) for s in symbols]
+            profiles = [p for p in profiles if p is not None]
+        self.put_all(profiles)
+        return len(profiles)
+
+    def has(self, symbol: str) -> bool:
+        return symbol in self._cache
+
+    def symbols(self) -> List[str]:
+        return list(self._cache.keys())