malkhut(e2e): memory-efficient long run — rolling stats, no episode accumulation

Fixed OOM kill by replacing all_episodes list accumulation with:
- Rolling stats (clear every 20 episodes)
- Only PnL history kept for characterization
- Peak/worst tracking without full episode storage
- Periodic stdout reports from rolling aggregates

100-opponent swarm + 9 assets × 30 scenarios = 270 scenarios per cycle.
CMA-ES every 5 cycles (3 evals). 3-hour target.

MALKHUT/malkhut/long_e2e_3h.py

@@ -291,7 +291,7 @@ def main():
     params = _baseline()
     factory = ScenarioFactory(exchange_id="bingx")
 
-    all_episodes: list = []
+    all_episodes: list = []  # DEPRECATED — use rolling stats
     cma_bests: list[PolicySnapshot] = []
     cycle = 0
     phase = "INIT"
@@ -309,6 +309,23 @@ def main():
     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
@@ -318,7 +335,6 @@ def main():
             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)
@@ -332,9 +348,31 @@ def main():
                     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
+                # 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,
@@ -347,8 +385,8 @@ def main():
             import traceback
             traceback.print_exc()
 
-        # Phase 2: Short CMA-ES optimization (every 5 cycles)
+        # Phase 2: CMA-ES every 5 cycles
-        if cycle % 5 == 0 and remaining > 600 and len(all_episodes) >= 20:
+        if cycle % 5 == 0 and remaining > 600 and recent_episodes >= 20:
             phase = f"CYCLE {cycle} — CMA-ES"
             cma_scenarios = rng.sample(all_scenarios, min(5, len(all_scenarios)))
             try:
@@ -363,81 +401,57 @@ def main():
             except Exception as e:
                 print(f"  CMA-ES error (cycle {cycle}): {e}", flush=True)
 
-        # Report
+        # Periodic report from rolling stats
         elapsed = time.time() - t_start
-        if elapsed > 0 and (len(all_episodes) % 20 == 0 or cycle % 3 == 0):
+        if elapsed > 0 and cycle % 5 == 0:
-            print_report(elapsed, phase, all_episodes, cma_bests, matrix)
+            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)
-
-    n = len(all_episodes)
     if n > 0:
-        pnls = [e["pnl_bps"] for e in all_episodes]
+        total_actions = sum(recent_ot_counts.values()) + recent_actions_total
-        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"  Total actions: {total_actions}")
         print(f"\n  PERFORMANCE")
-        print(f"  Avg PnL:        {sum(pnls)/n:+.1f} bps")
+        print(f"  Avg PnL:        {sum(all_pnls)/n:+.1f} bps")
-        print(f"  Median PnL:     {sorted(pnls)[n//2]:+.1f} bps")
+        print(f"  Median PnL:     {sorted(all_pnls)[n//2]:+.1f} bps")
-        print(f"  Best:           {max(pnls):+.1f} bps")
+        print(f"  Best:           {max(all_pnls):+.1f} bps")
-        print(f"  Worst:          {min(pnls):+.1f} bps")
+        print(f"  Worst:          {min(all_pnls):+.1f} bps")
-        print(f"  Std dev:        {math.sqrt(sum((p - sum(pnls)/n)**2 for p in pnls) / n):.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 pnls if p > 0)/n*100:.1f}%")
+        print(f"  Win rate:       {sum(1 for p in all_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"  Actions/sec:     {total_actions/max(elapsed,1):.0f}")
-        print(f"  Avg equity vol:  {sum(e.get('equity_volatility_bps', 0) for e in all_episodes)/n:.1f} bps")
+        print(f"  Episodes/hour:   {n/max(elapsed/3600,0.01):.0f}")
-        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
@@ -448,9 +462,11 @@ def main():
         "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,
+        "avg_pnl_bps": round(sum(all_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,
+        "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: