スポンサーリンク

【python】ストキャスティクスとMACDで反転寸前銘柄を探してみるテスト【逆張り】

python
スポンサーリンク
moomoo証券【WEB】

ストキャスティクスとは

ストキャスティクス(Stochastics)は、「株価の過熱感」を判断するためのテクニカル指標です。

買われすぎ・売られすぎのサインを示すのに使われ、特に短期の反発や調整を読むのに向いています。

以下のような%Kと%Dという値から算出されます。

それぞれ、80以上で「買われすぎ」、20以下で「売られすぎ」と判断されるようです。

ストキャスティクスの具体例

下記は(7013)IHIの日足チャートです。

画像下がストキャスティクスです。

赤線が%K、青線が%Dです。

白丸のように「%Kが10以下になったあと、%Dをクロスしたら」株価上昇になっています。

MACDとは

MACDは、トレンドの強さや転換点を捉えるためのテクニカル指標です。

「短期移動平均」と「長期移動平均」の差を利用して勢いを視覚化します。

MACDの具体例

下記は(7013)IHIの日足チャートです。

画像の下から二段目がMACDです。

緑線がMACDで、「傾きがマイナスからプラスに転じた」ところから株価が上向きです。

銘柄探しプログラムの方針

上記から、「ストキャスティクスの%Kが10以下」かつ、「MACDの傾きが徐々にプラス方向、もしくはマイナスからプラスに転じた」ら、上向きになると仮定しました。

プログラムの方針は下記です。

指標実施事項判定
ストキャスティクス直近営業日の%Kを算出10以下
MACD・3営業日前から直近営業日までのMACDを算出
・各営業日の差分を算出
MACDの傾きがプラスもしくは直近営業日までマイナスが少なくなる傾向

銘柄探しプログラム

作成したプログラムは下記です。

利用される場合、必要なモジュールはインポートください。

また、銘柄名として同じフォルダに「nk.csv」が必要です。

日本取引所(https://www.jpx.co.jp/markets/statistics-equities/misc/01.html)からダウンロードし、csvで保存ください(ETFなどの余計な銘柄もあるので、加工して保存した方が良いです)。

ストキャスティクスはSBI証券のアプリの表示に合うように調整して、直近営業日を含む5日間でカウントしています。

import warnings
warnings.filterwarnings("ignore", category=FutureWarning)

import datetime
import pandas as pd
import numpy as np
import yfinance as yf
from tqdm import tqdm
import time
import os
import shutil

# ==== 設定 ====
CACHE_DIR = "yf_macd_cache"

# ① フォルダが存在する場合は中身を全削除
if os.path.exists(CACHE_DIR):
    print(f"🧹 既存キャッシュフォルダ内のファイルを削除中: {CACHE_DIR}")
    for file in os.listdir(CACHE_DIR):
        path = os.path.join(CACHE_DIR, file)
        try:
            if os.path.isfile(path) or os.path.islink(path):
                os.remove(path)
            elif os.path.isdir(path):
                shutil.rmtree(path)
        except Exception as e:
            print(f"⚠️ {file} の削除に失敗しました: {e}")
else:
    os.makedirs(CACHE_DIR, exist_ok=True)

BATCH_SIZE = 300      # 高速バッチ処理単位
WAIT_TIME = 8         # Yahoo制限回避用のバッチ間ウェイト(秒)
TERM_STC = 9
start = "2022-10-01"
end = datetime.datetime.today().strftime('%Y-%m-%d')

# ==== nk.csv 読み込み ====
df_nk = pd.read_csv("nk.csv", dtype=str)
tickers = [str(c).strip() + ".T" for c in df_nk["コード"].tolist()]
names = df_nk["銘柄名"].tolist()
markets = df_nk["市場・商品区分"].tolist()
total = len(tickers)

results = []

# ==== 関数 ====
def calc_stochastic(df, term=5):
    """SBI仕様に近いストキャスティクス"""
    df = df.copy()
    if len(df) < term:
        return pd.DataFrame()

    # 当日を含む過去term日間でHigh/Lowを算出
    df["Low_prev"] = df["Low"].rolling(window=term, min_periods=term).min()
    df["High_prev"] = df["High"].rolling(window=term, min_periods=term).max()

    # %Kと%D
    df["%K"] = ((df["Close"] - df["Low_prev"]) / (df["High_prev"] - df["Low_prev"])) * 100
    df["%D"] = df["%K"].rolling(3).mean()
    df.dropna(inplace=True)
    return df

def calc_macd(df):
    """MACD計算"""
    short_ema = df["Close"].ewm(span=12, adjust=False).mean()
    long_ema  = df["Close"].ewm(span=26, adjust=False).mean()
    macd = short_ema - long_ema
    signal = macd.ewm(span=9, adjust=False).mean()
    return macd, signal

# ==== メイン処理 ====
for i in range(0, total, BATCH_SIZE):
    batch_tickers = tickers[i:i + BATCH_SIZE]
    batch_names = names[i:i + BATCH_SIZE]
    batch_markets = markets[i:i + BATCH_SIZE]
    batch_id = f"batch_{i // BATCH_SIZE + 1}"
    cache_path = os.path.join(CACHE_DIR, f"{batch_id}.csv")

    # --- キャッシュ利用 ---
    if os.path.exists(cache_path):
        data = pd.read_csv(cache_path, header=[0, 1], index_col=0, parse_dates=True)
        print(f"♻️ キャッシュ利用: {cache_path}")
    else:
        print(f"\n📡 {batch_id}: {len(batch_tickers)}銘柄を取得中...")
        try:
            data = yf.download(batch_tickers, start=start, end=end,
                               auto_adjust=False, group_by="ticker",
                               threads=True, progress=False) #マルチスレッド化
            data.to_csv(cache_path)
            time.sleep(WAIT_TIME)
        except Exception as e:
            print(f"⚠️ バッチ取得エラー: {e}")
            time.sleep(30)
            continue

    # --- 各銘柄処理 ---
    for ticker, name, market in tqdm(list(zip(batch_tickers, batch_names, batch_markets)),
                                     desc=f"処理中 {batch_id}", unit="銘柄"):
        try:
            if ticker not in data.columns.get_level_values(0):
                continue
            df = data[ticker].copy()
            if df.empty:
                continue

            df.sort_index(ascending=True, inplace=True)
            df = df[~df.index.duplicated(keep='last')]

            # 各指標計算
            df = calc_stochastic(df, 5)
            df["MACD"], df["Signal"] = calc_macd(df)
            df["VolMA5"] = df["Volume"].rolling(5).mean()  # 出来高平均

            df.dropna(inplace=True)
            if len(df) < 4:
                continue

            # --- MACD傾き ---
            macd_3d_ago = df["MACD"].iloc[-4]
            macd_2d_ago = df["MACD"].iloc[-3]
            macd_1d_ago = df["MACD"].iloc[-2]
            macd_latest = df["MACD"].iloc[-1]
            diff_3to2 = macd_2d_ago - macd_3d_ago
            diff_2to1 = macd_1d_ago - macd_2d_ago
            diff_1tolatest = macd_latest - macd_1d_ago

            # 下げ→減速→上昇転換
            macd_turn = (diff_2to1 > diff_3to2) and (diff_1tolatest > diff_2to1) and (diff_1tolatest > -2)

            # 最新データ
            latest = df.iloc[-1]
            k, d = latest["%K"], latest["%D"]
            close = latest["Close"]
            vol = latest["Volume"]
            volma5 = df["VolMA5"].iloc[-1]
            ticker = ticker.replace(".T", "")
            name = name.replace("ホールディングス","HD").replace("グループ","G")
            market = market.replace("(内国株式)","")

            # 条件判定
            if (k < 11 and d < 20) and macd_turn:
                results.append([
                    ticker, name, market, close, vol, round(volma5, 0),
                    round(k, 1), round(d, 1),
                    round(macd_latest, 1), round(diff_1tolatest, 1)
                ])

        except Exception as e:
            tqdm.write(f"⚠️ {name} ({ticker}) でエラー: {e}")

# ==== 出力 ====
today_str = datetime.datetime.today().strftime("%Y%m%d")
output_filename = f"{today_str}_screen.csv"

results_df = pd.DataFrame(results,
    columns=["コード", "銘柄名", "市場区分", "直近終値", "出来高", "5日平均出来高",
             "%K", "%D", "MACD", "MACD変化量"]
)
results_df.to_csv(output_filename, index=False, encoding="utf-8-sig")

print(f"\n✅ 完了しました! 抽出銘柄数: {len(results)} 件 -> {output_filename}")

出力例

最近は株価が上がりがちだったのであんまりヒットする銘柄はありませんでした。

表の銘柄をチェックしてみます。

たしかにストキャスティクスもMACDも狙い通りの形をピックアップできました!

ここから企業調査していい感じだったら買おうかなと考えています。

今回は買う銘柄はないかなと思っています。

買う場合、%Kが80を超えるか、50を超えずに折れたら売り時かな。

コメント

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