AIエージェントで為替のファンダメンタルズ分析を自動化する｜設計と実装の全体像

【データソース層】
├─ 経済カレンダーAPI（Forex Factory, Investing.com等）
├─ ニュースAPI（NewsAPI, Bloomberg等）
├─ 中央銀行公式サイト（FRB, 日銀, ECB等）
└─ 為替レートAPI（OANDA, Alpha Vantage等）
          ↓
【エージェント層】
├─ data-collector-agent（データ収集・前処理）
├─ indicator-analyzer-agent（経済指標分析）
├─ news-sentiment-agent（ニュースセンチメント）
├─ central-bank-watcher-agent（中央銀行モニタリング）
└─ report-generator-agent（レポート生成・統合）
          ↓
【出力層】
├─ 日次ブリーフィングレポート
├─ 異常シグナルアラート
├─ MT5 EAへのファンダメンタルズフィルター
└─ Slackまたはメール通知

import anthropic
import json
import asyncio
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional
import aiohttp

client = anthropic.Anthropic()

@dataclass
class EconomicEvent:
    """経済指標イベントのデータクラス"""
    name: str
    country: str
    currency: str
    time: datetime
    importance: str     # "high" / "medium" / "low"
    actual: Optional[float] = None
    forecast: Optional[float] = None
    previous: Optional[float] = None
    unit: str = ""

@dataclass
class FundamentalSignal:
    """ファンダメンタルズシグナルのデータクラス"""
    currency_pair: str
    bias: str           # "USD_STRONG" / "USD_WEAK" / "NEUTRAL"
    strength: int       # 1〜5
    confidence: float
    key_drivers: list[str] = field(default_factory=list)
    risk_factors: list[str] = field(default_factory=list)
    generated_at: datetime = field(default_factory=datetime.now)
    valid_until: Optional[datetime] = None
    disclaimer: str = "この分析は将来の市場動向を保証するものではありません"

class IndicatorAnalyzerAgent:
    """
    経済指標の発表結果を分析するエージェント
    """
    
    def __init__(self, model: str = "claude-sonnet-4-5"):
        self.model = model
        self.analysis_cache = {}
    
    def analyze_event(
        self,
        event: EconomicEvent,
        currency_pair: str = "USDJPY"
    ) -> dict:
        """
        単一の経済指標を分析する
        """
        
        if event.actual is None:
            return {"status": "not_released", "event": event.name}
        
        # 予想比乖離の計算
        if event.forecast is not None and event.forecast != 0:
            deviation_pct = (event.actual - event.forecast) / abs(event.forecast) * 100
        else:
            deviation_pct = None
        
        prompt = f"""
経済指標アナリストとして分析してください。

指標: {event.name}（{event.country}）
重要度: {event.importance}
発表値: {event.actual}{event.unit}
予想値: {event.forecast}{event.unit}（乖離: {f'{deviation_pct:+.1f}%' if deviation_pct is not None else 'N/A'}）
前回値: {event.previous}{event.unit}

{currency_pair}への影響を分析してJSON形式のみで回答:
{{
  "usd_impact_score": -10から+10の整数（プラスがUSD強気）,
  "surprise_classification": "large_beat/moderate_beat/inline/moderate_miss/large_miss",
  "immediate_market_reaction": "expected market reaction in 30 chars",
  "fade_risk": "high/medium/low",
  "time_horizon": "immediate/intraday/multi_day",
  "confidence": 0.0から1.0,
  "context": "重要な背景情報があれば50字以内"
}}
"""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=300,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            result["event_name"] = event.name
            result["analyzed_at"] = datetime.now().isoformat()
            return result
        except json.JSONDecodeError:
            return {"status": "parse_error", "event": event.name}
    
    def analyze_multiple_events(
        self,
        events: list[EconomicEvent],
        currency_pair: str = "USDJPY",
        period_hours: int = 24
    ) -> dict:
        """
        複数指標を統合して分析する
        """
        
        individual_analyses = []
        for event in events:
            if event.actual is not None:
                analysis = self.analyze_event(event, currency_pair)
                individual_analyses.append(analysis)
        
        if not individual_analyses:
            return {"status": "no_data", "signal": "NEUTRAL"}
        
        # 統合分析
        events_summary = "\n".join([
            f"- {a.get('event_name', 'N/A')}: スコア={a.get('usd_impact_score', 0)}, "
            f"サプライズ={a.get('surprise_classification', 'N/A')}"
            for a in individual_analyses
        ])
        
        prompt = f"""
過去{period_hours}時間の経済指標を統合分析してください。

【指標一覧】
{events_summary}

{currency_pair}の総合的なファンダメンタルズバイアスをJSON形式で:
{{
  "overall_bias": "STRONG_USD/MILD_USD/NEUTRAL/MILD_WEAK/STRONG_WEAK",
  "net_score": -10から+10の整数,
  "dominant_factor": "最も影響力の大きい指標を30字以内",
  "conflicting_signals": true/false,
  "recommended_time_horizon": "intraday/swing/neutral",
  "disclaimer": "バックテストなし。実市場での有効性は保証しません"
}}
"""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=300,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            result["individual_analyses"] = individual_analyses
            return result
        except json.JSONDecodeError:
            return {"status": "parse_error", "individual_analyses": individual_analyses}

class CentralBankWatcherAgent:
    """
    中央銀行の声明・会見を監視・分析するエージェント
    """
    
    CENTRAL_BANKS = {
        "FRB": {"currency": "USD", "currency_pair_impact": "USDJPY"},
        "日銀": {"currency": "JPY", "currency_pair_impact": "USDJPY"},
        "ECB": {"currency": "EUR", "currency_pair_impact": "EURJPY"},
        "BOE": {"currency": "GBP", "currency_pair_impact": "GBPJPY"},
    }
    
    def analyze_statement(
        self,
        bank_name: str,
        statement_text: str,
        previous_statement: Optional[str] = None
    ) -> dict:
        """
        中央銀行声明文を分析する
        前回声明と比較して変化を検出する
        """
        
        comparison_section = ""
        if previous_statement:
            comparison_section = f"""
【前回声明（比較用）】
{previous_statement[:1000]}...（以下省略）
"""
        
        prompt = f"""
中央銀行エコノミストとして{bank_name}の政策声明を分析してください。

【今回の声明】
{statement_text[:2000]}

{comparison_section}

以下の点を分析しJSON形式で回答:
{{
  "policy_stance": "hawkish/neutral/dovish",
  "stance_change_from_previous": "more_hawkish/unchanged/more_dovish/unknown",
  "key_language_changes": ["表現変化を3点以内で"],
  "rate_path_signal": "hike/hold/cut/unclear",
  "forward_guidance_clarity": "clear/vague/no_guidance",
  "inflation_assessment": "上昇懸念/目標達成/低すぎる/言及なし",
  "growth_assessment": "強い/堅調/減速/リスク/言及なし",
  "currency_implication": {{
    "direction": "{self.CENTRAL_BANKS.get(bank_name, {}).get('currency', 'USD')}_STRONG/WEAK/NEUTRAL",
    "magnitude": "large/moderate/small",
    "rationale": "50字以内"
  }},
  "surprise_level": "large_surprise/moderate_surprise/as_expected",
  "confidence": 0.0から1.0,
  "disclaimer": "この分析は将来の政策および市場動向を保証しません"
}}
"""
        
        response = client.messages.create(
            model="claude-sonnet-4-5",  # 長文処理のためSonnet推奨
            max_tokens=600,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            result["bank_name"] = bank_name
            result["analyzed_at"] = datetime.now().isoformat()
            return result
        except json.JSONDecodeError:
            return {"status": "parse_error", "bank_name": bank_name}

class FundamentalsReportAgent:
    """
    複数エージェントの出力を統合してレポートを生成するエージェント
    """
    
    def generate_daily_report(
        self,
        indicator_analysis: dict,
        news_sentiment: dict,
        cb_analysis: Optional[dict],
        currency_pair: str = "USDJPY"
    ) -> dict:
        """
        日次ファンダメンタルズレポートを生成する
        """
        
        # 各エージェントの出力をまとめる
        context_parts = []
        
        if indicator_analysis:
            bias = indicator_analysis.get("overall_bias", "NEUTRAL")
            score = indicator_analysis.get("net_score", 0)
            dominant = indicator_analysis.get("dominant_factor", "N/A")
            context_parts.append(
                f"経済指標: バイアス={bias}, スコア={score:+d}, 主因={dominant}"
            )
        
        if news_sentiment:
            sentiment = news_sentiment.get("sentiment", "NEUTRAL")
            confidence = news_sentiment.get("confidence", 0)
            context_parts.append(
                f"ニュースセンチメント: {sentiment} (確信度: {confidence:.2f})"
            )
        
        if cb_analysis:
            stance = cb_analysis.get("policy_stance", "neutral")
            change = cb_analysis.get("stance_change_from_previous", "unchanged")
            bank = cb_analysis.get("bank_name", "N/A")
            context_parts.append(
                f"中央銀行({bank}): スタンス={stance}, 前回比={change}"
            )
        
        context_summary = "\n".join(context_parts)
        
        prompt = f"""
FXファンダメンタルズアナリストとして{currency_pair}の本日レポートを作成してください。

【各エージェントの分析結果】
{context_summary}

以下の形式でJSON出力:
{{
  "overall_fundamental_bias": "STRONGLY_BULLISH/BULLISH/NEUTRAL/BEARISH/STRONGLY_BEARISH",
  "confidence_level": "high/medium/low",
  "key_drivers": ["主要ドライバー3点以内"],
  "contradictions": ["相反するシグナルがあれば"],
  "watch_list": ["本日注目すべきイベント・発言"],
  "ea_filter_recommendation": {{
    "long_filter": "allow/caution/block",
    "short_filter": "allow/caution/block",
    "rationale": "50字以内"
  }},
  "summary_for_trader": "200字以内の総括文",
  "valid_until": "HH:MM JST（有効期限、通常は次の主要イベントまで）",
  "disclaimer": "本レポートは情報提供目的であり、投資助言ではありません。FX取引には元本割れリスクがあります。"
}}
"""
        
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=800,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            result["generated_at"] = datetime.now().isoformat()
            result["currency_pair"] = currency_pair
            return result
        except json.JSONDecodeError:
            return {
                "status": "parse_error",
                "overall_fundamental_bias": "NEUTRAL",
                "disclaimer": "エラーのため分析不可"
            }

async def run_fundamentals_pipeline(currency_pair: str = "USDJPY"):
    """
    ファンダメンタルズ分析パイプラインの実行例
    """
    
    print(f"[{datetime.now().strftime('%H:%M')}] ファンダメンタルズ分析開始: {currency_pair}")
    
    # エージェントの初期化
    indicator_agent = IndicatorAnalyzerAgent()
    cb_watcher = CentralBankWatcherAgent()
    report_agent = FundamentalsReportAgent()
    
    # ダミーデータ（実際はAPIから取得）
    events_today = [
        EconomicEvent(
            name="米国消費者物価指数（CPI）",
            country="USA", currency="USD",
            time=datetime.now(),
            importance="high",
            actual=3.1, forecast=3.0, previous=3.2, unit="%"
        ),
        EconomicEvent(
            name="日本・鉱工業生産",
            country="Japan", currency="JPY",
            time=datetime.now() - timedelta(hours=3),
            importance="medium",
            actual=-1.2, forecast=-0.8, previous=0.5, unit="%"
        )
    ]
    
    # 各エージェント実行
    print("  経済指標分析中...")
    indicator_result = indicator_agent.analyze_multiple_events(events_today, currency_pair)
    
    # ニュースセンチメント（ダミー）
    news_result = {"sentiment": "NEUTRAL", "confidence": 0.5}
    
    # 統合レポート生成
    print("  レポート生成中...")
    daily_report = report_agent.generate_daily_report(
        indicator_analysis=indicator_result,
        news_sentiment=news_result,
        cb_analysis=None,
        currency_pair=currency_pair
    )
    
    # 結果の表示
    print(f"\n=== 本日のファンダメンタルズレポート: {currency_pair} ===")
    print(f"バイアス: {daily_report.get('overall_fundamental_bias', 'N/A')}")
    print(f"確信度: {daily_report.get('confidence_level', 'N/A')}")
    print(f"EAフィルター: {daily_report.get('ea_filter_recommendation', {})}")
    print(f"総括: {daily_report.get('summary_for_trader', 'N/A')}")
    print(f"注意: {daily_report.get('disclaimer', '')}")
    
    return daily_report


# 実行
if __name__ == "__main__":
    report = asyncio.run(run_fundamentals_pipeline("USDJPY"))

# コスト試算（2026年6月時点）
daily_cost_estimate = {
    "indicator_analysis": {
        "calls_per_day": 10,         # 指標10件
        "tokens_per_call": 500,       # input + output
        "cost_per_call_usd": 0.0015,
        "daily_cost_usd": 0.015
    },
    "news_sentiment": {
        "calls_per_day": 48,          # 30分ごと
        "tokens_per_call": 600,
        "cost_per_call_usd": 0.0018,
        "daily_cost_usd": 0.086
    },
    "daily_report": {
        "calls_per_day": 4,           # 4回/日
        "tokens_per_call": 1000,
        "cost_per_call_usd": 0.003,
        "daily_cost_usd": 0.012
    },
    "total_daily_usd": 0.113,         # 約17円/日、500円/月
}