""" Asset Behavior DSL — Pure Python composable behavior definitions. Decomposes asset behavior into orthogonal dimensions, each with empirically-validated parameters from live Binance/BingX API + academic literature (Bouchaud, Cont/Stoikov, Cartea/Jaimungal). 13 pre-defined assets, 3 composable templates, 10 orthogonal dimensions. Auto-compilable via asset_compiler.py for any Binance-listed symbol. Usage: from malkhut.training.asset_behavior import get_behavior, ASSET_BEHAVIORS btc = get_behavior("BTCUSDT") print(btc.depth.amplitude_usd) # $750,000 print(btc.expected_slippage_bps(100_000)) # ~2 bps """ from __future__ import annotations from dataclasses import dataclass from typing import Dict, List, Optional from malkhut.training.asset_classification import Sector, TokenRole @dataclass(frozen=True, slots=True) class DepthProfile: amplitude_usd: float alpha: float fragility_factor: float depth_at_10bps_usd: float depth_at_100bps_usd: float @dataclass(frozen=True, slots=True) class SpreadProfile: normal_bps: float stress_multiplier: float @dataclass(frozen=True, slots=True) class FlowProfile: orders_per_sec_normal: float orders_per_sec_stress: float cancel_fill_ratio: float median_order_usd: float p99_order_usd: float avg_trade_usd: float @dataclass(frozen=True, slots=True) class VolatilityProfile: annualized_normal: float annualized_crisis: float garch_alpha: float garch_beta: float half_life_hours: float @dataclass(frozen=True, slots=True) class IntradayProfile: peak_hour_utc: int trough_hour_utc: int ratio: float @dataclass(frozen=True, slots=True) class WeekendProfile: vol_mult: float volume_mult: float spread_mult: float @dataclass(frozen=True, slots=True) class CorrelationProfile: eth_beta: float btc_corr_normal: float btc_corr_crash: float @dataclass(frozen=True, slots=True) class MarketMakerProfile: max_inventory_usd: float skew_tolerance_bps: float pull_speed_ms: float margin_bps: float @dataclass(frozen=True, slots=True) class LiquidationProfile: oi_mcap_ratio: float trigger_pct: float speed: str recovery: str @dataclass(frozen=True, slots=True) class FundingProfile: mean_bps_8h: float std_bps_8h: float positive_pct: float basis_typical_bps: float @dataclass(frozen=True, slots=True) class RetailProfile: ratio: float inst_gap: float @dataclass(frozen=True, slots=True) class BingxProfile: spread_mult: float depth_ratio: float latency_ms: float taker_fee_bps: float maker_fee_bps: float funding_lag_hours: float @dataclass(frozen=True, slots=True) class BehaviorTemplate: name: str depth: DepthProfile spread: SpreadProfile flow: FlowProfile vol: VolatilityProfile intraday: IntradayProfile weekend: WeekendProfile correlation: CorrelationProfile market_maker: MarketMakerProfile liquidation: LiquidationProfile funding: FundingProfile retail: RetailProfile bingx: BingxProfile @dataclass(frozen=True, slots=True) class AssetBehavior: symbol: str depth: DepthProfile spread: SpreadProfile flow: FlowProfile vol: VolatilityProfile intraday: IntradayProfile weekend: WeekendProfile correlation: CorrelationProfile market_maker: MarketMakerProfile liquidation: LiquidationProfile funding: FundingProfile retail: RetailProfile bingx: BingxProfile template_name: str = "" reference_price: float = 0.0 @classmethod def from_template(cls, symbol: str, template: BehaviorTemplate, overrides: Optional[Dict[str, object]] = None, reference_price: float = 0.0) -> AssetBehavior: params = { "depth": template.depth, "spread": template.spread, "flow": template.flow, "vol": template.vol, "intraday": template.intraday, "weekend": template.weekend, "correlation": template.correlation, "market_maker": template.market_maker, "liquidation": template.liquidation, "funding": template.funding, "retail": template.retail, "bingx": template.bingx, } if overrides: for key, value in overrides.items(): if key in params: if isinstance(value, dict): base = params[key] params[key] = type(base)(**{**base.__dict__, **value}) else: params[key] = value return cls(symbol=symbol, template_name=template.name, reference_price=reference_price, **params) def depth_at_bps(self, bps: float) -> float: """D(d) = amplitude * d^(1-alpha)""" return self.depth.amplitude_usd * (bps ** (1.0 - self.depth.alpha)) def expected_slippage_bps(self, order_usd: float) -> float: """Estimate slippage for a market order of given notional.""" d = 1.0 cumulative = 0.0 while cumulative < order_usd and d < 1000: cumulative += self.depth.amplitude_usd * (d ** (-self.depth.alpha)) d += 1.0 return d if cumulative >= order_usd else 1000.0 TEMPLATES: Dict[str, BehaviorTemplate] = {} def _t(name: str, **kwargs) -> BehaviorTemplate: t = BehaviorTemplate(name=name, **kwargs) TEMPLATES[name] = t return t INSTITUTIONAL_BLUE_CHIP = _t("institutional_blue_chip", depth=DepthProfile(1_500_000, 0.70, 0.10, 7_000_000, 20_000_000), spread=SpreadProfile(0.01, 50.0), flow=FlowProfile(300, 5000, 20.0, 643, 200_000, 5_000), vol=VolatilityProfile(35.0, 100.0, 0.10, 0.88, 48), intraday=IntradayProfile(15, 19, 7.4), weekend=WeekendProfile(0.65, 0.52, 1.20), correlation=CorrelationProfile(1.00, 1.00, 1.00), market_maker=MarketMakerProfile(10_000_000, 15, 3.0, 0.5), liquidation=LiquidationProfile(0.005, 6.5, "slow", "fast"), funding=FundingProfile(0.59, 0.22, 100.0, 4.0), retail=RetailProfile(0.35, 0.04), bingx=BingxProfile(12.6, 0.048, 100, 5.0, 2.0, 4)) MID_CAP_L1 = _t("mid_cap_l1", depth=DepthProfile(200_000, 0.85, 0.15, 800_000, 3_000_000), spread=SpreadProfile(1.5, 10.0), flow=FlowProfile(100, 1500, 10.0, 800, 150_000, 2_000), vol=VolatilityProfile(75.0, 150.0, 0.12, 0.85, 36), intraday=IntradayProfile(15, 11, 5.0), weekend=WeekendProfile(0.70, 0.55, 1.25), correlation=CorrelationProfile(0.85, 0.60, 0.90), market_maker=MarketMakerProfile(2_000_000, 25, 15.0, 1.0), liquidation=LiquidationProfile(0.015, 5.0, "medium", "medium"), funding=FundingProfile(0.10, 0.30, 60.0, 3.0), retail=RetailProfile(0.65, 0.50), bingx=BingxProfile(3.0, 0.30, 100, 5.0, 2.0, 4)) RETAIL_MEME = _t("retail_meme", depth=DepthProfile(25_000, 1.00, 0.30, 100_000, 2_500_000), spread=SpreadProfile(1.35, 15.0), flow=FlowProfile(80, 800, 8.0, 96, 52_000, 200), vol=VolatilityProfile(78.0, 200.0, 0.15, 0.82, 24), intraday=IntradayProfile(15, 10, 4.5), weekend=WeekendProfile(0.70, 0.50, 1.30), correlation=CorrelationProfile(0.83, 0.45, 0.80), market_maker=MarketMakerProfile(500_000, 40, 25.0, 2.0), liquidation=LiquidationProfile(0.014, 4.0, "fast", "slow"), funding=FundingProfile(0.39, 0.34, 80.0, 2.0), retail=RetailProfile(0.80, 0.31), bingx=BingxProfile(7.0, 1.27, 100, 5.0, 2.0, 4)) ASSET_BEHAVIORS: Dict[str, AssetBehavior] = {} def _b(symbol: str, template_name: str, reference_price: float = 0.0, **overrides) -> AssetBehavior: tmpl = TEMPLATES[template_name] b = AssetBehavior.from_template(symbol, tmpl, overrides or None, reference_price=reference_price) ASSET_BEHAVIORS[symbol] = b return b BTC = _b("BTCUSDT", "institutional_blue_chip", reference_price=64000.0, depth=DepthProfile(750_000, 0.70, 0.10, 5_983_000, 20_000_000), vol=VolatilityProfile(35.0, 100.0, 0.10, 0.88, 48), bingx=BingxProfile(12.6, 0.048, 100, 5.0, 2.0, 4)) ETH = _b("ETHUSDT", "institutional_blue_chip", reference_price=1800.0, depth=DepthProfile(600_000, 0.75, 0.12, 2_095_000, 7_200_000), vol=VolatilityProfile(66.5, 130.0, 0.12, 0.86, 40), correlation=CorrelationProfile(1.00, 0.90, 0.97), liquidation=LiquidationProfile(0.019, 4.0, "medium", "medium"), bingx=BingxProfile(9.0, 9.06, 100, 5.0, 2.0, 4)) SOL = _b("SOLUSDT", "mid_cap_l1", reference_price=80.0, depth=DepthProfile(400_000, 0.85, 0.18, 954_000, 4_000_000), spread=SpreadProfile(1.26, 8.0), vol=VolatilityProfile(71.9, 150.0, 0.13, 0.84, 32), correlation=CorrelationProfile(0.88, 0.65, 0.92), retail=RetailProfile(0.72, 0.75), bingx=BingxProfile(1.7, 0.33, 80, 5.0, 2.0, 2)) DOGE = _b("DOGEUSDT", "retail_meme", reference_price=0.07, depth=DepthProfile(22_000, 1.00, 0.30, 432_000, 2_772_000), vol=VolatilityProfile(77.9, 200.0, 0.15, 0.82, 24), correlation=CorrelationProfile(0.83, 0.45, 0.80), bingx=BingxProfile(7.0, 1.27, 100, 5.0, 2.0, 4)) ADA = _b("ADAUSDT", "mid_cap_l1", reference_price=0.17, depth=DepthProfile(60_000, 0.90, 0.20, 110_000, 1_500_000), spread=SpreadProfile(5.95, 12.0), vol=VolatilityProfile(78.7, 160.0, 0.14, 0.83, 30), bingx=BingxProfile(8.0, 0.20, 120, 5.0, 2.0, 4)) AVAX = _b("AVAXUSDT", "mid_cap_l1", reference_price=7.0, depth=DepthProfile(55_000, 0.88, 0.18, 107_000, 1_200_000), spread=SpreadProfile(1.48, 10.0), vol=VolatilityProfile(80.1, 160.0, 0.13, 0.84, 32), correlation=CorrelationProfile(0.79, 0.55, 0.88), bingx=BingxProfile(4.0, 0.25, 110, 5.0, 2.0, 4)) UNI = _b("UNIUSDT", "mid_cap_l1", reference_price=3.6, depth=DepthProfile(30_000, 0.95, 0.25, 53_000, 800_000), spread=SpreadProfile(2.76, 12.0), vol=VolatilityProfile(95.3, 220.0, 0.16, 0.81, 22), correlation=CorrelationProfile(0.70, 0.50, 0.85), market_maker=MarketMakerProfile(300_000, 50, 30.0, 2.5), retail=RetailProfile(0.60, 0.25), bingx=BingxProfile(10.0, 0.15, 150, 5.0, 2.0, 6)) LINK = _b("LINKUSDT", "mid_cap_l1", reference_price=8.0, depth=DepthProfile(80_000, 0.87, 0.17, 129_000, 1_800_000), spread=SpreadProfile(1.25, 8.0), vol=VolatilityProfile(77.3, 155.0, 0.12, 0.85, 34), correlation=CorrelationProfile(0.88, 0.62, 0.91), bingx=BingxProfile(3.5, 0.28, 100, 5.0, 2.0, 4)) BNB = _b("BNBUSDT", "institutional_blue_chip", reference_price=575.0, depth=DepthProfile(500_000, 0.78, 0.12, 2_000_000, 8_000_000), spread=SpreadProfile(0.50, 15.0), vol=VolatilityProfile(50.0, 120.0, 0.11, 0.87, 42), funding=FundingProfile(0.40, 0.25, 90.0, 3.0), retail=RetailProfile(0.50, 0.20), bingx=BingxProfile(5.0, 0.50, 100, 5.0, 2.0, 4)) MATIC = _b("MATICUSDT", "mid_cap_l1", reference_price=0.5, depth=DepthProfile(35_000, 0.92, 0.22, 55_000, 900_000), spread=SpreadProfile(1.50, 10.0), vol=VolatilityProfile(82.0, 170.0, 0.14, 0.83, 28), correlation=CorrelationProfile(0.82, 0.58, 0.89), bingx=BingxProfile(5.0, 0.30, 110, 5.0, 2.0, 4)) AAVE = _b("AAVEUSDT", "mid_cap_l1", reference_price=100.0, depth=DepthProfile(20_000, 0.95, 0.25, 35_000, 600_000), spread=SpreadProfile(2.50, 12.0), vol=VolatilityProfile(85.0, 200.0, 0.15, 0.82, 26), correlation=CorrelationProfile(0.75, 0.52, 0.86), market_maker=MarketMakerProfile(200_000, 45, 28.0, 2.0), retail=RetailProfile(0.55, 0.20), bingx=BingxProfile(10.0, 0.12, 140, 5.0, 2.0, 6)) DOT = _b("DOTUSDT", "mid_cap_l1", reference_price=6.0, depth=DepthProfile(70_000, 0.88, 0.18, 120_000, 1_500_000), spread=SpreadProfile(1.00, 8.0), vol=VolatilityProfile(72.0, 145.0, 0.12, 0.85, 34), bingx=BingxProfile(4.0, 0.30, 110, 5.0, 2.0, 4)) ATOM = _b("ATOMUSDT", "mid_cap_l1", reference_price=8.0, depth=DepthProfile(25_000, 0.92, 0.22, 45_000, 700_000), spread=SpreadProfile(2.00, 10.0), vol=VolatilityProfile(76.0, 160.0, 0.13, 0.84, 30), correlation=CorrelationProfile(0.78, 0.48, 0.87), bingx=BingxProfile(6.0, 0.20, 130, 5.0, 2.0, 5)) # ============================================================================== # Query functions # ============================================================================== def get_behavior(symbol: str) -> Optional[AssetBehavior]: return ASSET_BEHAVIORS.get(symbol) def list_behavior_symbols() -> List[str]: return list(ASSET_BEHAVIORS.keys()) def get_behaviors_by_sector(sector: Sector) -> List[AssetBehavior]: from malkhut.training.asset_classification import get_assets_by_sector return [ASSET_BEHAVIORS[p.symbol] for p in get_assets_by_sector(sector) if p.symbol in ASSET_BEHAVIORS] def get_behaviors_by_role(role: TokenRole) -> List[AssetBehavior]: from malkhut.training.asset_classification import get_assets_by_token_role return [ASSET_BEHAVIORS[p.symbol] for p in get_assets_by_token_role(role) if p.symbol in ASSET_BEHAVIORS] def get_behaviors_by_template(template_name: str) -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if b.template_name == template_name] def get_behaviors_by_volatility_band(min_ann: float = 0.0, max_ann: float = 500.0) -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if min_ann <= b.vol.annualized_normal <= max_ann] def get_fast_cascade_assets() -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if b.liquidation.speed == "fast"] def get_institutional_assets() -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if b.retail.ratio < 0.5] def get_retail_dominated_assets() -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if b.retail.ratio >= 0.6] def get_high_vol_assets() -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if b.vol.annualized_normal >= 75.0] def get_thin_book_assets() -> List[AssetBehavior]: return [b for b in ASSET_BEHAVIORS.values() if b.depth.amplitude_usd < 100_000]