ClaudeとMT4をMCPサーバで連携させ、一目均衡表を使った逆指値トラップリピートEAを作りたい。
ClaudeとMT4をMCPサーバで連携させて一目均衡表を使った逆指値トラップリピートEAを作りたい。ソースコードを書いて。
一目均衡表を使った逆指値トラップリピートEAのMCPサーバー連携ソースコードを作成します。以下のように実装できます。
1. MCPサーバー実装 (Python)
```python
# mcp_ichimoku_server.py
import asyncio
import json
import logging
from typing import Dict, List, Optional
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class IchimokuTrapEA:
def __init__(self):
self.mt5_connected = False
self.symbols = ["USDJPY", "EURUSD", "GBPUSD"]
self.timeframe = mt5.TIMEFRAME_H1
self.period_tenkan = 9
self.period_kijun = 26
self.period_senkou = 52
async def connect_mt5(self) -> bool:
"""MT5に接続"""
try:
if not mt5.initialize():
logger.error("MT5 initialize failed")
return False
self.mt5_connected = True
logger.info("MT5 connected successfully")
return True
except Exception as e:
logger.error(f"MT5 connection error: {e}")
return False
def calculate_ichimoku(self, symbol: str, count: int = 100) -> Dict:
"""一目均衡表を計算"""
try:
rates = mt5.copy_rates_from_pos(symbol, self.timeframe, 0, count + self.period_senkou)
if rates is None or len(rates) < self.period_senkou:
return {}
df = pd.DataFrame(rates)
df['time'] = pd.to_datetime(df['time'], unit='s')
# 転換線 (Tenkan-sen)
high_tenkan = df['high'].rolling(window=self.period_tenkan).max()
low_tenkan = df['low'].rolling(window=self.period_tenkan).min()
df['tenkan_sen'] = (high_tenkan + low_tenkan) / 2
# 基準線 (Kijun-sen)
high_kijun = df['high'].rolling(window=self.period_kijun).max()
low_kijun = df['low'].rolling(window=self.period_kijun).min()
df['kijun_sen'] = (high_kijun + low_kijun) / 2
# 先行スパンA (Senkou Span A)
df['senkou_span_a'] = ((df['tenkan_sen'] + df['kijun_sen']) / 2).shift(self.period_kijun)
# 先行スパンB (Senkou Span B)
high_senkou = df['high'].rolling(window=self.period_senkou).max()
low_senkou = df['low'].rolling(window=self.period_senkou).min()
df['senkou_span_b'] = ((high_senkou + low_senkou) / 2).shift(self.period_kijun)
# 遅行スパン (Chikou Span)
df['chikou_span'] = df['close'].shift(-self.period_kijun)
latest = df.iloc[-1]
previous = df.iloc[-2]
return {
'symbol': symbol,
'tenkan_sen': float(latest['tenkan_sen']),
'kijun_sen': float(latest['kijun_sen']),
'senkou_span_a': float(latest['senkou_span_a']),
'senkou_span_b': float(latest['senkou_span_b']),
'chikou_span': float(latest['chikou_span']),
'price': float(latest['close']),
'cloud_top': max(latest['senkou_span_a'], latest['senkou_span_b']),
'cloud_bottom': min(latest['senkou_span_a'], latest['senkou_span_b']),
'trend': self._determine_trend(latest, previous),
'signal': self._generate_signal(latest, previous)
}
except Exception as e:
logger.error(f"Ichimoku calculation error for {symbol}: {e}")
return {}
def _determine_trend(self, latest, previous) -> str:
"""トレンド判定"""
if latest['close'] > latest['kijun_sen'] and latest['close'] > latest['senkou_span_a']:
return "bullish"
elif latest['close'] < latest['kijun_sen'] and latest['close'] < latest['senkou_span_a']:
return "bearish"
else:
return "neutral"
def _generate_signal(self, latest, previous) -> str:
"""シグナル生成"""
# 転換線と基準線のクロス
if (previous['tenkan_sen'] <= previous['kijun_sen'] and
latest['tenkan_sen'] > latest['kijun_sen']):
return "BUY"
elif (previous['tenkan_sen'] >= previous['kijun_sen'] and
latest['tenkan_sen'] < latest['kijun_sen']):
return "SELL"
# 価格と雲の関係
if latest['close'] > latest['cloud_top'] and previous['close'] <= previous['cloud_top']:
return "BUY"
elif latest['close'] < latest['cloud_bottom'] and previous['close'] >= previous['cloud_bottom']:
return "SELL"
return "HOLD"
async def calculate_trap_levels(self, symbol: str) -> Dict:
"""トラップレベルを計算"""
ichimoku_data = self.calculate_ichimoku(symbol)
if not ichimoku_data:
return {}
current_price = ichimoku_data['price']
# 逆指値レベルを計算
if ichimoku_data['signal'] == "BUY":
stop_loss = ichimoku_data['cloud_bottom']
take_profit = current_price + (current_price - stop_loss) * 1.5
trigger_price = current_price - (current_price - stop_loss) * 0.1
elif ichimoku_data['signal'] == "SELL":
stop_loss = ichimoku_data['cloud_top']
take_profit = current_price - (stop_loss - current_price) * 1.5
trigger_price = current_price + (stop_loss - current_price) * 0.1
else:
return {}
return {
'symbol': symbol,
'signal': ichimoku_data['signal'],
'current_price': current_price,
'trigger_price': round(trigger_price, 5),
'stop_loss': round(stop_loss, 5),
'take_profit': round(take_profit, 5),
'lot_size': 0.1
}
async def get_all_signals(self) -> List[Dict]:
"""全シンボルのシグナルを取得"""
if not self.mt5_connected:
await self.connect_mt5()
signals = []
for symbol in self.symbols:
try:
signal = await self.calculate_trap_levels(symbol)
if signal:
signals.append(signal)
except Exception as e:
logger.error(f"Error processing {symbol}: {e}")
return signals
# MCPサーバー設定
ea = IchimokuTrapEA()
async def handle_request(request: Dict) -> Dict:
"""MCPリクエストを処理"""
try:
method = request.get("method")
params = request.get("params", {})
if method == "get_ichimoku_signals":
signals = await ea.get_all_signals()
return {
"jsonrpc": "2.0",
"result": signals,
"id": request.get("id")
}
elif method == "calculate_trap_levels":
symbol = params.get("symbol", "USDJPY")
levels = await ea.calculate_trap_levels(symbol)
return {
"jsonrpc": "2.0",
"result": levels,
"id": request.get("id")
}
elif method == "health_check":
return {
"jsonrpc": "2.0",
"result": {"status": "healthy", "mt5_connected": ea.mt5_connected},
"id": request.get("id")
}
else:
return {
"jsonrpc": "2.0",
"error": {"code": -32601, "message": "Method not found"},
"id": request.get("id")
}
except Exception as e:
return {
"jsonrpc": "2.0",
"error": {"code": -32603, "message": f"Internal error: {str(e)}"},
"id": request.get("id")
}
async def main():
"""MCPサーバー起動"""
await ea.connect_mt5()
# 簡易HTTPサーバー（実際の実装では適切なMCPトランスポートを使用）
from aiohttp import web
import socketio
sio = socketio.AsyncServer(async_mode='aiohttp')
app = web.Application()
sio.attach(app)
@sio.event
async def connect(sid, environ):
logger.info(f"Client connected: {sid}")
@sio.event
async def disconnect(sid):
logger.info(f"Client disconnected: {sid}")
@sio.event
async def mcp_request(sid, data):
response = await handle_request(data)
return response
async def health_check(request):
return web.json_response({"status": "Ichimoku Trap EA MCP Server is running"})
app.router.add_get('/health', health_check)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, 'localhost', 8080)
await site.start()
logger.info("MCP Server started on http://localhost:8080")
# 定期的なシグナルチェック
async def periodic_signal_check():
while True:
try:
signals = await ea.get_all_signals()
active_signals = [s for s in signals if s.get('signal') in ['BUY', 'SELL']]
if active_signals:
logger.info(f"Active signals: {active_signals}")
await sio.emit('trading_signals', active_signals)
except Exception as e:
logger.error(f"Periodic check error: {e}")
await asyncio.sleep(60) # 1分ごとにチェック
asyncio.create_task(periodic_signal_check())
# サーバーを実行し続ける
await asyncio.Future()
if __name__ == "__main__":
asyncio.run(main())
```
2. MT4 EA実装 (MQL4)
```mql4
// IchimokuTrapEA.mq4
#property copyright "Ichimoku Trap EA"
#property version "1.00"
#property strict
// 外部パラメータ
input string MCP_SERVER_URL = "http://localhost:8080";
input double LotSize = 0.1;
input int MagicNumber = 12345;
input int Slippage = 3;
input int MaxOrders = 5;
// グローバル変数
int lastSignalCheck = 0;
bool tradingEnabled = true;
//+------------------------------------------------------------------+
//| Expert initialization function |
//+------------------------------------------------------------------+
int OnInit()
{
Print("Ichimoku Trap EA Started");
EventSetTimer(60); // 60秒ごとにチェック
return(INIT_SUCCEEDED);
}
//+------------------------------------------------------------------+
//| Expert deinitialization function |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
EventKillTimer();
Print("Ichimoku Trap EA Stopped");
}
//+------------------------------------------------------------------+
//| Expert tick function |
//+------------------------------------------------------------------+
void OnTick()
{
// 新しいバーでのみ処理
if(IsNewBar())
{
CheckForSignals();
ManageOrders();
}
}
//+------------------------------------------------------------------+
//| Timer function |
//+------------------------------------------------------------------+
void OnTimer()
{
CheckForSignals();
}
//+------------------------------------------------------------------+
//| MCPサーバーからシグナルを取得 |
//+------------------------------------------------------------------+
void CheckForSignals()
{
if(TimeCurrent() - lastSignalCheck < 60) // 60秒間隔でチェック
return;
string response = GetMCPSignal();
if(response == "")
return;
// JSONレスポンスを解析（簡易実装）
int signalPos = StringFind(response, "\"signal\":\"");
if(signalPos == -1)
return;
string signal = StringSubstr(response, signalPos + 10, 4);
if(StringSubstr(signal, 0, 4) == "BUY" || StringSubstr(signal, 0, 4) == "SELL")
{
ProcessSignal(response);
}
lastSignalCheck = TimeCurrent();
}
//+------------------------------------------------------------------+
//| MCPサーバーからシグナル取得 |
//+------------------------------------------------------------------+
string GetMCPSignal()
{
string headers = "Content-Type: application/json\r\n";
char data[], result[];
string sendData = "{\"jsonrpc\":\"2.0\",\"method\":\"calculate_trap_levels\",\"params\":{\"symbol\":\"" + Symbol() + "\"},\"id\":1}";
StringToCharArray(sendData, data);
int res = WebRequest("POST", MCP_SERVER_URL, headers, 5000, data, result, headers);
if(res == 200)
{
return CharArrayToString(result);
}
else
{
Print("MCP Server Error: ", res);
return "";
}
}
//+------------------------------------------------------------------+
//| シグナルを処理 |
//+------------------------------------------------------------------+
void ProcessSignal(string jsonResponse)
{
// JSON解析（簡易実装）
double triggerPrice = ExtractPrice(jsonResponse, "trigger_price");
double stopLoss = ExtractPrice(jsonResponse, "stop_loss");
double takeProfit = ExtractPrice(jsonResponse, "take_profit");
string signal = ExtractSignal(jsonResponse);
if(triggerPrice == 0 || stopLoss == 0 || takeProfit == 0)
return;
// 逆指値注文の発注
if(signal == "BUY" && CountOrders(OP_BUYSTOP) == 0)
{
PlaceBuyStopOrder(triggerPrice, stopLoss, takeProfit);
}
else if(signal == "SELL" && CountOrders(OP_SELLSTOP) == 0)
{
PlaceSellStopOrder(triggerPrice, stopLoss, takeProfit);
}
}
//+------------------------------------------------------------------+
//| 買い逆指値注文 |
//+------------------------------------------------------------------+
void PlaceBuyStopOrder(double triggerPrice, double stopLoss, double takeProfit)
{
double sl = stopLoss;
double tp = takeProfit;
if(triggerPrice > Ask)
{
int ticket = OrderSend(Symbol(), OP_BUYSTOP, LotSize, triggerPrice, Slippage, sl, tp,
"Ichimoku Trap BUY", MagicNumber, 0, clrGreen);
if(ticket > 0)
Print("Buy Stop Order Placed: ", triggerPrice);
}
}
//+------------------------------------------------------------------+
//| 売り逆指値注文 |
//+------------------------------------------------------------------+
void PlaceSellStopOrder(double triggerPrice, double stopLoss, double takeProfit)
{
double sl = stopLoss;
double tp = takeProfit;
if(triggerPrice < Bid)
{
int ticket = OrderSend(Symbol(), OP_SELLSTOP, LotSize, triggerPrice, Slippage, sl, tp,
"Ichimoku Trap SELL", MagicNumber, 0, clrRed);
if(ticket > 0)
Print("Sell Stop Order Placed: ", triggerPrice);
}
}
//+------------------------------------------------------------------+
//| 注文管理 |
//+------------------------------------------------------------------+
void ManageOrders()
{
// トリガー価格に達した注文の監視
for(int i = OrdersTotal() - 1; i >= 0; i--)
{
if(OrderSelect(i, SELECT_BY_POS, MODE_TRADES))
{
if(OrderMagicNumber() == MagicNumber && OrderSymbol() == Symbol())
{
// 約定した注文の管理
if(OrderType() == OP_BUY || OrderType() == OP_SELL)
{
// 利益確定や損切りの管理ロジックをここに追加
}
}
}
}
}
//+------------------------------------------------------------------+
//| ユーティリティ関数 |
//+------------------------------------------------------------------+
bool IsNewBar()
{
static datetime lastBar = 0;
if(Time[0] != lastBar)
{
lastBar = Time[0];
return true;
}
return false;
}
int CountOrders(int type)
{
int count = 0;
for(int i = 0; i < OrdersTotal(); i++)
{
if(OrderSelect(i, SELECT_BY_POS, MODE_TRADES))
{
if(OrderMagicNumber() == MagicNumber && OrderSymbol() == Symbol() && OrderType() == type)
count++;
}
}
return count;
}
double ExtractPrice(string json, string field)
{
int pos = StringFind(json, "\"" + field + "\":");
if(pos == -1) return 0;
int start = StringFind(json, ":", pos) + 1;
int end = StringFind(json, ",", start);
if(end == -1) end = StringFind(json, "}", start);
if(end > start)
{
string value = StringSubstr(json, start, end - start);
StringTrimLeft(value);
StringTrimRight(value);
return StringToDouble(value);
}
return 0;
}
string ExtractSignal(string json)
{
int pos = StringFind(json, "\"signal\":\"");
if(pos == -1) return "";
int start = pos + 10;
int end = StringFind(json, "\"", start);
if(end > start)
{
return StringSubstr(json, start, end - start);
}
return "";
}
```
3. Claude連携スクリプト
```python
# claude_mcp_client.py
import requests
import json
import asyncio
import logging
from typing import Dict, List, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ClaudeMCPClient:
def __init__(self, mcp_server_url: str = "http://localhost:8080"):
self.mcp_server_url = mcp_server_url
self.session = requests.Session()
def get_ichimoku_signals(self) -> List[Dict]:
"""一目均衡表シグナルを取得"""
try:
payload = {
"jsonrpc": "2.0",
"method": "get_ichimoku_signals",
"id": 1
}
response = self.session.post(
self.mcp_server_url,
json=payload,
timeout=10
)
if response.status_code == 200:
data = response.json()
return data.get('result', [])
else:
logger.error(f"MCP Server error: {response.status_code}")
return []
except Exception as e:
logger.error(f"Error getting signals: {e}")
return []
def calculate_trap_levels(self, symbol: str) -> Optional[Dict]:
"""特定シンボルのトラップレベルを計算"""
try:
payload = {
"jsonrpc": "2.0",
"method": "calculate_trap_levels",
"params": {"symbol": symbol},
"id": 2
}
response = self.session.post(
self.mcp_server_url,
json=payload,
timeout=10
)
if response.status_code == 200:
data = response.json()
return data.get('result')
else:
logger.error(f"MCP Server error: {response.status_code}")
return None
except Exception as e:
logger.error(f"Error calculating trap levels: {e}")
return None
def generate_trading_analysis(self) -> str:
"""取引分析を生成"""
signals = self.get_ichimoku_signals()
if not signals:
return "現在有効なシグナルはありません。"
analysis = "📊 **一目均衡表トラップEA分析レポート**\n\n"
for signal in signals:
symbol = signal.get('symbol', 'Unknown')
trend = signal.get('trend', 'neutral')
action = signal.get('signal', 'HOLD')
analysis += f"**{symbol}**\n"
analysis += f"📈 トレンド: {trend}\n"
analysis += f"🎯 シグナル: {action}\n"
if action in ['BUY', 'SELL']:
analysis += f"💵 現在価格: {signal.get('price', 0):.5f}\n"
analysis += f"⚡ トリガー: {signal.get('trigger_price', 0):.5f}\n"
analysis += f"🛡️ ストップ: {signal.get('stop_loss', 0):.5f}\n"
analysis += f"🎯 テイクプロフィット: {signal.get('take_profit', 0):.5f}\n"
analysis += "─" * 30 + "\n"
# 戦略的なアドバイスを追加
analysis += "\n**🎯 戦略的アドバイス:**\n"
buy_signals = [s for s in signals if s.get('signal') == 'BUY']
sell_signals = [s for s in signals if s.get('signal') == 'SELL']
if buy_signals:
analysis += f"• {len(buy_signals)}つの買いシグナルが検出されました\n"
if sell_signals:
analysis += f"• {len(sell_signals)}つの売りシグナルが検出されました\n"
if not buy_signals and not sell_signals:
analysis += "• 現在は待機が適切です。明確なシグナルを待ちましょう。\n"
return analysis
async def main():
"""Claudeクライアントデモ"""
client = ClaudeMCPClient()
print("🔍 一目均衡表トラップEA分析を開始します...\n")
# ヘルスチェック
try:
health_response = client.session.get(f"{client.mcp_server_url}/health")
if health_response.status_code == 200:
print("✅ MCPサーバーに正常に接続しました\n")
else:
print("❌ MCPサーバーに接続できませんでした")
return
except:
print("❌ MCPサーバーに接続できませんでした")
return
# 分析レポートを生成
analysis = client.generate_trading_analysis()
print(analysis)
# 特定のシンボルの詳細分析
print("\n🔍 USDJPYの詳細分析:")
usdjpy_analysis = client.calculate_trap_levels("USDJPY")
if usdjpy_analysis:
print(json.dumps(usdjpy_analysis, indent=2, ensure_ascii=False))
if __name__ == "__main__":
asyncio.run(main())
```
4. 設定ファイル
```yaml
# config.yaml
mcp_server:
host: "localhost"
port: 8080
timeout: 30
trading:
symbols: ["USDJPY", "EURUSD", "GBPUSD", "USDCHF", "AUDUSD"]
timeframe: "H1"
lot_size: 0.1
max_orders: 3
risk_per_trade: 0.02
ichimoku:
tenkan_period: 9
kijun_period: 26
senkou_period: 52
claude:
api_key: "your-claude-api-key"
model: "claude-3-sonnet-20240229"
```
5. インストール手順
```bash
# 必要なPythonパッケージのインストール
pip install metaTrader5 pandas numpy aiohttp python-socketio requests pyyaml
# MT4の設定
# 1. IchimokuTrapEA.mq4をMT4のExpertsフォルダに配置
# 2. MT4を再起動してEAをチャートにアタッチ
# 3. MCPサーバーURLを設定
# 起動手順
# 1. MCPサーバーを起動
python mcp_ichimoku_server.py
# 2. Claudeクライアントを実行
python claude_mcp_client.py
# 3. MT4でEAを有効化
```
このシステムは以下の特徴を持っています：
1. 一目均衡表ベースのシグナル生成
2. 逆指値注文によるリスク管理
3. MCPサーバーによるClaude連携
4. 自動的なトラップレベル計算
5. リアルタイム監視と注文管理
実際の運用前に、デモ口座で十分にテストすることをお勧めします。
コメント
コメントを投稿