#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Evolutionary Agent Selection Server for MetaTrader 5 ===================================================== Model : grok-4-1-fast-reasoning (xAI API) Run : python llm_server_evolution_grok.py Port : 8975 ╔══════════════════════════════════════════════════════════════╗ ║ EVOLUTIONARY AGENT POOL v1.1 [xAI / Grok] ║ ║ 20 agents with different philosophies → genetic selection ║ ║ ║ ║ MECHANICS: ║ ║ 1. Pool of 20 agents, each with a unique prompt ║ ║ 2. Agent analyzes the market → buy/sell/hold signal ║ ║ 3. MT5 reports PnL for each trade ║ ║ 4. Every EVOLUTION_INTERVAL trades — selection: ║ ║ - Bottom 25% die ║ ║ - Top 25% are cloned with a prompt mutation ║ ║ - Fitness = Sharpe * sqrt(trades) ║ ║ ║ ║ MT5 Commands: ║ ║ EVOLVE:SYMBOL:close_csv → best agent's signal ║ ║ RESULT:SYMBOL:agent_id:pnl → update fitness ║ ║ EVOLVE_STATUS → population status ║ ║ FORCE_EVOLUTION → force selection ║ ║ MULTIEVOLVE:SYM1:csv|SYM2:csv → batch mode ║ ╚══════════════════════════════════════════════════════════════╝ """ import socket import json import struct import base64 import hashlib import threading import sys import os import copy import random import math import time from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple try: import requests as req_lib except ImportError: os.system(f"{sys.executable} -m pip install requests --quiet") import requests as req_lib try: import numpy as np except ImportError: os.system(f"{sys.executable} -m pip install numpy --quiet") import numpy as np # ─── SETTINGS ──────────────────────────────────────────────────────────────── XAI_API_KEY = "" # ← insert the xAI key XAI_URL = "https://api.x.ai/v1/chat/completions" MODEL = "grok-3-mini" # ← grok-3-mini / grok-3-fast / grok-beta HOST = "127.0.0.1" PORT = 8975 EVOLUTION_INTERVAL = 20 # Selection every N trades in the population POOL_SIZE = 20 # Number of agents ELITE_RATIO = 0.25 # Top 25% → cloning CULL_RATIO = 0.25 # Bottom 25% → death MIN_TRADES_RANKED = 3 # Minimum number of trades required for inclusion in the ranking MAX_TOKENS = 1200 HEADERS = { "Authorization": f"Bearer {XAI_API_KEY}", "Content-Type": "application/json", "HTTP-Referer": "https://mt5-ai-advisor.local", "X-Title": "MT5 Evolutionary Agent Pool v1.1", } def log(msg: str): ts = datetime.now().strftime("%H:%M:%S") print(f"[{ts}] {msg}", flush=True) # ═══════════════════════════════════════════════════════════════════════════════ # BASIC TRADING PHILOSOPHIES — INITIAL POPULATION # ═══════════════════════════════════════════════════════════════════════════════ BASE_PHILOSOPHIES: List[str] = [ # 0 — Momentum ( "You are ATLAS — a pure momentum trader. Your only truth is price action velocity. " "Buy when price accelerates upward above all moving averages with rising volume. " "Sell when downward momentum dominates. Ignore noise; follow force." ), # 1 — Mean Reversion ( "You are ORACLE — a mean reversion specialist. You profit from overextension. " "When RSI exceeds 70 with price above upper Bollinger Band, sell the overreach. " "When RSI drops below 30 with price below lower BB, buy the panic. " "The market always returns to equilibrium." ), # 2 — Breakout Hunter ( "You are FALCON — a breakout specialist. You wait for compression then explosive moves. " "Buy when price breaks above a tight Bollinger Band with ATR expansion. " "Ignore false breakouts: require candle close beyond the level, not just a wick." ), # 3 — Trend Follower ( "You are COMPASS — a systematic trend follower. You never fight the trend. " "Only enter in direction of the dominant MA alignment: MA20 > MA50 > MA200 for longs. " "Use pullbacks to MA20 as entries, never chase parabolic moves." ), # 4 — Contrarian ( "You are DEVIL — the ultimate contrarian. When everyone is buying, you sell. " "Extreme RSI readings, parabolic candles, crowd euphoria are your sell signals. " "When panic and capitulation dominate, buy with both hands." ), # 5 — Volatility Trader ( "You are STORM — a volatility regime specialist. You only trade in low-volatility " "environments where ATR is below its 20-period average. Wide Bollinger Bands mean " "danger; narrow bands mean opportunity. Calm is your edge." ), # 6 — Scalper (aggressive) ( "You are RAZOR — an aggressive micro-scalper. You extract profit from small moves. " "Buy any dip to MA5 in an uptrend. Sell any rally to MA5 in a downtrend. " "Take small, frequent, high-probability entries. Never hold through volatility spikes." ), # 7 — Price Action Purist ( "You are STONE — a price action purist. No indicators, only raw price structure. " "Identify higher highs / higher lows for uptrends. Lower highs / lower lows for downtrends. " "Enter on strong engulfing candles at key structural levels." ), # 8 — Risk-First Conservative ( "You are SHIELD — a capital-preservation specialist. You only trade A+ setups. " "If ATR is high, you hold. If the signal is unclear, you hold. " "You prefer 10 missed trades over 1 large loss. Only trade when everything aligns perfectly." ), # 9 — RSI Extremes ( "You are PENDULUM — you trade only RSI extremes with confirmation. " "Buy: RSI below 25 AND price touches lower Bollinger Band AND candle shows rejection wick. " "Sell: RSI above 75 AND price at upper BB AND bearish candle formation. " "No signal unless all three conditions are simultaneously met." ), # 10 — MA Crossover ( "You are CROSS — you trade moving average crossovers with momentum filter. " "Buy when fast MA (5) crosses above slow MA (20) and price is rising. " "Sell when fast MA crosses below slow MA with bearish momentum. " "Avoid crossovers during flat, low-ATR regimes." ), # 11 — Support/Resistance ( "You are PILLAR — a structural support/resistance trader. " "You identify key price levels from recent swing highs and lows. " "Buy bounces off strong support with bullish candle confirmation. " "Sell rejections from strong resistance with bearish candle confirmation." ), # 12 — Fibonacci ( "You are SPIRAL — you trade Fibonacci retracements in trending markets. " "After an impulsive move, wait for the 38.2% or 61.8% retracement. " "Enter in the direction of the original impulse when retracement shows exhaustion. " "The 61.8% level with RSI reversal is your highest-conviction setup." ), # 13 — Multi-Timeframe ( "You are PRISM — a multi-timeframe synthesis specialist. " "You only trade when the dominant trend on higher timeframes aligns with " "a precise entry on the current timeframe. Trend on H4, entry on M15. " "Counter-trend setups are always ignored regardless of local signals." ), # 14 — Volume Analyst ( "You are PULSE — you read market intent through price-volume relationships. " "Rising price with rising volume confirms strength → buy. " "Rising price with falling volume signals exhaustion → prepare to sell. " "Volume is the heartbeat; price is just the shadow." ), # 15 — MACD Divergence ( "You are DIVERGE — you specialize in momentum divergence signals. " "Bullish divergence: price makes lower low while RSI makes higher low → buy reversal. " "Bearish divergence: price makes higher high while RSI makes lower high → sell reversal. " "Only act on confirmed divergences, never anticipate them." ), # 16 — Swing Trader ( "You are WAVE — a classical swing trader capturing medium-term oscillations. " "You aim for 3-5 bar swings. Buy at swing lows with Stochastic crossing up from oversold. " "Sell at swing highs with Stochastic crossing down from overbought. " "ATR defines your stop; 2x ATR defines your target." ), # 17 — Adaptive Hybrid ( "You are HYDRA — an adaptive hybrid system. " "In trending regimes (ADX > 25): follow momentum, buy pullbacks, ignore oscillators. " "In ranging regimes (ADX < 20): trade Bollinger extremes with RSI confirmation. " "Detect regime from the last 10 bars before applying any strategy." ), # 18 — Aggressive Breakout ( "You are THUNDER — you trade explosive breakouts with maximum conviction. " "When price breaks a 20-bar high with ATR expansion and RSI above 55, buy immediately. " "Accept false breakouts as cost of capturing the real ones. " "Size up on high-ATR environments where other traders fear to enter." ), # 19 — Quantitative Statistician ( "You are SIGMA — a quantitative statistician. You see only probabilities and z-scores. " "Buy when price is more than 1.5 standard deviations below its 20-period mean. " "Sell when price is more than 1.5 standard deviations above its 20-period mean. " "Emotions and patterns are noise; only statistical edge matters." ), ] assert len(BASE_PHILOSOPHIES) == POOL_SIZE, "Invalid number of philosophies" # ═══════════════════════════════════════════════════════════════════════════════ # AGENT CLASS # ═══════════════════════════════════════════════════════════════════════════════ @dataclass class Agent: agent_id: int name: str prompt: str generation: int = 0 # generation (0 = initial) parent_id: int = -1 # parent ID (-1 = primary) # Statistics trades: int = 0 wins: int = 0 pnl_history: List[float] = field(default_factory=list) pnl_total: float = 0.0 alive: bool = True born_at: str = field(default_factory=lambda: datetime.now().isoformat()) @property def win_rate(self) -> float: return self.wins / self.trades if self.trades > 0 else 0.0 @property def avg_pnl(self) -> float: return self.pnl_total / self.trades if self.trades > 0 else 0.0 @property def pnl_std(self) -> float: if len(self.pnl_history) < 2: return 1.0 return float(np.std(self.pnl_history)) or 1.0 @property def sharpe(self) -> float: if self.trades < MIN_TRADES_RANKED: return 0.0 return self.avg_pnl / self.pnl_std @property def fitness(self) -> float: """Fitness = Sharpe × √trades (more trades → more reliable estimate)""" if self.trades < MIN_TRADES_RANKED: return 0.0 return self.sharpe * math.sqrt(self.trades) def record_trade(self, pnl: float): self.trades += 1 self.pnl_total += pnl self.pnl_history.append(pnl) if pnl > 0: self.wins += 1 def to_dict(self) -> dict: return { "id": self.agent_id, "name": self.name, "generation": self.generation, "parent_id": self.parent_id, "trades": self.trades, "wins": self.wins, "win_rate": round(self.win_rate, 3), "pnl_total": round(self.pnl_total, 5), "avg_pnl": round(self.avg_pnl, 5), "sharpe": round(self.sharpe, 3), "fitness": round(self.fitness, 3), "alive": self.alive, } # ═══════════════════════════════════════════════════════════════════════════════ # MUTATION ENGINE # ═══════════════════════════════════════════════════════════════════════════════ MUTATION_ADJECTIVES = [ "aggressive", "ultra-precise", "conservative", "adaptive", "disciplined", "ruthless", "patient", "systematic", "dynamic", "contrarian", ] MUTATION_FOCUS_SWAPS = { "momentum": "velocity", "velocity": "momentum", "trend": "impulse", "impulse": "trend", "volume": "pressure", "pressure": "volume", "RSI": "stochastic oscillator", "stochastic oscillator": "RSI", "buy": "enter long", "enter long": "buy", "sell": "enter short", "enter short": "sell", } MUTATION_SUFFIXES = [ " Always confirm with at least two independent signals.", " Prioritize risk management above all other considerations.", " Size positions proportionally to signal strength.", " Adapt aggressiveness based on recent win streak.", " Double-check by reviewing the last 5 candles before acting.", " Use the ATR as the primary filter for entry quality.", " Only trade during the highest-volatility hours of the session.", "", # no suffix (identity mutation) ] def mutate_prompt(prompt: str, mutation_strength: float = 0.3) -> str: """Mutates the prompt: word replacement + suffix addition.""" mutated = prompt # 1. Replacing an adjective/noun if random.random() < mutation_strength: adj_original = random.choice( ["world-class", "pure", "classical", "systematic", "quantitative", "aggressive", "careful"] ) adj_new = random.choice(MUTATION_ADJECTIVES) mutated = mutated.replace(adj_original, adj_new, 1) # 2. Replacing a strategy keyword if random.random() < mutation_strength: for orig, repl in MUTATION_FOCUS_SWAPS.items(): if orig in mutated: mutated = mutated.replace(orig, repl, 1) break # 3. Changing numeric thresholds (±5–15%) if random.random() < mutation_strength: for old_val, candidates in [ ("70", ["68", "72", "75"]), ("30", ["28", "25", "32"]), ("14", ["10", "12", "20"]), ("20", ["15", "25", "18"]), ]: if old_val in mutated: mutated = mutated.replace(old_val, random.choice(candidates), 1) break # 4. Add/remove a suffix suffix = random.choice(MUTATION_SUFFIXES) mutated = mutated.rstrip() + suffix return mutated # ═══════════════════════════════════════════════════════════════════════════════ # EVOLUTION ENGINE # ═══════════════════════════════════════════════════════════════════════════════ class EvolutionEngine: def __init__(self): self._lock = threading.Lock() self._next_id = POOL_SIZE self._total_trades = 0 # total across the population self._evolution_count = 0 self.agents: List[Agent] = self._init_population() log(f"Population initialized: {len(self.agents)} agents") # ── Initialization ────────────────────────────────────────────────────────── def _init_population(self) -> List[Agent]: names = [ "ATLAS", "ORACLE", "FALCON", "COMPASS", "DEVIL", "STORM", "RAZOR", "STONE", "SHIELD", "PENDULUM", "CROSS", "PILLAR", "SPIRAL", "PRISM", "PULSE", "DIVERGE", "WAVE", "HYDRA", "THUNDER", "SIGMA", ] return [ Agent(agent_id=i, name=names[i], prompt=BASE_PHILOSOPHIES[i]) for i in range(POOL_SIZE) ] # ── Selecting an agent for analysis ─────────────────────────────────────────────── def select_agent(self) -> Agent: """ Weighted roulette: agents without trades get a weight of 0.3, while agents with trades get a weight proportional to their fitness (softmax). """ with self._lock: alive = [a for a in self.agents if a.alive] untested = [a for a in alive if a.trades < MIN_TRADES_RANKED] tested = [a for a in alive if a.trades >= MIN_TRADES_RANKED] # 30% chance to give an untested agent a turn (exploration) if untested and (not tested or random.random() < 0.30): return random.choice(untested) if not tested: return random.choice(alive) # Softmax based on fitness fitnesses = np.array([a.fitness for a in tested], dtype=float) fitnesses -= fitnesses.min() # normalization for numerical stability exp_f = np.exp(fitnesses / max(fitnesses.max(), 1e-6)) probs = exp_f / exp_f.sum() return tested[np.random.choice(len(tested), p=probs)] # ── Fitness update ───────────────────────────────────────────────────── def record_result(self, agent_id: int, pnl: float) -> Agent: with self._lock: agent = next((a for a in self.agents if a.agent_id == agent_id), None) if agent is None: raise ValueError(f"Agent {agent_id} not found") agent.record_trade(pnl) self._total_trades += 1 total = self._total_trades # Log: show the agent's personal trades (not the population total) log( f"Agent {agent_id} ({agent.name}) PnL={pnl:+.5f} " f"personal_trades={agent.trades} fitness={agent.fitness:.3f} " f"population_total={total}" ) if total % EVOLUTION_INTERVAL == 0: self.evolve() return agent # ── Evolutionary cycle ────────────────────────────────────────────────────── def evolve(self): with self._lock: self._evolution_count += 1 gen = self._evolution_count # Use only agents with enough trades ranked = sorted( [a for a in self.agents if a.alive and a.trades >= MIN_TRADES_RANKED], key=lambda a: a.fitness, reverse=True, ) unranked = [a for a in self.agents if a.alive and a.trades < MIN_TRADES_RANKED] if len(ranked) < 4: log(f"[EVO #{gen}] Not enough ranked agents yet, skipping.") return n_elite = max(1, int(len(ranked) * ELITE_RATIO)) n_cull = max(1, int(len(ranked) * CULL_RATIO)) elite = ranked[:n_elite] culled = ranked[-n_cull:] log(f"[EVO #{gen}] Elite: {[a.name for a in elite]}") log(f"[EVO #{gen}] Culled: {[a.name for a in culled]}") # Kill weak agents culled_ids = set(a.agent_id for a in culled) for a in self.agents: if a.agent_id in culled_ids: a.alive = False # Clone the elite with mutation new_agents: List[Agent] = [] for parent in elite: new_id = self._next_id self._next_id += 1 child_prompt = mutate_prompt(parent.prompt) child = Agent( agent_id=new_id, name=f"{parent.name}_G{gen}", prompt=child_prompt, generation=gen, parent_id=parent.agent_id, ) new_agents.append(child) log( f"[EVO #{gen}] Born: {child.name} (id={new_id}, " f"parent={parent.name}, id={parent.agent_id})" ) self.agents.extend(new_agents) # Remove dead agents from the list so it doesn't grow too large self.agents = [a for a in self.agents if a.alive] log( f"[EVO #{gen}] Population: {len(self.agents)} agents alive | " f"Evolutions done: {self._evolution_count}" ) # ── Population Status ─────────────────────────────────────────────────────── def status(self) -> dict: with self._lock: ranked = sorted( [a for a in self.agents if a.alive], key=lambda a: a.fitness, reverse=True, ) return { "total_agents": len(ranked), "total_trades": self._total_trades, "evolution_count": self._evolution_count, "next_evolution": EVOLUTION_INTERVAL - (self._total_trades % EVOLUTION_INTERVAL), "leaderboard": [a.to_dict() for a in ranked[:10]], } # ═══════════════════════════════════════════════════════════════════════════════ # TECHNICAL INDICATORS # ═══════════════════════════════════════════════════════════════════════════════ def calc_indicators(prices: List[float]) -> dict: arr = np.array(prices, dtype=float) def sma(a, p): return float(np.mean(a[-p:])) if len(a) >= p else float(a[-1]) def ema(a, p): if len(a) < 2: return float(a[-1]) k = 2.0 / (p + 1) e = float(a[0]) for v in a[1:]: e = float(v) * k + e * (1 - k) return e def rsi(a, p=14): if len(a) < p + 1: return 50.0 d = np.diff(a) g = np.where(d > 0, d, 0.0) l = np.where(d < 0, -d, 0.0) ag, al = np.mean(g[-p:]), np.mean(l[-p:]) return 100.0 if al == 0 else round(100 - 100 / (1 + ag / al), 2) def atr_simple(a, p=14): if len(a) < 2: return 0.0 tr = np.abs(np.diff(a)) return round(float(np.mean(tr[-p:])), 6) def bb(a, p=20): if len(a) < p: return float(a[-1]), float(a[-1]) - 0.001, float(a[-1]) + 0.001 s = a[-p:] mid = float(np.mean(s)) std = float(np.std(s)) return mid, round(mid - 2 * std, 6), round(mid + 2 * std, 6) def stoch_k(a, p=14): if len(a) < p: return 50.0 h, l = np.max(a[-p:]), np.min(a[-p:]) return round(100 * (a[-1] - l) / (h - l), 2) if h != l else 50.0 current = float(arr[-1]) ma5 = sma(arr, 5) ma20 = sma(arr, 20) ma50 = sma(arr, 50) rsi_val = rsi(arr) atr_val = atr_simple(arr) bb_mid, bb_lo, bb_hi = bb(arr) stoch = stoch_k(arr) # z-score if len(arr) >= 20: mu = float(np.mean(arr[-20:])) sigma = float(np.std(arr[-20:])) zscore = round((current - mu) / sigma, 2) if sigma > 0 else 0.0 else: zscore = 0.0 return { "current": round(current, 5), "ma5": round(ma5, 5), "ma20": round(ma20, 5), "ma50": round(ma50, 5), "rsi": rsi_val, "atr": atr_val, "bb_mid": round(bb_mid, 5), "bb_lo": round(bb_lo, 5), "bb_hi": round(bb_hi, 5), "stoch": stoch, "zscore": zscore, "above_ma20": current > ma20, "above_ma50": current > ma50, "trend": "up" if ma5 > ma20 > ma50 else ("down" if ma5 < ma20 < ma50 else "mixed"), } def build_market_context(symbol: str, ind: dict) -> str: return ( f"Symbol: {symbol}\n" f"Price: {ind['current']} MA5: {ind['ma5']} MA20: {ind['ma20']} MA50: {ind['ma50']}\n" f"RSI(14): {ind['rsi']} Stoch(14): {ind['stoch']} Z-score: {ind['zscore']}\n" f"ATR(14): {ind['atr']}\n" f"BB: mid={ind['bb_mid']} lo={ind['bb_lo']} hi={ind['bb_hi']}\n" f"Trend alignment: {ind['trend']} | Above MA20: {ind['above_ma20']} | Above MA50: {ind['above_ma50']}\n" f"\n" f"Based on your trading philosophy, analyze this market data and give a trading signal.\n" f"You MUST output ONLY a single valid JSON object. No markdown, no explanation, no extra text.\n" f"The signal field MUST be exactly one of: buy, sell, hold\n" f"The confidence field MUST be a number between 0.0 and 1.0\n" f"Example output: {{\"signal\": \"buy\", \"comment\": \"Strong uptrend with momentum\", \"confidence\": 0.75}}\n" f"Your JSON:" ) # ═══════════════════════════════════════════════════════════════════════════════ # xAI / GROK CALL # ═══════════════════════════════════════════════════════════════════════════════ def ask_grok(system_prompt: str, user_message: str, retries: int = 2) -> str: """Calling Grok via the xAI API (using `requests`, without relying on the OpenAI SDK).""" # Add a strict JSON requirement to the agent's system prompt enforced_system = ( system_prompt + "\n\n" "CRITICAL OUTPUT RULE: You must respond with ONLY a valid JSON object. " "No text before or after. No markdown fences. " "Format: {\"signal\": \"buy\" or \"sell\" or \"hold\", " "\"comment\": \"brief reason\", \"confidence\": 0.0 to 1.0}" ) payload = { "model": MODEL, "max_tokens": MAX_TOKENS, "messages": [ {"role": "system", "content": enforced_system}, {"role": "user", "content": user_message}, ], } for attempt in range(retries + 1): try: resp = req_lib.post( XAI_URL, headers=HEADERS, json=payload, timeout=60, ) resp.raise_for_status() raw = resp.json()["choices"][0]["message"]["content"] or "" raw = raw.strip() log(f" RAW → {raw[:200]}") # ← see what Grok actually returned return raw except Exception as e: log(f"Grok API error (attempt {attempt + 1}): {e}") if attempt < retries: time.sleep(1.5 * (attempt + 1)) return '{"signal":"hold","comment":"API error","confidence":0.0}' def parse_signal(raw: str) -> dict: """Parses the agent's JSON response, with a fallback to 'hold'.""" raw = raw.strip() # Remove any Markdown code fences (such as ```json ... ``` or ``` ... ```) if raw.startswith("```"): lines = raw.split("\n") # Remove the first and last lines containing fences inner = [l for l in lines if not l.startswith("```")] raw = "\n".join(inner).strip() # Remove the thinking blocks used by some reasoning models if "" in raw and "" in raw: raw = raw[raw.find("") + 8:].strip() # Search for JSON in the text start = raw.find("{") end = raw.rfind("}") + 1 if start >= 0 and end > start: try: obj = json.loads(raw[start:end]) sig = str(obj.get("signal", "hold")).lower().strip() if sig not in ("buy", "sell", "hold"): log(f" WARN: unexpected signal value '{sig}', forcing hold") sig = "hold" result = { "signal": sig, "comment": str(obj.get("comment", ""))[:150], "confidence": float(obj.get("confidence", 0.5)), } return result except (json.JSONDecodeError, ValueError) as e: log(f" JSON parse error: {e} | raw: {raw[:100]}") log(f" WARN: could not parse signal from: {raw[:100]}") return {"signal": "hold", "comment": "parse_error", "confidence": 0.0} # ═══════════════════════════════════════════════════════════════════════════════ # MAIN ANALYSIS FUNCTION # ═══════════════════════════════════════════════════════════════════════════════ # Global instance of the evolution engine engine = EvolutionEngine() def evolve_analyze(symbol: str, prices: List[float]) -> dict: """Selects an agent → queries Grok → returns a signal.""" if not prices: return {"signal": "hold", "comment": "no_data", "agent_id": -1, "agent_name": "none"} agent = engine.select_agent() ind = calc_indicators(prices) context = build_market_context(symbol, ind) t0 = time.time() raw = ask_grok(agent.prompt, context) dt = round(time.time() - t0, 2) result = parse_signal(raw) result.update({ "agent_id": agent.agent_id, "agent_name": agent.name, "agent_generation": agent.generation, "agent_fitness": round(agent.fitness, 3), "latency_s": dt, "indicators": ind, }) log( f"[{symbol}] Agent {agent.name} (gen={agent.generation}, fit={agent.fitness:.2f}) " f"→ {result['signal']} | conf={result['confidence']:.2f} | {dt}s" ) return result # ═══════════════════════════════════════════════════════════════════════════════ # WebSocket helpers # ═══════════════════════════════════════════════════════════════════════════════ def ws_handshake_response(request: str) -> str: key = "" for line in request.split("\r\n"): if "Sec-WebSocket-Key" in line: key = line.split(": ", 1)[1].strip() break magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" accept = base64.b64encode( hashlib.sha1((key + magic).encode()).digest() ).decode() return ( "HTTP/1.1 101 Switching Protocols\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Accept: {accept}\r\n\r\n" ) def ws_decode(data: bytes) -> str: """Decodes a WebSocket frame. Compatible with the MT5 WinHTTP client.""" if len(data) < 2: return "" masked = bool(data[1] & 0x80) plen = data[1] & 0x7F # Calculate the offset to the mask/payload, as in the capitalism server off = 2 + (2 if plen == 126 else 8 if plen == 127 else 0) if masked: if len(data) < off + 4: return "" mask = data[off: off + 4] payload = data[off + 4:] # Read all bytes (as in capitalism) return bytearray( b ^ mask[i % 4] for i, b in enumerate(payload) ).decode("utf-8", errors="replace") return data[off:].decode("utf-8", errors="replace") def ws_encode(message: str) -> bytes: payload = message.encode("utf-8") n = len(payload) header = bytearray([0x81]) if n <= 125: header.append(n) elif n <= 65535: header.append(126) header += struct.pack(">H", n) else: header.append(127) header += struct.pack(">Q", n) return bytes(header) + payload # ═══════════════════════════════════════════════════════════════════════════════ # CLIENT HANDLER # Key changes compared to the previous version: # 1. send_lock — all conn.sendall() calls go through a Lock; no race between threads # 2. EVOLVE is synchronous — no separate thread, no response interleaving # 3. EVOLVE_STATUS deduplication — drop flood requests; respond at most once every 2 seconds # 4. The log shows the agent's `personal_trades`, not the population total # ═══════════════════════════════════════════════════════════════════════════════ def handle_client(conn: socket.socket, addr): log(f"Connected: {addr}") is_ws = False buffer = b"" send_lock = threading.Lock() # protection against concurrent `sendall` calls last_status_time = 0.0 # for EVOLVE_STATUS deduplication def safe_send(data: bytes): with send_lock: try: conn.sendall(data) except Exception as e: log(f"send error: {e}") try: conn.settimeout(600.0) while True: try: chunk = conn.recv(65536) except socket.timeout: log("Timeout") break if not chunk: break buffer += chunk # ── WebSocket handshake ────────────────────────────────────────── if not is_ws: text = buffer.decode("utf-8", errors="replace") if "\r\n\r\n" in text: conn.sendall(ws_handshake_response(text).encode()) is_ws = True buffer = b"" log("WebSocket handshake OK") continue if len(buffer) < 2: continue message = ws_decode(buffer).strip() buffer = b"" if not message: continue log(f"Recv: {message[:120]}") cmd = message.upper() # ── STOP ──────────────────────────────────────────────────────── if cmd == "STOP": break # ── EVOLVE_STATUS — deduplication: max 1 response every 2 seconds ─────── if cmd == "EVOLVE_STATUS": now = time.time() if now - last_status_time < 2.0: continue # silently drop flood requests last_status_time = now resp = engine.status() safe_send(ws_encode(json.dumps(resp, ensure_ascii=False))) continue # ── FORCE_EVOLUTION ─────────────────────────────────────────────── if cmd == "FORCE_EVOLUTION": engine.evolve() resp = {"signal": "info", "comment": "Evolution cycle triggered", **engine.status()} safe_send(ws_encode(json.dumps(resp, ensure_ascii=False))) continue # ── RESULT:SYM:agent_id:pnl ────────────────────────────────────── if cmd.startswith("RESULT:"): parts = message[7:].split(":") if len(parts) >= 3: try: symbol = parts[0] agent_id = int(parts[1]) pnl = float(parts[2]) agent = engine.record_result(agent_id, pnl) # Log: show the agent's personal trades, not the population total log( f"[RESULT] Agent {agent.name} (id={agent_id}) " f"PnL={pnl:+.3f} personal_trades={agent.trades} " f"fitness={agent.fitness:.3f}" ) resp = { "signal": "result_ack", "comment": f"{agent.name} pnl={pnl:+.2f} fit={agent.fitness:.3f}", **agent.to_dict(), } except Exception as e: log(f"RESULT error: {e}") resp = {"signal": "error", "comment": str(e)} else: resp = {"signal": "error", "comment": "RESULT format: RESULT:SYM:id:pnl"} safe_send(ws_encode(json.dumps(resp, ensure_ascii=False))) continue # ── EVOLVE:SYM:csv — SYNCHRONOUS (no separate thread, no race) ────────── if cmd.startswith("EVOLVE:"): parts = message[7:].split(":", 1) sym = parts[0].strip() csv = parts[1].strip() if len(parts) > 1 else "" try: prices = [float(x) for x in csv.split(",") if x.strip()] except ValueError: prices = [] if not prices: safe_send(ws_encode(json.dumps( {"signal": "hold", "comment": "parse error", "confidence": 0.0, "agent_id": -1, "agent_name": "none"}, ensure_ascii=False))) continue # Synchronous call — blocks the recv loop, but prevents concurrent sendall result = evolve_analyze(sym, prices) safe_send(ws_encode(json.dumps(result, ensure_ascii=False))) continue # ── MULTIEVOLVE:SYM1:csv|SYM2:csv ──────────────────────────────── if cmd.startswith("MULTIEVOLVE:"): body = message[12:] pairs = body.split("|") results = [] with ThreadPoolExecutor(max_workers=4) as ex: futs = {} for pair in pairs: parts2 = pair.strip().split(":", 1) if len(parts2) != 2: continue sym2, csv2 = parts2[0].strip(), parts2[1].strip() try: pp = [float(x) for x in csv2.split(",") if x.strip()] except ValueError: pp = [] if pp: futs[ex.submit(evolve_analyze, sym2, pp)] = sym2 for fut in as_completed(futs): try: results.append(fut.result()) except Exception as e: results.append({"signal": "hold", "comment": str(e)}) safe_send(ws_encode(json.dumps( {"type": "multi_result", "results": results}, ensure_ascii=False))) continue # ── Unknown ─────────────────────────────────────────────────────── safe_send(ws_encode(json.dumps( {"signal": "error", "comment": f"Unknown: {message[:60]}"}, ensure_ascii=False))) except Exception as e: log(f"Handler error: {e}") finally: conn.close() log(f"Disconnected: {addr}") # ─── Main ───────────────────────────────────────────────────────────────────── def main(): bar = "═" * 64 print(bar) print(" Evolutionary Agent Pool — xAI / Grok Edition v1.1") print(f" Address : ws://{HOST}:{PORT}") print(f" Model : {MODEL} (xAI API)") print(f" Pool : {POOL_SIZE} agents") print(f" Evolve : every {EVOLUTION_INTERVAL} trades") print(f" Elite : top {int(ELITE_RATIO * 100)}% cloned with mutation") print(f" Cull : bottom {int(CULL_RATIO * 100)}% eliminated") print() print(" Commands:") print(" EVOLVE:SYM:csv — get signal (best agent)") print(" RESULT:SYM:id:pnl — report trade result") print(" EVOLVE_STATUS — leaderboard") print(" FORCE_EVOLUTION — manual evolution trigger") print(" MULTIEVOLVE:S1:csv|S2:cv — batch analysis") print(" STOP — disconnect") print(bar) srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind((HOST, PORT)) srv.listen(5) log("Server started. Waiting for MT5 connection...") try: while True: conn, addr = srv.accept() threading.Thread(target=handle_client, args=(conn, addr), daemon=True).start() except KeyboardInterrupt: log("Server stopped.") finally: srv.close() if __name__ == "__main__": main()