【オシレーター】ストキャスティクスをPythonで計算!売買サインの実装

Python実装・コード

※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。

テクニカル指標の学習を進める中で、ストキャスティクス(Stochastic Oscillator)の実装に挑戦しようとしている方は多いはずです。

ストキャスティクスは、一定期間における価格の相対的な位置を0〜100の範囲で示すオシレーター系指標です。「買われすぎ」「売られすぎ」の判断に加え、%Kと%Dのクロスによる売買シグナルを生成できるため、実践的な価値が高い指標です。

しかし、%Kと%Dの計算式は理解できても、pandasでの実装段階で「ローリング処理の適用順序がわからない」「スローストキャスティクスとの違いが曖昧」といった壁にぶつかるケースが目立ちます。

原因は、ファストストキャスティクスとスローストキャスティクスの計算手順の違い、そしてゴールデンクロス(Golden Cross)・デッドクロス(Dead Cross)の判定ロジックを明確に分離できていないことにあります。

本記事では、ストキャスティクスの%K・%Dをpandasで算出し、買われすぎ・売られすぎゾーンでのクロスシグナルを自動検知するPythonコードを提供します。さらに、移動平均線のゴールデンクロスとの複合フィルタで精度を高める応用コードも紹介します。

コードはすべてコピペで動作します。設定エリアのパラメータを変更するだけで、任意の銘柄・期間に適用できる設計です。

ストキャスティクスの計算理論

%Kと%Dの算出式

ストキャスティクスの核となる2つのラインの計算式は以下のとおりです。

%K(ファストライン)は、直近N日間の値幅に対する現在価格の位置を示します。計算式は(終値 − N日間最安値) ÷ (N日間最高値 − N日間最安値) × 100です。一般的にN=14が使用されます。

%D(スローライン)は、%Kの移動平均です。通常3日間の単純移動平均(SMA:Simple Moving Average)で平滑化します。%Dは%Kよりも滑らかに動くため、クロスシグナルの基準線として機能します。

ファストとスローの違い

ストキャスティクスには2つのバリエーションがあります。

* ファストストキャスティクス:上記の%Kと%Dをそのまま使用する。感度が高い分、だましシグナルが多い

* スローストキャスティクス:ファストの%Dを新たな%Kとし、それをさらに平滑化したものを%Dとする。ノイズが軽減され、実用性が高い

本記事ではスローストキャスティクスを採用します。実務では圧倒的にスローが多く使われており、特に理由がない限りスローを選択してください。

売買シグナルの基本ルール

ストキャスティクスの売買判断には、以下の閾値とクロスを組み合わせます。

条件シグナル意味
%K・%Dが80以上買われすぎゾーン上昇余地が限定的
%K・%Dが20以下売られすぎゾーン下落余地が限定的
売られすぎゾーンで%Kが%Dを上抜け買いシグナル反転上昇の可能性
買われすぎゾーンで%Kが%Dを下抜け売りシグナル反転下落の可能性

ゾーン外でのクロスはノイズが多いため、必ずゾーン内のクロスに限定してフィルタリングしてください。

【コピペOK】ストキャスティクス算出と売買シグナル検知コード

まず必要なライブラリをインストールしてください。


pip install yfinance pandas numpy matplotlib japanize-matplotlib

以下がメインのコードです。


from typing import List, Tuple

import japanize_matplotlib  # noqa: F401
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import yfinance as yf

# ==============================
# 設定エリア
# ==============================
TICKER: str = "7203.T"              # 分析対象の銘柄コード
START_DATE: str = "2023-01-01"      # データ取得開始日
END_DATE: str = "2025-01-01"        # データ取得終了日
STOCH_K_PERIOD: int = 14            # %K算出期間
STOCH_D_PERIOD: int = 3             # %D平滑化期間(ファスト→スロー変換にも使用)
OVERBOUGHT: float = 80.0            # 買われすぎ閾値
OVERSOLD: float = 20.0              # 売られすぎ閾値
SMA_SHORT: int = 25                 # 短期移動平均(複合フィルタ用)
SMA_LONG: int = 75                  # 長期移動平均(複合フィルタ用)

# ==============================
# データ取得関数
# ==============================
def fetch_ohlc(ticker: str, start: str, end: str) -> pd.DataFrame:
    '\"yfinanceからOHLCデータを取得する'\"
    df: pd.DataFrame = yf.download(
        ticker, start=start, end=end, auto_adjust=True, progress=False
    )
    if df.empty:
        raise ValueError(f"データ取得に失敗しました: {ticker}")
    df.columns = [c[0] if isinstance(c, tuple) else c for c in df.columns]
    return df


# ==============================
# ストキャスティクス算出関数
# ==============================
def calc_stochastic(
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    k_period: int,
    d_period: int,
) -> Tuple[pd.Series, pd.Series]:
    '\"スローストキャスティクスの%Kと%Dを算出する'\"
    lowest_low: pd.Series = low.rolling(window=k_period).min()
    highest_high: pd.Series = high.rolling(window=k_period).max()

    fast_k: pd.Series = (
        (close - lowest_low) / (highest_high - lowest_low) * 100
    )

    # スローストキャスティクス: ファスト%Dがスロー%Kになる
    slow_k: pd.Series = fast_k.rolling(window=d_period).mean()
    slow_d: pd.Series = slow_k.rolling(window=d_period).mean()

    return slow_k, slow_d


# ==============================
# 移動平均算出関数
# ==============================
def calc_sma(close: pd.Series, period: int) -> pd.Series:
    '\"単純移動平均を算出する'\"
    return close.rolling(window=period).mean()


# ==============================
# クロスシグナル検知関数
# ==============================
def detect_cross_signals(
    slow_k: pd.Series,
    slow_d: pd.Series,
    overbought: float,
    oversold: float,
) -> pd.DataFrame:
    '\"買われすぎ・売られすぎゾーンでのクロスシグナルを検知する'\"
    signals: List[dict] = []
    k_vals: np.ndarray = slow_k.values
    d_vals: np.ndarray = slow_d.values
    dates = slow_k.index

    for i in range(1, len(k_vals)):
        if np.isnan(k_vals[i]) or np.isnan(d_vals[i]):
            continue
        if np.isnan(k_vals[i - 1]) or np.isnan(d_vals[i - 1]):
            continue

        prev_diff: float = k_vals[i - 1] - d_vals[i - 1]
        curr_diff: float = k_vals[i] - d_vals[i]

        # 売られすぎゾーンでの%K上抜け → 買いシグナル
        if prev_diff <= 0 and curr_diff > 0:
            if k_vals[i] <= oversold or d_vals[i] <= oversold:
                signals.append({
                    "date": dates[i],
                    "type": "buy",
                    "slow_k": round(k_vals[i], 2),
                    "slow_d": round(d_vals[i], 2),
                })

        # 買われすぎゾーンでの%K下抜け → 売りシグナル
        if prev_diff >= 0 and curr_diff < 0:
            if k_vals[i] >= overbought or d_vals[i] >= overbought:
                signals.append({
                    "date": dates[i],
                    "type": "sell",
                    "slow_k": round(k_vals[i], 2),
                    "slow_d": round(d_vals[i], 2),
                })

    return pd.DataFrame(signals)


# ==============================
# 複合フィルタ関数
# ==============================
def apply_sma_filter(
    signal_df: pd.DataFrame,
    close: pd.Series,
    sma_short: pd.Series,
    sma_long: pd.Series,
) -> pd.DataFrame:
    '\"移動平均のトレンド方向でシグナルをフィルタリングする'\"
    if signal_df.empty:
        return signal_df

    filtered: List[dict] = []
    for _, row in signal_df.iterrows():
        date = row["date"]
        if date not in close.index:
            continue
        short_val: float = sma_short.loc[date] if date in sma_short.index else np.nan
        long_val: float = sma_long.loc[date] if date in sma_long.index else np.nan

        if np.isnan(short_val) or np.isnan(long_val):
            continue

        # 買いシグナル: 短期SMA > 長期SMA(上昇トレンド)のみ採用
        if row["type"] == "buy" and short_val > long_val:
            filtered.append(row.to_dict())
        # 売りシグナル: 短期SMA < 長期SMA(下降トレンド)のみ採用
        elif row["type"] == "sell" and short_val < long_val:
            filtered.append(row.to_dict())

    return pd.DataFrame(filtered)


# ==============================
# 可視化関数
# ==============================
def plot_stochastic(
    df: pd.DataFrame,
    slow_k: pd.Series,
    slow_d: pd.Series,
    signal_df: pd.DataFrame,
) -> None:
    '\"価格チャートとストキャスティクスを2段構成で描画する'\"
    fig, (ax1, ax2) = plt.subplots(
        2, 1, figsize=(14, 9), sharex=True,
        gridspec_kw={"height_ratios": [2, 1]},
    )

    # 価格チャート
    ax1.plot(df.index, df["Close"], color="black", linewidth=0.8, label="終値")
    ax1.set_title(f"{TICKER} ストキャスティクス売買シグナル")
    ax1.set_ylabel("価格")

    if not signal_df.empty:
        buys: pd.DataFrame = signal_df[signal_df["type"] == "buy"]
        sells: pd.DataFrame = signal_df[signal_df["type"] == "sell"]
        for _, row in buys.iterrows():
            if row["date"] in df.index:
                ax1.annotate(
                    "▲買", xy=(row["date"], df.loc[row["date"], "Close"]),
                    fontsize=8, color="blue", ha="center",
                )
        for _, row in sells.iterrows():
            if row["date"] in df.index:
                ax1.annotate(
                    "▼売", xy=(row["date"], df.loc[row["date"], "Close"]),
                    fontsize=8, color="red", ha="center",
                )

    ax1.legend(loc="upper left")
    ax1.grid(True, alpha=0.3)

    # ストキャスティクスチャート
    ax2.plot(slow_k.index, slow_k, label="%K(スロー)", linewidth=0.9)
    ax2.plot(slow_d.index, slow_d, label="%D(スロー)", linewidth=0.9)
    ax2.axhline(y=OVERBOUGHT, color="red", linestyle="--", alpha=0.5, label=f"買われすぎ({OVERBOUGHT})")
    ax2.axhline(y=OVERSOLD, color="blue", linestyle="--", alpha=0.5, label=f"売られすぎ({OVERSOLD})")
    ax2.fill_between(slow_k.index, OVERBOUGHT, 100, alpha=0.05, color="red")
    ax2.fill_between(slow_k.index, 0, OVERSOLD, alpha=0.05, color="blue")
    ax2.set_ylabel("ストキャスティクス")
    ax2.set_ylim(-5, 105)
    ax2.legend(loc="upper left", fontsize=8)
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig("stochastic_signals.png", dpi=150)
    plt.show()


# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    # データ取得
    ohlc_df: pd.DataFrame = fetch_ohlc(TICKER, START_DATE, END_DATE)

    # ストキャスティクス算出
    slow_k, slow_d = calc_stochastic(
        ohlc_df["High"], ohlc_df["Low"], ohlc_df["Close"],
        STOCH_K_PERIOD, STOCH_D_PERIOD,
    )
    ohlc_df["slow_k"] = slow_k
    ohlc_df["slow_d"] = slow_d

    # 移動平均算出
    sma_short: pd.Series = calc_sma(ohlc_df["Close"], SMA_SHORT)
    sma_long: pd.Series = calc_sma(ohlc_df["Close"], SMA_LONG)

    # シグナル検知(ゾーン内クロスのみ)
    raw_signals: pd.DataFrame = detect_cross_signals(
        slow_k, slow_d, OVERBOUGHT, OVERSOLD
    )

    # 複合フィルタ適用
    filtered_signals: pd.DataFrame = apply_sma_filter(
        raw_signals, ohlc_df["Close"], sma_short, sma_long
    )

    # 結果出力
    print("=" * 55)
    print("ストキャスティクス シグナル検知結果")
    print("=" * 55)
    print(f"分析期間              : {START_DATE} ~ {END_DATE}")
    print(f"フィルタ前シグナル数  : {len(raw_signals)}件")
    print(f"フィルタ後シグナル数  : {len(filtered_signals)}件")
    print("=" * 55)

    if not raw_signals.empty:
        print("\n【フィルタ前】全シグナル:")
        for _, row in raw_signals.iterrows():
            label: str = "買い" if row["type"] == "buy" else "売り"
            print(f"  [{label}] {row['date'].strftime('%Y-%m-%d')} "
                  f"%K={row['slow_k']} %D={row['slow_d']}")

    if not filtered_signals.empty:
        print("\n【フィルタ後】SMAトレンド一致シグナル:")
        for _, row in filtered_signals.iterrows():
            label = "買い" if row["type"] == "buy" else "売り"
            print(f"  [{label}] {row['date'].strftime('%Y-%m-%d')} "
                  f"%K={row['slow_k']} %D={row['slow_d']}")

    # 可視化(フィルタ前シグナルを表示)
    plot_stochastic(ohlc_df, slow_k, slow_d, raw_signals)

コードの処理フロー解説

上記のコードは、以下の6ステップで構成されています。

* ステップ1(データ取得)fetch_ohlcでyfinanceから高値・安値・終値を含むOHLCデータを取得する

* ステップ2(ストキャスティクス算出)calc_stochasticでファスト%Kを算出し、2回の移動平均でスロー%K・スロー%Dに変換する

* ステップ3(シグナル検知)detect_cross_signalsで前日と当日の%K-%D差分の符号反転を検出し、ゾーン内クロスのみをシグナルとして記録する

* ステップ4(複合フィルタ)apply_sma_filterで25日SMAと75日SMAのトレンド方向と一致するシグナルのみを残す

* ステップ5(結果出力):フィルタ前後のシグナル数と詳細をコンソールに表示する

* ステップ6(可視化):価格チャートとストキャスティクスの2段構成で、シグナル位置を矢印で描画する

STOCH_K_PERIOD9に変更すると短期トレード向けの感度が得られます。スイングトレードには14、中期には21を試してください。

シグナル精度を高めるための実践知識

ゾーン閾値の調整指針

デフォルトの80/20は汎用的な閾値ですが、銘柄のボラティリティによって最適値は異なります。

ボラティリティの高い銘柄(新興株・小型株)では、ストキャスティクスが頻繁に80や20に到達するため、85/15に広げるとだましが減少します。逆に、値動きが穏やかな大型株では75/25に狭めることで、シグナルの検出頻度を確保できます。

実際のデータでフィルタ前のシグナル数を確認し、分析期間に対して多すぎる(月10回以上)場合はゾーンを広げてください。

SMAフィルタが有効な理由

ストキャスティクス単体のシグナルは「逆張り」の性質を持ちます。強いトレンドが継続している局面では、買われすぎゾーンに張り付いたまま上昇が続くケースが頻発し、売りシグナルがだましになります。

SMAのゴールデンクロス(短期SMA>長期SMA)で上昇トレンドを確認し、その方向と一致する買いシグナルのみを採用することで、トレンドに逆らうエントリーを排除できます。フィルタ前後のシグナル数を比較し、半分以下に絞れていれば適切に機能しています。

【コピペOK】シグナル勝率の簡易バックテスト追加

検知したシグナルの有効性を定量的に検証する追加コードです。メインコードのif __name__ == "__main__":ブロック末尾に追加してください。


    # ==============================
    # シグナル勝率バックテスト(追加分析)
    # ==============================
    HOLD_DAYS: List[int] = [5, 10, 20]

    def backtest_signals(
        signal_df: pd.DataFrame, close: pd.Series, hold_days: List[int]
    ) -> None:
        '\"シグナル発生後のリターンを集計する'\"
        if signal_df.empty:
            print("\nシグナルが0件のためバックテストをスキップします。")
            return

        for sig_type in ["buy", "sell"]:
            subset: pd.DataFrame = signal_df[signal_df["type"] == sig_type]
            if subset.empty:
                continue

            print(f"\n--- {('買い' if sig_type == 'buy' else '売り')}シグナル バックテスト ---")
            close_idx: pd.Index = close.index

            for hd in hold_days:
                returns: List[float] = []
                for _, row in subset.iterrows():
                    date = row["date"]
                    if date not in close_idx:
                        continue
                    pos: int = close_idx.get_loc(date)
                    if pos + hd >= len(close):
                        continue
                    entry_price: float = close.iloc[pos]
                    exit_price: float = close.iloc[pos + hd]
                    ret: float = (exit_price - entry_price) / entry_price * 100
                    if sig_type == "sell":
                        ret = -ret
                    returns.append(ret)

                if returns:
                    wins: int = sum(1 for r in returns if r > 0)
                    avg_ret: float = np.mean(returns)
                    print(f"  {hd}日保有: 勝率 {wins}/{len(returns)} "
                          f"({wins / len(returns) * 100:.1f}%) "
                          f"平均リターン {avg_ret:.2f}%")

    print("\n【フィルタ前シグナル】")
    backtest_signals(raw_signals, ohlc_df["Close"], HOLD_DAYS)

    print("\n【フィルタ後シグナル】")
    backtest_signals(filtered_signals, ohlc_df["Close"], HOLD_DAYS)

フィルタ前後の勝率を比較し、フィルタ後の勝率が55%以上に改善されていれば、SMAフィルタが有効に機能していると判断できます。勝率に変化がない場合はSMAの期間を調整するか、別のフィルタ(出来高・ATRなど)を検討してください。

よくあるエラーと対処法

%Kと%Dの値が常に0または100に張り付く

STOCH_K_PERIODに対してデータ数が不足している場合、ローリング計算の初期値が極端になります。また、出来高が極端に少ない銘柄では高値と安値が同値になり、ゼロ除算が発生して異常値になります。

以下を試してください。

* START_DATEを早めに設定し、分析開始前に十分なウォームアップ期間(最低50営業日)を確保する

* highest_high - lowest_lowがゼロの場合にNaNを返すガード処理をcalc_stochastic内に追加する

* 出来高が日常的に1,000株未満の銘柄は分析対象から除外する

シグナルが多すぎてチャートが読めない

OVERBOUGHTOVERSOLDの閾値が緩すぎるか、ボラティリティの高い銘柄でゾーンの出入りが頻繁に発生していることが原因です。

以下を試してください。

* OVERBOUGHT85OVERSOLD15に変更して再実行する

* STOCH_K_PERIOD21に引き上げて感度を下げる

* SMAフィルタを適用したfiltered_signalsのみをplot_stochasticに渡すよう変更する

FutureWarningがpandasで大量に表示される

pandas 2.x以降で一部のローリング操作に対して表示される警告です。動作には影響しませんが、出力が見づらくなります。

コードの冒頭(import直後)に以下を追加してください。


import warnings
warnings.simplefilter("ignore", FutureWarning)

根本的な解決にはpandasのバージョンに合わせた記法への修正が必要ですが、本コードの範囲では上記の対処で問題ありません。

まとめ

この記事では、ストキャスティクスの%K・%Dをpandasで算出し、買われすぎ・売られすぎゾーンでのクロスシグナルを自動検知する方法を解説しました。

要点を整理します。

* スローストキャスティクスはファスト%Kを2回平滑化した指標であり、ノイズが少なく実用的

* 売買シグナルはゾーン内(80以上・20以下)のクロスに限定し、ゾーン外のクロスは無視する

* SMA(25日・75日)のトレンド方向と一致するシグナルのみを採用する複合フィルタが有効

* STOCH_K_PERIODは日足で9(短期)〜21(中期)の範囲で銘柄特性に合わせて調整する

* フィルタ後の勝率が55%以上かどうかを簡易バックテストで必ず検証する

次のステップとして、RSI(Relative Strength Index:相対力指数)やMACD(Moving Average Convergence Divergence)とストキャスティクスを組み合わせた複合シグナルシステムの構築に挑戦してみてください。

detect_cross_signals関数と同じインターフェースでRSIやMACDのシグナル検知関数を作成し、複数の関数が同日にシグナルを出した場合のみエントリーする設計にすれば、単一指標では得られない精度の向上が期待できます。

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