pip install transformers torch newsapi-python pandas requests vaderSentiment

import os
NEWS_API_KEY = os.environ.get("NEWS_API_KEY")

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import torch.nn.functional as F
from newsapi import NewsApiClient

# FinBERTのロード
MODEL_NAME = "ProsusAI/finbert"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
model.eval()

def analyze_sentiment(text: str) -> dict:
    """テキストのセンチメントスコアを返す"""
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
    with torch.no_grad():
        outputs = model(**inputs)
    scores = F.softmax(outputs.logits, dim=-1).squeeze()
    labels = ["positive", "negative", "neutral"]
    return {label: float(score) for label, score in zip(labels, scores)}

# NewsAPIでFX関連ニュースを取得
newsapi = NewsApiClient(api_key=NEWS_API_KEY)

def fetch_fx_news(query: str = "USD JPY Federal Reserve", page_size: int = 20) -> list:
    response = newsapi.get_everything(
        q=query,
        language="en",
        sort_by="publishedAt",
        page_size=page_size
    )
    return response.get("articles", [])

# センチメントスコアを計算
articles = fetch_fx_news()
results = []
for article in articles:
    title = article.get("title", "")
    sentiment = analyze_sentiment(title)
    results.append({
        "publishedAt": article["publishedAt"],
        "title": title,
        **sentiment
    })

import pandas as pd

def calculate_composite_score(results: list, window_minutes: int = 60) -> float:
    """直近N分のニュースから複合センチメントスコアを計算"""
    df = pd.DataFrame(results)
    df["publishedAt"] = pd.to_datetime(df["publishedAt"])
    df = df.sort_values("publishedAt")

    # 直近window分を抽出
    cutoff = df["publishedAt"].max() - pd.Timedelta(minutes=window_minutes)
    recent = df[df["publishedAt"] >= cutoff]

    if recent.empty:
        return 0.0

    # ポジティブ - ネガティブ の加重平均
    composite = (recent["positive"] - recent["negative"]).mean()
    return composite

def generate_signal(score: float, threshold: float = 0.2) -> str:
    """スコアをBUY/SELL/HOLDシグナルに変換"""
    if score > threshold:
        return "BUY"
    elif score < -threshold:
        return "SELL"
    else:
        return "HOLD"

def backtest_sentiment_signal(
    sentiment_df: pd.DataFrame,
    price_df: pd.DataFrame,
    forward_hours: int = 4
) -> pd.DataFrame:
    """
    センチメントシグナル発生後のforward_hours後のリターンを計算
    sentiment_df: timestamp, signal列を持つDataFrame
    price_df: timestamp, close列を持つDataFrame
    """
    results = []
    for _, row in sentiment_df.iterrows():
        entry_time = row["timestamp"]
        exit_time = entry_time + pd.Timedelta(hours=forward_hours)

        entry_price = price_df[price_df["timestamp"] >= entry_time]["close"].iloc[0]
        exit_prices = price_df[price_df["timestamp"] >= exit_time]["close"]

        if exit_prices.empty:
            continue

        exit_price = exit_prices.iloc[0]
        pnl = (exit_price - entry_price) if row["signal"] == "BUY" else (entry_price - exit_price)
        results.append({"signal": row["signal"], "pnl_pips": pnl * 100})

    return pd.DataFrame(results)