diff --git a/MALKHUT/malkhut/long_learning_5h.py b/MALKHUT/malkhut/long_learning_5h.py new file mode 100644 index 0000000..628ed67 --- /dev/null +++ b/MALKHUT/malkhut/long_learning_5h.py @@ -0,0 +1,304 @@ +#!/usr/bin/env python3 +""" +MALKHUT 5-Hour Learning Test — with OOM/CPU/results monitoring. + +All features: calibrated slippage, urgency, fee+slippage threshold, +chase, REQUOTE, all order types, configurable friction per scenario. + +Monitors: OOM (RSS), CPU usage, results (fill_value, PnL, WR). + +Usage: + PYTHONUNBUFFERED=1 NUMBA_CACHE_DIR=/tmp/numba_cache python -m malkhut.long_learning_5h +""" +from __future__ import annotations + +import json +import math +import os +import random +import resource +import sys +import threading +import time +from collections import defaultdict +from typing import Any, Dict, List, Optional + +_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 = 5 * 3600 # 5 hours +ASSETS = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT", "ADAUSDT", + "AVAXUSDT", "UNIUSDT", "LINKUSDT", "BNBUSDT"] +STEPS_PER_EPISODE = 20 +SEED = 42 +N_OPPONENTS = 1500 + + +class ResourceMonitor: + def __init__(self): + self._start = time.time() + self._peak_rss = 0 + self._peak_cpu = 0 + self._samples = [] + + def sample(self): + usage = resource.getrusage(resource.RUSAGE_SELF) + rss_mb = usage.ru_maxrss / 1024 + try: + with open("/proc/self/stat") as f: + fields = f.read().split() + utime = int(fields[13]) + int(fields[14]) + total = sum(int(f) for f in fields[13:22]) + cpu_pct = utime / max(total, 1) * 100 + except: + cpu_pct = 0.0 + self._peak_rss = max(self._peak_rss, rss_mb) + self._peak_cpu = max(self._peak_cpu, cpu_pct) + self._samples.append({"time": time.time() - self._start, "rss_mb": rss_mb, "cpu_pct": cpu_pct}) + return rss_mb, cpu_pct + + def report(self): + if not self._samples: + return {} + avg_cpu = sum(s["cpu_pct"] for s in self._samples) / len(self._samples) + avg_rss = sum(s["rss_mb"] for s in self._samples) / len(self._samples) + return { + "peak_rss_mb": self._peak_rss, "avg_rss_mb": avg_rss, + "peak_cpu_pct": self._peak_cpu, "avg_cpu_pct": avg_cpu, + "samples": len(self._samples), + } + + +def _build_swarm(n: int = 1500) -> tuple: + 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)), + ] + rng = random.Random(99) + return tuple(rng.choice(pool)() for _ in range(n)) + + +def _gen(state, rng, params): + r = rng.random() + if r < 0.10: return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) + elif r < 0.22: + return FulfilmentAction(ActionKind.CROSS_SPREAD, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.MARKET, 0, rng.uniform(0.01,0.10), 50) + elif r < 0.35: + return FulfilmentAction(ActionKind.PLACE, Side.BUY if rng.random()<0.55 else Side.SELL, OrderType.LIMIT, rng.randint(0,5), rng.uniform(0.05,0.25), 200, post_only=True) + elif r < 0.42: + if state.open_orders: + oo = rng.choice(state.open_orders) + return FulfilmentAction(ActionKind.CANCEL_REPLACE, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.LIMIT, rng.randint(0,3), 0.25, params.passive_ttl_ms, cancel_order_id=oo.client_order_id, post_only=True, metadata={'requote': True}) + return FulfilmentAction(ActionKind.PLACE, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.LIMIT, rng.randint(0,3), 0.10, 200, post_only=True) + elif r < 0.50: + return FulfilmentAction(ActionKind.PLACE, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.LIMIT, rng.randint(0,params.chase_offset_ticks), rng.uniform(0.03,0.15), params.wait_to_retry_ms, post_only=True, metadata={'chase': True}) + elif r < 0.58: + 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.65: + pos = state.account.positions.get(state.venue.symbol) + if pos and abs(pos.qty) > 0.001: + return FulfilmentAction(ActionKind.FULL_EXIT, Side.SELL if pos.qty>0 else Side.BUY, OrderType.STOP_MARKET, 0, 1.0, 0, reduce_only=True) + return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) + elif r < 0.72: + pos = state.account.positions.get(state.venue.symbol) + if pos and abs(pos.qty) > 0.001: + return FulfilmentAction(ActionKind.FULL_EXIT, Side.SELL if pos.qty>0 else Side.BUY, OrderType.TRIGGER_MARKET, 0, 1.0, 0, reduce_only=True) + return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) + elif r < 0.79: + pos = state.account.positions.get(state.venue.symbol) + if pos and abs(pos.qty) > 0.001: + return FulfilmentAction(ActionKind.FULL_EXIT, Side.SELL if pos.qty>0 else Side.BUY, OrderType.TRAILING_STOP, 0, 1.0, 0, reduce_only=True) + return FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) + elif r < 0.86: + return FulfilmentAction(ActionKind.PLACE, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.STOP_MARKET, rng.randint(-5,5), rng.uniform(0.01,0.05), 200) + elif r < 0.93: + return FulfilmentAction(ActionKind.PLACE, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.TRIGGER_MARKET, rng.randint(-5,5), rng.uniform(0.01,0.05), 200) + else: + return FulfilmentAction(ActionKind.PLACE, Side.BUY if rng.random()<0.5 else Side.SELL, OrderType.TRAILING_STOP, rng.randint(-3,3), rng.uniform(0.02,0.08), 300) + + +def _params(): + return FulfilmentPolicyParams( + version='learn', 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, + wait_to_retry_ms=100, chase_enabled=True, + chase_offset_ticks=2, chase_max_retries=3, + urgency_taker_threshold=0.65, urgency_taker_penalty_bps=2.0, + execution_friction_threshold_bps=3.0, + ) + + +def run_episode(cwm, scenario, params, steps, seed, rng, risk_gate): + state = scenario.initial_state + fq_s=0; fq_e=0; fq_sr=0; fq_fv=0; fq_fc=0; fq_t=0 + for step in range(steps): + action = _gen(state, rng, params) + cp = tuple(c.rollout_action(state, rng) for c in scenario.counterparties) + if action.kind != ActionKind.NOOP: + planned = PlannedPolicy(actions=(action,), probabilities=(1.0,), selected_action=action, diagnostics={}) + if not risk_gate.validate(state, planned, params).approved: + action = FulfilmentAction(ActionKind.NOOP, None, None, 0, 0.0, 0) + fq_t += 1 + state = cwm.transition(state, (action, *cp)) + if state.fill_quality: + fq = state.fill_quality; fq_s += fq.slippage_bps; fq_e += fq.expected_slippage_bps + fq_sr += fq.slippage_bps - fq.expected_slippage_bps; fq_fv += fq.fill_value_score + if fq.filled: fq_fc += 1 + n = max(fq_t, 1) + pnl = (state.account.equity - 10000.0) / 10000.0 * 10_000 + return pnl, fq_s/n, fq_e/n, fq_sr/n, fq_fv/n, fq_fc/max(fq_t,1) + + +def main(): + t_start = time.time() + t_end = t_start + DURATION_S + monitor = ResourceMonitor() + + print("=" * 80) + print("MALKHUT 5-HOUR LEARNING TEST — OOM/CPU MONITORED") + print(f" Duration: 5h ({DURATION_S}s)") + print(f" CWM: HftBacktestCWM (calibrated slippage, urgency, chase)") + print(f" Swarm: {N_OPPONENTS} opponents") + print(f" Features: all order types, configurable friction per scenario") + print("=" * 80, flush=True) + + swarm = _build_swarm(N_OPPONENTS) + print(f"Swarm: {len(swarm)} opponents built", flush=True) + + cwm_factory = lambda: HftBacktestCWM(use_queue_model=True) + risk_gate = RiskGate() + params = _params() + factory = ScenarioFactory(exchange_id="bingx") + rng = random.Random(SEED) + + all_scenarios = [] + for sym in ASSETS: + all_scenarios.extend(factory.build_suite(symbols=[sym], steps_per_scenario=STEPS_PER_EPISODE, seed=SEED)) + print(f"Scenarios: {len(all_scenarios)} ({len(ASSETS)} assets)", flush=True) + + # Stats + all_pnls=[]; fq_s_sum=0; fq_e_sum=0; fq_sr_sum=0; fq_fv_sum=0; fq_fc_sum=0; fq_t_sum=0 + fq_n_reports=0; fq_history=[]; cycle=0 + cma_bests=[]; codec=CMAParameterCodec(); pool=SelfPlayPool(max_size=20) + monitor.sample() + + print(f"\nStarting 5-hour run...", flush=True) + + while time.time() < t_end: + cycle += 1 + remaining = t_end - time.time() + if remaining < 60: break + + try: + selected = rng.sample(all_scenarios, min(len(all_scenarios), 20)) + for i, scenario in enumerate(selected): + if time.time() > t_end - 30: break + ep = run_episode(cwm=cwm_factory(), scenario=scenario, params=params, + steps=STEPS_PER_EPISODE, seed=SEED + int(time.time()*1000) + i, + rng=random.Random(SEED + int(time.time()*1000) + i), risk_gate=risk_gate) + all_pnls.append(ep[0]) + fq_s_sum += ep[1]; fq_e_sum += ep[2]; fq_sr_sum += ep[3] + fq_fv_sum += ep[4]; fq_fc_sum += ep[5]; fq_t_sum += 1 + fq_n_reports += 1 + except Exception as e: + print(f" Error: {e}", flush=True) + + # CMA-ES every 15 reports + if fq_n_reports % 15 == 0 and remaining > 600: + try: + 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=rng.sample(all_scenarios, min(3, len(all_scenarios))), + budget_evals=2, seed=SEED + fq_n_reports) + cma_bests.append({"cycle": fq_n_reports, "score": best.score}) + if best.score > params.w_expected_pnl * 10: params = best.params + del best; import gc; gc.collect() + except Exception as e: + print(f" CMA-ES error: {e}", flush=True) + + # Monitor + report + elapsed = time.time() - t_start + rss, cpu = monitor.sample() + if fq_n_reports > 0 and fq_n_reports % 10 == 0: + n = fq_n_reports + h=elapsed/3600; m=(elapsed%3600)/60 + avg_fv=fq_fv_sum/n; avg_sl=fq_s_sum/n; avg_sr=fq_sr_sum/n + avg_fr=fq_fc_sum/max(fq_t_sum,1) + fq_history.append({"fv":avg_fv,"sl":avg_sl,"sr":avg_sr,"fr":avg_fr,"rss":rss,"cpu":cpu}) + trend="IMP" if len(fq_history)>=4 and fq_history[-1]["fv"]>fq_history[-4]["fv"] else "STB" + print(f" [{h:.1f}h{m:.0f}m] ep={len(all_pnls)} | fv={avg_fv:.3f}({trend}) | " + f"sl={avg_sl:.3f} | sr={avg_sr:+.3f} | fr={avg_fr:.1%} | " + f"RSS={rss:.0f}MB CPU={cpu:.0f}%", flush=True) + + # Final report + elapsed = time.time() - t_start + n = len(all_pnls) + mon = monitor.report() + print(f"\n{'='*80}\n 5-HOUR LEARNING TEST — FINAL REPORT\n{'='*80}") + if n > 0: + print(f" Duration: {elapsed/3600:.1f}h ({elapsed:.0f}s) | Episodes: {n} | Cycles: {cycle}") + print(f" PnL: avg={sum(all_pnls)/n:+.1f} best={max(all_pnls):+.1f} worst={min(all_pnls):+.1f} WR={sum(1 for p in all_pnls if p>0)/n*100:.1f}%") + if fq_history: + f=fq_history[0]; l=fq_history[-1] + print(f" Fill Value: {f['fv']:.3f} -> {l['fv']:.3f} ({(l['fv']-f['fv'])/max(abs(f['fv']),0.01)*100:+.1f}%)") + print(f" Slippage: {f['sl']:.3f} -> {l['sl']:.3f}") + print(f" Surprise: {f['sr']:+.3f} -> {l['sr']:+.3f}") + print(f" Fill Rate: {f['fr']:.1%} -> {l['fr']:.1%}") + print(f" CMA-ES: {len(cma_bests)} cycles") + print(f"\n RESOURCE USAGE:") + print(f" Peak RSS: {mon['peak_rss_mb']:.0f} MB") + print(f" Avg RSS: {mon['avg_rss_mb']:.0f} MB") + print(f" Peak CPU: {mon['peak_cpu_pct']:.0f}%") + print(f" Avg CPU: {mon['avg_cpu_pct']:.0f}%") + print(f"{'='*80}", flush=True) + + +if __name__ == "__main__": + try: main() + except Exception as e: + import traceback; print(f"\nFATAL: {e}", flush=True); traceback.print_exc(); sys.exit(1)