3b1dae6cbe
1. Friction configurable per scenario: Scenario gains maker_fee_bps, taker_fee_bps, adverse_cost_bps ScenarioFactory accepts friction overrides All 34 scenario builder calls updated System learns in ALL conditions (free maker, BingX real, Binance-like) 2. 5h learning test (long_learning_5h.py): - 1500 opponents, 9 assets, 270 scenarios - OOM/CPU monitoring (ResourceMonitor) - CMA-ES every 15 reports - Rolling fill_value/surprise/fill_rate tracking - Needs screen/tmux on host for 5h execution Note: background redirect issue on this shell session. Run via: cd MALKHUT && PYTHONUNBUFFERED=1 NUMBA_CACHE_DIR=/tmp/numba_cache python -m malkhut.long_learning_5h
305 lines
14 KiB
Python
305 lines
14 KiB
Python
#!/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)
|