#!/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 = [] 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() while time.time() < t_end: cycle += 1 elapsed = time.time() - t_start remaining = t_end - time.time() if remaining < 60: break try: # Phase 1: Run episodes with swarm 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, ) all_episodes.append(ep) # Record to matrix 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: Short CMA-ES optimization (every 5 cycles) if cycle % 5 == 0 and remaining > 600 and len(all_episodes) >= 20: phase = f"CYCLE {cycle} — CMA-ES" cma_scenarios = rng.sample(all_scenarios, min(5, len(all_scenarios))) try: best = run_cma_cycle( cwm_factory, cma_scenarios, params, codec, pool, n_evals=3, seed=SEED + cycle * 100, ) cma_bests.append(best) if best.score > params.w_expected_pnl * 10: params = best.params except Exception as e: print(f" CMA-ES error (cycle {cycle}): {e}", flush=True) # Report elapsed = time.time() - t_start if elapsed > 0 and (len(all_episodes) % 20 == 0 or cycle % 3 == 0): print_report(elapsed, phase, all_episodes, cma_bests, matrix) # Final report elapsed = time.time() - t_start print() print("=" * 80) print(" FINAL REPORT — 3-HOUR RUN COMPLETE") print("=" * 80) n = len(all_episodes) if n > 0: 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 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} (non-noop: {total_non_noop})") print(f"\n PERFORMANCE") print(f" Avg PnL: {sum(pnls)/n:+.1f} bps") print(f" Median PnL: {sorted(pnls)[n//2]:+.1f} bps") print(f" Best: {max(pnls):+.1f} bps") print(f" Worst: {min(pnls):+.1f} bps") print(f" Std dev: {math.sqrt(sum((p - sum(pnls)/n)**2 for p in pnls) / n):.1f} bps") print(f" Win rate: {sum(1 for p in pnls if p > 0)/n*100:.1f}%") print(f" Avg max DD: {sum(dds)/n:.1f} bps") print(f"\n ORDER FLOW") print(f" Fill rate: {sum(fills)/max(total_non_noop,1)*100:.1f}% (non-noop)") print(f" Aggressive: {sum(aggressive)} ({sum(aggressive)/max(total_non_noop,1)*100:.1f}%)") print(f" Passive: {sum(passive)} ({sum(passive)/max(total_non_noop,1)*100:.1f}%)") print(f" Cancels: {sum(e['cancels'] for e in all_episodes)}") print(f" Post-only: {sum(post_onlys)} ({sum(post_onlys)/max(total_non_noop,1)*100:.1f}%)") print(f" Reduce-only: {sum(reduce_onlys)} ({sum(reduce_onlys)/max(total_non_noop,1)*100:.1f}%)") print(f"\n ORDER TYPES") for ot, c in sorted(all_ots.items(), key=lambda x: -x[1]): print(f" {ot:20s} {c:6d} ({c/max(total_actions,1)*100:5.1f}%)") print(f"\n TIME-IN-FORCE") for t, c in sorted(all_tifs.items(), key=lambda x: -x[1]): print(f" {t:20s} {c:6d} ({c/max(total_actions,1)*100:5.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" Improvement: {(scores[-1] - scores[0]) / max(abs(scores[0]), 1) * 100:+.1f}%") print(f"\n MARKET CHARACTERIZATION") print(f" Avg spread: {sum(spreads)/n:.2f} bps") print(f" Avg equity vol: {sum(e.get('equity_volatility_bps', 0) for e in all_episodes)/n:.1f} bps") print(f" Agg/Passive: {sum(aggressive)/max(sum(passive),1):.2f}") print() print(f" Performance: {n} episodes in {elapsed/60:.1f} min = {n/max(elapsed/60,0.01):.1f} ep/min") print(f" Actions/sec: {total_actions/max(elapsed,1):.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(pnls)/n, 1) if n else 0, "win_rate_pct": round(sum(1 for p in pnls if p > 0)/n*100, 1) if n else 0, "total_actions": total_actions, } 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__": main()