LLM × FX API × MT5 連携ガイド｜Claude/GPTをMT5に繋ぐ実装手順

# llm_signal_writer.py
import anthropic
import json
import time
import schedule
from pathlib import Path
from datetime import datetime

# MT5のCommonフォルダパスを設定
MT5_COMMON = Path(r"C:\Users\USER\AppData\Roaming\MetaQuotes\Terminal\Common\Files")

client = anthropic.Anthropic()

def generate_market_bias(news_headlines: list[str]) -> dict:
    """
    最新ニュースからLLMで市場バイアスを生成
    """
    headlines_text = "\n".join([f"- {h}" for h in news_headlines[:8]])
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"""
以下の為替市場ニュースヘッドラインを分析してください。

{headlines_text}

USD/JPYの今後のバイアスをJSON形式のみで出力（説明不要）:
{{
  "bias": "BULLISH_USD/BEARISH_USD/NEUTRAL のいずれか",
  "strength": 1から5の整数,
  "confidence": 0.0から1.0,
  "valid_minutes": 60（有効期間、分）,
  "key_driver": "主要因を30字以内",
  "generated_at": "{datetime.now().isoformat()}"
}}
"""
        }]
    )
    
    try:
        return json.loads(response.content[0].text)
    except json.JSONDecodeError:
        return {"bias": "NEUTRAL", "strength": 0, "confidence": 0,
                "valid_minutes": 30, "key_driver": "解析失敗",
                "generated_at": datetime.now().isoformat()}


def fetch_news_headlines() -> list[str]:
    """
    ニュース取得（実装はRSSまたは有料データプロバイダー推奨）
    ここでは例示用のダミーデータ
    """
    return [
        "FRB、今年の利下げ回数を2回と予測－6月会合後声明",
        "日銀、次回会合で政策金利の引き上げ検討か",
        "米国雇用統計、非農業部門雇用者数が予想を上回る"
    ]


def update_signal():
    """シグナルファイルを更新する"""
    print(f"[{datetime.now().strftime('%H:%M:%S')}] シグナル更新中...")
    
    headlines = fetch_news_headlines()
    signal_data = generate_market_bias(headlines)
    
    output_path = MT5_COMMON / "llm_bias_usdjpy.json"
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(signal_data, f, ensure_ascii=False, indent=2)
    
    print(f"  バイアス: {signal_data['bias']} (強度: {signal_data['strength']}, "
          f"確信度: {signal_data['confidence']:.2f})")
    print(f"  主要因: {signal_data['key_driver']}")


# 15分ごとに更新
schedule.every(15).minutes.do(update_signal)

if __name__ == "__main__":
    update_signal()  # 起動時に即時実行
    while True:
        schedule.run_pending()
        time.sleep(60)

//+------------------------------------------------------------------+
//| LLMバイアスフィルター付きEA                                        |
//| ファイル経由でLLMシグナルを読み取る                                  |
//+------------------------------------------------------------------+
#include <Trade\Trade.mqh>

CTrade trade;

input int    MagicNumber   = 20240601;
input double RiskPercent   = 1.0;
input int    RSI_Period    = 14;
input double RSI_Overbought = 65.0;
input double RSI_Oversold   = 35.0;
input int    EMA_Fast      = 20;
input int    EMA_Slow      = 50;

// LLMシグナルの構造体
struct LLMBias {
    string bias;           // "BULLISH_USD" / "BEARISH_USD" / "NEUTRAL"
    int    strength;       // 1〜5
    double confidence;     // 0.0〜1.0
    bool   is_valid;       // シグナルが有効期限内か
};

//--- インジケータハンドル
int handle_rsi;
int handle_ema_fast;
int handle_ema_slow;

int OnInit() {
    handle_rsi      = iRSI(Symbol(), PERIOD_CURRENT, RSI_Period, PRICE_CLOSE);
    handle_ema_fast = iMA(Symbol(), PERIOD_CURRENT, EMA_Fast, 0, MODE_EMA, PRICE_CLOSE);
    handle_ema_slow = iMA(Symbol(), PERIOD_CURRENT, EMA_Slow, 0, MODE_EMA, PRICE_CLOSE);
    
    if (handle_rsi == INVALID_HANDLE ||
        handle_ema_fast == INVALID_HANDLE ||
        handle_ema_slow == INVALID_HANDLE) {
        Print("インジケータ初期化失敗");
        return INIT_FAILED;
    }
    return INIT_SUCCEEDED;
}

void OnDeinit(const int reason) {
    IndicatorRelease(handle_rsi);
    IndicatorRelease(handle_ema_fast);
    IndicatorRelease(handle_ema_slow);
}

//+------------------------------------------------------------------+
//| LLMバイアスファイルを読み込む                                        |
//+------------------------------------------------------------------+
LLMBias ReadLLMBias() {
    LLMBias result;
    result.bias       = "NEUTRAL";
    result.strength   = 0;
    result.confidence = 0.0;
    result.is_valid   = false;
    
    // Common フォルダのファイルを読む
    int handle = FileOpen("llm_bias_usdjpy.json",
                          FILE_READ | FILE_TXT | FILE_COMMON);
    
    if (handle == INVALID_HANDLE) {
        Print("LLMシグナルファイルなし。フィルターを無効化して動作。");
        result.is_valid = true;  // フェイルオープン
        result.bias = "NEUTRAL";
        return result;
    }
    
    string content = "";
    while (!FileIsEnding(handle)) {
        content += FileReadString(handle);
    }
    FileClose(handle);
    
    // 簡易JSONパース
    // 注: 本番実装には json4mql等のライブラリ推奨
    
    // bias の抽出
    if (StringFind(content, "\"BULLISH_USD\"") >= 0)
        result.bias = "BULLISH_USD";
    else if (StringFind(content, "\"BEARISH_USD\"") >= 0)
        result.bias = "BEARISH_USD";
    else
        result.bias = "NEUTRAL";
    
    // confidence の抽出（簡易版）
    int conf_pos = StringFind(content, "\"confidence\":");
    if (conf_pos >= 0) {
        string conf_str = StringSubstr(content, conf_pos + 14, 4);
        result.confidence = StringToDouble(conf_str);
    }
    
    // strength の抽出
    int str_pos = StringFind(content, "\"strength\":");
    if (str_pos >= 0) {
        result.strength = (int)StringToInteger(
            StringSubstr(content, str_pos + 11, 1)
        );
    }
    
    result.is_valid = (result.confidence > 0);
    return result;
}

//+------------------------------------------------------------------+
//| テクニカルシグナルの判定                                             |
//+------------------------------------------------------------------+
int GetTechnicalSignal() {
    double rsi_buf[], ema_fast_buf[], ema_slow_buf[];
    ArraySetAsSeries(rsi_buf, true);
    ArraySetAsSeries(ema_fast_buf, true);
    ArraySetAsSeries(ema_slow_buf, true);
    
    if (CopyBuffer(handle_rsi, 0, 0, 3, rsi_buf) < 3) return 0;
    if (CopyBuffer(handle_ema_fast, 0, 0, 3, ema_fast_buf) < 3) return 0;
    if (CopyBuffer(handle_ema_slow, 0, 0, 3, ema_slow_buf) < 3) return 0;
    
    double rsi = rsi_buf[1];        // 前バー確定値
    double ema_f = ema_fast_buf[1];
    double ema_s = ema_slow_buf[1];
    
    if (ema_f > ema_s && rsi > 50 && rsi < RSI_Overbought)
        return 1;  // ロングシグナル
    
    if (ema_f < ema_s && rsi < 50 && rsi > RSI_Oversold)
        return -1; // ショートシグナル
    
    return 0;
}

//+------------------------------------------------------------------+
//| ロット計算（口座残高の1%リスク）                                      |
//+------------------------------------------------------------------+
double CalcLotSize(double stop_pips) {
    double account_balance = AccountInfoDouble(ACCOUNT_BALANCE);
    double risk_amount = account_balance * RiskPercent / 100.0;
    
    double tick_value = SymbolInfoDouble(Symbol(), SYMBOL_TRADE_TICK_VALUE);
    double tick_size  = SymbolInfoDouble(Symbol(), SYMBOL_TRADE_TICK_SIZE);
    
    if (tick_value == 0 || tick_size == 0) return 0.01;
    
    double point = SymbolInfoDouble(Symbol(), SYMBOL_POINT);
    double stop_points = stop_pips * point * 10;  // pipsをpointsに変換
    
    double lot = risk_amount / (stop_points / tick_size * tick_value);
    
    // ブローカー制限に合わせてクランプ
    double min_lot = SymbolInfoDouble(Symbol(), SYMBOL_VOLUME_MIN);
    double max_lot = SymbolInfoDouble(Symbol(), SYMBOL_VOLUME_MAX);
    double lot_step = SymbolInfoDouble(Symbol(), SYMBOL_VOLUME_STEP);
    
    lot = MathMax(min_lot, MathMin(max_lot, lot));
    lot = MathRound(lot / lot_step) * lot_step;
    
    return lot;
}

//+------------------------------------------------------------------+
//| メインロジック                                                       |
//+------------------------------------------------------------------+
void OnTick() {
    // 新しいバーの開始時のみ実行
    static datetime last_bar_time = 0;
    datetime current_bar_time = iTime(Symbol(), PERIOD_CURRENT, 0);
    if (current_bar_time == last_bar_time) return;
    last_bar_time = current_bar_time;
    
    // 既存ポジションチェック
    if (PositionsTotal() > 0) return;
    
    // テクニカルシグナル取得
    int tech_signal = GetTechnicalSignal();
    if (tech_signal == 0) return;
    
    // LLMバイアスフィルター
    LLMBias llm = ReadLLMBias();
    
    // LLMバイアスとテクニカルシグナルの整合性チェック
    bool llm_ok_long  = (llm.bias != "BEARISH_USD" || llm.strength <= 2);
    bool llm_ok_short = (llm.bias != "BULLISH_USD" || llm.strength <= 2);
    
    double atr_buf[];
    ArraySetAsSeries(atr_buf, true);
    int handle_atr = iATR(Symbol(), PERIOD_CURRENT, 14);
    if (CopyBuffer(handle_atr, 0, 1, 1, atr_buf) < 1) return;
    IndicatorRelease(handle_atr);
    
    double atr = atr_buf[0];
    double stop_pips = atr * 1.5 / SymbolInfoDouble(Symbol(), SYMBOL_POINT) / 10;
    
    double ask = SymbolInfoDouble(Symbol(), SYMBOL_ASK);
    double bid = SymbolInfoDouble(Symbol(), SYMBOL_BID);
    double lot = CalcLotSize(stop_pips);
    
    if (tech_signal == 1 && llm_ok_long) {
        double sl = ask - atr * 1.5;
        double tp = ask + atr * 2.5;
        trade.SetExpertMagicNumber(MagicNumber);
        trade.Buy(lot, Symbol(), ask, sl, tp,
                  StringFormat("LLM:%s S:%d", llm.bias, llm.strength));
        
        Print(StringFormat("BUY実行: lot=%.2f, LLMバイアス=%s, 強度=%d",
                           lot, llm.bias, llm.strength));
    }
    
    if (tech_signal == -1 && llm_ok_short) {
        double sl = bid + atr * 1.5;
        double tp = bid - atr * 2.5;
        trade.SetExpertMagicNumber(MagicNumber);
        trade.Sell(lot, Symbol(), bid, sl, tp,
                   StringFormat("LLM:%s S:%d", llm.bias, llm.strength));
        
        Print(StringFormat("SELL実行: lot=%.2f, LLMバイアス=%s, 強度=%d",
                           lot, llm.bias, llm.strength));
    }
}

# middleware_server.py（FastAPI）
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import anthropic
import json
from functools import lru_cache
from datetime import datetime, timedelta

app = FastAPI()
client = anthropic.Anthropic()

# キャッシュ（同じリクエストを連続して送らないため）
_cache: dict = {}
CACHE_TTL_MINUTES = 15

class MarketContext(BaseModel):
    symbol: str
    timeframe: str
    rsi: float
    ema_diff_pct: float  # (EMA20 - EMA50) / EMA50 * 100
    recent_news: list[str] = []

class LLMResponse(BaseModel):
    filter_action: str  # "ALLOW_LONG" / "ALLOW_SHORT" / "BLOCK_LONG" / "BLOCK_SHORT" / "NEUTRAL"
    confidence: float
    reason: str
    cached: bool = False

def get_cache_key(ctx: MarketContext) -> str:
    news_hash = hash(tuple(ctx.recent_news))
    return f"{ctx.symbol}_{ctx.timeframe}_{news_hash}"

@app.post("/filter", response_model=LLMResponse)
async def filter_signal(ctx: MarketContext):
    """
    MT5 EAからのシグナルフィルタリングリクエストを処理
    """
    cache_key = get_cache_key(ctx)
    
    # キャッシュチェック
    if cache_key in _cache:
        cached_time, cached_result = _cache[cache_key]
        if datetime.now() - cached_time < timedelta(minutes=CACHE_TTL_MINUTES):
            cached_result.cached = True
            return cached_result
    
    # ニュースがない場合はNEUTRALを返す
    if not ctx.recent_news:
        return LLMResponse(
            filter_action="NEUTRAL",
            confidence=0.5,
            reason="ニュースデータなし"
        )
    
    news_text = "\n".join([f"- {n}" for n in ctx.recent_news[:5]])
    
    prompt = f"""
{ctx.symbol}のトレードシグナルをフィルタリングしてください。

【市場状況】
テクニカル: RSI={ctx.rsi:.1f}, EMA差={ctx.ema_diff_pct:+.2f}%

【直近ニュース】
{news_text}

以下のJSON形式のみで回答:
{{
  "filter_action": "ALLOW_LONG/ALLOW_SHORT/BLOCK_LONG/BLOCK_SHORT/NEUTRAL のいずれか",
  "confidence": 0.0から1.0,
  "reason": "判断理由を50字以内"
}}

注意: 情報が不明確な場合はNEUTRALを優先してください。
"""
    
    try:
        response = client.messages.create(
            model="claude-haiku-4-5",  # コスト最適化のためHaikuを使用
            max_tokens=150,
            messages=[{"role": "user", "content": prompt}]
        )
        
        data = json.loads(response.content[0].text)
        result = LLMResponse(
            filter_action=data.get("filter_action", "NEUTRAL"),
            confidence=data.get("confidence", 0.5),
            reason=data.get("reason", ""),
            cached=False
        )
        
        # キャッシュ保存
        _cache[cache_key] = (datetime.now(), result)
        
        return result
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
def health_check():
    return {"status": "ok", "timestamp": datetime.now().isoformat()}

//+------------------------------------------------------------------+
//| Claude APIへの直接リクエスト（簡易版）                               |
//+------------------------------------------------------------------+

// 注意: APIキーを直接コードに書かない。
//       DLLまたはファイルから読み込む方式を推奨。
string API_KEY = ""; // 環境変数または設定ファイルから読み込む

string CallClaudeAPI(string prompt) {
    string url = "https://api.anthropic.com/v1/messages";
    
    string payload = StringFormat(
        "{\"model\":\"claude-haiku-4-5\","
        "\"max_tokens\":200,"
        "\"messages\":[{\"role\":\"user\",\"content\":\"%s\"}]}",
        StringReplace(prompt, "\"", "\\\"")
    );
    
    char post_data[];
    StringToCharArray(payload, post_data, 0, StringLen(payload));
    
    string headers = StringFormat(
        "Content-Type: application/json\r\n"
        "x-api-key: %s\r\n"
        "anthropic-version: 2023-06-01\r\n",
        API_KEY
    );
    
    char response[];
    string response_headers;
    int timeout = 10000; // 10秒
    
    int res = WebRequest("POST", url, headers, timeout,
                         post_data, response, response_headers);
    
    if (res == 200) {
        return CharArrayToString(response);
    } else {
        Print("API呼び出しエラー: HTTP ", res);
        return "";
    }
}

# cost_optimizer.py
from datetime import datetime, timedelta
from collections import deque

class APICallOptimizer:
    """
    LLM APIコールのコスト最適化クラス
    """
    
    def __init__(
        self,
        min_interval_seconds: int = 300,   # 最小呼び出し間隔（5分）
        cache_ttl_minutes: int = 15,        # キャッシュ有効期間
        daily_call_limit: int = 100         # 1日の最大呼び出し数
    ):
        self.min_interval = timedelta(seconds=min_interval_seconds)
        self.cache_ttl = timedelta(minutes=cache_ttl_minutes)
        self.daily_limit = daily_call_limit
        
        self._last_call_time = {}
        self._cache = {}
        self._daily_calls = deque()  # タイムスタンプのキュー
    
    def should_call_api(self, cache_key: str) -> tuple[bool, str]:
        """APIを呼ぶべきかを判定（bool, 理由）"""
        
        # キャッシュが有効なら呼ばない
        if cache_key in self._cache:
            cached_time, _ = self._cache[cache_key]
            if datetime.now() - cached_time < self.cache_ttl:
                return False, "キャッシュ有効"
        
        # 最小間隔チェック
        if cache_key in self._last_call_time:
            elapsed = datetime.now() - self._last_call_time[cache_key]
            if elapsed < self.min_interval:
                remaining = (self.min_interval - elapsed).seconds
                return False, f"インターバル制限（残{remaining}秒）"
        
        # 日次制限チェック
        now = datetime.now()
        day_start = now.replace(hour=0, minute=0, second=0)
        today_calls = sum(1 for t in self._daily_calls if t >= day_start)
        
        if today_calls >= self.daily_limit:
            return False, f"日次制限到達（{today_calls}/{self.daily_limit}）"
        
        return True, "呼び出し可能"
    
    def record_call(self, cache_key: str, result: dict):
        """API呼び出しを記録"""
        now = datetime.now()
        self._last_call_time[cache_key] = now
        self._cache[cache_key] = (now, result)
        self._daily_calls.append(now)
    
    def get_daily_stats(self) -> dict:
        """本日のAPI利用統計"""
        now = datetime.now()
        day_start = now.replace(hour=0, minute=0, second=0)
        today_calls = sum(1 for t in self._daily_calls if t >= day_start)
        
        return {
            "today_calls": today_calls,
            "limit": self.daily_limit,
            "remaining": max(0, self.daily_limit - today_calls)
        }