@rS_alonewolf

Python×株式投資｜仕事終わりでも投資を諦めない。スクリーニング結果を自動通知するBotを作る（中編）

Last updated at 2025-07-05Posted at 2025-07-05

素人が生成AI無料期間中に作る！毎日自動で銘柄スクリーニング＆X自動通知Bot

これまでの経緯

本記事は、Pythonによる株式スクリーニング自動化・実践の続編です。これまでの背景や検証の流れは、以下の記事をご確認ください。

現在構築中のスクリーニングモデルの全体像と今回やること

はじめに

　毎日自動で銘柄スクリーニングの結果を知れたら、仕事が終わった後の疲れたの脳でも、めんどくさがらずに、お布団に吸引されることもなく、定期的に投資が続けられる。→自動かつ通知機能を持つスクリーニングシステム（bot）を構築したい。この野望を叶えるために下記のようなこと行ってきました。

以前はyfinance都度DL＆ループで地獄のバックテスト3日間
ローカルParquetデータセット構築で高速化のベースを構築
バックテストは高速化できたが、毎日運用には「最新データセットの再構築」が必要
生成AI（Google Gemini, Gemini CLI、Cursor）を無料期間の7/14までに活用し、毎日運用用スクリーニングロジックを生成・統合する方針を決めた。
その結果、データセット構築を当日の最初に行って、構築されたファイルを爆速選抜コードに読ませることで、毎日最新データでスクリーニング→CSV出力が自動化できるようになった（地味なのであえて記事にはしない）。

本記事では、スクリーニング結果（CSV）をもとに、画像を生成し、X（Twitter）APIを使って自動投稿するまでの流れを紹介します。

作りたいBotの概略図：

この記事で紹介すること

これまで作ってきた銘柄スクリーニング系が出力するスクリーニング結果CSVをグラフ化し、Xbot（X API）で自動投稿する部分の実装例を紹介。
スクリーニングロジックやデータ取得の詳細は過去記事を参考にしてください。省略している部分などもございますので、筆者のXなどでご質問お受けします。

コードの概説

コード全体の流れ

今回作成したのは下記のような流れのコードです。

スクリーニング結果csv取得・作図用のテクニカル指標計算
相場の状況に応じて、適した選抜パターンで選ばれた銘柄を抽出
作図・作表
X（Twitter）自動投稿

1. データ取得・テクニカル指標計算の関数

これまで構築したスクリーニング系は、

東証プライム1600銘柄のうち、
1.一定の財務基準を満たした銘柄（割安で業績が健全）をまず選抜
2. 移動平均線の傾きと乖離でゴールデンクロスを狙い、さらに、RSIで買われすぎではない銘柄を選抜
3. ただし、相場状況（マクロ）に合わて判定基準を変える

このように、ファンダ・テクニカル・マクロの3要素を踏まえた抽出を行なっています。

本記事のコードでは、このスクリーニングの後に出力されるcsvをまず読み込み、
作図・作表用にyfinance等で株価データを取得し、RSIや移動平均などのテクニカル指標を計算します。（この部分は一般的なので詳細は割愛。）

スクリーニングコードが出力するcsvはこのような状態でデータが保存されてます。

2. 地合いスコア判定＆パターン分岐の関数

地合いスコア（相場の状況）に応じて、最適なスクリーニングパターンを選択し、適したパターンで選ばれた銘柄を抽出するために行います。

このスクリーニングパターンとは、移動平均の乖離度合いを2-8%の範囲で振り分けたものです。

これまでの検討で、わかっていることは次のとおりです。
相場が下落し切った時→移動平均の乖離が大きめだと１ヶ月後のリターンが大きい傾向
相場が回復中の時は→乖離が小さい方が良い
相場が加熱し切った時→適したパターンがない

スクリーニング結果は全パターン出力されるので、Xbotでの通知の前に最適なものを選択する必要があります。今回は、過去のバックテストの結果をもとに下記のようにパターンを選んでいます。

地合いスコアごとのパターン選択

-3：pattern2
-2：pattern4（選抜銘柄数0ならpattern9）
-1：pattern5
0：pattern1（選抜銘柄数0ならpattern5）
1：pattern5
2：pattern5
3：選定なし(下落する可能性が高い)

判定の参考にしているデータ（過去記事より引用）

平均リターンをヒートマップで可視化しています。
★マークは、「シャープレシオが0.8以上かつ選抜銘柄数が20未満」の条件に付与しており、リスクが低く、実際の投資や機械学習に用いる候補として有望なものです。

def select_stocks(df, jiai_score):
    print(f"地合いスコア: {jiai_score} に基づいて銘柄を選定します。")
    if jiai_score == -3:
        selected_df = df[df['Pattern'] == 'pattern2']
    elif jiai_score == -2:
        selected_df = df[df['Pattern'] == 'pattern4']
        if selected_df.empty:
            print("パターン4の銘柄が見つからないため、パターン9を選びます。")
            selected_df = df[df['Pattern'] == 'pattern9']
    elif jiai_score in [-1, 1, 2]:
        selected_df = df[df['Pattern'] == 'pattern5']
    elif jiai_score == 0:
        selected_df = df[df['Pattern'] == 'pattern1']
        if selected_df.empty:
            print("パターン1の銘柄が見つからないため、パターン5を選びます。")
            selected_df = df[df['Pattern'] == 'pattern5']
    elif jiai_score == 3:
        print("地合いスコア3はリスクが高いため、銘柄は選定しません。")
        return pd.DataFrame()
    else:
        print("該当する地合いスコアではありません。")
        return pd.DataFrame()
    return selected_df

先ほどのcsvのpatternカラムがその日の相場条件を満たす銘柄をdfに入れていく仕様です。

3. 補助関数（企業名・コメント取得など）

get_company_name：証券コードから企業名を取得
get_jiai_message：地合いスコアに応じたコメントを生成

（このあたりは表や画像に情報を載せるための補助的な役割です）

4. 作図・作表の自動化の関数

＜表の作成＞

選定銘柄リストを表として出力し、それを画像化、地合いスコアやコメントも自動で埋め込みます。
matplotlibで作表・PILでコメント含めて画像化してます。

def create_summary_image(date_str, jiai_score, stocks_df):
    # 財務データテーブルの行数に応じてfigsize高さを自動調整
    num_rows = len(stocks_df)
    base_height = 3.5  # ヘッダーや余白用
    row_height = 0.5   # 1行あたりの高さ
    fig_height = base_height + num_rows * row_height
    fig, axes = plt.subplots(3, 1, figsize=(12, fig_height), gridspec_kw={'height_ratios': [1, 3, 1]})
    for ax in axes:
        ax.axis('off')
    jiai_message = get_jiai_message(jiai_score)
    ax_header = axes[0]
    header_text = (
        f"■ {date_str} のスクリーニング結果\n"
        f"■ 地合いスコア: {jiai_score}\n"
        f"■ 地合いコメント:{jiai_message}\n"
         "■ 本日の抽出銘柄とその財務指標↓"
    )
    ax_header.text(0.05, 0.95, header_text, transform=ax_header.transAxes,
                   fontsize=14, verticalalignment='top', horizontalalignment='left', linespacing=1.8)
    # 会社名を取得して追加
    all_stocks_df = stocks_df.copy()
    all_stocks_df['会社名'] = all_stocks_df['Ticker'].apply(lambda x: get_company_name(str(x)))
    for col in ['Close', 'PER', 'ROE(%)', 'OPM(%)']:
        if col in all_stocks_df.columns:
            all_stocks_df[col] = all_stocks_df[col].apply(lambda v: f"{v:.2f}" if pd.notna(v) else v)
    fin_data = all_stocks_df[['会社名', 'Ticker', 'Close', 'PER', 'ROE(%)', 'OPM(%)']]
    fin_data.rename(columns={
        '会社名': '会社名',
        'Ticker': 'ティッカー',
        'Close': '終値',
        'PER': 'PER',
        'ROE(%)': 'ROE(%)',
        'OPM(%)': 'OPM(%)',
    }, inplace=True)
    fin_table = ax_fin.table(cellText=fin_data.values, colLabels=fin_data.columns,
                             loc='center', cellLoc='center', colColours=['#f2f2f2']*len(fin_data.columns),
                             colWidths=[0.35] + [0.11]*(len(fin_data.columns)-1))
    fin_table.auto_set_font_size(True)
    fin_table.set_fontsize(11)
    fin_table.scale(1, 1.8)
    # 罫線カスタマイズ・レイアウト調整は省略（詳細は全体コード参照）
    fig.tight_layout(pad=3.0)
    output_dir = "summary_reports"
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    file_path = os.path.join(output_dir, f"summary_{date_str}_jiai{jiai_score}.jpeg")
    fig.savefig(file_path, format='jpeg', dpi=150, bbox_inches='tight')
    plt.close(fig)
    print(f"\nサマリー画像を保存しました: {file_path}")
    return file_path

出力される表が埋め込まれた画像

<チャートの作成>
mplfinance・yfinance・PILを使い、各銘柄のチャート画像を自動生成・連結します。

def create_technical_charts(stocks_df, output_dir, date_str):
    import yfinance as yf
    import mplfinance as mpf
    from PIL import Image
    import io
    import os

    images = []
    for _, row in stocks_df.iterrows():
        ticker = row['Ticker']
        yf_code = f"{str(ticker)}.T"
        df = yf.download(yf_code, period='130d', progress=False, auto_adjust=False)
        if df is None or not isinstance(df, pd.DataFrame):
            continue
        if not all(col in df.columns for col in ["Open", "High", "Low", "Close", "Volume"]):
            continue
        try:
            df = df[["Open", "High", "Low", "Close", "Volume"]].astype(float).dropna()
            close_series = df["Close"]
            df["MA5"] = close_series.rolling(5).mean()
            df["MA25"] = close_series.rolling(25).mean()
            df["MA75"] = close_series.rolling(75).mean()
            if len(df) >= 60:
                xlim = (df.index[-60], df.index[-1])
            else:
                xlim = (df.index[0], df.index[-1])
            fig, _ = mpf.plot(
                df,
                type="candle",
                mav=(5, 25, 75),
                style='yahoo',
                volume=False,
                title=f"{ticker}",
                figscale=1.0,
                returnfig=True,
                xlim=xlim
            )
            buf = io.BytesIO()
            fig.savefig(buf, format='jpeg', dpi=150, bbox_inches='tight')
            buf.seek(0)
            img = Image.open(buf)
            images.append(img)
            plt.close(fig)
        except Exception as e:
            print(f"[WARN] {yf_code} チャート描画エラー: {e}")
            continue

画像連結・保存処理

    # 画像を縦に連結
    if images:
        widths, heights = zip(*(i.size for i in images))
        total_height = sum(heights)
        max_width = max(widths)
        combined = Image.new('RGB', (max_width, total_height), (255,255,255))
        y_offset = 0
        for im in images:
            combined.paste(im, (0, y_offset))
            y_offset += im.size[1]
        out_path = os.path.join(output_dir, f"tech_charts_{date_str}.jpeg")
        combined.save(out_path, format='JPEG')
        print(f"テクニカル指標グラフ画像を保存しました: {out_path}")
    else:
        print("[WARN] 画像が生成されませんでした")
        out_path = None
    return out_path

出力される各銘柄のテクニカルチャート

5. X（Twitter）API認証・画像付き投稿の関数

.envでAPIキーを安全管理し、tweepy v1.1/v2のハイブリッドで画像付きの投稿。

画像はv1.1でアップロード、ツイートはv2で投稿
エラー時はprintで内容を出力

def post_tweet(text, image_paths=None):
    try:
        # Tweepy v2 (Client) を使用した認証
        client = tweepy.Client(
            consumer_key=API_KEY,
            consumer_secret=API_KEY_SECRET,
            access_token=ACCESS_TOKEN,
            access_token_secret=ACCESS_TOKEN_SECRET
        )

        media_ids = []
        if image_paths:
            # 画像をアップロードするためのAPI v1.1クライアント
            auth = tweepy.OAuth1UserHandler(API_KEY, API_KEY_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
            api_v1 = tweepy.API(auth)
            for image_path in image_paths:
                print(f"Uploading image from: {image_path}")
                media = api_v1.media_upload(image_path)
                media_ids.append(media.media_id)
                print(f"Image uploaded with media_id: {media.media_id}")

        # ツイートを投稿 (media_idsがあれば画像も添付)
        response = client.create_tweet(text=text, media_ids=media_ids if media_ids else None)
        print(f"Tweet posted successfully!")
        print(f"Response: {response}")

    except Exception as e:
        print(f"Error posting tweet: {e}")

6. メイン処理部

全体をまとめて実行。
CSV読み込み
↓
地合いスコア算出
↓
銘柄選定（パターン分岐）
↓
銘柄数制限（最大5件）
↓
サマリー画像とチャート画像生成
↓
X投稿（画像2枚添付）

if __name__ == '__main__':
    import pandas as pd
    import datetime

    # 1. スクリーニング結果CSVを読み込み
    screener_df = pd.read_csv("日々のスクリーニニングで出力される.csv")

    # 2. 地合いスコアを算出
    if 'score_today' in screener_df.columns:
        JIAI_SCORE = screener_df['score_today'].mean()
    else:
        JIAI_SCORE = 0
    print(f"地合いスコア: {JIAI_SCORE} (score_todayカラムの平均値) に基づいて銘柄を選定します。")

    # 3. 地合いスコアごとにパターン分岐し銘柄選定
    selected_stocks = select_stocks(screener_df, JIAI_SCORE)

    # 4. 銘柄を最大5件に絞り込み
    if not selected_stocks.empty:
        if len(selected_stocks) > 5:
            print(f"候補が{len(selected_stocks)}件見つかりました。5銘柄に絞ります。")
            selected_for_chart = selected_stocks.head(5)
        else:
            selected_for_chart = selected_stocks
        # 5. サマリー画像を生成
        today_str = datetime.date.today().strftime('%Y年%m月%d日')
        summary_image_path = create_summary_image(today_str, JIAI_SCORE, selected_stocks)
        # 6. チャート画像を生成
        tech_image_path = create_technical_charts(selected_for_chart, output_dir="summary_reports", date_str=today_str)
        # 7. サマリー画像とチャート画像を組み合わせてツイート
        tweet_text = (
            f"{today_str}の銘柄スクリーニングで抽出された銘柄は画像の通りです。\n"
            "チャートは抽出された銘柄のうち5つ分を表示しています。\n"
            "#株式投資 #テクニカル分析 #チャート分析\n"
            "本情報はあくまで参考情報です。\n"
            "現在テスト運用中。"
        )
        image_paths = [summary_image_path, tech_image_path]
        if image_paths:
            print(f"投稿する画像: {image_paths}")
            post_tweet(tweet_text, image_paths)
        else:
            print("投稿する画像が見つかりませんでした。")
    else:
        print("画像生成の対象となる銘柄はありませんでした。")
    print("\n処理を終了します。")

実際に投稿された様子

まとめ・課題

このように、スクリーニングの結果を画像化して、Xに自動投稿できるベースが整いました。
今後は、Macのcronを使って毎日自動で実行できれば、目標達成となります。次回の記事にしようと思います。

また、今後の改善策として、

MySQL等への日々のデータ蓄積で運用効率化
通知内容のカスタマイズや分析手法の追加
Botの安定運用・エラー監視

などを予定してます。

さらに、今回の自動化は、移動平均とRSIを基準としたシンプルなスクリーニングですが、
現在カーネルPCAなどを用いた、多特徴量のシステムもbot化運用を目指して、改善構築中です。記事にするかどうかは、需要次第です。

参考・関連リンク

その他参考文献

データ取得・API関連

チャート描画・デザイン

X(Twitter)連携・自動化

ご質問などの対応・お礼

V3でバックテストは高速化できたが、毎日運用には「最新データセットの再構築」が必要
生成AI（Google Gemini, Gemini CLI、Cursor）を無料期間に活用し、V3ベースの毎日運用用スクリーニングロジックを生成・統合した。
その結果、データセット構築を当日の最初に行って、構築されたファイルをV3ベースのコードに読ませることで、毎日最新データでスクリーニング→CSV出力が自動化できるようになった（地味なのであえて記事にはしない）。

→ この部分割愛しましたが、ご要望があればお伝えします。
コメントなどありましたら、Xやこの投稿のコメント欄に、何なりとお申し付けください。
素人なもので、10年前のスキルでコードを書いています。
完璧には答えられないかもしれませが、自身のできる最大限のことはさせていただきたいです。

最近ありがたいことに、記事を読んでくださった方が、Xでコメントしてきてくださっています。嬉しくて嬉しくて涙と汗がちょちょ切れます。いつもありがとうございます。
今後とも、ご指導ご鞭撻のほどよろしくお願いします。

補足・全体コード

なぜ「画像付き投稿はv1.1で画像アップロード→v2でツイート」なのか？

問題点

Twitter API v2だけでは画像付きツイートができない（画像ファイルを直接アップロードする機能がない）

各バージョンの特徴

v1.1 API
- 画像（メディア）をアップロードできる（media_uploadメソッド）
- 画像をアップロードすると「media_id」が返る
v2 API
- 新しいAPI。ツイート投稿や取得はできる
- 画像ファイルを直接アップロードする機能は（2024年現在）未提供
- ただし、ツイート投稿時に「media_id」を指定すれば画像付きツイートができる

なぜ組み合わせると画像投稿できるのか

まずv1.1 APIで画像をアップロードし、「media_id」を取得する
次にv2 APIでツイートを投稿する際、その「media_id」を指定する
これにより、画像付きツイートが実現できる

def post_tweet(text, image_paths=None):
    try:
        # Tweepy v2 (Client) を使用した認証
        client = tweepy.Client(
            consumer_key=API_KEY,
            consumer_secret=API_KEY_SECRET,
            access_token=ACCESS_TOKEN,
            access_token_secret=ACCESS_TOKEN_SECRET
        )

        media_ids = []
        if image_paths:
            # 画像をアップロードするためのAPI v1.1クライアント
            auth = tweepy.OAuth1UserHandler(API_KEY, API_KEY_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
            api_v1 = tweepy.API(auth)
            for image_path in image_paths:
                print(f"Uploading image from: {image_path}")
                media = api_v1.media_upload(image_path)
                media_ids.append(media.media_id)
                print(f"Image uploaded with media_id: {media.media_id}")

        # ツイートを投稿 (media_idsがあれば画像も添付)
        response = client.create_tweet(text=text, media_ids=media_ids if media_ids else None)
        print(f"Tweet posted successfully!")
        print(f"Response: {response}")

    except Exception as e:
        print(f"Error posting tweet: {e}")

全体コード

【マシン環境】

今後紹介する過程は、以下の環境で実行しています。

コンポーネント	スペック
マシン	MacBook Pro (14-inch, Nov 2024)
プロセッサ	Apple M4 Pro
メモリ	24 GB
OS	macOS 15.5 Sequoia (24F74)

【Anaconda/condaバージョン】

conda: 24.11.3
Python 3.11.11

【主要ライブラリとバージョン例】

pandas: 2.2.2
yfinance: 0.2.40
matplotlib: 3.8.4
datetime: （標準ライブラリ）
os: （標準ライブラリ）
numpy: 2.2.5
mplfinance: 0.12.10b0
Pillow (PIL): 10.3.0
io: （標準ライブラリ）
glob: （標準ライブラリ）
tweepy: 4.14.0
python-dotenv: 1.1.1

import pandas as pd
import yfinance as yf
import matplotlib.pyplot as plt
import datetime
import os
import numpy as np
import mplfinance as mpf
from PIL import Image
import io
import glob
import tweepy
from dotenv import load_dotenv

# --- Twitter API認証・投稿関数（twitter_bot.pyから移植） ---
# Xbottoken.envファイルから環境変数を読み込み
load_dotenv('Xbottoken.env')

# 環境変数からAPI認証情報を取得
API_KEY = os.getenv('TWITTER_API_KEY')
API_KEY_SECRET = os.getenv('TWITTER_API_SECRET')
ACCESS_TOKEN = os.getenv('TWITTER_ACCESS_TOKEN')
ACCESS_TOKEN_SECRET = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')

# 認証情報の確認
if not all([API_KEY, API_KEY_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET]):
    raise ValueError("環境変数が正しく設定されていません。Xbottoken.envファイルを確認してください。")

def post_tweet(text, image_paths=None):
    """
    指定されたテキストと画像リストでツイートを投稿する関数
    """
    try:
        # Tweepy v2 (Client) を使用した認証
        client = tweepy.Client(
            consumer_key=API_KEY,
            consumer_secret=API_KEY_SECRET,
            access_token=ACCESS_TOKEN,
            access_token_secret=ACCESS_TOKEN_SECRET
        )

        media_ids = []
        if image_paths:
            # 画像をアップロードするためのAPI v1.1クライアント
            auth = tweepy.OAuth1UserHandler(API_KEY, API_KEY_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
            api_v1 = tweepy.API(auth)
            for image_path in image_paths:
                print(f"Uploading image from: {image_path}")
                media = api_v1.media_upload(image_path)
                media_ids.append(media.media_id)
                print(f"Image uploaded with media_id: {media.media_id}")

        # ツイートを投稿 (media_idsがあれば画像も添付)
        response = client.create_tweet(text=text, media_ids=media_ids if media_ids else None)
        print(f"Tweet posted successfully!")
        print(f"Response: {response}")

    except Exception as e:
        print(f"Error posting tweet: {e}")

# --- テクニカル指標自作関数 ---
def calc_rsi(series, period=14):
    delta = series.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    avg_gain = gain.rolling(window=period, min_periods=period).mean()
    avg_loss = loss.rolling(window=period, min_periods=period).mean()
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

def calc_macd(series, fast=12, slow=26, signal=9):
    ema_fast = series.ewm(span=fast, adjust=False).mean()
    ema_slow = series.ewm(span=slow, adjust=False).mean()
    macd = ema_fast - ema_slow
    signal_line = macd.ewm(span=signal, adjust=False).mean()
    macd_hist = macd - signal_line
    return macd, signal_line, macd_hist

def calc_rci(series, period=9):
    rci = []
    for i in range(len(series)):
        window = series[i - period + 1:i + 1]
        if (i < period - 1) or (window.isnull().any()) or (len(window) != period):
            rci.append(np.nan)
        else:
            try:
                rank_price = window.rank(ascending=False)
                rank_time = pd.Series(range(period, 0, -1), index=window.index)
                d = ((rank_price - rank_time).abs()).sum()
                rci_value = (1 - (6 * d) / (period * (period**2 - 1))) * 100
                rci.append(float(rci_value))
            except Exception:
                rci.append(np.nan)
    return pd.Series(rci, index=series.index, dtype=float)

def is_df_empty(df):
    try:
        result = df.empty
        if isinstance(result, bool):
            return result
        return True
    except Exception:
        return True

# Matplotlibのフォント設定（日本語文字化け対策）
plt.rcParams['font.sans-serif'] = ['Hiragino Maru Gothic Pro', 'Yu Gothic', 'MS Gothic', 'sans-serif']
plt.rcParams['axes.unicode_minus'] = False # マイナス記号の文字化け対策

def select_stocks(df, jiai_score):
    """
    地合いスコアに応じて銘柄を選定する関数
    """
    print(f"地合いスコア: {jiai_score} に基づいて銘柄を選定します。")
    if jiai_score == -3:
        selected_df = df[df['Pattern'] == 'pattern2']
    elif jiai_score == -2:
        selected_df = df[df['Pattern'] == 'pattern4']
        if is_df_empty(selected_df):
            print("パターン4の銘柄が見つからないため、パターン9を選びます。")
            selected_df = df[df['Pattern'] == 'pattern9']
    elif jiai_score in [-1, 1, 2]:
        selected_df = df[df['Pattern'] == 'pattern5']
    elif jiai_score == 0:
        selected_df = df[df['Pattern'] == 'pattern1']
        if is_df_empty(selected_df):
            print("パターン1の銘柄が見つからないため、パターン5を選びます。")
            selected_df = df[df['Pattern'] == 'pattern5']
    elif jiai_score == 3:
        print("地合いスコア3はリスクが高いため、銘柄は選定しません。")
        return pd.DataFrame()
    else:
        print("該当する地合いスコアではありません。")
        return pd.DataFrame()
    return selected_df

def get_technical_indicators(stock_code):
    """
    銘柄コードから最新のテクニカル指標を取得する
    """
    yf_code = f"{str(stock_code)}.T"
    end_date = datetime.date.today()
    start_date = end_date - datetime.timedelta(days=90)
    try:
        df = yf.download(yf_code, start=start_date, end=end_date, progress=False, auto_adjust=False)
        if df is None or not isinstance(df, pd.DataFrame):
            print(f"[WARN] {yf_code} DataFrameがNoneまたはDataFrame型でない")
            return {'RSI': 'N/A', 'MACD_Hist': 'N/A', 'RCI': 'N/A'}
        print(f"[DEBUG] {yf_code} columns: {df.columns}")
        print(f"[DEBUG] {yf_code} head:\n{df.head()}")
        if df.empty:
            print(f"[WARN] {yf_code} DataFrameが空")
            return {'RSI': 'N/A', 'MACD_Hist': 'N/A', 'RCI': 'N/A'}
        close = None
        if 'Close' in df.columns:
            close = df['Close']
            if isinstance(close, pd.DataFrame):
                close = close.iloc[:, 0]
        elif 'Adj Close' in df.columns:
            close = df['Adj Close']
            if isinstance(close, pd.DataFrame):
                close = close.iloc[:, 0]
        else:
            print(f"[WARN] {yf_code} 'Close'も'Adj Close'もカラムが無い")
            return {'RSI': 'N/A', 'MACD_Hist': 'N/A', 'RCI': 'N/A'}
        print(f"[DEBUG] {yf_code} Close head:\n{close.head()}")
        rsi = calc_rsi(close, period=14)
        _, _, macd_hist = calc_macd(close, fast=12, slow=26, signal=9)
        rci = calc_rci(close, period=9)
        if isinstance(rsi, pd.Series):
            print(f"[DEBUG] {yf_code} RSI tail:\n{rsi.tail()}")
        if isinstance(macd_hist, pd.Series):
            print(f"[DEBUG] {yf_code} MACD_Hist tail:\n{macd_hist.tail()}")
        if isinstance(rci, pd.Series):
            print(f"[DEBUG] {yf_code} RCI tail:\n{rci.tail()}")
        def get_last_value(x):
            try:
                if isinstance(x, pd.Series):
                    return float(x.iloc[-1])
                elif isinstance(x, (np.ndarray, list)):
                    return float(x[-1])
                else:
                    return float(x)
            except Exception:
                return np.nan
        rsi_val = get_last_value(rsi)
        macd_hist_val = get_last_value(macd_hist)
        rci_val = get_last_value(rci)
        return {
            'RSI': f"{rsi_val:.2f}" if pd.notna(rsi_val) else 'N/A',
            'MACD_Hist': f"{macd_hist_val:.2f}" if pd.notna(macd_hist_val) else 'N/A',
            'RCI': f"{rci_val:.2f}" if pd.notna(rci_val) else 'N/A',
        }
    except Exception as e:
        print(f"[ERROR] get_technical_indicators: {e}")
        return {'RSI': 'N/A', 'MACD_Hist': 'N/A', 'RCI': 'N/A'}

# --- 企業名取得関数 ---
def get_company_name(ticker):
    try:
        info = yf.Ticker(f"{ticker}.T").info
        return info.get("shortName", ticker)
    except Exception:
        return ticker

def get_jiai_message(score):
    score_str = str(int(round(float(score))))
    messages = {
        '-3': "相場は大きく下落済み。一般的に反発しやすい水準です。",
        '-2': "相場は下落傾向。反発も見られる局面です。",
        '-1': "相場はやや下落傾向。",
         '0': "相場は中立的。",
         '1': "相場はやや上昇傾向です。",
         '2': "相場は上昇傾向。調整に注意。",
         '3': "相場は大きく上昇済み。一般的に調整しやすい水準です。",
    }
    return messages.get(score_str, "本情報は参考情報です。投資判断はご自身でお願いします。")

def create_summary_image(date_str, jiai_score, stocks_df):
    """
    選定された銘柄のサマリー画像を生成する
    """
    # 財務データテーブルの行数に応じてfigsize高さを自動調整
    num_rows = len(stocks_df)
    base_height = 3.5  # ヘッダーや余白用
    row_height = 0.5   # 1行あたりの高さ
    fig_height = base_height + num_rows * row_height
    fig, axes = plt.subplots(3, 1, figsize=(12, fig_height), gridspec_kw={'height_ratios': [1, 3, 1]})
    
    # 全ての軸を非表示に
    for ax in axes:
        ax.axis('off')

    # 地合いコメントを先に取得
    jiai_message = get_jiai_message(jiai_score)

    # --- 1. ヘッダー情報 ---
    ax_header = axes[0]
    header_text = (
        f"■ {date_str} のスクリーニング結果\n"
        f"■ 地合いスコア: {jiai_score}\n"
        f"■ 地合いコメント:{jiai_message}\n"
         "■ 本日の抽出銘柄とその財務指標↓"
    )
    ax_header.text(0.05, 0.95, header_text, transform=ax_header.transAxes,
                   fontsize=14, verticalalignment='top', horizontalalignment='left', linespacing=1.8)

    # --- 2. 財務データテーブル ---
    ax_fin = axes[1]
    #ax_fin.set_title("抽出銘柄の財務データ", fontsize=14, loc='left', pad=20)
    # 会社名を取得して追加
    all_stocks_df = stocks_df.copy()  # 全銘柄を表示
    all_stocks_df['会社名'] = all_stocks_df['Ticker'].apply(lambda x: get_company_name(str(x)))
    # 小数点以下2桁に丸める
    for col in ['Close', 'PER', 'ROE(%)', 'OPM(%)']:
        if col in all_stocks_df.columns:
            all_stocks_df[col] = all_stocks_df[col].apply(lambda v: f"{v:.2f}" if pd.notna(v) else v)
    fin_data = all_stocks_df[['会社名', 'Ticker', 'Close', 'PER', 'ROE(%)', 'OPM(%)']]
    fin_data.rename(columns={
        '会社名': '会社名',
        'Ticker': 'ティッカー',
        'Close': '終値',
        'PER': 'PER',
        'ROE(%)': 'ROE(%)',
        'OPM(%)': 'OPM(%)',
    }, inplace=True)
    fin_table = ax_fin.table(cellText=fin_data.values, colLabels=fin_data.columns,
                             loc='center', cellLoc='center', colColours=['#f2f2f2']*len(fin_data.columns),
                             colWidths=[0.35] + [0.11]*(len(fin_data.columns)-1))  # 会社名だけ広く
    fin_table.auto_set_font_size(True)
    fin_table.set_fontsize(11)
    fin_table.scale(1, 1.8)
    # --- 罫線カスタマイズ ---
    # まず全ての罫線を消す
    for key, cell in fin_table.get_celld().items():
        cell.set_linewidth(0)
    nrows = fin_data.shape[0]
    ncols = fin_data.shape[1]
    # 1. ヘッダー上（表の一番上）
    for col in range(ncols):
        cell = fin_table[(0, col)]
        cell.visible_edges = 'T'
        cell.set_linewidth(2.0)
        cell.set_edgecolor('black')
    # 2. ヘッダー下（データ1行目の上側）
    if nrows > 0:
        for col in range(ncols):
            cell = fin_table[(1, col)]
            cell.visible_edges = 'T'
            cell.set_linewidth(2.0)
            cell.set_edgecolor('black')
    # 3. テーブル一番下（最終行の下側）
    for col in range(ncols):
        cell = fin_table[(nrows, col)]
        cell.visible_edges = 'B'
        cell.set_linewidth(2.0)
        cell.set_edgecolor('black')
    # 縦罫線は何もしない（linewidth=0のまま）
    # テーブルの位置調整（中央寄せ＆幅広）
    ax_fin.set_position([0.05, 0.4, 0.9, 0.01])
    ax_fin.axis('off')

    # 財務データテーブルの上に地合いスコアメッセージを表示
    ax_fin.text(0.5, -0.10, "当日の財務指標・RSI・移動平均線の推移が一定の基準を満たす銘柄を抽出しています", fontsize=11, ha='center', va='top', transform=ax_fin.transAxes)
    ax_fin.text(0.5, -0.15, "本情報は参考情報です。投資判断はご自身でお願いします。", fontsize=11, ha='center', va='top', transform=ax_fin.transAxes)
    # --- テクニカル指標テーブルは削除 ---

    # レイアウトと保存
    fig.tight_layout(pad=3.0)
    
    output_dir = "summary_reports"
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
        
    file_path = os.path.join(output_dir, f"summary_{date_str}_jiai{jiai_score}.jpeg")
    fig.savefig(file_path, format='jpeg', dpi=150, bbox_inches='tight')
    plt.close(fig)
    print(f"\nサマリー画像を保存しました: {file_path}")
    return file_path

# --- テクニカル指標推移グラフ作成関数 ---
def create_technical_charts(stocks_df, output_dir, date_str):
    images = []
    for _, row in stocks_df.iterrows():
        ticker = row['Ticker']
        yf_code = f"{str(ticker)}.T"
        # 100営業日分のデータを取得
        df = yf.download(yf_code, period='130d', progress=False, auto_adjust=False)
        if df is None or not isinstance(df, pd.DataFrame):
            print(f"[WARN] {yf_code} DataFrameがNoneまたはDataFrame型でない")
            continue
        if isinstance(df.columns, pd.MultiIndex):
            ticker_full = str(ticker) + '.T'
            if ticker_full in df.columns.levels[1]:
                df = df.xs(ticker_full, axis=1, level=1)
            else:
                print(f"[WARN] {ticker_full} not found in columns")
                continue
        if not all(col in df.columns for col in ["Open", "High", "Low", "Close", "Volume"]):
            print(f"[WARN] {yf_code} 必要なカラムが不足しています: {df.columns}")
            continue
        try:
            df = df[["Open", "High", "Low", "Close", "Volume"]].astype(float).dropna()
            close_series = df["Close"]
            if not isinstance(close_series, pd.Series):
                close_series = pd.Series(close_series)
            # 移動平均線（全期間で計算）
            df["MA5"] = close_series.rolling(5).mean()
            df["MA25"] = close_series.rolling(25).mean()
            df["MA75"] = close_series.rolling(75).mean()
            # RCI 3本
            df["RCI_9"] = calc_rci(close_series, period=9)
            df["RCI_26"] = calc_rci(close_series, period=26)
            df["RCI_52"] = calc_rci(close_series, period=52)
            # MACD
            ema12 = close_series.ewm(span=12, adjust=False).mean()
            ema26 = close_series.ewm(span=26, adjust=False).mean()
            macd = ema12 - ema26
            signal = macd.ewm(span=9, adjust=False).mean()
            hist = macd - signal
            df["MACD"] = macd
            df["Signal"] = signal
            df["Hist"] = hist
            # xlim用に直近60営業日分のインデックス範囲を取得
            if len(df) >= 60:
                xlim = (df.index[-60], df.index[-1])
            else:
                xlim = (df.index[0], df.index[-1])
            addplots = [
                mpf.make_addplot(df["MACD"], panel=1, color="blue", width=0.8, ylabel="MACD"),
                mpf.make_addplot(df["Signal"], panel=1, color="orange", width=0.8),
                mpf.make_addplot(df["Hist"], panel=1, type="bar", color="gray", alpha=0.3),
                mpf.make_addplot(df["RCI_9"], panel=2, color="purple", width=0.8, ylabel="RCI"),
                mpf.make_addplot(df["RCI_26"], panel=2, color="orange", width=0.8),
                mpf.make_addplot(df["RCI_52"], panel=2, color="green", width=0.8),
            ]
            mc = mpf.make_marketcolors(up='tab:red', down='tab:green', edge='inherit', wick='gray')
            s = mpf.make_mpf_style(marketcolors=mc, gridstyle=':', facecolor='#f9f9f9')
            company_name = get_company_name(str(ticker))
            fig, _ = mpf.plot(
                df,
                type="candle",
                mav=(5, 25, 75),
                addplot=addplots,
                style=s,
                volume=False,
                title=f"{company_name} ({ticker})",
                panel_ratios=(3, 2, 2),
                figscale=1.0,
                returnfig=True,
                xlim=xlim
            )
            buf = io.BytesIO()
            fig.savefig(buf, format='jpeg', dpi=150, bbox_inches='tight')
            buf.seek(0)
            img = Image.open(buf)
            images.append(img)
            plt.close(fig)
        except Exception as e:
            print(f"[WARN] {yf_code} テクニカル指標計算・描画エラー: {e}")
            continue
    # 画像を縦に連結
    if images:
        widths, heights = zip(*(i.size for i in images))
        total_height = sum(heights)
        max_width = max(widths)
        combined = Image.new('RGB', (max_width, total_height), (255,255,255))
        y_offset = 0
        for im in images:
            combined.paste(im, (0, y_offset))
            y_offset += im.size[1]
        out_path = os.path.join(output_dir, f"tech_charts_{date_str}.jpeg")
        combined.save(out_path, format='JPEG')
        print(f"テクニカル指標グラフ画像を保存しました: {out_path}")
    else:
        print("[WARN] 画像が生成されませんでした")
    return out_path

# --- メイン処理 ---
if __name__ == '__main__':
    # 1. 地合いスコアをCSVから取得
    screener_df = pd.read_csv("スクリーニングデータのディレクトリ.csv")
    if 'score_today' in screener_df.columns:
        JIAI_SCORE = screener_df['score_today'].mean()
    else:
        JIAI_SCORE = 0
    print(f"地合いスコア: {JIAI_SCORE} (score_todayカラムの平均値) に基づいて銘柄を選定します。")
    # 3. 条件に合う銘柄を選定
    selected_stocks = select_stocks(screener_df, JIAI_SCORE)
    # 4. 銘柄を最大5件に絞り込み
    if not selected_stocks.empty:
        if len(selected_stocks) > 5:
            print(f"候補が{len(selected_stocks)}件見つかりました。5銘柄に絞ります。")
            selected_for_chart = selected_stocks.head(5)
        else:
            selected_for_chart = selected_stocks
        # 5. サマリー画像を生成
        today_str = datetime.date.today().strftime('%Y年%m月%d日')
        summary_image_path = create_summary_image(today_str, JIAI_SCORE, selected_stocks)
        # 6. チャート画像を生成
        tech_image_path = create_technical_charts(selected_for_chart, output_dir="summary_reports", date_str=today_str)
        # 7. サマリー画像とチャート画像を組み合わせてツイート
        tweet_text = (
            f"{today_str}の銘柄スクリーニングで抽出された銘柄は画像の通りです。\n"
            "チャートは抽出された銘柄のうち5つ分を表示しています。\n"
            "#株式投資 #テクニカル分析 #チャート分析\n"
            "本情報はあくまで参考情報です。\n"
            "現在テスト運用中。"
        )
        image_paths = [summary_image_path, tech_image_path]
        if image_paths:
            print(f"投稿する画像: {image_paths}")
            post_tweet(tweet_text, image_paths)
        else:
            print("投稿する画像が見つかりませんでした。")
    else:
        print("画像生成の対象となる銘柄はありませんでした。")
    print("\n処理を終了します。")

運用予定のX Botについて

銘柄スクリーニングbotは、以下の方針で運用を行う予定です。

運用目的

モデル検証: 構築したスクリーニングモデルの精度と有効性を検証
透明性確保: 分析手法と結果を公開し、再現可能性を担保
技術開発: 自動化システムの実装と運用ノウハウの蓄積
自己学習: 将来的な自己資金運用に向けた知識と経験の習得

免責事項

本アカウントで公開する銘柄スクリーニング結果について：

投資助言ではありません: 第三者への売買判断の助言を意図したものではありません
研究目的: 予測モデルの透明性・検証可能性を担保するためのログとして開示
無償提供: 情報の提供に対して金銭的対価を受け取ることは一切ありません
法規制準拠: 金融商品取引法における「投資助言・代理業」には該当しない範囲で運用します

投稿内容

テクニカル指標によるスクリーニング結果（移動平均、RSI等）
財務指標によるスクリーニング結果（PER、PBR、ROE等）
市場全体の動向と指標の変化
スクリーニングシステムの稼働状況
分析手法の改善や技術的な課題の共有

このbotの運用を通じて、技術的な知見の共有と、投資分析の自動化に関する実践的なノウハウを蓄積し、皆様に公開していく予定です。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up