株式ニュース感情分析をChatGPT APIで実装する方法

※本記事のコードは学習・検証目的です。ニュース感情は遅れて市場に織り込まれることが多く、スコア単体で発注しないでください。APIキーは環境変数で管理します。

先に結論です。ChatGPT APIで株式ニュースを強気・弱気・中立に分類することはできます。そのスコアだけで翌日の値動きを当てることは、ほぼできません。旧記事のようにjson.loadsで自由文をパースし、当日の5日モメンタムと足してBUY/SELLを出す実装は、デモにはなりますが検証にはなりません。

検索意図は「Pythonでニュース感情分析してシグナルにしたい」です。本記事では、(1) Structured Outputsで壊れるJSONを止める、(2) 発表時刻の翌日リターンと突き合わせる、(3) モメンタムは確認用であって予測の主役にしない、の3点をコードで示します。Hugging Faceの分類器比較は本筋ではないので、API実装に絞ります。

📘 外部参考：OpenAI API ドキュメント ／ Structured Outputs ／ Discord Incoming Webhook

感情分析が売買で弱い理由

ニュースは、見出しが強いほど既に織り込まれていることが多いです。「上方修正」がトップに出た時点で、寄り付きはギャップしている。終値時点の感情スコアを、同じ日のリターンに掛けてしまうと、lookaheadになります。Vibe Codingのクロス戦略と同じ穴です（Vibe CodingでBOTを作る方法）。

使い方	妥当か	理由
翌朝のウォッチリストを削る	妥当	人間の読む量を減らすフィルタ
スコア>0.3で即買い	不適	発表済み情報。ギャップを後追いしやすい
弱気ニュースのあとの押し目を機械買い	要検証	仮説にはなる。翌日リターンで期間分割する
モメンタムと平均してcomposite	危険	価格側が強く、感情の寄与が分からなくなる

したがって本記事のシグナルは「発注」ではなくアラートです。BUYは「今日この銘柄の見出しを優先して読む」、HOLDは「スキップ」です。自動発注に接続するなら、別途コスト込みバックテスト（移動平均クロスの検証）が先です。

環境とキー（直書きしない）

pip install openai yfinance pandas python-dotenv requests

# .env（gitignoreする）
OPENAI_API_KEY=sk-...
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...

旧コードのopenai.api_key = "YOUR_OPENAI_API_KEY"は、リポジトリ漏洩の典型です。公式SDKは環境変数OPENAI_API_KEYを自動で読みます。モデルは分類用途ならgpt-4o-miniで十分です。Structured OutputsはGPT-4o系以降で使えます。見出し3分類に上位モデルは不要で、先にコストとレート制限を見てください。

Structured Outputsで感情を固定スキーマにする

自由文JSONは、キー名がscoreになったりsentimentになったりして落ちます。公式のStructured Outputsは、与えたJSON Schemaに出力を合わせます。enumでラベルを固定します。

import os
from typing import Literal
from pydantic import BaseModel, Field
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()  # OPENAI_API_KEY

class NewsSentiment(BaseModel):
    ticker: str = Field(description="推定される銘柄コード。不明なら UNKNOWN")
    label: Literal["bullish", "bearish", "neutral"]
    score: float = Field(description="-1.0(弱気)〜1.0(強気)")
    already_priced_in: bool = Field(description="既に市場に織り込まれていそうなら true")
    reason: str = Field(description="40字以内の根拠")

SYSTEM = """あなたは日本株のヘッドライン分類器です。
投資助言はしない。新規情報か既報の再掲かを already_priced_in で区別する。
スコアは見出しのトーンであり、将来リターンの予測ではない。"""

def classify(headline: str, model: str = "gpt-4o-mini") -> NewsSentiment:
    resp = client.beta.chat.completions.parse(
        model=model,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": headline},
        ],
        response_format=NewsSentiment,
        temperature=0,
    )
    parsed = resp.choices[0].message.parsed
    if parsed is None:
        raise RuntimeError("structured output failed")
    return parsed

client.beta.chat.completions.parseはPydanticモデルをそのまま渡せます。json.loadsは使いません。SDKのバージョンでclient.chat.completions.parseに移っている場合は、そちらに置き換えてください。公式ガイドはモデル名とフィールド名が更新されるので、動かす直前にStructured Outputsを見てください。

バッチでヘッドラインを回す

import pandas as pd

HEADLINES = [
    {"time": "2024-08-01 15:30", "ticker": "7203.T",
     "text": "トヨタ、通期営業益の見通しを上方修正。北米の販売が想定超"},
    {"time": "2024-08-02 08:50", "ticker": "7203.T",
     "text": "トヨタ、認証不正の影響で一部出荷停止が継続との報道"},
    {"time": "2024-08-05 16:00", "ticker": "9984.T",
     "text": "ソフトバンクG、AI投資の評価損が四半期赤字の主因と説明"},
]

def classify_batch(rows: list[dict]) -> pd.DataFrame:
    out = []
    for row in rows:
        s = classify(row["text"])
        out.append({
            "time": row["time"],
            "ticker": row["ticker"],
            "label": s.label,
            "score": s.score,
            "priced_in": s.already_priced_in,
            "reason": s.reason,
        })
    return pd.DataFrame(out)

print(classify_batch(HEADLINES))

本番のニュース取得は、契約した配信（TDnet、NewsAPI、証券会社のRSSなど）を使います。Webページの無断スクレイピングは利用規約とrobots.txtを確認してください。ここでは分類器の検証に集中するため、見出しは手入力です。

翌日リターンと突き合わせる（感情単体の実力）

ここが独自の検証です。見出しのtimeが属する営業日の翌日リターンとスコアの符号が一致したかを数えます。当日リターンを使うと、発表後の値動きを「当たった」と誤認します。

import numpy as np
import yfinance as yf

def next_day_return(ticker: str, ts: str) -> float:
    px = yf.download(ticker, start="2020-01-01", auto_adjust=True, progress=False)
    if isinstance(px.columns, pd.MultiIndex):
        px.columns = px.columns.get_level_values(0)
    close = px["Close"].dropna()
    close.index = pd.to_datetime(close.index).tz_localize(None)
    t = pd.Timestamp(ts)
    # 発表日当日の終値は使わず、当日より後の最初の営業日→その翌日
    after = close[close.index.normalize() > t.normalize()]
    if len(after) < 2:
        return float("nan")
    return float(after.iloc[1] / after.iloc[0] - 1)

def hit_rate(df: pd.DataFrame) -> dict:
    rows = []
    for r in df.itertuples():
        nxt = next_day_return(r.ticker, r.time)
        if np.isnan(nxt) or abs(r.score) < 0.2:
            continue  # 中立は除外
        pred = 1 if r.score > 0 else -1
        real = 1 if nxt > 0 else -1
        rows.append({"hit": int(pred == real), "nxt": nxt, "score": r.score})
    tab = pd.DataFrame(rows)
    if tab.empty:
        return {"n": 0, "hit_rate": None}
    return {"n": int(len(tab)), "hit_rate": round(float(tab["hit"].mean()), 3)}

少数サンプルではhit_rateは安定しません。数十件以上、かつ前半/後半で符号が残るかまで見てください。多くの場合、中立を除いても0.5前後に戻ります。それが「感情分析BOTが勝てない」理由です。スコアはフィルタ、エッジの主張は数字が出てからにします。

already_priced_in を落とす

モデルに「既に織り込み済みか」を聞いても正解ではありません。ただし運用では使えます。trueの見出しはアラート優先度を下げ、false（新規性が高そう）だけをDiscordに流します。人間の読む量を半分にする、というのがこのフラグの仕事です。

def alert_rows(df: pd.DataFrame) -> pd.DataFrame:
    return df[(df["priced_in"] == False) & (df["score"].abs() >= 0.4)]

モメンタムは「確認」にだけ使う

旧記事のcomposite = sentiment * 0.5 + tanh(momentum) * 0.5は、価格が強い日に感情が消えます。混ぜるなら重みを検証し、少なくとも「感情だけ」「価格だけ」「混合」の3列で翌日ヒット率を出してください。

def momentum_5d(ticker: str) -> float:
    h = yf.download(ticker, period="10d", auto_adjust=True, progress=False)
    if isinstance(h.columns, pd.MultiIndex):
        h.columns = h.columns.get_level_values(0)
    c = h["Close"].dropna()
    if len(c) < 6:
        return float("nan")
    return float(c.iloc[-1] / c.iloc[-6] - 1)

def three_way(ticker: str, score: float) -> dict:
    m = momentum_5d(ticker)
    return {
        "sent_only": "BUY" if score > 0.3 else "SELL" if score < -0.3 else "HOLD",
        "px_only": "BUY" if m > 0.02 else "SELL" if m < -0.02 else "HOLD",
        "mix": "BUY" if (0.5 * score + 0.5 * np.tanh(m * 10)) > 0.3 else "HOLD",
    }

3列が全部BUYなら、それは「強いトレンドの日に強気見出しが重なった」だけです。稀な一致を戦略と思わないでください。アラート文には3列を全部載せ、人間が捨てられるようにします。

Discord通知（キーは環境変数）

import requests

def notify(df: pd.DataFrame) -> None:
    url = os.getenv("DISCORD_WEBHOOK_URL")
    if not url:
        print(df.to_string(index=False))
        return
    if df.empty:
        return
    lines = ["【ニュース感情アラート（発注ではない）】"]
    for r in df.itertuples():
        lines.append(f"{r.ticker} {r.label} score={r.score:.2f} {r.reason}")
    requests.post(url, json={"content": "\n".join(lines)}, timeout=10)

Webhookをソースに直書きしないでください。チャンネルは自分だけが見るプライベートにします。BOT名で「推奨」と書かない。コンプライアンス以前に、自分の判断が鈍ります。

毎朝の実行例

if __name__ == "__main__":
    classified = classify_batch(HEADLINES)
    print("hit_rate", hit_rate(classified))
    notify(alert_rows(classified))

Windowsならタスクスケジューラ、VPSならcronです。環境構築はPython環境構築ガイド。ニュース取得と分類と通知を1ファイルに混ぜると、API障害のときに全部止まります。分類関数と通知関数は分けてください。

コストとレート制限

項目	目安	対策
モデル	gpt-4o-mini で見出し1本は安価	本文全文を投げない。見出し+リード2文まで
件数	1銘柄あたり1日数十本で十分	同一文のハッシュで重複排除
temperature	0	分類の再現性。創作させない
失敗	パース失敗・429	リトライは指数バックオフ。落ちたらHOLD

import hashlib, time

def dedupe_key(text: str) -> str:
    return hashlib.sha256(text.strip().encode("utf-8")).hexdigest()[:16]

def classify_retry(text: str, tries: int = 3) -> NewsSentiment:
    last = None
    for i in range(tries):
        try:
            return classify(text)
        except Exception as e:
            last = e
            time.sleep(2 ** i)
    raise RuntimeError(last)

分類器そのものの精度を測る

翌日リターンとのhit_rateが0.5前後でも、それは「相場が当たらない」だけかもしれません。分類器が壊れているのか、分類は正しいがエッジがないのかは、分けて測る必要があります。前者は人手ラベルとの一致率で測ります。

やることは単純です。自分で見出しを30本読み、bullish/bearish/neutralを手で付ける。それとモデル出力を突き合わせます。時間はかかりますが、一度作れば以後モデルを変えるたびに使い回せます。

GOLD = [
    {"text": "トヨタ、通期営業益を上方修正", "gold": "bullish"},
    {"text": "認証不正で一部車種の出荷停止が継続", "gold": "bearish"},
    {"text": "定時株主総会の招集通知を発送", "gold": "neutral"},
    # ... 30本以上ためる
]

def confusion(gold_rows: list[dict], model: str = "gpt-4o-mini") -> pd.DataFrame:
    labels = ["bullish", "bearish", "neutral"]
    mat = pd.DataFrame(0, index=labels, columns=labels, dtype=int)
    mat.index.name = "gold"
    for row in gold_rows:
        pred = classify_retry(row["text"]).label
        mat.loc[row["gold"], pred] += 1
    return mat

def accuracy(mat: pd.DataFrame) -> dict:
    total = int(mat.values.sum())
    correct = int(sum(mat.loc[l, l] for l in mat.index))
    # neutral を bullish/bearish と誤るのが実運用では一番痛い
    false_alarm = int(mat.loc["neutral", "bullish"] + mat.loc["neutral", "bearish"])
    return {
        "n": total,
        "accuracy": round(correct / total, 3) if total else None,
        "neutral_false_alarm": false_alarm,
    }

見るべきは全体正解率よりもneutral_false_alarmです。中立の事務連絡を強気と判定する分類器は、アラートをノイズで埋めます。読む量を減らすために作ったのに、増えます。ここが高いなら、システムプロンプトに「決算・業績・需給に直接触れない発表はneutral」と具体例を足してください。

プロンプトを変えたら必ず再測定する

変更内容	再測定するもの	ありがちな副作用
system文を追記	混同行列・中立誤検知	強気が減り、全部neutralに寄る
モデルを上位に変更	正解率・1件あたりコスト	精度は微増、費用は数倍
スコア閾値を0.3→0.4	アラート件数・翌日hit_rate	件数が減り、n不足で評価不能
本文リードを追加投入	正解率・トークン数	関係ない社名に引っ張られる

閾値をいじって翌日hit_rateが上がったときは、まずサンプル数を見てください。nが10件台なら、それは偶然です。閾値の最適化は、パラメータの過学習と同じ罠です。

結果を貯めないと検証できない

感情分析BOTが続かない一番の理由は、通知して終わりで記録が残らないことです。あとから「あの見出しは当たっていたのか」を調べられません。SQLiteに1行ずつ入れるだけで、3か月後に検証できます。

import sqlite3
from contextlib import closing

DDL = """
CREATE TABLE IF NOT EXISTS news_sentiment (
    key         TEXT PRIMARY KEY,
    published   TEXT NOT NULL,
    ticker      TEXT NOT NULL,
    headline    TEXT NOT NULL,
    label       TEXT NOT NULL,
    score       REAL NOT NULL,
    priced_in   INTEGER NOT NULL,
    model       TEXT NOT NULL,
    created_at  TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

def save(rows: pd.DataFrame, db: str = "news.db", model: str = "gpt-4o-mini") -> int:
    with closing(sqlite3.connect(db)) as conn:
        conn.execute(DDL)
        n = 0
        for r in rows.itertuples():
            try:
                conn.execute(
                    "INSERT INTO news_sentiment"
                    " (key, published, ticker, headline, label, score, priced_in, model)"
                    " VALUES (?,?,?,?,?,?,?,?)",
                    (dedupe_key(r.reason + r.ticker), r.time, r.ticker,
                     r.reason, r.label, float(r.score), int(r.priced_in), model),
                )
                n += 1
            except sqlite3.IntegrityError:
                pass  # 同じ見出しは無視
        conn.commit()
    return n

主キーに重複排除キーを使うので、同じ見出しを二度分類しても課金されるのは初回だけです。model列を持たせておくと、モデルを乗り換えたときに前後比較ができます。

貯めたデータで月次レビューする

def monthly_review(db: str = "news.db") -> pd.DataFrame:
    with closing(sqlite3.connect(db)) as conn:
        df = pd.read_sql_query(
            "SELECT published, ticker, label, score, priced_in FROM news_sentiment",
            conn,
        )
    if df.empty:
        return df
    df["month"] = pd.to_datetime(df["published"]).dt.to_period("M").astype(str)
    return (
        df.groupby(["month", "label"])
          .agg(n=("score", "size"), mean_score=("score", "mean"))
          .reset_index()
    )

月次で強気の本数が急増しているのに株価が動いていないなら、分類器が甘くなっているか、相場が強気材料を無視しています。どちらでも、アラートの閾値を上げる根拠になります。判断の根拠を数字で持つ、というのは長期運用の基本で、自作アルゴの長期運用ロードマップと同じ考え方です。

やってはいけない実装

• 当日終値リターンに当日スコアを掛ける（リーク）
• 英語の金融BERTを日本語見出しにそのまま当てる（トークナイザ不一致）
• 「勝てる銘柄をJSONで20個」と頼む（幻覚。存在しないティッカーが出る）
• スクレイピング結果を無制限にAPIへ流す（コストと規約）
• composite閾値0.3を最適化して過去にフィットさせる（閾値の過学習）

機械学習で騰落を当てたい場合は、感情APIとは別記事の問題です。ラベルは翌日騰落、特徴量は価格と出来高を先に置き、ニュースは後から1列足す、の順が検証しやすいです。先にニュースを入れると、価格だけで説明できる部分まで感情の手柄に見えます。

もう一つ、見落とされがちなのが時刻です。15時の大引け後に出た開示と、翌朝8時50分の報道では、反映される営業日が違います。publishedを保存せずラベルだけ貯めると、あとから営業日を割り当て直せません。時刻は必ず生の文字列で残してください。

まとめ：分類器は作れる。エッジは別問題

ChatGPT APIとStructured Outputsで、見出しをbullish/bearish/neutralに固定できます。旧実装の自由JSONとキー直書きは捨ててください。シグナルとして使うなら、翌日リターンとのhit_rateを先に出し、0.5から離れないなら発注に繋がない。

残る使い道は、読むニュースを減らすアラートです。Discordに流すのは「織り込み済みでない、絶対値0.4以上」だけ。価格モメンタムは参考列であり、感情と平均して隠さない。

本の型で金融指標を固めるなら株分析を学ぶ本3選、ルール売買の検証は移動平均クロスのバックテストへ進んでください。ニュースBOTは、その後の周辺ツールです。