diff --git a/MALKHUT/malkhut/cwm/hft_cwm.py b/MALKHUT/malkhut/cwm/hft_cwm.py index b84c84b..44e094d 100644 --- a/MALKHUT/malkhut/cwm/hft_cwm.py +++ b/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) - if self._use_dynamic_book and book.bids and book.asks: + # 4b. Dynamic book refresh + 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)) diff --git a/MALKHUT/malkhut/tests/test_asset_book_profile.py b/MALKHUT/malkhut/tests/test_asset_book_profile.py new file mode 100644 index 0000000..58d62f1 --- /dev/null +++ b/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" diff --git a/MALKHUT/malkhut/training/asset_book_profile.py b/MALKHUT/malkhut/training/asset_book_profile.py new file mode 100644 index 0000000..dab2d04 --- /dev/null +++ b/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, + ) diff --git a/MALKHUT/malkhut/training/asset_registry.py b/MALKHUT/malkhut/training/asset_registry.py new file mode 100644 index 0000000..3c76e72 --- /dev/null +++ b/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())