LLM × 為替ニュース センチメント分析：研究成果と実践実装の完全解説

import feedparser
import time
from datetime import datetime

FX_NEWS_FEEDS = [
    "https://www.forexlive.com/feed/news",
    "https://www.fxstreet.com/rss/news",
    # 日本語ソース
    "https://jp.reuters.com/rssFeed/businessNews",
]

def fetch_recent_news(max_age_minutes: int = 30) -> list[dict]:
    """直近N分以内のFXニュースを収集する"""
    articles = []
    cutoff = time.time() - (max_age_minutes * 60)
    
    for feed_url in FX_NEWS_FEEDS:
        feed = feedparser.parse(feed_url)
        for entry in feed.entries:
            pub_time = time.mktime(entry.published_parsed) if hasattr(entry, 'published_parsed') else 0
            if pub_time > cutoff:
                articles.append({
                    "title": entry.get("title", ""),
                    "summary": entry.get("summary", "")[:500],  # 500字制限
                    "published": datetime.fromtimestamp(pub_time).isoformat(),
                    "source": feed_url
                })
    
    return articles

import anthropic
import json

client = anthropic.Anthropic()

def score_news_sentiment(articles: list[dict], target_currency: str = "USDJPY") -> dict:
    """ニュース記事リストを分析してセンチメントスコアを生成する"""
    
    if not articles:
        return {"score": 0, "confidence": "low", "reason": "no_data"}
    
    # 記事をまとめてプロンプトに変換
    news_text = "\n".join([
        f"[{a['published']}] {a['title']}: {a['summary']}"
        for a in articles[:10]  # 最新10件に絞る
    ])
    
    prompt = f"""以下のFXニュース記事を分析し、{target_currency}への影響をスコアリングしてください。

<news_articles>
{news_text}
</news_articles>

分析要件:
- ヘッドラインの煽り文句に惑わされず、実際の政策・経済への影響を評価する
- 複数の記事が矛盾する場合は信頼性の高い情報源を優先する
- 短期（1〜4時間）と中期（1〜3日）を分けて評価する

JSON形式のみで回答:
{{
  "short_term_score": -10から+10（USD側が+）,
  "medium_term_score": -10から+10,
  "confidence": "high/medium/low",
  "key_driver": "最も影響が大きいニュースのタイトル",
  "conflicting_signal": "矛盾するシグナルがあれば記述、なければnull",
  "high_impact_risk": "今後4時間以内の高インパクト予定イベントがあれば記述"
}}"""

    message = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=256,
        thinking={"type": "adaptive"},
        messages=[{"role": "user", "content": prompt}]
    )
    
    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:
        return {"short_term_score": 0, "confidence": "low", "error": "parse_failed"}

def should_trade_based_on_sentiment(
    sentiment: dict,
    signal_direction: str,  # "long" or "short"
    minimum_score: float = 3.0
) -> bool:
    """センチメントスコアに基づいてトレードを承認するか判断する"""
    
    # 低信頼度のスコアはスキップ
    if sentiment.get("confidence") == "low":
        return False
    
    short_score = sentiment.get("short_term_score", 0)
    
    # ロングシグナルの場合：スコアが+3以上
    if signal_direction == "long" and short_score >= minimum_score:
        return True
    
    # ショートシグナルの場合：スコアが-3以下
    if signal_direction == "short" and short_score <= -minimum_score:
        return True
    
    # テクニカルとファンダメンタルズが逆方向 → スキップ
    return False