【出来高分析】出来高急増をフックに「注目銘柄」を検知するPythonコード

Python実装・コード

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

Pythonで株価チャートを描画したり、移動平均のクロスを検出したりできるようになった段階の方が次に注目すべき指標が出来高(Volume)です。

出来高は「市場参加者の関心度」を数値で示す唯一の指標です。価格が動く前に出来高が急増するケースは多く、初動を捉えるための最も基本的なフィルターになります。

しかし、数百〜数千の銘柄を毎日目視で確認するのは現実的ではありません。「急増しているかどうか」の判断基準も曖昧なまま、スクリーニングを自動化できずにいる方が大半です。

その原因は、「何日間の平均に対して何倍なら急増とみなすか」という閾値の設計と、複数銘柄を一括処理するコード構成の両方が不足している点にあります。

本記事では、出来高の5日移動平均からの乖離率(Volume Ratio)を計算し、閾値を超えた銘柄を自動検出するPythonコードを提供します。単一銘柄版と複数銘柄スクリーニング版の2本立てで、段階的に実装を進められる構成にしました。

コードはすべてコピペで動きます。閾値や銘柄リストを自分の運用方針に合わせて調整してください。

出来高急増の定義と分析の考え方

出来高乖離率(Volume Ratio)とは

出来高乖離率(Volume Ratio)とは、当日の出来高がN日間の平均出来高に対して何倍かを示す指標です。計算式は以下のとおりです。

出来高乖離率 = 当日出来高 ÷ N日間平均出来高

たとえば、5日平均出来高が100万株の銘柄で当日出来高が300万株なら、乖離率は3.0倍です。この値が大きいほど「普段より注目が集まっている」と解釈できます。

閾値の目安と判断基準

乖離率の閾値は分析の目的によって変わります。一般的な目安を整理します。

乖離率 解釈 用途の例
1.5倍以上 やや活況 広めのスクリーニング
2.0倍以上 明確な急増 標準的な注目銘柄検出
3.0倍以上 異常値レベル 材料出現・大口参入の可能性

本記事のコードでは初期値を2.0倍に設定しています。最初は2.0倍で運用し、検出数が多すぎれば2.5〜3.0倍に引き上げてください。

N日間の期間設定

平均期間のNは、短すぎると直近の急増に引きずられ、長すぎると季節的な出来高変動を拾えません。

* 5日(1週間):直近の相場環境を反映しやすく、短期トレード向き

* 20日(約1か月):中期的な平均との比較で、より「異常」な急増を検出できる

本記事では5日を基本とし、応用版で20日との併用も紹介します。

【コピペOK】出来高急増を検出するPythonコード

まず必要なライブラリをインストールします。


pip install yfinance pandas

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


import sys
from datetime import datetime, timedelta
from typing import Optional

import yfinance as yf
import pandas as pd

# ==============================
# 設定エリア
# ==============================
TICKER_SYMBOL: str = "7203.T"          # 対象銘柄(Yahoo Finance形式)
TICKER_LABEL: str = "トヨタ自動車(7203)"
VOLUME_MA_PERIOD: int = 5              # 出来高移動平均の期間(日)
VOLUME_SPIKE_THRESHOLD: float = 2.0    # 急増とみなす乖離率の閾値(倍)
LOOKBACK_DAYS: int = 30                # 株価データ取得期間(日)
DISPLAY_LAST_N: int = 10               # コンソールに表示する直近N日分


# ==============================
# データ取得
# ==============================
def fetch_stock_data(ticker: str, days: int) -> pd.DataFrame:
    '株価データをYahoo Financeから取得する'
    end_date: datetime = datetime.now()
    start_date: datetime = end_date - timedelta(days=days)
    df: pd.DataFrame = yf.download(
        ticker,
        start=start_date.strftime("%Y-%m-%d"),
        end=end_date.strftime("%Y-%m-%d"),
        progress=False,
    )
    if df.empty:
        raise ValueError(f"株価データを取得できませんでした: {ticker}")
    return df


# ==============================
# 出来高乖離率の計算
# ==============================
def calculate_volume_ratio(
    df: pd.DataFrame, period: int
) -> pd.DataFrame:
    '出来高のN日移動平均と乖離率を算出する'
    df = df.copy()
    df["Volume_MA"] = df["Volume"].rolling(window=period).mean()
    df["Volume_Ratio"] = df["Volume"] / df["Volume_MA"]
    df.dropna(inplace=True)
    return df


# ==============================
# 急増判定
# ==============================
def detect_volume_spikes(
    df: pd.DataFrame, threshold: float
) -> pd.DataFrame:
    '乖離率が閾値を超えた日を抽出する'
    spikes: pd.DataFrame = df[df["Volume_Ratio"] >= threshold].copy()
    return spikes


# ==============================
# 結果表示
# ==============================
def display_results(
    df: pd.DataFrame,
    spikes: pd.DataFrame,
    label: str,
    period: int,
    threshold: float,
    last_n: int,
) -> None:
    '分析結果をコンソールに表示する'
    print("=" * 60)
    print(f"銘柄          : {label}")
    print(f"移動平均期間  : {period}日")
    print(f"急増閾値      : {threshold}倍")
    print(f"分析対象期間  : {df.index[0].date()} 〜 {df.index[-1].date()}")
    print(f"急増検出日数  : {len(spikes)}日")
    print("=" * 60)

    print(f"n--- 直近{last_n}日の出来高乖離率 ---")
    recent: pd.DataFrame = df.tail(last_n)
    for idx, row in recent.iterrows():
        date_str: str = idx.strftime("%Y-%m-%d")
        volume: int = int(row["Volume"])
        ratio: float = float(row["Volume_Ratio"])
        marker: str = " ★急増" if ratio >= threshold else '
        print(f"  {date_str}  出来高: {volume:>12,}  乖離率: {ratio:.2f}倍{marker}")

    if not spikes.empty:
        print(f"n--- 急増日一覧({threshold}倍以上) ---")
        for idx, row in spikes.iterrows():
            date_str = idx.strftime("%Y-%m-%d")
            close: float = float(row["Close"])
            volume = int(row["Volume"])
            ratio = float(row["Volume_Ratio"])
            print(
                f"  {date_str}  終値: ¥{close:,.1f}  "
                f"出来高: {volume:>12,}  乖離率: {ratio:.2f}倍"
            )


# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    print(f"[INFO] {TICKER_LABEL} のデータを取得中...")
    stock_df: pd.DataFrame = fetch_stock_data(TICKER_SYMBOL, LOOKBACK_DAYS)

    print(f"[INFO] 出来高乖離率を計算中({VOLUME_MA_PERIOD}日平均)...")
    analyzed_df: pd.DataFrame = calculate_volume_ratio(
        stock_df, VOLUME_MA_PERIOD
    )

    spikes_df: pd.DataFrame = detect_volume_spikes(
        analyzed_df, VOLUME_SPIKE_THRESHOLD
    )

    display_results(
        analyzed_df,
        spikes_df,
        TICKER_LABEL,
        VOLUME_MA_PERIOD,
        VOLUME_SPIKE_THRESHOLD,
        DISPLAY_LAST_N,
    )

    if analyzed_df.empty:
        sys.exit(1)

コードの処理フロー解説

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

* ステップ1 データ取得yfinanceで指定銘柄の過去30日分のOHLCVデータをダウンロードする

* ステップ2 乖離率計算:出来高の5日移動平均を算出し、当日出来高を平均で割って乖離率を求める

* ステップ3 急増判定:乖離率が閾値(2.0倍)以上の日を抽出する

* ステップ4 結果表示:直近10日分の乖離率一覧と、急増日の詳細(終値・出来高・乖離率)を表示する

detect_volume_spikes関数に価格変動率(前日比)の条件を追加すれば、「出来高急増かつ株価上昇」というフィルターも実装できます。

【コピペOK】複数銘柄を一括スクリーニングする応用版

複数銘柄スクリーニングコード

実運用では、監視リスト全体から出来高急増銘柄を一括で抽出します。


pip install yfinance pandas

import sys
import time
from datetime import datetime, timedelta
from typing import Optional

import yfinance as yf
import pandas as pd

# ==============================
# 設定エリア
# ==============================
WATCHLIST: list[dict] = [
    {"ticker": "7203.T", "label": "トヨタ自動車"},
    {"ticker": "6758.T", "label": "ソニーG"},
    {"ticker": "9984.T", "label": "ソフトバンクG"},
    {"ticker": "6861.T", "label": "キーエンス"},
    {"ticker": "8306.T", "label": "三菱UFJ"},
    {"ticker": "9432.T", "label": "NTT"},
    {"ticker": "6902.T", "label": "デンソー"},
    {"ticker": "4063.T", "label": "信越化学"},
]
VOLUME_MA_PERIOD: int = 5
VOLUME_SPIKE_THRESHOLD: float = 2.0
LOOKBACK_DAYS: int = 30
REQUEST_INTERVAL_SEC: float = 1.0   # API呼び出し間隔(秒)


# ==============================
# データ取得
# ==============================
def fetch_stock_data(ticker: str, days: int) -> pd.DataFrame:
    '株価データをYahoo Financeから取得する'
    end_date: datetime = datetime.now()
    start_date: datetime = end_date - timedelta(days=days)
    df: pd.DataFrame = yf.download(
        ticker,
        start=start_date.strftime("%Y-%m-%d"),
        end=end_date.strftime("%Y-%m-%d"),
        progress=False,
    )
    return df


# ==============================
# 出来高乖離率の計算
# ==============================
def calculate_volume_ratio(
    df: pd.DataFrame, period: int
) -> pd.DataFrame:
    '出来高のN日移動平均と乖離率を算出する'
    df = df.copy()
    df["Volume_MA"] = df["Volume"].rolling(window=period).mean()
    df["Volume_Ratio"] = df["Volume"] / df["Volume_MA"]
    df.dropna(inplace=True)
    return df


# ==============================
# 直近日の急増判定
# ==============================
def check_latest_spike(
    df: pd.DataFrame, threshold: float
) -> Optional[dict]:
    '直近営業日の出来高乖離率が閾値を超えているか判定する'
    if df.empty:
        return None

    latest = df.iloc[-1]
    ratio: float = float(latest["Volume_Ratio"])

    if ratio < threshold:
        return None

    return {
        "date": str(latest.name.date()),
        "close": float(latest["Close"]),
        "volume": int(latest["Volume"]),
        "volume_ma": int(latest["Volume_MA"]),
        "volume_ratio": ratio,
    }


# ==============================
# スクリーニング実行
# ==============================
def run_screening(
    watchlist: list[dict],
    period: int,
    threshold: float,
    days: int,
    interval: float,
) -> list[dict]:
    'ウォッチリスト全体をスクリーニングする'
    results: list[dict] = []

    for item in watchlist:
        ticker: str = item["ticker"]
        label: str = item["label"]

        try:
            df: pd.DataFrame = fetch_stock_data(ticker, days)
            if df.empty:
                print(f"  [WARN] {label}: データ取得失敗")
                continue

            analyzed: pd.DataFrame = calculate_volume_ratio(df, period)
            spike: Optional[dict] = check_latest_spike(analyzed, threshold)

            if spike is not None:
                spike["ticker"] = ticker
                spike["label"] = label
                results.append(spike)
                print(f"  [HIT]  {label}: {spike['volume_ratio']:.2f}倍")
            else:
                print(f"  [----] {label}: 閾値未満")

        except Exception as e:
            print(f"  [ERROR] {label}: {e}")

        time.sleep(interval)

    return results


# ==============================
# 結果レポート表示
# ==============================
def display_report(
    results: list[dict], period: int, threshold: float
) -> None:
    'スクリーニング結果をレポート形式で表示する'
    print("n" + "=" * 65)
    print(f"  出来高急増スクリーニング結果({period}日平均 × {threshold}倍以上)")
    print("=" * 65)

    if not results:
        print("  該当銘柄なし")
        return

    sorted_results: list[dict] = sorted(
        results, key=lambda x: x["volume_ratio"], reverse=True
    )

    header: str = (
        f"  {'銘柄':<14} {'日付':<12} {'終値':>10} "
        f"{'出来高':>14} {'平均出来高':>14} {'乖離率':>8}"
    )
    print(header)
    print("  " + "-" * 61)

    for r in sorted_results:
        row: str = (
            f"  {r['label']:<12} {r['date']:<12} "
            f"¥{r['close']:>9,.1f} "
            f"{r['volume']:>13,} "
            f"{r['volume_ma']:>13,} "
            f"{r['volume_ratio']:>7.2f}倍"
        )
        print(row)

    print("=" * 65)
    print(f"  検出数: {len(sorted_results)} / {len(WATCHLIST)} 銘柄")


# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
    print(f"[INFO] {len(WATCHLIST)}銘柄のスクリーニングを開始...")
    detected: list[dict] = run_screening(
        WATCHLIST,
        VOLUME_MA_PERIOD,
        VOLUME_SPIKE_THRESHOLD,
        LOOKBACK_DAYS,
        REQUEST_INTERVAL_SEC,
    )

    display_report(detected, VOLUME_MA_PERIOD, VOLUME_SPIKE_THRESHOLD)

コードの処理フロー解説

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

* ステップ1 ウォッチリスト定義WATCHLISTに監視対象の銘柄を辞書形式で列挙する

* ステップ2 一括スクリーニング:各銘柄のデータを取得し、直近営業日の出来高乖離率が閾値を超えているか判定する

* ステップ3 レート制限対策REQUEST_INTERVAL_SEC(1.0秒)の待機を挟み、Yahoo FinanceへのリクエストをAPI制限内に抑える

* ステップ4 レポート出力:検出された銘柄を乖離率の降順でソートし、一覧表形式で表示する

WATCHLISTに銘柄を追加するだけで監視対象を拡張できます。Discord Webhookと組み合わせれば、急増検出時に自動通知を送るシステムも構築できます。

出来高急増シグナルの活用と注意点

出来高急増=買いシグナルではない

出来高の急増は「市場参加者の関心が集まった」ことを示すだけです。上昇を伴う急増もあれば、悪材料による急落時の急増もあります。

出来高急増を検出した後は、必ず以下の情報を確認してください。

* 株価の方向:終値が前日比プラスかマイナスか

* ニュース・IR:決算発表・業務提携・自社株買い等の材料の有無

* 板情報・歩み値:大口の成行注文が入っていないか

出来高急増だけでエントリーすることは危険です。あくまでスクリーニングの第一フィルターとして使い、他の分析と組み合わせてください。

業種・時価総額による出来高特性の違い

銘柄ごとに平常時の出来高は大きく異なります。時価総額が大きいメガキャップ銘柄は1日数千万株の出来高がありますが、小型株は数万株にとどまることもあります。

同じ「2.0倍」でも意味合いは異なります。小型株の2.0倍は個人投資家の集中で発生しやすく、大型株の2.0倍は機関投資家の大口注文を示唆する場合があります。閾値を銘柄群ごとに分けて設定するのも有効な運用方法です。

よくあるエラーと対処法

Volume列が全て0になる

yfinanceで一部の銘柄(特にETFや新興市場銘柄)を取得すると、出来高が0で返るケースがあります。Yahoo Finance側のデータ欠損が原因です。

以下を試してください。

* Yahoo FinanceのWebサイトで該当銘柄の出来高が表示されるか手動で確認する

* LOOKBACK_DAYSを60〜90に広げて取得し、0でない日が含まれるか確認する

* yfinanceを最新版にアップデートする(pip install -U yfinance

ZeroDivisionError: float division by zero

移動平均が0のとき、乖離率の計算でゼロ除算が発生します。出来高が0の日が連続すると移動平均も0になるためです。

calculate_volume_ratio関数内で、Volume_MAが0の行を除外する処理を追加してください。具体的には、df.dropna(inplace=True)の直後にdf = df[df["Volume_MA"] > 0]を1行追加するだけで解決します。

スクリーニング結果が常に0件になる

閾値の設定が高すぎるか、取得期間が短すぎることが原因です。

以下を試してください。

* VOLUME_SPIKE_THRESHOLDを1.5に下げて検出されるか確認する

* LOOKBACK_DAYSを60に広げ、直近だけでなく過去の急増日も含めてテストする

* 流動性の高い銘柄(7203.T、9984.T等)で試し、コードの動作自体に問題がないか切り分ける

まとめ

この記事では、出来高乖離率を用いた急増検出ロジックと、複数銘柄を一括スクリーニングするPythonコードを解説しました。

要点を整理します。

* 出来高乖離率は「当日出来高 ÷ N日平均出来高」で計算し、2.0倍以上を急増の基準とする

* yfinanceで過去30日分のデータを取得すれば、5日移動平均ベースの乖離率を算出できる

* 複数銘柄の一括スクリーニングはWATCHLISTに銘柄を追加するだけで拡張可能

* 出来高急増は「市場の注目」を示す指標であり、買いシグナルと同義ではない

* 閾値は銘柄群の特性(大型・小型)や運用スタイルに合わせて調整する必要がある

次のステップとして、検出結果をDiscord WebhookやCSVファイルに出力する仕組みを追加してください。cronやタスクスケジューラで毎営業日の引け後に自動実行すれば、日次スクリーニングの完全自動化が実現します。

さらに、出来高乖離率と株価変動率(前日比)を組み合わせた複合フィルターを実装すれば、「出来高急増かつ株価上昇」の銘柄だけを抽出する高精度なスクリーニングへ発展させることができます。

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