LLMトレードシグナルをFXに活用する｜仕組み・実装・精度の現実

入力: 「FRBパウエル議長、インフレ目標達成に自信を示す」
出力: {"signal": "USD_BULLISH", "score": +7, "confidence": 0.72}

入力（テキスト化した価格データ）:
"USD/JPY 1時間足: 始値155.20, 高値155.48, 安値155.15, 終値155.42
 前バー: 始値155.00, 高値155.25, 安値154.95, 終値155.20
 RSI(14): 62.4, EMA(20): 155.18, EMA(50): 154.85"

質問: "現在のテクニカル状況を評価し、シグナルを出してください"

import anthropic
import json
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class TradingSignal:
    currency_pair: str
    direction: str          # "LONG" / "SHORT" / "NEUTRAL"
    strength: int           # 1〜10
    confidence: float       # 0.0〜1.0
    time_horizon: str       # "intraday" / "swing" / "position"
    rationale: str
    generated_at: datetime
    disclaimer: str = "このシグナルは投資助言ではありません"

class LLMSignalGenerator:
    
    def __init__(self, model: str = "claude-sonnet-4-5"):
        self.client = anthropic.Anthropic()
        self.model = model
        self.signal_history = []
    
    def generate_from_news(
        self,
        news_items: list[dict],
        currency_pair: str
    ) -> Optional[TradingSignal]:
        """
        ニュース記事リストからトレードシグナルを生成
        """
        
        # ニュースを構造化テキストに変換
        news_text = "\n".join([
            f"[{item.get('time', 'N/A')}] {item['headline']}"
            for item in news_items[:10]  # 最新10件に絞る
        ])
        
        prompt = f"""
あなたはFX市場の上級アナリストです。以下のニュースヘッドライン群を分析し、
{currency_pair}のトレードシグナルを生成してください。

【分析対象ニュース】
{news_text}

【出力形式】JSON形式のみで回答（説明文不要）:
{{
  "direction": "LONG/SHORT/NEUTRAL のいずれか",
  "strength": 1から10の整数（10が最強シグナル）,
  "confidence": 0.0から1.0の実数,
  "time_horizon": "intraday/swing/position のいずれか",
  "rationale": "シグナル根拠を100字以内で",
  "key_risks": "主なリスク要因を50字以内で",
  "conflicting_signals": true/false（相反するシグナルが混在するか）
}}

重要: 情報が不十分な場合や、相反するシグナルが強い場合はNEUTRALを優先してください。
"""
        
        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=300,
                messages=[{"role": "user", "content": prompt}]
            )
            
            data = json.loads(response.content[0].text)
            
            signal = TradingSignal(
                currency_pair=currency_pair,
                direction=data["direction"],
                strength=data["strength"],
                confidence=data["confidence"],
                time_horizon=data["time_horizon"],
                rationale=data["rationale"],
                generated_at=datetime.now()
            )
            
            self.signal_history.append(signal)
            return signal
            
        except (json.JSONDecodeError, KeyError) as e:
            print(f"シグナル生成エラー: {e}")
            return None
    
    def generate_from_indicator(
        self,
        indicator_name: str,
        actual: float,
        forecast: float,
        previous: float,
        currency_pair: str
    ) -> Optional[TradingSignal]:
        """
        経済指標の発表結果からシグナルを生成
        """
        
        surprise = actual - forecast
        surprise_pct = (surprise / abs(forecast) * 100) if forecast != 0 else 0
        
        prompt = f"""
経済指標アナリストとして以下の指標発表を分析してください。

指標: {indicator_name}
発表値: {actual}（予想: {forecast}、前回: {previous}）
予想との乖離: {surprise:+.2f}（{surprise_pct:+.1f}%）

{currency_pair}への影響をJSON形式で:
{{
  "direction": "LONG/SHORT/NEUTRAL",
  "strength": 1から10の整数,
  "confidence": 0.0から1.0,
  "time_horizon": "immediate/intraday のいずれか",
  "rationale": "根拠を80字以内で",
  "fade_probability": "high/medium/low（初期反応が逆転する可能性）"
}}
"""
        
        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=200,
                messages=[{"role": "user", "content": prompt}]
            )
            
            data = json.loads(response.content[0].text)
            
            return TradingSignal(
                currency_pair=currency_pair,
                direction=data["direction"],
                strength=data["strength"],
                confidence=data["confidence"],
                time_horizon=data["time_horizon"],
                rationale=data["rationale"],
                generated_at=datetime.now()
            )
        except Exception as e:
            print(f"指標シグナルエラー: {e}")
            return None

class SignalFilter:
    """
    LLMシグナルの品質フィルター
    """
    
    def __init__(
        self,
        min_confidence: float = 0.65,
        min_strength: int = 6,
        require_no_conflict: bool = True
    ):
        self.min_confidence = min_confidence
        self.min_strength = min_strength
        self.require_no_conflict = require_no_conflict
    
    def is_actionable(self, signal: TradingSignal, raw_data: dict) -> tuple[bool, str]:
        """
        シグナルが実行可能かを判定する
        
        Returns:
            (bool: 実行可否, str: 判定理由)
        """
        
        if signal.direction == "NEUTRAL":
            return False, "NEUTRAL シグナル"
        
        if signal.confidence < self.min_confidence:
            return False, f"確信度不足: {signal.confidence:.2f} < {self.min_confidence}"
        
        if signal.strength < self.min_strength:
            return False, f"シグナル強度不足: {signal.strength} < {self.min_strength}"
        
        if self.require_no_conflict and raw_data.get("conflicting_signals"):
            return False, "相反するシグナルが混在"
        
        return True, "フィルター通過"
    
    def get_position_size_multiplier(self, signal: TradingSignal) -> float:
        """
        シグナル強度に基づくポジションサイズ調整（0.5〜1.0倍）
        ※ベースロットに掛け合わせる係数
        """
        base = signal.confidence * (signal.strength / 10)
        # 0.5〜1.0の範囲にクランプ
        return max(0.5, min(1.0, base))


# 使用例
generator = LLMSignalGenerator()
filter_obj = SignalFilter(min_confidence=0.65, min_strength=6)

signal = generator.generate_from_news(news_items, "USDJPY")

if signal:
    is_ok, reason = filter_obj.is_actionable(signal, raw_data={})
    if is_ok:
        multiplier = filter_obj.get_position_size_multiplier(signal)
        print(f"シグナル実行: {signal.direction}, ポジションサイズ係数: {multiplier:.2f}")
    else:
        print(f"シグナルスキップ: {reason}")

テクニカルシグナル（例: EMAクロス + RSIフィルター）
           ↓
      シグナル発生
           ↓
  LLMセンチメントチェック（非同期）
  ・直近2時間のニュースセンチメント
  ・直前の経済指標の影響方向
           ↓
  テクニカルとセンチメントが一致？
    │
    ├─ 一致（または中立） → エントリー許可
    └─ 逆方向の強いシグナル → エントリー保留

# signal_server.py（Python側）
import asyncio
import json
from anthropic import Anthropic

client = Anthropic()
current_signal = {"direction": "NEUTRAL", "strength": 0, "confidence": 0}

async def update_signal_loop(currency_pair: str, interval_seconds: int = 300):
    """5分ごとにシグナルを更新してファイルに書き出す"""
    while True:
        # ニュース取得（実装省略）
        news = fetch_latest_news(currency_pair)
        
        # LLMでシグナル生成
        signal = generate_signal(client, news, currency_pair)
        
        if signal:
            current_signal.update({
                "direction": signal.direction,
                "strength": signal.strength,
                "confidence": signal.confidence,
                "updated_at": datetime.now().isoformat()
            })
            
            # MT5が読めるファイルに書き出す
            with open(f"llm_signal_{currency_pair}.json", "w") as f:
                json.dump(current_signal, f)
        
        await asyncio.sleep(interval_seconds)

// EA側（OnTick内）
bool CheckLLMFilter(string symbol, string direction) {
    string filename = "llm_signal_" + symbol + ".json";
    
    int handle = FileOpen(filename, FILE_READ | FILE_TXT | FILE_COMMON);
    if (handle == INVALID_HANDLE) {
        // ファイルなし = フィルターをパス（フェイルオープン）
        return true;
    }
    
    string content = "";
    while (!FileIsEnding(handle)) {
        content += FileReadString(handle);
    }
    FileClose(handle);
    
    // JSON解析（簡易版）
    // 本格実装にはjson4mql等のライブラリ推奨
    if (StringFind(content, "\"direction\": \"NEUTRAL\"") >= 0) return true;
    if (direction == "LONG" && StringFind(content, "\"direction\": \"SHORT\"") >= 0) {
        int strength_pos = StringFind(content, "\"strength\": ");
        if (strength_pos >= 0) {
            int strength = (int)StringToInteger(StringSubstr(content, strength_pos + 12, 2));
            if (strength >= 7) return false;  // 強い逆シグナルはブロック
        }
    }
    return true;
}

class SignalTracker:
    """シグナルの精度をトラッキングする"""
    
    def __init__(self):
        self.signals = []
    
    def record_signal(self, signal: TradingSignal, actual_outcome: str, pips: float):
        """シグナルの結果を記録"""
        self.signals.append({
            "generated_at": signal.generated_at,
            "direction": signal.direction,
            "confidence": signal.confidence,
            "actual_outcome": actual_outcome,  # "WIN" / "LOSS" / "BREAKEVEN"
            "pips": pips,
            "correct": signal.direction == actual_outcome
        })
    
    def get_accuracy_by_confidence(self) -> dict:
        """確信度ごとの精度を計算"""
        buckets = {"high (≥0.8)": [], "medium (0.6-0.8)": [], "low (<0.6)": []}
        
        for s in self.signals:
            if s["direction"] != "NEUTRAL":
                if s["confidence"] >= 0.8:
                    buckets["high (≥0.8)"].append(s["correct"])
                elif s["confidence"] >= 0.6:
                    buckets["medium (0.6-0.8)"].append(s["correct"])
                else:
                    buckets["low (<0.6)"].append(s["correct"])
        
        return {
            bucket: {
                "accuracy": sum(items) / len(items) if items else 0,
                "sample_count": len(items)
            }
            for bucket, items in buckets.items()
        }