Claude APIでFXファンダメンタルズ分析を自動化する：プロンプト設計と実装の全手順

経済カレンダー / ニュースフィード
    ↓  スクレイピング or RSS取得（Python）
テキストデータの前処理
    ↓
Claude APIへのリクエスト（プロンプト設計）
    ↓
構造化された分析結果（JSON）
    ↓
MT5 EAへのシグナル送信 / ダッシュボード表示

あなたは為替市場専門のマクロ経済アナリストです。
以下の経済指標発表を分析してください。

指標: {指標名}
発表値: {発表値}
予想値: {予想値}
前回値: {前回値}

以下のJSON形式で回答してください：
{
  "score": -10から+10の整数（+が対象通貨に強気、-が弱気）,
  "rationale": "100字以内の根拠説明",
  "time_horizon": "immediate/short/medium（影響の持続時間）",
  "risk_note": "リスク要因があれば記載、なければnull"
}

以下は{中央銀行名}の{日付}声明文です。
FXトレーダーの視点から、前回声明文と比較した際の政策方向性の変化を分析してください。

【今回の声明】
{声明文全文}

【前回の声明（比較用）】
{前回声明文}

分析形式：
1. 政策スタンスの変化（ハト派化/鷹派化/変化なし）
2. 変化のシグナルとなったキーフレーズを3つ抜き出す
3. 通貨への影響スコア（-10〜+10）
4. 今後2週間の相場へのインプリケーション（100字以内）

以下の発言を分析してください。
発言者: {名前・役職}
発言日時: {日時}
発言内容: "{発言全文}"

この人物の発言の文脈で、以下を評価してください：
- 実際の政策示唆の強度（1=示唆弱い、5=明確なシグナル）
- ヘッドラインと実際の意図のギャップがあれば指摘する
- 市場が過剰反応または過小反応している場合はその旨を記す
- USD/JPYへの影響方向と推定持続時間

以下の材料を総合して、{通貨ペア}の{時間軸}トレードの環境を評価してください。

【材料リスト】
1. {材料1}
2. {材料2}
3. {材料3}（以降追加）

評価してください：
- 総合的な相場の方向性バイアス（強気/弱気/中立）
- 最もリスクの高い材料（逆方向の場合の影響が大きいもの）
- 現在のトレードを推奨しない条件（これがあれば見送る、というもの）

import anthropic
import json

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY は環境変数から

def analyze_economic_indicator(
    indicator_name: str,
    actual: float,
    forecast: float,
    previous: float,
    currency: str
) -> dict:
    """経済指標を分析してスコアとJSON結果を返す"""
    
    prompt = f"""あなたは為替市場専門のマクロ経済アナリストです。
以下の経済指標発表を分析してください。

指標: {indicator_name}（{currency}）
発表値: {actual}
予想値: {forecast}
前回値: {previous}

以下のJSON形式のみで回答してください：
{{
  "score": -10から+10の整数,
  "rationale": "根拠説明（100字以内）",
  "time_horizon": "immediate/short/medium",
  "risk_note": "リスク要因またはnull"
}}"""

    message = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=256,
        thinking={"type": "adaptive"},
        messages=[{"role": "user", "content": prompt}]
    )
    
    # thinkingブロックを除いた最終テキストを取得
    response_text = ""
    for block in message.content:
        if block.type == "text":
            response_text = block.text
            break
    
    try:
        return json.loads(response_text)
    except json.JSONDecodeError:
        # JSONパースに失敗した場合のフォールバック
        return {"score": 0, "rationale": "パースエラー", "raw": response_text}

import time
from datetime import datetime

INDICATORS_TODAY = [
    {"name": "米・非農業部門雇用者数", "actual": 256000, "forecast": 230000, "previous": 227000, "currency": "USD"},
    {"name": "米・失業率", "actual": 4.1, "forecast": 4.2, "previous": 4.2, "currency": "USD"},
    {"name": "日・東京CPI", "actual": 2.8, "forecast": 2.5, "previous": 2.6, "currency": "JPY"},
]

def batch_analyze(indicators: list) -> dict:
    results = {}
    currency_scores = {}
    
    for ind in indicators:
        result = analyze_economic_indicator(
            ind["name"], ind["actual"], ind["forecast"], ind["previous"], ind["currency"]
        )
        results[ind["name"]] = result
        
        currency = ind["currency"]
        if currency not in currency_scores:
            currency_scores[currency] = []
        currency_scores[currency].append(result.get("score", 0))
        
        time.sleep(0.5)  # レート制限対策
    
    # 通貨別の総合スコア
    summary = {
        cur: sum(scores) / len(scores) 
        for cur, scores in currency_scores.items()
    }
    
    return {"details": results, "currency_summary": summary}

if __name__ == "__main__":
    analysis = batch_analyze(INDICATORS_TODAY)
    print(f"USD総合スコア: {analysis['currency_summary'].get('USD', 0):.1f}")
    print(f"JPY総合スコア: {analysis['currency_summary'].get('JPY', 0):.1f}")