AI×自動売買シグナルをXに自動投稿する仕組みを作った【tweepy × Python実装】

X自動投稿とAI自動売買システムの連携イメージ AI×自動売買

はじめに：なぜXにシグナルを自動投稿するのか

AI自動売買システムを運用していると、日々大量のシグナルが生成されます。「今日SPYはBUYだった」「BTCはHOLDが続いている」——そうした記録を残す方法として、最初はDiscordのWebhookを使っていました。

ただDiscordは「身内向け」の印象が強く、外部への情報発信には向いていません。一方でX（旧Twitter）なら、ハッシュタグで検索されやすく、シグナルの実績を公開記録として残せるというメリットがあります。

また、投稿を積み上げることでバックテスト代わりにもなります。「3ヶ月前にBUYシグナルが出ていたあの銘柄、実際どうなったっけ？」を振り返る手段として、Xのタイムラインは非常に使い勝手が良いのです。

今回は、GeminiがニュースからBUY/SELL/HOLDシグナルを生成 → PythonのtweepyでそのままXに自動投稿する仕組みの構築手順を、実装コードつきで解説します。

全体の仕組み

システム全体のフローはシンプルです。

【全体フロー】

  ニュース取得（RSS / NewsAPI）
         ↓
  Gemini AI で分析・スコアリング
         ↓
  BUY / SELL / HOLD シグナル生成
         ↓
  tweepy（X API v2）で自動ポスト
         ↓
  Xのタイムラインにシグナルが流れる ✅

2時間おきにスケジューラーが動き、最新ニュースをGeminiに投げてスコアを取得。そのスコアが閾値を超えたらXに投稿する設計です。HerokuやRailwayのようなクラウド環境で動かすことを想定しています。

X Developer APIキーの取得手順

まずdeveloper.twitter.comにアクセスし、Xアカウントでログインします。

① アプリを作成する

  1. 「Projects & Apps」→「Create App」
  2. アプリ名を入力（例：algo-trading-bot）
  3. 作成完了後、「Keys and Tokens」タブへ

② User authentication settings を「Read and Write」に変更

デフォルトでは「Read only」になっています。これだと投稿（ツイート）ができません。

  1. 「User authentication settings」→「Edit」
  2. 「App permissions」を Read and Write に変更
  3. 「Type of App」は Web App, Automated App or Bot を選択
  4. Callback URLは https://localhost など適当なものを入力
  5. 「Save」をクリック

③ 4つのキーを取得する

「Keys and Tokens」タブから以下の4つをメモします：

  • API Key（Consumer Key）
  • API Key Secret（Consumer Secret）
  • Access Token
  • Access Token Secret

⚠️ よくあるつまずき：権限変更後はAccess Tokenを再生成する必要あり

「Read and Write」に変更する前に生成したAccess Tokenは、旧権限（Read only）のままです。権限変更後に「Regenerate」で必ず再生成してください。これを見落とすと401エラーが出続けます。

Pythonコードの実装

インストール

pip install tweepy python-dotenv

adapters/x_notifier.py

X投稿専用のアダプタークラスを作成します。

import tweepy
import os
from dotenv import load_dotenv

load_dotenv()

class XNotifier:
    def __init__(self):
        api_key = os.getenv("X_API_KEY")
        api_secret = os.getenv("X_API_SECRET")
        access_token = os.getenv("X_ACCESS_TOKEN")
        access_token_secret = os.getenv("X_ACCESS_TOKEN_SECRET")
        if not all([api_key, api_secret, access_token, access_token_secret]):
            self.client = None
            return
        self.client = tweepy.Client(
            consumer_key=api_key,
            consumer_secret=api_secret,
            access_token=access_token,
            access_token_secret=access_token_secret
        )

    def post(self, text: str) -> bool:
        if not self.client:
            print("[X] APIキー未設定 - スキップ")
            return False
        try:
            if len(text) > 140:
                text = text[:137] + "..."
            self.client.create_tweet(text=text)
            print(f"[X] 投稿成功: {text[:50]}...")
            return True
        except Exception as e:
            print(f"[X] 投稿エラー: {e}")
            return False

    def notify_signal(self, symbol: str, decision: str, score: float, reasoning: str):
        emoji = {"BUY": "📈", "SELL": "📉", "HOLD": "⏸"}.get(decision, "❓")
        reason_short = reasoning[:40] + "…" if len(reasoning) > 40 else reasoning
        text = f"{emoji}【{symbol}】{decision} スコア:{score:+.2f}
{reason_short}
#AI自動売買 #アルゴトレード"
        return self.post(text)

ポイントを解説します：

  • tweepy.ClientはAPI v2を使用します（旧tweepy.APIはv1.1）
  • create_tweet()で投稿。シンプルなテキスト投稿であればこれだけでOKです
  • 140字を超える場合は末尾を「…」に切り詰める処理を入れています
  • 環境変数が未設定の場合はself.client = Noneとし、gracefulにスキップします

.envの設定

プロジェクトルートに.envファイルを作成し、以下を記入します：

X_API_KEY=your_api_key
X_API_SECRET=your_api_secret
X_ACCESS_TOKEN=your_access_token
X_ACCESS_TOKEN_SECRET=your_access_token_secret

.envは必ず.gitignoreに追加してください。APIキーをGitHubに上げてしまうと悪用されるリスクがあります。

main_trader.py への組み込み方

既存のメインロジックに組み込む際のコード例です：

from adapters.x_notifier import XNotifier

# 起動時に1度だけインスタンス化
x_notifier = XNotifier()

# シグナル生成後に呼び出す
def run_analysis(symbol: str):
    # ... Gemini分析ロジック ...
    decision = "BUY"       # 例
    score = 0.72           # 例
    reasoning = "米国雇用統計が予想を上回り強気相場の継続が見込まれる"

    # Xに投稿
    x_notifier.notify_signal(
        symbol=symbol,
        decision=decision,
        score=score,
        reasoning=reasoning
    )

投稿される内容の例

実際にXに投稿されるテキストのイメージです：

# BUYシグナルの場合
📈【SPY】BUY スコア:+0.72
米国雇用統計が予想を上回り強気…
#AI自動売買 #アルゴトレード

# SELLシグナルの場合
📉【NVDA】SELL スコア:-0.58
半導体輸出規制の強化懸念が高まり…
#AI自動売買 #アルゴトレード

# HOLDシグナルの場合
⏸【BTC】HOLD スコア:+0.05
方向感に乏しく様子見が続いている…
#AI自動売買 #アルゴトレード

notify_signal()の中でreasoningを40字に切り詰め、ハッシュタグを付加した上で140字制限に収める仕組みになっています。日本語は1文字2バイトですが、Xの文字数カウントは全角・半角ともに1文字＝1カウントなのでシンプルにlen(text) > 140で判定できます。

実際に動かしてみてわかったこと

① 401 Unauthorized エラー

最初にハマったのがこのエラーです。原因は権限変更前のAccess Tokenをそのまま使っていたこと。

解決策：Developer Portalで「Read and Write」に変更したあと、Access Tokenを必ずRegenerateする。これだけで解決します。

② 403 Forbidden（duplicate content）

同じ内容のツイートを連続して投稿しようとすると、Xのスパム防止機能が働いて403エラーになります。

tweepy.errors.Forbidden: 403 Forbidden
You are not allowed to create a Tweet with duplicate content.

これは実は正常動作の証拠です。テスト中に同じシグナルを何度も投げているときに出ます。本番運用では2時間ごとに異なるデータが入ってくるので問題になりません。テスト時はシンボルや日時をテキストに含めて重複を避けましょう。

③ Windowsで絵文字が文字化けする

Windowsのコマンドプロンプトや一部のターミナルでは、絵文字の出力が化けることがあります。

# Pythonスクリプトの先頭に追加
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

またはpython実行時に環境変数を指定します：

set PYTHONIOENCODING=utf-8
python main_trader.py

X側への投稿自体は問題なく行われているので、これはあくまでローカルのコンソール表示の問題です。

まとめ

tweepyを使えば、X APIの認証さえ通ってしまえば数十行のコードで自動投稿が実現できます。今回の仕組みで実現できていること：

  • ✅ 2時間ごとにGeminiがニュースを自動分析
  • ✅ BUY/SELL/HOLDシグナルをXに自動ポスト
  • ✅ 140字制限を自動的に処理
  • ✅ ハッシュタグで検索・集計が可能
  • ✅ シグナルの記録として活用可能（投稿ログ = バックテストデータ）

次のステップとして考えているのは以下です：

  • 📊 バックテスト機能の強化：過去のシグナル投稿を収集して勝率・損益を計算
  • 💹 実口座との接続：Interactive BrokersやalpacaのAPIと連携して、シグナルを実際の注文に変換
  • 📱 Xのリプライ自動処理：フォロワーからの「このシグナルの根拠は？」という質問にGeminiが自動返答

Discordからの移行を検討している方、シグナルの外部公開を考えている方はぜひ試してみてください。コードはGitHubで公開予定です。

タイトルとURLをコピーしました