AI エージェントで FX 取引戦略を自動化する：マルチエージェント設計の実践ガイド【2026年版】

if EMA20 > EMA50 and RSI < 70:
    enter_long()

データ → LLMによる状況推論 → アクションの選択 → 実行

[Agent 1: 市場環境分類エージェント]
    入力: テクニカルデータ（EMA/RSI/ADX等）+ 直近の主要ファンダメンタルズ
    出力: 相場環境の分類（トレンド/レンジ/警戒）+ スコア
    
    ↓ 環境が"トレンド"の場合のみ次へ進む
    
[Agent 2: シグナル評価エージェント]
    入力: 環境スコア + 具体的な売買シグナル（テクニカル条件の充足状況）
    出力: エントリーの妥当性評価（0〜100点） + 否定的な根拠
    
    ↓ 評価スコアが閾値（例: 70点）以上の場合のみ次へ進む
    
[Agent 3: リスク管理エージェント]
    入力: 評価スコア + 口座残高 + 既存ポジション
    出力: 推奨ロットサイズ + ストップロス位置
    
    ↓ 承認された場合のみ実行層へ
    
[実行層（EA）: MT5]
    入力: Agent 3の出力（ロット・SL・TP）
    出力: 実際の注文

import anthropic
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

client = anthropic.Anthropic()

# 共有状態の定義
class TradingState(TypedDict):
    market_data: dict
    environment: str
    environment_score: float
    signal_score: float
    lot_size: float
    stop_loss: float
    decision: Literal["execute", "skip", "error"]

# Agent 1: 市場環境分類
def classify_market_environment(state: TradingState) -> TradingState:
    data = state["market_data"]
    
    prompt = f"""<market_data>
EMA20: {data['ema20']}, EMA50: {data['ema50']}
RSI14: {data['rsi']}, ADX14: {data['adx']}
直近指標: {data.get('news', 'なし')}
</market_data>

JSON形式のみで回答:
{{"environment": "trend/range/caution", "score": 0-100の整数, "reason": "30字以内"}}"""

    message = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=128,
        messages=[{"role": "user", "content": prompt}]
    )
    
    import json
    result = json.loads(message.content[0].text)
    state["environment"] = result["environment"]
    state["environment_score"] = result["score"]
    return state

# 条件分岐：環境が適切かチェック
def should_continue_after_environment(state: TradingState) -> str:
    if state["environment"] == "trend" and state["environment_score"] >= 60:
        return "evaluate_signal"
    return "skip_trade"

# スキップ処理
def skip_trade(state: TradingState) -> TradingState:
    state["decision"] = "skip"
    return state

# ワークフロー構築
workflow = StateGraph(TradingState)
workflow.add_node("classify_environment", classify_market_environment)
workflow.add_node("skip_trade", skip_trade)
# ... （シグナル評価・リスク管理エージェントを同様に追加）

workflow.set_entry_point("classify_environment")
workflow.add_conditional_edges(
    "classify_environment",
    should_continue_after_environment,
    {"evaluate_signal": "evaluate_signal", "skip_trade": "skip_trade"}
)

def safe_parse_json(response_text: str, fallback: dict) -> dict:
    """LLM応答のJSONパース失敗時にフォールバック値を返す"""
    try:
        import json
        return json.loads(response_text)
    except json.JSONDecodeError:
        # ログに記録してフォールバック
        print(f"JSON parse failed: {response_text[:100]}")
        return fallback

# 使用例: パース失敗時は"skip"判断にフォールバック
result = safe_parse_json(
    message.content[0].text, 
    {"environment": "caution", "score": 0, "reason": "parse_error"}
)