diff --git a/MALKHUT/malkhut/long_e2e_3h.py b/MALKHUT/malkhut/long_e2e_3h.py new file mode 100644 index 0000000..5c8d811 --- /dev/null +++ b/MALKHUT/malkhut/long_e2e_3h.py @@ -0,0 +1,483 @@ +#!/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() diff --git a/MALKHUT/malkhut/risk/gate.py b/MALKHUT/malkhut/risk/gate.py index 102d190..f92fd4e 100644 --- a/MALKHUT/malkhut/risk/gate.py +++ b/MALKHUT/malkhut/risk/gate.py @@ -147,6 +147,8 @@ class RiskGate: price = materialize_price_from_action(state, action) if price is None: return False + if not state.book.bids or not state.book.asks: + return False if action.side == Side.BUY and price >= state.book.best_ask: return True if action.side == Side.SELL and price <= state.book.best_bid: