#!/usr/bin/env python3 """ MALKHUT 3-Hour E2E Long Run — HftBacktestCWM + CMA-ES + Swarm + Full Characterization. Runs for ~3 hours with: - CMA-ES optimization with HftBacktestCWM (queue model) - 13 assets × 30 scenarios = 390 scenarios - 11-agent swarm opponents - All order types exercised - Periodic reports every 10 minutes - Final comprehensive market characterization Usage: python -m malkhut.long_e2e_3h """ from __future__ import annotations import json import math import os import random import sys import time from collections import defaultdict from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple _HERE = os.path.dirname(os.path.abspath(__file__)) if _HERE not in sys.path: sys.path.insert(0, _HERE) from malkhut.state import ( AccountState, ActionKind, FulfilmentPolicyParams, MarketWorldState, OrderType, PositionState, Side, ) from malkhut.actions import FulfilmentAction, PlannedPolicy from malkhut.cwm.hft_cwm import HftBacktestCWM from malkhut.risk.gate import RiskGate from malkhut.training.cma_trainer import ( CMAESTrainer, CMAParameterCodec, PolicyEvaluator, ScenarioFactory, SelfPlayPool, PolicySnapshot, ) from malkhut.training.selector import PerformanceMatrix, MarketRegime from malkhut.counterparties import ( ToxicTakerPolicy, PassiveMakerPolicy, LatencyArbPolicy, NoiseTraderPolicy, ) from malkhut.counterparties_extended import ( MomentumTakerPolicy, MeanReversionTakerPolicy, InventoryMarketMakerPolicy, LiquidationFlowPolicy, StaleQuoteAttackerPolicy, ) DURATION_S = 3 * 3600 # 3 hours REPORT_INTERVAL_S = 600 # report every 10 minutes ASSETS = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT", "ADAUSDT", "AVAXUSDT", "UNIUSDT", "LINKUSDT", "BNBUSDT"] STEPS_PER_EPISODE = 20 SEED = 42 # ── Swarm ──────────────────────────────────────────────────────────────────── def _build_swarm(n: int = 100) -> tuple: """Build a swarm of n diverse opponents.""" pool = [ lambda: ToxicTakerPolicy(sensitivity=random.uniform(0.1, 0.8)), lambda: PassiveMakerPolicy(join_probability=random.uniform(0.3, 0.9)), lambda: LatencyArbPolicy(lead_threshold=random.uniform(0.3, 0.7)), lambda: NoiseTraderPolicy(), lambda: MomentumTakerPolicy(threshold=random.uniform(0.1, 0.5)), lambda: MeanReversionTakerPolicy(threshold=random.uniform(0.2, 0.8)), lambda: InventoryMarketMakerPolicy(max_inventory=random.uniform(0.02, 0.15)), lambda: LiquidationFlowPolicy(trigger_bps=random.uniform(20, 80)), lambda: StaleQuoteAttackerPolicy(stale_threshold_s=random.uniform(2, 10)), ] swarm = [] rng = random.Random(99) for i in range(n): factory = rng.choice(pool) swarm.append(factory()) return tuple(swarm) SWARM = _build_swarm(100) # ── Action generator ───────────────────────────────────────────────────────── def _generate_action(state: MarketWorldState, rng: random.Random) -> FulfilmentAction: r = rng.random() if r < 0.12: return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) elif r < 0.28: side = Side.BUY if rng.random() < 0.5 else Side.SELL tif = rng.choice(["IOC", "GTC"]) return FulfilmentAction(ActionKind.CROSS_SPREAD, side, OrderType.LIMIT, 0, rng.uniform(0.01, 0.10), 50, time_in_force=tif) elif r < 0.48: side = Side.BUY if rng.random() < 0.55 else Side.SELL return FulfilmentAction(ActionKind.PLACE, side, OrderType.LIMIT, rng.randint(0, 5), rng.uniform(0.05, 0.25), 200, post_only=True) elif r < 0.62: side = Side.BUY if rng.random() < 0.5 else Side.SELL return FulfilmentAction(ActionKind.PLACE, side, OrderType.LIMIT, rng.randint(0, 3), rng.uniform(0.05, 0.20), 200) elif r < 0.72: if state.open_orders: oo = rng.choice(state.open_orders) return FulfilmentAction(ActionKind.CANCEL, None, None, 0, 0.0, 0, cancel_order_id=oo.client_order_id) return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) elif r < 0.82: pos = state.account.positions.get(state.venue.symbol) if pos and abs(pos.qty) > 0.001: side = Side.SELL if pos.qty > 0 else Side.BUY return FulfilmentAction(ActionKind.REDUCE, side, OrderType.MARKET, 0, rng.uniform(0.1, 0.5), 0, reduce_only=True) return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) elif r < 0.92: pos = state.account.positions.get(state.venue.symbol) if pos and abs(pos.qty) > 0.001: side = Side.SELL if pos.qty > 0 else Side.BUY return FulfilmentAction(ActionKind.FULL_EXIT, side, OrderType.MARKET, 0, 1.0, 0, reduce_only=True) return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) else: side = Side.BUY if rng.random() < 0.5 else Side.SELL return FulfilmentAction(ActionKind.PLACE, side, OrderType.STOP_MARKET, rng.randint(-5, 5), rng.uniform(0.01, 0.05), 200) # ── Episode runner ─────────────────────────────────────────────────────────── def run_episode(cwm, scenario, params, steps, seed, rng, risk_gate, matrix=None): state = scenario.initial_state cp_policies = scenario.counterparties fills = 0; noops = 0; cancels = 0; post_onlys = 0; reduce_onlys = 0 aggressive = 0; passive = 0; peak_eq = state.account.equity max_dd = 0.0; total_steps = steps ot_counts = defaultdict(int); tif_counts = defaultdict(int) spreads = []; equities = [] for step in range(steps): spread_bps = state.book.spread_bps if state.book.bids and state.book.asks else 0.0 spreads.append(spread_bps) equities.append(state.account.equity) action = _generate_action(state, rng) cp_actions = tuple(cp.rollout_action(state, rng) for cp in cp_policies) if risk_gate and action.kind != ActionKind.NOOP: planned = PlannedPolicy(actions=(action,), probabilities=(1.0,), selected_action=action, diagnostics={}) decision = risk_gate.validate(state, planned, params) if not decision.approved: action = FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) prev_eq = state.account.equity state = cwm.transition(state, (action, *cp_actions)) eq = state.account.equity peak_eq = max(peak_eq, eq) dd = (peak_eq - eq) / max(peak_eq, 1e-12) * 10_000 max_dd = max(max_dd, dd) ot = action.order_type.value if action.order_type else "NONE" ot_counts[ot] += 1 tif_counts[getattr(action, 'time_in_force', 'GTC')] += 1 if action.kind == ActionKind.NOOP: noops += 1 elif action.kind in (ActionKind.CANCEL, ActionKind.CANCEL_REPLACE): cancels += 1 elif action.kind == ActionKind.CROSS_SPREAD: aggressive += 1 else: passive += 1 if action.post_only: post_onlys += 1 if action.reduce_only: reduce_onlys += 1 if eq != prev_eq and action.kind != ActionKind.NOOP: fills += 1 pnl = (state.account.equity - 10000.0) / 10000.0 * 10_000 avg_spread = sum(spreads) / max(len(spreads), 1) eq_vol = (max(equities) - min(equities)) / max(max(equities), 1e-12) * 10_000 if len(equities) > 1 else 0 pos = state.account.positions.get(state.venue.symbol, PositionState("", 0, 0, 0, 0, None, 0, None)) return { "scenario_id": scenario.scenario_id, "pnl_bps": pnl, "max_dd_bps": max_dd, "fills": fills, "noops": noops, "cancels": cancels, "aggressive": aggressive, "passive": passive, "post_onlys": post_onlys, "reduce_onlys": reduce_onlys, "order_types": dict(ot_counts), "tifs": dict(tif_counts), "avg_spread_bps": avg_spread, "equity_volatility_bps": eq_vol, "final_pos": pos.qty, "final_eq": state.account.equity, "peak_eq": peak_eq, "steps": total_steps, } # ── CMA-ES optimization cycle ──────────────────────────────────────────────── def run_cma_cycle(cwm_factory, scenarios, params, codec, pool, n_evals=10, seed=42): """Run a short CMA-ES optimization cycle.""" evaluator = PolicyEvaluator(cwm_factory=cwm_factory, scoring_mode="fast") trainer = CMAESTrainer(codec=codec, evaluator=evaluator, pool=pool, workers=0) best = trainer.train( incumbent=params, scenarios=scenarios, budget_evals=n_evals, seed=seed, ) return best # ── Reporting ──────────────────────────────────────────────────────────────── def print_report(elapsed, phase, all_episodes, cma_bests, matrix): n = len(all_episodes) if n == 0: return pnls = [e["pnl_bps"] for e in all_episodes] dds = [e["max_dd_bps"] for e in all_episodes] fills = [e["fills"] for e in all_episodes] spreads = [e["avg_spread_bps"] for e in all_episodes] aggressive = [e["aggressive"] for e in all_episodes] passive = [e["passive"] for e in all_episodes] post_onlys = [e["post_onlys"] for e in all_episodes] reduce_onlys = [e["reduce_onlys"] for e in all_episodes] all_ots = defaultdict(int) all_tifs = defaultdict(int) for e in all_episodes: for ot, c in e["order_types"].items(): all_ots[ot] += c for t, c in e["tifs"].items(): all_tifs[t] += c total_actions = sum(all_ots.values()) total_noops = sum(e["noops"] for e in all_episodes) total_non_noop = total_actions - total_noops h = elapsed / 3600 m = (elapsed % 3600) / 60 print() print(f"{'='*80}") print(f" PERIODIC REPORT — {phase} — {h:.1f}h {m:.0f}m elapsed") print(f"{'='*80}") print(f" Episodes: {n} | Actions: {total_actions} | Non-noop: {total_non_noop}") print(f" Avg PnL: {sum(pnls)/n:+.1f} bps | Win rate: {sum(1 for p in pnls if p > 0)/n*100:.0f}%") print(f" Best: {max(pnls):+.1f} bps | Worst: {min(pnls):+.1f} bps") print(f" Avg max DD: {sum(dds)/n:.1f} bps | Avg spread: {sum(spreads)/n:.2f} bps") print(f" Fill rate (non-noop): {sum(fills)/max(total_non_noop,1)*100:.1f}%") print(f" Aggressive: {sum(aggressive)} | Passive: {sum(passive)} | Ratio: {sum(aggressive)/max(sum(passive),1):.2f}") print(f" Post-only: {sum(post_onlys)} | Reduce-only: {sum(reduce_onlys)}") print(f" CMA-ES cycles: {len(cma_bests)} | Best CMA score: {cma_bests[-1].score:.1f}" if cma_bests else "") if all_ots: print(f"\n Order types:") for ot, c in sorted(all_ots.items(), key=lambda x: -x[1]): print(f" {ot:20s} {c:5d} ({c/max(total_actions,1)*100:5.1f}%)") if all_tifs: print(f"\n TimeInForce:") for t, c in sorted(all_tifs.items(), key=lambda x: -x[1]): print(f" {t:20s} {c:5d} ({c/max(total_actions,1)*100:5.1f}%)") print(f"{'='*80}") # ── Main ───────────────────────────────────────────────────────────────────── def main(): t_start = time.time() t_end = t_start + DURATION_S print("=" * 80) print("MALKHUT 3-HOUR LONG E2E RUN") print(f" Duration: 3 hours ({DURATION_S}s)") print(f" CWM: HftBacktestCWM (PowerProbQueueModel)") print(f" Swarm: {len(SWARM)} diverse opponents") print(f" Assets: {', '.join(ASSETS)}") print(f" Steps/episode: {STEPS_PER_EPISODE}") print("=" * 80) print() # Initialize cwm_factory = lambda: HftBacktestCWM(use_queue_model=True) risk_gate = RiskGate() codec = CMAParameterCodec() pool = SelfPlayPool(max_size=20) matrix = PerformanceMatrix() params = _baseline() factory = ScenarioFactory(exchange_id="bingx") all_episodes: list = [] # DEPRECATED — use rolling stats cma_bests: list[PolicySnapshot] = [] cycle = 0 phase = "INIT" # Build all scenarios print("Building scenarios...") all_scenarios = [] for sym in ASSETS: scenarios = factory.build_suite(symbols=[sym], steps_per_scenario=STEPS_PER_EPISODE, seed=SEED) all_scenarios.extend(scenarios) print(f" Total scenarios: {len(all_scenarios)} ({len(ASSETS)} assets)") rng = random.Random(SEED) print(f"\nStarting 3-hour run...") print() # Rolling stats (memory-efficient — don't accumulate full episodes) recent_pnls: list = [] recent_fills: int = 0 recent_noops: int = 0 recent_cancels: int = 0 recent_aggressive: int = 0 recent_passive: int = 0 recent_post_onlys: int = 0 recent_reduce_onlys: int = 0 recent_ot_counts: Dict[str, int] = defaultdict(int) recent_tif_counts: Dict[str, int] = defaultdict(int) recent_actions_total: int = 0 recent_episodes: int = 0 all_pnls: list = [] # keep only PnL for final characterization peak_pnl = -float("inf") worst_pnl = float("inf") while time.time() < t_end: cycle += 1 elapsed = time.time() - t_start remaining = t_end - time.time() if remaining < 60: break try: phase = f"CYCLE {cycle} — EPISODES" n_episodes = min(len(all_scenarios), 20) selected = rng.sample(all_scenarios, n_episodes) for i, scenario in enumerate(selected): if time.time() > t_end - 30: break ep_rng = random.Random(SEED + cycle * 1000 + i) ep = run_episode( cwm=cwm_factory(), scenario=scenario, params=params, steps=STEPS_PER_EPISODE, seed=SEED + cycle * 1000 + i, rng=ep_rng, risk_gate=risk_gate, matrix=matrix, ) # Accumulate rolling stats (memory efficient) recent_pnls.append(ep["pnl_bps"]) all_pnls.append(ep["pnl_bps"]) peak_pnl = max(peak_pnl, ep["pnl_bps"]) worst_pnl = min(worst_pnl, ep["pnl_bps"]) recent_fills += ep["fills"] recent_noops += ep["noops"] recent_cancels += ep["cancels"] recent_aggressive += ep["aggressive"] recent_passive += ep["passive"] recent_post_onlys += ep["post_onlys"] recent_reduce_onlys += ep["reduce_onlys"] recent_episodes += 1 for ot, c in ep["order_types"].items(): recent_ot_counts[ot] += c for t, c in ep["tifs"].items(): recent_tif_counts[t] += c recent_actions_total += ep["steps"] # Keep only last 200 PnLs for rolling stats if len(recent_pnls) > 200: recent_pnls = recent_pnls[-200:] tag = scenario.tags[0] if scenario.tags else "normal" matrix.record( strategy_id=params.version, regime=tag, score=ep["pnl_bps"], venue=scenario.venue, ) except Exception as e: print(f" Episode error (cycle {cycle}): {e}", flush=True) import traceback traceback.print_exc() # Phase 2: CMA-ES every 10 cycles (memory-heavy — keep infrequent) if cycle % 10 == 0 and remaining > 600 and recent_episodes >= 20: phase = f"CYCLE {cycle} — CMA-ES" cma_scenarios = rng.sample(all_scenarios, min(3, len(all_scenarios))) try: best = run_cma_cycle( cwm_factory, cma_scenarios, params, codec, pool, n_evals=2, seed=SEED + cycle * 100, ) cma_bests.append(best) if best.score > params.w_expected_pnl * 10: params = best.params # Explicitly free CMA-ES state del best import gc; gc.collect() except Exception as e: print(f" CMA-ES error (cycle {cycle}): {e}", flush=True) # Periodic report from rolling stats elapsed = time.time() - t_start if elapsed > 0 and cycle % 5 == 0: n = recent_episodes if n > 0: avg_pnl = sum(recent_pnls[-min(200, len(recent_pnls)):]) / min(200, len(recent_pnls)) h = elapsed / 3600 m = (elapsed % 3600) / 60 print(f"\n [{h:.1f}h{m:.0f}m] Cycle {cycle} | {recent_episodes} ep | " f"PnL {avg_pnl:+.0f} bps | fill {recent_fills/max(recent_actions_total-recent_noops,1)*100:.0f}% | " f"agg/pass {recent_aggressive/max(recent_passive,1):.2f} | " f"reduce_only {recent_reduce_onlys} | " f"peak {peak_pnl:+.0f} worst {worst_pnl:+.0f}", flush=True) recent_pnls.clear() recent_episodes = 0 recent_fills = recent_noops = recent_cancels = 0 recent_aggressive = recent_passive = 0 recent_post_onlys = recent_reduce_onlys = 0 recent_ot_counts.clear() recent_tif_counts.clear() recent_actions_total = 0 # Final report elapsed = time.time() - t_start n = len(all_pnls) print() print("=" * 80) print(" FINAL REPORT — 3-HOUR RUN COMPLETE") print("=" * 80) if n > 0: total_actions = sum(recent_ot_counts.values()) + recent_actions_total print(f"\n Duration: {elapsed/3600:.1f}h ({elapsed:.0f}s)") print(f" Cycles: {cycle}") print(f" Total episodes: {n}") print(f" Total actions: {total_actions}") print(f"\n PERFORMANCE") print(f" Avg PnL: {sum(all_pnls)/n:+.1f} bps") print(f" Median PnL: {sorted(all_pnls)[n//2]:+.1f} bps") print(f" Best: {max(all_pnls):+.1f} bps") print(f" Worst: {min(all_pnls):+.1f} bps") print(f" Std dev: {math.sqrt(sum((p - sum(all_pnls)/n)**2 for p in all_pnls) / n):.1f} bps") print(f" Win rate: {sum(1 for p in all_pnls if p > 0)/n*100:.1f}%") print(f"\n CMA-ES OPTIMIZATION") print(f" Cycles: {len(cma_bests)}") if cma_bests: scores = [b.score for b in cma_bests] print(f" Best score: {max(scores):.1f}") print(f" Final score: {scores[-1]:.1f}") print(f"\n MARKET CHARACTERIZATION") print(f" Actions/sec: {total_actions/max(elapsed,1):.0f}") print(f" Episodes/hour: {n/max(elapsed/3600,0.01):.0f}") print("=" * 80) # Save report os.makedirs("malkhut/results", exist_ok=True) report = { "duration_s": round(elapsed, 1), "cycles": cycle, "n_episodes": n, "n_scenarios": len(all_scenarios), "cma_cycles": len(cma_bests), "avg_pnl_bps": round(sum(all_pnls)/n, 1) if n else 0, "win_rate_pct": round(sum(1 for p in all_pnls if p > 0)/n*100, 1) if n else 0, "total_actions": total_actions, "peak_pnl": round(peak_pnl, 1) if n else 0, "worst_pnl": round(worst_pnl, 1) if n else 0, } path = f"malkhut/results/long_e2e_{int(time.time())}.json" with open(path, "w") as f: json.dump(report, f, indent=2) print(f"\nReport: {path}") def _baseline(): return FulfilmentPolicyParams( version="long_e2e", ucb_c=1.414, max_sims=64, max_depth=2, rollout_depth=2, root_temperature=0.5, min_root_entropy=0.25, quote_offsets_ticks=(0, 1, 2), quote_size_fractions=(0.10, 0.25, 0.50), passive_ttl_ms=200, aggressive_ttl_ms=50, maker_edge_min_bps=0.5, cross_spread_edge_min_bps=5.0, adverse_toxicity_cancel_threshold=0.5, queue_churn_cancel_threshold=0.5, mae_tail_cut_bps=50.0, mfe_giveback_cut_fraction=0.5, max_time_in_loss_s=300.0, failed_recovery_cut_count=3, recovery_velocity_min_bps_per_s=0.0, max_symbol_notional_fraction=0.20, max_single_order_notional_fraction=0.05, reduce_when_global_up_fraction=0.30, session_profit_lock_fraction=0.02, w_expected_pnl=1.0, w_fill_probability=0.5, w_adverse_selection=2.0, w_queue_priority=0.5, w_inventory_risk=1.5, w_tail_loss=5.0, w_fee_quality=0.5, w_time_decay=0.3, w_policy_entropy=0.5, robust_tail_weight=2.0, toxic_counterparty_weight=3.0, low_liquidity_weight=2.0, latency_stress_weight=1.0, ) if __name__ == "__main__": try: main() except Exception as e: import traceback print(f"\nFATAL ERROR: {e}") traceback.print_exc() sys.exit(1)