MQL5とデータ処理パッケージの統合（第7回）：銘柄間連携のためのマルチエージェント環境の構築

目次：

1. はじめに
2. システムの概要と理解
3. 導入手順
4. MQL5での統合実装
5. ライブデモ
6. 結論

はじめに

第6回の「市場フィードバックとモデル適応の統合」では、市場の挙動と意思決定ロジックの間にフィードバックループを構築することに焦点を当てました。固定的なシグナルだけに依存するのではなく、取引システム自身の運用成績を継続的に評価し、市場環境の変化に対応し、内部パラメータを適応的に調整できる仕組みを導入しました。これには、取引結果、ボラティリティ変化、市場構造の変化といったリアルタイムフィードバックを活用し、MQL5–Pythonハイブリッドアーキテクチャ内でシグナルの解釈方法と実行方法を継続的に改善する仕組みが含まれています。

今回は、この統合をさらに拡張し、銘柄間連携が可能なマルチエージェント環境を構築します。目的は、それぞれ独立したエージェントが異なる市場や銘柄を分析し、相互に洞察を共有しながら、連携的に取引判断へ影響を与えるフレームワークを設計することです。このアプローチでは、通貨間の相関関係やリスクセンチメントなどの銘柄間の関係性を活用することで、シグナル品質の向上、ダマシのシグナルの削減、そして単独の価格変動だけではなく、より広範な市場コンテキストに適応する堅牢な取引システムの構築を目指します。

システムの概要と理解

銘柄間の連携を可能にするマルチエージェント環境を構築する場合、中核となる考え方は、銘柄ごとに分離された戦略から、ネットワーク化されたインテリジェンスアーキテクチャへと移行します。この構造では、各エージェントは集合的な意思決定グラフ内の専門ノードとして機能します。各エージェントは、EURUSD、ゴールド、株価指数など、特定の銘柄または市場セグメントに対する専門的な知識を持ちます。一方で、他のエージェントとの動的な通信チャネルも維持します。これにより、システムは市場間の関係性を単なる固定的な相関としてではなく、リアルタイムで活用可能な情報経路として認識できます。

この環境の動作ロジックの中心となるのは「連携型自律性」です。エージェントは局所的な実行権限を持ちますが、システム全体の制約や相乗効果を管理するグローバル調整レイヤーに接続されています。この調整レイヤーは、共有コンテキストバッファ、報酬設計、軽量なコンセンサスメカニズムを通じて、個々のエージェントのアクションを調整し、最大相関エクスポージャーや銘柄間のボラティリティ目標など、ポートフォリオ全体の目標を維持します。たとえば、複数のエージェントが相反するシグナルを検出した場合（例：あるエージェントがGBPUSDに対して強気判断を出し、別のエージェントがドル指数に対して弱気判断を出す場合）、調整レイヤーは構造化された推論プロセスを開始できます。その際、証拠の階層性や過去の相互作用パターンを踏まえて評価し、最終的な方向性バイアスを決定します。

導入手順

import MetaTrader5 as mt5
import pandas as pd
import numpy as np
import json
import time
import schedule
from datetime import datetime, timedelta
from flask import Flask, jsonify, request
import threading
import warnings
warnings.filterwarnings('ignore')

account = '123456'
password = 'YourPass@MT5'
server = 'YourBrokersServer-Demo'

print("=" * 80)
print(" MULTI-AGENT TRADING SYSTEM - LIVE DEPLOYMENT")
print("=" * 80)

まず、PythonとMetaTrader 5を接続し、データ管理、タイミング制御、外部通信を行うために必要となる基本環境を構築します。市場アクセスにはMetaTrader 5モジュールを使用し、データ処理にはpandasおよびNumPyを利用します。また、Flaskを使用することで、軽量なAPIを通じてシグナルやシステム状態を外部へ公開します。さらに、スケジューリング、スレッド処理、タイムスタンプ管理、JSONシリアライズなどを処理する追加モジュールを導入します。これにより、取引システムは継続的に動作しながら、MQL5とリアルタイムかつ安全に連携できます。最後に、account、password、serverの各変数は、MetaTrader 5へのライブ口座またはデモ口座接続を確立するために必要な認証情報を定義します。

class LiveMultiAgentSystem:
    def __init__(self, account_number, password, server):
        self.account_number = account_number
        self.password = password
        self.server = server
        self.symbols = ["XAUUSD", "EURUSD", "GBPUSD", "USDJPY"]
        self.agents = {}
        self.positions = {}
        self.decision_log = []
        
        # Initialize MT5
        self.init_mt5()
        
        # Initialize agents
        self.init_agents()
        
        print(":/ Live Multi-Agent System Initialized")
    
    def init_mt5(self):
        """Initialize connection to MT5"""
        if not mt5.initialize():
            print("X MT5 initialization failed")
            print(f"   Error: {mt5.last_error()}")
            return False
        
        # Login to account
        authorized = mt5.login(
            login=self.account_number,
            password=self.password,
            server=self.server
        )
        
        if authorized:
            print(f":/ Connected to MT5 Account: {self.account_number}")
            account_info = mt5.account_info()
            print(f"   Balance: ${account_info.balance:.2f}")
            print(f"   Equity: ${account_info.equity:.2f}")
            return True
        else:
            print(f"X Login failed: {mt5.last_error()}")
            return False

このクラスでは、ライブ環境で動作するマルチエージェント取引環境を管理するための基本構造を定義します。コンストラクタでは、アカウント認証情報の設定、あらかじめ定義された銘柄リストの割り当て、エージェント、保有ポジション、意思決定履歴を管理するためのコンテナの準備を行います。その後、ターミナルの初期化と指定された取引アカウントへのログイン処理を実行することで、MetaTrader 5への接続を確立します。接続状態を検証し、基本的なアカウント情報も取得します。このセットアップにより、すべてのエージェントは共有された認証済みのMetaTrader 5セッション上で動作できるようになります。これが、連携的かつリアルタイムな複数銘柄意思決定を実現するための基盤となります。

    def init_agents(self):
        """Initialize trading agents"""
        print("\n Initializing Live Agents...")
             
        from collections import defaultdict
        
        self.agents = {
            "XAUUSD": {
                "name": "Gold Agent",
                "weights": {
                    "trend": 1.5,
                    "momentum": 1.2,
                    "volatility": 1.0
                },
                "params": {
                    "ema_period": 20,
                    "rsi_period": 14,
                    "atr_period": 14
                }
            },
            "CONTEXT": {
                "name": "Context Agent",
                "weights": {
                    "usd_strength": 1.3,
                    "risk_sentiment": 1.1
                }
            },
            "LIQUIDITY": {
                "name": "Liquidity Agent",
                "weights": {
                    "volatility": 1.2,
                    "volume": 1.0
                }
            }
        }
        
        print(f":/ Created {len(self.agents)} agents")
    
    def get_live_data(self, symbol, timeframe=mt5.TIMEFRAME_M5, bars=100):
        """Get live data from MT5"""
        try:
            rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bars)
            if rates is None or len(rates) == 0:
                print(f"❌ No data for {symbol}")
                return None
            
            df = pd.DataFrame(rates)
            df['time'] = pd.to_datetime(df['time'], unit='s')
            df.set_index('time', inplace=True)
            
            # Calculate features
            df['returns'] = df['close'].pct_change()
            df['ema_20'] = df['close'].ewm(span=20).mean()
            df['ema_50'] = df['close'].ewm(span=50).mean()
            df['rsi'] = self.calculate_rsi(df['close'])
            df['atr'] = self.calculate_atr(df)
            
            return df
        except Exception as e:
            print(f"X Error getting data for {symbol}: {e}")
            return None
    
    def calculate_rsi(self, prices, period=14):
        """Calculate RSI"""
        delta = prices.diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        return rsi
    
    def calculate_atr(self, df, period=14):
        """Calculate ATR"""
        high = df['high']
        low = df['low']
        close = df['close']
        
        tr1 = high - low
        tr2 = abs(high - close.shift())
        tr3 = abs(low - close.shift())
        
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(period).mean()
        return atr
    
    def analyze_symbol(self, symbol):
        """Agent analysis for a symbol"""
        data = self.get_live_data(symbol)
        if data is None or len(data) < 50:
            return {"action": 0, "confidence": 0.0, "reason": "Insufficient data"}
        
        current_price = data['close'].iloc[-1]
        ema_20 = data['ema_20'].iloc[-1]
        ema_50 = data['ema_50'].iloc[-1]
        rsi = data['rsi'].iloc[-1]
        atr = data['atr'].iloc[-1]
        
        # Decision logic
        bullish = 0
        bearish = 0
        
        # Trend
        if current_price > ema_20 > ema_50:
            bullish += 2
        elif current_price < ema_20 < ema_50:
            bearish += 2
        
        # Momentum
        if 40 < rsi < 70:
            bullish += 1
        elif 30 < rsi < 60:
            bearish += 1
        
        # Volatility (using ATR)
        if atr / current_price < 0.002:  # Low volatility
            bullish += 0.5
            bearish += 0.5
        
        # Make decision
        if bullish > bearish and bullish >= 2:
            action = 1
            confidence = min(bullish / 4, 0.8)
            reason = f"Bullish: trend={bullish}, RSI={rsi:.1f}"
        elif bearish > bullish and bearish >= 2:
            action = -1
            confidence = min(bearish / 4, 0.8)
            reason = f"Bearish: trend={bearish}, RSI={rsi:.1f}"
        else:
            action = 0
            confidence = 0.1
            reason = f"Neutral: Bull={bullish}, Bear={bearish}"
        
        return {
            "symbol": symbol,
            "action": action,
            "confidence": confidence,
            "reason": reason,
            "price": current_price,
            "timestamp": datetime.now().isoformat()
        }

このセクションでは、マルチエージェントシステムを構成するライブ取引エージェントの初期化を行います。各エージェントには、それぞれ明確な役割と責任が定義されています。例えば、銘柄固有の分析を担当するGold Agent、より広範な市場状況を分析するContext Agent、ボラティリティや出来高を考慮するLiquidity Agentなどがあります。これらのエージェントには、重み付けされた意思決定要素とインジケーターパラメータが設定されます。これにより、単一のモノリシックな戦略に依存するのではなく、各エージェントが専門的なインテリジェンスを提供し、システム全体の判断に貢献できる構成になります。

次に、データ処理用のメソッドでは、MetaTrader 5を基盤としたリアルタイム市場データパイプラインを構築します。価格データはMetaTrader 5から直接取得され、構造化されたpandas DataFrameへ変換されます。その後、リターン、指数移動平均(EMA)、RSI、ATRなどのテクニカル特徴量を追加してデータを拡張します。専用のヘルパー関数では、RSIおよびATRの計算を手動で実装しています。これにより、インジケーターの動作を完全に把握しながら柔軟に調整でき、さらにすべての前処理ロジックをエージェント間で共有可能な形で一元管理し、再利用できます。

analyze_symbolメソッドは、各銘柄専用エージェントにおける中核的な意思決定エンジンです。このメソッドでは、トレンド構造、モメンタム、ボラティリティを評価し、強気条件と弱気条件をそれぞれスコアリングします。これにより、生の市場データを実行可能な取引シグナルへ変換します。これらのスコアに基づき、エージェントは方向性アクション、信頼度、判断理由を出力します。生成された情報は、後続処理で他のエージェントと共有したり、実行レイヤーへ渡したりすることで、連携的な複数銘柄取引判断に利用できます。

    def analyze_context(self):
        """Context agent analysis"""
        # Get USD strength from major pairs
        eur_data = self.get_live_data("EURUSD", bars=50)
        gbp_data = self.get_live_data("GBPUSD", bars=50)
        jpy_data = self.get_live_data("USDJPY", bars=50)
        
        if not all([eur_data is not None, gbp_data is not None, jpy_data is not None]):
            return {"action": 0, "confidence": 0.0, "reason": "Missing data"}
        
        # Calculate USD strength
        eur_change = (eur_data['close'].iloc[-1] / eur_data['close'].iloc[-20] - 1)
        gbp_change = (gbp_data['close'].iloc[-1] / gbp_data['close'].iloc[-20] - 1)
        jpy_change = (jpy_data['close'].iloc[-1] / jpy_data['close'].iloc[-20] - 1)
        
        # EURUSD and GBPUSD down = USD strong, USDJPY up = USD strong
        usd_strength = (-eur_change - gbp_change + jpy_change) / 3
        
        if usd_strength > 0.01:
            action = -1  # USD strong → bearish for gold
            confidence = min(abs(usd_strength) * 10, 0.7)
            reason = f"USD Strong: {usd_strength:.2%}"
        elif usd_strength < -0.01:
            action = 1  # USD weak → bullish for gold
            confidence = min(abs(usd_strength) * 10, 0.7)
            reason = f"USD Weak: {usd_strength:.2%}"
        else:
            action = 0
            confidence = 0.1
            reason = "USD Neutral"
        
        return {
            "agent": "CONTEXT",
            "action": action,
            "confidence": confidence,
            "reason": reason,
            "usd_strength": usd_strength
        }
    
    def run_decision_pipeline(self, target_symbol="XAUUSD"):
        """Run complete decision pipeline"""
        print(f"\nO- {datetime.now().strftime('%H:%M:%S')} - Analyzing market...")
        
        # Collect agent decisions
        decisions = []
        
        # Symbol agents
        for symbol in self.symbols:
            if symbol == target_symbol:
                decision = self.analyze_symbol(symbol)
                decisions.append(decision)
                print(f"   {symbol}: {decision['action']} ({decision['confidence']:.1%}) - {decision['reason']}")
        
        # Context agent
        context_decision = self.analyze_context()
        decisions.append(context_decision)
        print(f"   CONTEXT: {context_decision['action']} ({context_decision['confidence']:.1%}) - {context_decision['reason']}")
        
        # Calculate consensus
        weighted_sum = 0
        total_weight = 0
        
        weights = {
            "XAUUSD": 1.5,
            "EURUSD": 1.0,
            "GBPUSD": 1.0,
            "USDJPY": 1.0,
            "CONTEXT": 1.3
        }
        
        for decision in decisions:
            symbol = decision.get('symbol', decision.get('agent', 'UNKNOWN'))
            weight = weights.get(symbol, 1.0)
            weighted_sum += decision['action'] * decision['confidence'] * weight
            total_weight += weight * decision['confidence']
        
        # Make final decision
        if total_weight == 0:
            final_action = 0
            confidence = 0.0
            reason = "No consensus"
        else:
            consensus = weighted_sum / total_weight
            
            if consensus > 0.3:
                final_action = 1
                confidence = min(abs(consensus), 0.9)
                reason = f"Bullish consensus: {consensus:.2f}"
            elif consensus < -0.3:
                final_action = -1
                confidence = min(abs(consensus), 0.9)
                reason = f"Bearish consensus: {consensus:.2f}"
            else:
                final_action = 0
                confidence = 0.2
                reason = f"Neutral: {consensus:.2f}"
        
        final_decision = {
            "timestamp": datetime.now().isoformat(),
            "symbol": target_symbol,
            "action": final_action,
            "confidence": confidence,
            "reason": reason,
            "consensus": consensus if total_weight > 0 else 0,
            "agent_decisions": decisions
        }
        
        self.decision_log.append(final_decision)
        
        # Save decision to file for MQL5 to read
        self.save_decision(final_decision)
        
        # Execute trade if confidence is high enough
        if confidence > 0.5:
            self.execute_trade(final_decision)
        
        return final_decision
        
    def write_signal(signal):
        with open("multi_agent_signal.json", "w") as f:
            json.dump(signal, f, indent=2)
    
    def save_decision(self, decision):
        """Save decision to JSON file for MQL5"""
        try:
            with open('multi_agent_signal.json', 'w') as f:
                json.dump(decision, f, indent=2)
            print(f"💾 Signal saved to file")
        except Exception as e:
            print(f"X Error saving signal: {e}")
    
    def execute_trade(self, decision):
        """Execute trade via MT5"""
        symbol = decision['symbol']
        action = decision['action']
        confidence = decision['confidence']
        
        # Check existing positions
        positions = mt5.positions_get(symbol=symbol)
        if positions:
            print(f"!! Existing position found for {symbol}")
            # Check if we should close it
            current_position = positions[0]
            if (action == 1 and current_position.type == 1) or (action == -1 and current_position.type == 0):
                print(f"   Same direction, considering adding to position")
                return
            else:
                print(f"   Opposite direction, closing position first")
                self.close_position(symbol)
        
        # Prepare trade request
        symbol_info = mt5.symbol_info(symbol)
        if symbol_info is None:
            print(f"X Symbol {symbol} not found")
            return
        
        point = symbol_info.point
        price = mt5.symbol_info_tick(symbol).ask if action == 1 else mt5.symbol_info_tick(symbol).bid
        
        # Calculate position size based on risk
        account_info = mt5.account_info()
        balance = account_info.balance
        risk_amount = balance * 0.01 * confidence  # 1% risk adjusted by confidence
        
        # Calculate stop loss based on ATR
        data = self.get_live_data(symbol, bars=50)
        if data is not None and 'atr' in data.columns:
            atr = data['atr'].iloc[-1]
            stop_distance = atr * 1.5
        else:
            stop_distance = price * 0.01  # 1% stop
        
        # Calculate volume
        volume = risk_amount / stop_distance
        volume = round(volume, 2)  # Round to 2 decimal places
        
        # Validate volume
        min_volume = symbol_info.volume_min
        max_volume = symbol_info.volume_max
        volume = max(min_volume, min(volume, max_volume))
        
        # Prepare order
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": volume,
            "type": mt5.ORDER_TYPE_BUY if action == 1 else mt5.ORDER_TYPE_SELL,
            "price": price,
            "sl": price - stop_distance if action == 1 else price + stop_distance,
            "tp": price + stop_distance * 2 if action == 1 else price - stop_distance * 2,
            "deviation": 10,
            "magic": 234000,
            "comment": f"Multi-Agent: {decision['reason'][:30]}",
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
        }
        
        # Send order
        result = mt5.order_send(request)
        
        if result.retcode == mt5.TRADE_RETCODE_DONE:
            print(f":/ Trade executed: {'BUY' if action == 1 else 'SELL'} {volume} {symbol} at {price}")
            print(f"   SL: {result.request.sl:.5f}, TP: {result.request.tp:.5f}")
            print(f"   Order ID: {result.order}")
        else:
            print(f"X Trade failed: {result.comment}")
    
    def close_position(self, symbol):
        """Close all positions for a symbol"""
        positions = mt5.positions_get(symbol=symbol)
        if not positions:
            return
        
        for position in positions:
            tick = mt5.symbol_info_tick(symbol)
            
            request = {
                "action": mt5.TRADE_ACTION_DEAL,
                "symbol": symbol,
                "volume": position.volume,
                "type": mt5.ORDER_TYPE_BUY if position.type == 1 else mt5.ORDER_TYPE_SELL,
                "position": position.ticket,
                "price": tick.ask if position.type == 1 else tick.bid,
                "deviation": 10,
                "magic": 234000,
                "comment": "Close by Multi-Agent",
                "type_time": mt5.ORDER_TIME_GTC,
                "type_filling": mt5.ORDER_FILLING_IOC,
            }
            
            result = mt5.order_send(request)
            if result.retcode == mt5.TRADE_RETCODE_DONE:
                print(f":/ Position {position.ticket} closed")
            else:
                print(f"X Failed to close position {position.ticket}: {result.comment}")
    
    def monitor_positions(self):
        """Monitor and manage open positions"""
        positions = mt5.positions_get()
        
        for position in positions:
            symbol = position.symbol
            current_price = mt5.symbol_info_tick(symbol).bid
            
            # Check stop loss and take profit
            # These are managed by MT5 automatically, but we can log them
            unrealized_pnl = position.profit
            
            print(f" {symbol}: {position.type} {position.volume} | P&L: ${unrealized_pnl:.2f}")
    
    def run_continuously(self, interval_minutes=5):
        """Run the system continuously"""
        print(f"\n<=> Starting continuous monitoring every {interval_minutes} minutes...")
        
        def job():
            try:
                # Run analysis
                decision = self.run_decision_pipeline("XAUUSD")
                
                # Monitor positions
                self.monitor_positions()
                
                # Log status
                print(f" Decision: {decision['action']} ({decision['confidence']:.1%}) - {decision['reason']}")
                
            except Exception as e:
                print(f"X Error in scheduled job: {e}")
        
        # Schedule the job
        schedule.every(interval_minutes).minutes.do(job)
        
        # Run immediately first time
        job()
        
        # Keep running
        try:
            while True:
                schedule.run_pending()
                time.sleep(1)
        except KeyboardInterrupt:
            print("\n!!! Stopping system...")
            mt5.shutdown()

このコード部分では、コンテキスト分析レイヤーを導入します。このレイヤーは、特定の銘柄を取引するエージェントではなく、より上位の文脈判断を担うエージェントとして機能します。analyze_contextメソッドでは、EURUSD、EURUSD、GBPUSD、USDJPYといった主要通貨ペアの動きを観測することで、全体的な米ドルの強弱を評価します。これらの市場における直近の変化率を計測し、それらを単一のドル強度指標へ統合することで、システムはXAUUSD（ゴールド）の判断に影響を与えるマクロレベルのバイアスを算出します。これにより、戦略は個別銘柄の値動きだけに依存するのではなく、市場間の相互関係を考慮した分析が可能になります。

次に、run_decision_pipelineメソッドでは、マルチエージェントワークフロー全体を統括します。このメソッドは、銘柄専用エージェント(XAUUSD)とコンテキストエージェントから判断結果を収集し、それぞれの出力を記録します。その後、重み付きコンセンサスモデルを使用して、これらの判断を統合します。各エージェントは、信頼度と事前に定義された重要度ウェイトの両方に基づいて最終判断へ貢献します。これにより、より強力または信頼性の高いシグナルが、最終的な意思決定により大きな影響を与えることが保証されます。このコンセンサスメカニズムにより、複数の独立した判断を、信頼度スコア付きの単一で一貫した取引アクションへ変換できます。

最終的な判断が形成されると、システムはその結果を透明性と永続性のために記録します。判断結果は内部ログへ追加されると同時に、MQL5のEAが外部から取得できるJSONファイルへ書き込まれます。信頼度が事前に設定された閾値を超えた場合、システムは取引実行へ進みます。これにより、分析インテリジェンスと実際の市場での取引アクションを橋渡しします。この設計では、意思決定の生成、保存、実行を明確に分離しながら、それぞれの処理を緊密に同期させています。

残りのメソッドでは、MetaTrader 5を介したライブ取引実行とポジション管理を直接処理します。取引数量は、口座残高、信頼度に応じて調整されたリスク量、ATRベースのストップ幅を基準として動的に算出され、一貫したリスク管理を実現します。また、システムは反対方向のポジション決済、保有ポジションの監視、一定間隔での継続実行にも対応しています。これらのコンポーネントを組み合わせることで、市場コンテキストを分析し、マルチエージェントによる合意形成を行い、規律ある取引を実行しながら、リアルタイムでパフォーマンスを継続的に監視する完全な適応型トレーディングループが構築されます。

# ================================================
# FLASK API FOR EXTERNAL CONTROL 
# ================================================

from flask import Flask, request, jsonify
import threading
import time
import json
import MetaTrader5 as mt5

app = Flask(__name__)

trading_system = None
system_thread = None
system_running = False
system_lock = threading.Lock()

@app.route('/')
def home():
    return jsonify({
        "status": "Multi-Agent Trading System API",
        "version": "1.1",
        "running": system_running
    })

@app.route('/start', methods=['POST'])
def start_system():
    global trading_system, system_thread, system_running
    global account, password, serve
    
    with system_lock:
        if system_running:
            return jsonify({"error": "System already running"}), 400

        data = request.json or {} 
        account = data.get('account') 
        password = data.get('password')
        server = data.get('server', 'SpaceMarkets-Live')

        if not all([account, password]):
            return jsonify({"error": "Missing account or password"}), 400

        # Create system
        trading_system = LiveMultiAgentSystem(account, password, server)
        system_running = True

        # Background execution loop
        def runner():
            while system_running:
                try:
                    trading_system.run_once()
                    time.sleep(5)
                except Exception as e:
                    print("X System error:", e)
                    time.sleep(5)

        system_thread = threading.Thread(target=runner, daemon=True)
        system_thread.start()

        return jsonify({
            "status": "System started",
            "account": account,
            "server": server
        })

@app.route('/stop', methods=['POST'])
def stop_system():
    global system_running

    with system_lock:
        if not system_running:
            return jsonify({"error": "System not running"}), 400

        system_running = False

    return jsonify({"status": "System stopping"})

@app.route('/status', methods=['GET'])
def get_status():
    if not system_running or trading_system is None:
        return jsonify({"status": "stopped"})

    last_decision = (
        trading_system.decision_log[-1]
        if trading_system.decision_log
        else None
    )

    return jsonify({
        "status": "running",
        "last_decision": last_decision,
        "total_decisions": len(trading_system.decision_log)
    })

このコードセクションでは、Flaskアプリケーション、共有状態変数、同期メカニズムを初期化し、取引システムの安全かつ制御された実行環境を構築します。システムの稼働状態を追跡するためにグローバルフラグを導入し、取引エンジンの開始・停止処理時に競合状態（レースコンディション）が発生しないようロック機構を使用します。「/」ルートエンドポイントは軽量なヘルスチェックとして機能し、現在のAPIバージョンとマルチエージェントシステムが稼働中であるかどうかを返します。これにより、デバッグ作業や外部監視システムからの状態確認に利用できます。

残りのFlaskルートでは、取引システムのライフサイクルを外部から制御するための機能を提供します。/startエンドポイントでは、リクエストで提供された認証情報を使用してマルチエージェントエンジンを安全に初期化し、バックグラウンドスレッド上で起動することで継続的な動作を可能にします。一方、/stopエンドポイントでは、API自体を終了させることなく、取引処理を安全に停止します。/statusエンドポイントでは、最新の判断結果や実行された判断数など、リアルタイムのシステム状態を外部へ公開します。これにより、MQL5やその他のクライアントは、Pythonベースのマルチエージェントロジックと同期しながら、非ブロッキングかつ制御された形でシステム状態を取得できます。

MQL5での統合実装

//+------------------------------------------------------------------+
//|                                                 Multi-Agents.mq5 |
//|                        GIT under Copyright 2025, MetaQuotes Ltd. |
//|                     https://www.mql5.com/ja/users/johnhlomohang/ |
//+------------------------------------------------------------------+
#property copyright "GIT under Copyright 2025, MetaQuotes Ltd."
#property link      "https://www.mql5.com/ja/users/johnhlomohang/"
#property version   "1.00"

#include <Trade/Trade.mqh>
#include "Json.mqh"  // Include our JSON parser

//--- Input parameters
input string   PythonServer = "http://127.0.0.1:5000";  // Python server address
input double   RiskPercent  = 1.0;                      // Risk percentage
input int      SignalCheckInterval = 30;                // Check signal every N seconds
input bool     EnableTrading = true;                    // Enable trading
input string   CommentText   = "Multi-Agent";           // Order comment

//--- Global variables
CTrade trade;
datetime lastSignalCheck = 0;
string currentSignal = "";
double lastSignalPrice = 0;
int magicNumber = 234000;

//+------------------------------------------------------------------+
//| Expert initialization function                                   |
//+------------------------------------------------------------------+
int OnInit()
{
   // Set magic number for order identification
   trade.SetExpertMagicNumber(magicNumber);
   
   // Set asynchronous mode
   trade.SetAsyncMode(true);
   
   // Print initialization message
   Print("Multi-Agent EA Initialized - Simple JSON Version");
   Print("Python Server: ", PythonServer);
   Print("Risk: ", RiskPercent, "%");
   Print("Magic Number: ", magicNumber);
   
   return(INIT_SUCCEEDED);
}

//+------------------------------------------------------------------+
//| Expert deinitialization function                                 |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
   Print("Multi-Agent EA Deinitialized");
}

//+------------------------------------------------------------------+
//| Expert tick function                                             |
//+------------------------------------------------------------------+
void OnTick()
{
   // Check if it's time to check for new signals
   if(TimeCurrent() - lastSignalCheck >= SignalCheckInterval)
   {
      CheckSignal();
      lastSignalCheck = TimeCurrent();
   }
   
   // Manage existing positions
   ManagePositions();
}

//+------------------------------------------------------------------+
//| Check signal from Python server                                  |
//+------------------------------------------------------------------+
void CheckSignal()
{  
   string url = PythonServer + "/signal";
   
   string payload = "{}";  // empty body is fine
   char data[];
   StringToCharArray(payload, data);
   
   char result[];
   string headers = "Content-Type: application/json\r\n";
   string result_headers;
   
   int res = WebRequest(
      "POST",
      url,
      headers,
      5000,
      data,
      result,
      result_headers
   );
   
   if(res != 200)
   {
      Print("HTTP request failed: ", res);
      return;
   }
   Print("LastError=", GetLastError());
   
   if(res == 200) // HTTP OK
   {
      string jsonStr = CharArrayToString(result);
      Print("Received JSON: ", jsonStr);
      
      // Parse using our simple JSON parser
      string symbol = CJson::ParseString(jsonStr, "symbol", "");
      int action = CJson::ParseInteger(jsonStr, "action", 0);
      double confidence = CJson::ParseNumber(jsonStr, "confidence", 0);
      string reason = CJson::ParseString(jsonStr, "reason", "");
      double price = CJson::ParseNumber(jsonStr, "price", 0);
      
      Print("Parsed Signal: Symbol=", symbol, " Action=", action, 
            " Confidence=", confidence, " Reason=", reason);
      
      // Only trade if confidence is high enough
      if(confidence >= 0.5 && EnableTrading)
      {
         // Check if we should trade this symbol
         if(symbol == _Symbol || symbol == "")
         {
            ExecuteSignal(action, confidence, reason, price);
         }
      }
      
      // Update current signal display
      currentSignal = reason;
   }
   else if(res == -1)
   {
      Print("WebRequest failed. Error: ", GetLastError());
      Print("Make sure to add URL to allowed list: ", url);
      
      // To add URL to allowed list in MT5:
      // 1. Go to Tools -> Options -> Expert Advisors
      // 2. Click "Add" under "Allowed URLs"
      // 3. Add: http://localhost:5000
   }
   else
   {
      Print("HTTP request failed: ", res);
   }
}

ここでは、取引ターミナルと外部のPythonベースのマルチエージェントシステムを接続するブリッジとして機能するMetaTrader 5用EAをMQL5で定義します。まず、リスク設定、シグナル取得間隔、注文追跡用のマジックナンバーなどの取引パラメータを初期化し、非同期実行用にCTradeオブジェクトを設定します。各ティックの処理では、EAが一定間隔でPythonサーバーへHTTPリクエストを送信し、取引シグナルを取得します。返されたJSONペイロードを解析し、銘柄、アクション、信頼度、価格などのフィールドを抽出します。その後、基本的なフィルタリングロジックを適用し、十分な信頼度があり、かつ取引が有効化されている場合のみ注文候補として扱います。

残りのロジックでは、実行時の動作とシステムの堅牢性を管理します。EAは、新しいシグナルを待機している間も既存ポジションを継続的に管理し、透明性を確保するために重要なイベントをログへ記録します。また、Webリクエストの失敗や権限設定の不備などに対する安全対策も含まれています。この設計では、シグナル生成処理をPythonサーバー側へ移し、MetaTrader 5側の実行ロジックを軽量に保つことで、意思決定と取引実行を明確に分離しています。これにより、マルチエージェントインテリジェンスの進化を取引ターミナルから独立して進めることが可能になります。

//+------------------------------------------------------------------+
//| Execute trading signal                                           |
//+------------------------------------------------------------------+
void ExecuteSignal(int action, double confidence, string reason, double signalPrice = 0)
{
   // Get current price
   double ask = SymbolInfoDouble(_Symbol, SYMBOL_ASK);
   double bid = SymbolInfoDouble(_Symbol, SYMBOL_BID);
   double price = (action == 1) ? ask : bid;
   
   // Use signal price if provided and valid
   if(signalPrice > 0 && MathAbs(signalPrice - price) / price < 0.05) // Within 5%
   {
      price = signalPrice;
   }
   
   // Check if price has changed significantly
   if(MathAbs(price - lastSignalPrice) / price < 0.001) // Less than 0.1% change
   {
      Print("Price hasn't changed enough, skipping trade");
      return;
   }
   
   lastSignalPrice = price;
   
   // Close opposite positions first
   if(action == 1) // Buy signal
   {
      CloseSellPositions();
   }
   else if(action == -1) // Sell signal
   {
      CloseBuyPositions();
   }
   
   // Calculate position size
   double volume = CalculateVolume(confidence);
   
   if(volume <= 0)
   {
      Print("Volume too small, skipping trade");
      return;
   }
   
   // Calculate stop loss and take profit based on ATR
   double atr = CalculateATR(14);
   double stopLoss = 0;
   double takeProfit = 0;
   
   if(action == 1) // Buy
   {
      stopLoss = price - (atr * 1.5);
      takeProfit = price + (atr * 3.0);
   }
   else // Sell
   {
      stopLoss = price + (atr * 1.5);
      takeProfit = price - (atr * 3.0);
   }
   
   // Normalize SL/TP to tick size
   stopLoss = NormalizePrice(stopLoss);
   takeProfit = NormalizePrice(takeProfit);
   
   // Place order
   if(action == 1)
   {
      if(trade.Buy(volume, _Symbol, price, stopLoss, takeProfit, reason))
      {
         Print("BUY order placed: ", DoubleToString(volume, 2), " ", _Symbol, 
               " at ", DoubleToString(price, 5));
         Print("SL: ", DoubleToString(stopLoss, 5), 
               " TP: ", DoubleToString(takeProfit, 5));
      }
      else
      {
         Print("Failed to place BUY order: ", trade.ResultRetcodeDescription());
      }
   }
   else if(action == -1)
   {
      if(trade.Sell(volume, _Symbol, price, stopLoss, takeProfit, reason))
      {
         Print("SELL order placed: ", DoubleToString(volume, 2), " ", _Symbol, 
               " at ", DoubleToString(price, 5));
         Print("SL: ", DoubleToString(stopLoss, 5), 
               " TP: ", DoubleToString(takeProfit, 5));
      }
      else
      {
         Print("Failed to place SELL order: ", trade.ResultRetcodeDescription());
      }
   }
}

//+------------------------------------------------------------------+
//| Calculate ATR                                                    |
//+------------------------------------------------------------------+
double CalculateATR(int period)
{
   double atr = 0;
   
   // Try to get ATR from indicator
   int atrHandle = iATR(_Symbol, PERIOD_M5, period);
   if(atrHandle != INVALID_HANDLE)
   {
      double atrArray[];
      if(CopyBuffer(atrHandle, 0, 0, 1, atrArray) > 0)
      {
         atr = atrArray[0];
      }
      IndicatorRelease(atrHandle);
   }
   
   if(atr <= 0)
   {
      // Fallback: use percentage of price
      atr = SymbolInfoDouble(_Symbol, SYMBOL_BID) * 0.002; // 0.2%
   }
   
   return atr;
}

//+------------------------------------------------------------------+
//| Normalize price to tick size                                     |
//+------------------------------------------------------------------+
double NormalizePrice(double price)
{
   double tickSize = SymbolInfoDouble(_Symbol, SYMBOL_TRADE_TICK_SIZE);
   if(tickSize > 0)
   {
      price = MathRound(price / tickSize) * tickSize;
   }
   return NormalizeDouble(price, (int)SymbolInfoInteger(_Symbol, SYMBOL_DIGITS));
}

ここでは、方向性シグナルを制御されたリスク管理付きの成行注文へ変換する、完全な取引実行パイプラインを実装します。ExecuteSignal関数では、まず現在のBidまたはAsk価格を使用して適切な約定価格を決定します。また、有効な外部シグナル価格が提供されている場合は、その価格で上書きすることも可能です。その後、前回のシグナルから価格が十分に変化していることを確認し、不要な重複取引を防止します。さらに、反対方向のポジションが存在する場合は先に決済することで、方向性の一貫性を維持します。その後、シグナルの信頼度に基づいてポジションサイズを動的に計算し、ATRを使用して現在の市場ボラティリティを反映した適応型のストップロス(SL)およびテイクプロフィット(TP)レベルを設定します。

注文を出す前に、すべての価格レベルは銘柄のティックサイズおよび小数点精度に合わせて正規化され、ブローカー側の仕様に適合するよう処理されます。その後、買いまたは売り注文が送信され、透明性を確保するために実行内容が完全にログへ記録されます。補助関数では、ATRインジケーターを利用したボラティリティ測定と、安全なフォールバック処理を実装しています。また、正確な価格正規化処理も提供します。これらの仕組みにより、取引実行ロジックは堅牢性、適応性を備え、一般的な取引上の問題やブローカー側のエラーに対して耐性を持つ設計になっています。

//+------------------------------------------------------------------+
//|                                                         Json.mqh |
//|                        GIT under Copyright 2025, MetaQuotes Ltd. |
//|                     https://www.mql5.com/ja/users/johnhlomohang/ |
//+------------------------------------------------------------------+
#property copyright "GIT under Copyright 2025, MetaQuotes Ltd."
#property link      "https://www.mql5.com/ja/users/johnhlomohang/"

#ifndef JSON_MQH
#define JSON_MQH

class CJson
{
private:
   string m_json;
   
   string ExtractValue(string key)
   {
      int keyPos = StringFind(m_json, "\"" + key + "\"");
      if(keyPos == -1) return "";
      
      int colonPos = StringFind(m_json, ":", keyPos);
      if(colonPos == -1) return "";
      
      // Find the start of the value
      int valueStart = colonPos + 1;
      while(valueStart < StringLen(m_json) && 
            (m_json[valueStart] == ' ' || m_json[valueStart] == '\t' || 
             m_json[valueStart] == '\n' || m_json[valueStart] == '\r'))
      {
         valueStart++;
      }
      
      if(valueStart >= StringLen(m_json)) return "";
      
      char firstChar = m_json[valueStart];
      
      // String value
      if(firstChar == '\"')
      {
         int endQuote = StringFind(m_json, "\"", valueStart + 1);
         if(endQuote == -1) return "";
         return StringSubstr(m_json, valueStart + 1, endQuote - valueStart - 1);
      }
      
      // Number or boolean value
      int valueEnd = valueStart;
      while(valueEnd < StringLen(m_json) && 
            ((m_json[valueEnd] >= '0' && m_json[valueEnd] <= '9') || 
             m_json[valueEnd] == '-' || m_json[valueEnd] == '.' || 
             m_json[valueEnd] == 'e' || m_json[valueEnd] == 'E' ||
             m_json[valueEnd] == 't' || m_json[valueEnd] == 'r' ||
             m_json[valueEnd] == 'u' || m_json[valueEnd] == 'e' ||
             m_json[valueEnd] == 'f' || m_json[valueEnd] == 'a' ||
             m_json[valueEnd] == 'l' || m_json[valueEnd] == 's' ||
             m_json[valueEnd] == 'n' || m_json[valueEnd] == 'u' ||
             m_json[valueEnd] == 'l'))
      {
         valueEnd++;
      }
      
      return StringSubstr(m_json, valueStart, valueEnd - valueStart);
   }
   
public:
   void SetJson(string json)
   {
      m_json = json;
   }
   
   string GetString(string key, string defaultValue = "")
   {
      string value = ExtractValue(key);
      if(value == "") return defaultValue;
      
      // Check if it's actually a string (starts with quote)
      if(StringGetCharacter(value, 0) == '\"')
      {
         return StringSubstr(value, 1, StringLen(value) - 2);
      }
      
      return value;
   }
   
   double GetNumber(string key, double defaultValue = 0)
   {
      string value = ExtractValue(key);
      if(value == "") return defaultValue;
      
      // Check for true/false
      if(value == "true") return 1;
      if(value == "false") return 0;
      
      return StringToDouble(value);
   }
   
   int GetInteger(string key, int defaultValue = 0)
   {
      string value = ExtractValue(key);
      if(value == "") return defaultValue;
      
      // Check for true/false
      if(value == "true") return 1;
      if(value == "false") return 0;
      
      return (int)StringToInteger(value);
   }
   
   bool GetBool(string key, bool defaultValue = false)
   {
      string value = ExtractValue(key);
      if(value == "") return defaultValue;
      
      if(value == "true") return true;
      if(value == "false") return false;
      if(value == "1") return true;
      if(value == "0") return false;
      
      return defaultValue;
   }
   
   // Static helper methods for quick parsing
   static string ParseString(string json, string key, string defaultValue = "")
   {
      CJson parser;
      parser.SetJson(json);
      return parser.GetString(key, defaultValue);
   }
   
   static double ParseNumber(string json, string key, double defaultValue = 0)
   {
      CJson parser;
      parser.SetJson(json);
      return parser.GetNumber(key, defaultValue);
   }
   
   static int ParseInteger(string json, string key, int defaultValue = 0)
   {
      CJson parser;
      parser.SetJson(json);
      return parser.GetInteger(key, defaultValue);
   }
   
   static bool ParseBool(string json, string key, bool defaultValue = false)
   {
      CJson parser;
      parser.SetJson(json);
      return parser.GetBool(key, defaultValue);
   }
};

#endif

Json.mqhファイルでは、外部ライブラリに依存せず、単純なキーと値のペアを安全に取得できる、軽量で自己完結型のMQL5用JSON解析ユーティリティを定義します。CJsonクラスは、生のJSON文字列を保持し、制御された文字列スキャン方式を使用してキーの検索、空白文字のスキップ、文字列・数値・ブール値の正確な解釈を行います。このクラスは、型付きアクセサー（GetString、GetNumber、GetInteger、GetBool）を提供します。これらのメソッドは、キーが存在しない場合やデータ形式が不正な場合でも、デフォルト値へ安全にフォールバックするため、ライブ取引環境における堅牢性を確保できます。さらに、静的ヘルパーメソッドを利用することで、JSONペイロードから1行で直接解析できるようになり、APIレスポンス、設定メッセージ、プロセス間通信など、パフォーマンス・安全性・依存関係の最小化が重要となる用途に適したクラスとなっています。

ライブデモ

以下では、Flaskサーバーが正常に初期化され、ポート5000のlocalhost上で稼働しています。また、Jupyter LabからMetaTrader 5へシグナルを正常に送信しています。

結論

本記事では、独立したエージェントが共有Pythonサービスを介して通信し、単独で動作するのではなく、複数の銘柄間で連携するマルチエージェント取引環境を構築しました。各エージェントは構造化されたシグナルを生成し、それらはJSON形式で送信されます。その後、MQL5のEA内部で安全に解析され、信頼度しきい値、リスク管理制御、銘柄検証を通じて評価されます。このアーキテクチャにより、複数銘柄の状況認識、連携的な意思決定、フィードバックに基づく適応が可能になります。結果として、複数の金融商品の市場インテリジェンスを、MetaTrader 5内の単一で一貫した実行レイヤーへ統合できます。

結論として、このアプローチは、視野の狭い判断を減らし、より広範な市場コンテキストや相関資産間の集合的な動きを戦略へ取り込むことで、取引能力を大きく向上させる可能性があります。インテリジェンス処理を連携するエージェントへ移譲する一方で、実行、リスク管理、検証処理をEA側で厳密に制御することで、トレーダーはより柔軟で拡張性が高く、改善しやすいシステムを構築できます。その結果、ロジックと実行処理の分離がより明確になり、実験や改善のスピードが向上するとともに、市場環境の変化へより自然に適応できる、効率的なスマートな取引ワークフローを実現できます。

ファイル名	ファイルの説明
Multi Agents.mq5	Pythonサーバーへ接続し、JSON形式でマルチエージェント取引シグナルを受信し、リスク管理を行いながらMetaTrader 5上で取引を実行するメインのEA
Json.mq5	EAがPythonで生成されたシグナルレスポンスから文字列、数値、ブール値を安全に取得するために使用する軽量なカスタムJSONパーサー
MultiAgentsLab.ipynb	マルチエージェント環境を構築・実行し、複数銘柄の取引シグナルを生成して、ローカルAPIを通じてMetaTraderへ提供するPython Notebook