AI×為替ニュース自動分析：テキストマイニングでファンダメンタルズを定量化する

[データ収集層]
   ├── ニュースAPI（Reuters, Bloomberg, Dow Jones等）
   ├── 中央銀行声明・議事録
   └── SNS（X/Twitterの金融アカウント）
         ↓
[前処理層]
   ├── クリーニング（HTML除去・記号処理）
   ├── 言語検出・翻訳（英語記事への統一）
   └── トークン化・ストップワード除去
         ↓
[特徴量抽出層]
   ├── TF-IDF ベクトル化
   ├── Word2Vec / FastText 埋め込み
   └── FinBERT / BERTエンベディング
         ↓
[センチメント分類層]
   ├── FinBERT: 金融特化型分類（Positive / Negative / Neutral）
   └── スコア化（-1 〜 +1の連続値）
         ↓
[シグナル統合層]
   └── センチメントスコア + テクニカル指標 → トレードシグナル

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch

# FinBERTのロード（HuggingFace Hub経由）
model_name = "ProsusAI/finbert"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)

# センチメント分析パイプライン
sentiment_pipeline = pipeline(
    "text-classification",
    model=model,
    tokenizer=tokenizer,
    return_all_scores=True
)

def analyze_sentiment(text: str) -> dict:
    """
    金融テキストのセンチメントスコアを返す。
    出力: {'positive': 0.xx, 'negative': 0.xx, 'neutral': 0.xx}
    """
    results = sentiment_pipeline(text[:512])  # BERTの入力長制限
    scores = {r['label']: r['score'] for r in results[0]}
    
    # 複合スコア（-1〜+1）
    composite = scores['positive'] - scores['negative']
    scores['composite'] = composite
    
    return scores

# 使用例
news_text = "The Federal Reserve kept interest rates unchanged, signaling potential cuts later this year."
result = analyze_sentiment(news_text)
print(f"センチメントスコア: {result['composite']:.3f}")
# スコア +0.7以上: 強気、-0.7以下: 弱気

import feedparser
import pandas as pd
from datetime import datetime

# 主要FXニュースのRSSフィード
RSS_FEEDS = {
    'reuters_fx': 'https://feeds.reuters.com/reuters/businessNews',
    'investing_com': 'https://www.investing.com/rss/news_25.rss',
    'forexfactory': 'https://www.forexfactory.com/rss-news',
}

def fetch_news(feed_url: str, max_items: int = 50) -> list:
    """RSSフィードからニュースを取得"""
    feed = feedparser.parse(feed_url)
    news_items = []
    
    for entry in feed.entries[:max_items]:
        news_items.append({
            'title': entry.get('title', ''),
            'summary': entry.get('summary', ''),
            'published': entry.get('published', ''),
            'link': entry.get('link', '')
        })
    
    return news_items

# 複数フィードから収集
all_news = []
for source, url in RSS_FEEDS.items():
    items = fetch_news(url)
    for item in items:
        item['source'] = source
    all_news.extend(items)

news_df = pd.DataFrame(all_news)

# investing.comなどからスクレイピングした経済指標データ
# （実際の実装はAPIの利用規約を確認すること）

def filter_fx_relevant_news(df: pd.DataFrame, keywords: list) -> pd.DataFrame:
    """為替関連ニュースのフィルタリング"""
    fx_keywords = ['dollar', 'yen', 'euro', 'Fed', 'BOJ', 'ECB', 
                   'interest rate', 'inflation', 'GDP', 'employment',
                   '円', 'ドル', '日銀', 'FRB', '金利', 'インフレ']
    
    pattern = '|'.join(fx_keywords)
    mask = (df['title'].str.contains(pattern, case=False, na=False) |
            df['summary'].str.contains(pattern, case=False, na=False))
    
    return df[mask]

import numpy as np
from scipy.signal import savgol_filter

def calculate_rolling_sentiment(news_df: pd.DataFrame, 
                                  window_hours: int = 24) -> pd.Series:
    """
    過去N時間のニュースのセンチメント平均を計算。
    重み付き平均（直近のニュースを重く）を使用。
    """
    news_df['published_dt'] = pd.to_datetime(news_df['published'])
    news_df = news_df.sort_values('published_dt')
    
    # センチメントスコアを計算（事前にfinBERTで計算済みの前提）
    # news_df['sentiment'] はcompositスコア（-1〜+1）
    
    # 時間加重ローリング平均
    news_df['weight'] = np.exp(
        -0.1 * (news_df['published_dt'].max() - news_df['published_dt']).dt.total_seconds() / 3600
    )
    
    weighted_sentiment = (news_df['sentiment'] * news_df['weight']).sum() / news_df['weight'].sum()
    return weighted_sentiment

def generate_signal(sentiment_score: float, 
                    threshold_long: float = 0.3,
                    threshold_short: float = -0.3) -> str:
    """センチメントスコアからシグナルを生成"""
    if sentiment_score > threshold_long:
        return 'LONG'
    elif sentiment_score < threshold_short:
        return 'SHORT'
    else:
        return 'NEUTRAL'

def combined_signal(sentiment_score: float, 
                    rsi: float, 
                    price_trend: str) -> str:
    """
    センチメント + テクニカルの組み合わせシグナル
    """
    # センチメントが強気かつRSI過売り かつ 上昇トレンド
    if (sentiment_score > 0.3 and rsi < 40 and price_trend == 'UP'):
        return 'STRONG_LONG'
    
    # センチメントが弱気かつRSI過買い かつ 下降トレンド
    elif (sentiment_score < -0.3 and rsi > 60 and price_trend == 'DOWN'):
        return 'STRONG_SHORT'
    
    # シグナルが一致しない
    else:
        return 'WAIT'