※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。
寄り付き前の気配値には、その日の市場参加者の熱量が凝縮されています。オーダーインバランスや前日比乖離率を見ることで、寄り付き後の方向感をある程度推定できます。ということで、Pythonで気配値を分析する実装をまとめます。
📘 外部参考：Python 公式サイト（ダウンロード） ／ Python 公式ドキュメント（日本語）
しかし「気配値をどうやってPythonで取得するのか」「リアルタイムデータには高額なAPIが必要なのでは」「取得できたとして何をどう数値化すればいいのか」という疑問で手が止まる人が大半です。
原因は、日本株の板情報（Order Book）をリアルタイムで取得できる無料APIが極めて限られていること、そして気配値の「読み方」と「数値化手法」が体系的にまとめられていないことにあります。
ということで、この記事では、無料で利用可能なデータソースを活用して寄り付き前の気配情報を取得し、買い気配と売り気配の偏り（オーダーインバランス）を数値化するPythonコードを提供します。さらに、前日終値との乖離率から当日の戦略を判断するロジックまで実装します。
コードはすべてコピペで動作する構成です。銘柄コードや閾値を自分のトレードルールに合わせて調整し、朝のルーティンに組み込んでください。
寄り付き前の気配値と市場の「熱量」
気配値が示す市場参加者の意図
気配値とは、取引所のオーダーブック（Order Book：注文板）に蓄積された未約定の指値注文です。東京証券取引所では、8:00から注文受付が開始され、9:00の寄り付き（板寄せ方式）まで約1時間かけて注文が積み上がります。
買い気配が売り気配を大きく上回る状態は、買い需要の超過を意味します。逆に売り気配が優勢な場合は、売り圧力が強いことを示します。
この需給の偏りをオーダーインバランス（Order Imbalance）と呼びます。寄り付き前のインバランスは、当日の初動方向と相関する傾向があります。
気配値の数値化指標
気配値から読み取れる情報を、以下の3つの指標に集約します。
| 指標名 | 計算式 | 意味 |
|---|---|---|
| オーダーインバランス比率 | (買い数量 − 売り数量) ÷ (買い数量 + 売り数量) | 需給の偏り（−1〜+1） |
| 気配スプレッド率 | (最良売り気配 − 最良買い気配) ÷ 前日終値 × 100 | 流動性の目安（%） |
| 前日比乖離率 | (気配中値 − 前日終値) ÷ 前日終値 × 100 | ギャップの大きさ（%） |
オーダーインバランス比率が+0.3以上なら買い優勢、−0.3以下なら売り優勢と判断するのが一般的な目安です。
【コピペOK】気配値取得と分析のメインコード
pip install requests pandas numpy matplotlib japanize-matplotlib yfinance
# ==============================
# 寄り付き前 気配値モニタリング & 分析
# ==============================
import time
import datetime
from typing import Any
import numpy as np
import pandas as pd
import requests
import yfinance as yf
import matplotlib.pyplot as plt
import japanize_matplotlib # noqa: F401
# ==============================
# 設定エリア
# ==============================
TICKER_LIST: list[str] = ["7203", "9984", "6758", "8306", "9432"]
MARKET_SUFFIX: str = ".T" # 東証サフィックス
POLL_INTERVAL_SEC: int = 30 # 監視間隔（秒）
POLL_COUNT: int = 5 # 監視回数（デモ用に少数）
IMBALANCE_THRESHOLD: float = 0.3 # インバランス判定閾値
GAP_THRESHOLD_PCT: float = 1.0 # ギャップ判定閾値（%）
SPREAD_WARNING_PCT: float = 0.5 # スプレッド警告閾値（%）
# ==============================
# 関数定義: 前日終値の取得
# ==============================
def fetch_previous_close(ticker_list: list[str], suffix: str) -> dict[str, float]:
'yfinanceで前日終値を取得する'
result: dict[str, float] = {}
for code in ticker_list:
symbol: str = f"{code}{suffix}"
try:
info: dict = yf.Ticker(symbol).fast_info
prev_close: float = float(info.get("previous_close", 0))
if prev_close > 0:
result[code] = prev_close
except Exception as e:
print(f"前日終値取得エラー ({code}): {e}")
return result
# ==============================
# 関数定義: 気配値のシミュレーション取得
# ==============================
def fetch_quote_data(
ticker_list: list[str],
prev_closes: dict[str, float],
) -> list[dict[str, Any]]:
'気配値データを取得する（デモ用シミュレーション）
【重要】日本株のリアルタイム板情報を無料で取得できるAPIは
限られています。実運用では以下のいずれかを使用してください。
- kabuステーションAPI（auカブコム証券・口座保有者向け・無料）
- J-Quants API（JPX提供・無料プランあり）
- 楽天RSS（楽天証券・Excel/DDE連携）
本関数は動作確認用にランダムデータを生成します。
'
records: list[dict[str, Any]] = []
for code in ticker_list:
prev: float = prev_closes.get(code, 0)
if prev <= 0:
continue
np.random.seed(int(time.time() * 1000) % (2**31) + hash(code) % 1000)
gap_pct: float = np.random.normal(0, 0.8)
mid_price: float = prev * (1 + gap_pct / 100)
spread: float = prev * np.random.uniform(0.001, 0.005)
best_bid: float = round(mid_price - spread / 2, 1)
best_ask: float = round(mid_price + spread / 2, 1)
bid_volume: int = int(np.random.lognormal(10, 1))
ask_volume: int = int(np.random.lognormal(10, 1))
records.append({
"code": code,
"prev_close": prev,
"best_bid": best_bid,
"best_ask": best_ask,
"bid_volume": bid_volume,
"ask_volume": ask_volume,
"mid_price": round((best_bid + best_ask) / 2, 1),
"timestamp": datetime.datetime.now().strftime("%H:%M:%S"),
})
return records
# ==============================
# 関数定義: 指標算出
# ==============================
def calc_indicators(records: list[dict[str, Any]]) -> pd.DataFrame:
'気配値データから分析指標を算出する'
rows: list[dict[str, Any]] = []
for r in records:
total_vol: int = r["bid_volume"] + r["ask_volume"]
imbalance: float = (
(r["bid_volume"] - r["ask_volume"]) / total_vol
if total_vol > 0 else 0.0
)
spread_pct: float = (
(r["best_ask"] - r["best_bid"]) / r["prev_close"] * 100
if r["prev_close"] > 0 else 0.0
)
gap_pct: float = (
(r["mid_price"] - r["prev_close"]) / r["prev_close"] * 100
if r["prev_close"] > 0 else 0.0
)
rows.append({
"銘柄": r["code"],
"前日終値": r["prev_close"],
"買い気配": r["best_bid"],
"売り気配": r["best_ask"],
"買い数量": r["bid_volume"],
"売り数量": r["ask_volume"],
"インバランス": round(imbalance, 4),
"スプレッド率(%)": round(spread_pct, 4),
"乖離率(%)": round(gap_pct, 3),
"時刻": r["timestamp"],
})
return pd.DataFrame(rows)
# ==============================
# 関数定義: 戦略判定
# ==============================
def judge_strategy(
df: pd.DataFrame,
imb_thresh: float,
gap_thresh: float,
spread_warn: float,
) -> pd.DataFrame:
'指標に基づいて当日の戦略方針を判定する'
signals: list[str] = []
for _, row in df.iterrows():
parts: list[str] = []
if row["インバランス"] >= imb_thresh:
parts.append("買い優勢")
elif row["インバランス"] <= -imb_thresh:
parts.append("売り優勢")
else:
parts.append("中立")
if abs(row["乖離率(%)"]) >= gap_thresh:
direction: str = "GU" if row["乖離率(%)"] > 0 else "GD"
parts.append(f"ギャップ{direction}")
if row["スプレッド率(%)"] >= spread_warn:
parts.append("低流動性注意")
signals.append(" / ".join(parts))
df["戦略シグナル"] = signals
return df
# ==============================
# 関数定義: モニタリングループ
# ==============================
def run_monitoring(
ticker_list: list[str],
prev_closes: dict[str, float],
poll_count: int,
interval_sec: int,
imb_thresh: float,
gap_thresh: float,
spread_warn: float,
) -> list[pd.DataFrame]:
'指定回数だけ気配値を取得し分析結果を蓄積する'
history: list[pd.DataFrame] = []
for i in range(poll_count):
print(f"n--- 取得 {i + 1}/{poll_count} ---")
records: list[dict[str, Any]] = fetch_quote_data(ticker_list, prev_closes)
df: pd.DataFrame = calc_indicators(records)
df = judge_strategy(df, imb_thresh, gap_thresh, spread_warn)
print(df[["銘柄", "インバランス", "乖離率(%)", "戦略シグナル", "時刻"]].to_string(index=False))
history.append(df)
if i < poll_count - 1:
time.sleep(interval_sec)
return history
# ==============================
# 関数定義: インバランス推移チャート
# ==============================
def plot_imbalance_history(history: list[pd.DataFrame]) -> None:
'監視期間中のインバランス推移をグラフ表示する'
if not history:
print("描画データがありません。")
return
combined: pd.DataFrame = pd.concat(history, ignore_index=True)
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
for code in combined["銘柄"].unique():
mask = combined["銘柄"] == code
axes[0].plot(
combined.loc[mask, "時刻"].values,
combined.loc[mask, "インバランス"].values,
marker="o", label=code,
)
axes[0].axhline(y=0, color="gray", linestyle="--", alpha=0.5)
axes[0].set_title("オーダーインバランス推移")
axes[0].set_ylabel("インバランス比率")
axes[0].set_xlabel("時刻")
axes[0].legend()
axes[0].grid(True, alpha=0.3)
for code in combined["銘柄"].unique():
mask = combined["銘柄"] == code
axes[1].plot(
combined.loc[mask, "時刻"].values,
combined.loc[mask, "乖離率(%)"].values,
marker="s", label=code,
)
axes[1].axhline(y=0, color="gray", linestyle="--", alpha=0.5)
axes[1].set_title("前日比乖離率推移")
axes[1].set_ylabel("乖離率（%）")
axes[1].set_xlabel("時刻")
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# ==============================
# メイン処理
# ==============================
if __name__ == "__main__":
print("前日終値を取得中...")
prev_closes: dict[str, float] = fetch_previous_close(TICKER_LIST, MARKET_SUFFIX)
if not prev_closes:
raise RuntimeError("前日終値を1件も取得できませんでした。ネットワークを確認してください。")
print(f"取得完了: {len(prev_closes)}銘柄")
history: list[pd.DataFrame] = run_monitoring(
TICKER_LIST, prev_closes, POLL_COUNT, POLL_INTERVAL_SEC,
IMBALANCE_THRESHOLD, GAP_THRESHOLD_PCT, SPREAD_WARNING_PCT,
)
plot_imbalance_history(history)
コードの処理フロー解説
上記のコードは、以下の6ステップで構成されています。
* fetch_previous_close関数でyfinance経由の前日終値を銘柄ごとに取得し、比較の基準値とする
* fetch_quote_data関数で気配値データを取得する（本コードではデモ用にシミュレーションデータを生成。実運用時はkabuステーションAPI等に差し替える）
* calc_indicators関数でオーダーインバランス比率・スプレッド率・前日比乖離率の3指標を算出する
* judge_strategy関数で閾値に基づいて「買い優勢」「売り優勢」「ギャップGU/GD」「低流動性注意」のシグナルを判定する
* run_monitoring関数で指定回数のポーリングを行い、時系列の分析結果を蓄積する
* plot_imbalance_history関数でインバランスと乖離率の推移チャートを描画する
fetch_quote_data関数の中身をkabuステーションAPIやJ-Quants APIの呼び出しに置き換えれば、リアルタイムの気配値で動作します。
戦略シグナルの読み解き方と判断基準
シグナル別の戦略方針
算出されたシグナルに基づく判断基準を以下にまとめます。
| シグナル | 解釈 | 戦略方針 |
|---|---|---|
| 買い優勢 | 買い注文が売り注文を大幅に上回る | 寄り付き後の上昇初動を狙う順張り候補 |
| 売り優勢 | 売り注文が買い注文を大幅に上回る | 空売り候補、または買い見送り |
| ギャップGU | 前日終値より1%以上高い気配 | ギャップアップ後の押し目買い or 反落狙い |
| ギャップGD | 前日終値より1%以上低い気配 | 投げ売り後のリバウンド狙い or 追随売り |
| 低流動性注意 | スプレッドが0.5%以上 | 成行注文はスリッページが大きくなるため指値推奨 |
シグナルが複数重なる場合は、それぞれの意味を組み合わせて判断してください。「買い優勢 / ギャップGU」であれば強い買い需要が確認できますが、「買い優勢 / 低流動性注意」の場合は流動性リスクに警戒が必要です。
気配値分析の限界
寄り付き前の気配値は、9:00の板寄せまでに大きく変動します。8:00時点と8:50時点では注文状況がまったく異なることが珍しくありません。
特に、見せ板（約定意思のない大量注文）が混在する可能性を常に疑ってください。インバランスが極端に偏っている場合、寄り付き直前に大量のキャンセルが入るケースがあります。
気配値だけで売買を決定するのは危険です。前日の出来高、海外市場の動向、先物の値動きなど、複数の情報源と組み合わせて総合的に判断することが鉄則です。
【コピペOK】実運用向けAPI接続のテンプレート
実運用で気配値を取得する場合、デモ用のfetch_quote_data関数を以下のテンプレートに差し替えてください。ここではkabuステーションAPI（auカブコム証券）の接続例を示します。
# ==============================
# kabuステーションAPI接続テンプレート
# ==============================
import requests
from typing import Any
# --- 設定エリア ---
KABU_API_BASE: str = "http://localhost:18080/kabusapi"
KABU_API_PASSWORD: str = "your_api_password" # APIパスワードを設定
# ==============================
# 関数定義: APIトークン取得
# ==============================
def get_kabu_token(base_url: str, password: str) -> str:
'kabuステーションAPIのトークンを取得する'
url: str = f"{base_url}/token"
payload: dict[str, str] = {"APIPassword": password}
response: requests.Response = requests.post(url, json=payload)
response.raise_for_status()
token: str = response.json().get("Token", ')
if not token:
raise ValueError("トークン取得に失敗しました。パスワードを確認してください。")
return token
# ==============================
# 関数定義: 板情報取得
# ==============================
def fetch_board_from_kabu(
base_url: str,
token: str,
symbol: str,
exchange: int = 1,
) -> dict[str, Any]:
'指定銘柄の板情報を取得する'
url: str = f"{base_url}/board/{symbol}@{exchange}"
headers: dict[str, str] = {"X-API-KEY": token}
response: requests.Response = requests.get(url, headers=headers)
response.raise_for_status()
data: dict[str, Any] = response.json()
return {
"code": symbol,
"best_bid": data.get("BidPrice", 0),
"best_ask": data.get("AskPrice", 0),
"bid_volume": data.get("BidQty", 0),
"ask_volume": data.get("AskQty", 0),
"current_price": data.get("CurrentPrice", 0),
"prev_close": data.get("PreviousClose", 0),
}
# ==============================
# 関数定義: 複数銘柄一括取得
# ==============================
def fetch_multi_board(
base_url: str,
token: str,
ticker_list: list[str],
) -> list[dict[str, Any]]:
'複数銘柄の板情報を一括取得する'
records: list[dict[str, Any]] = []
for code in ticker_list:
try:
board: dict[str, Any] = fetch_board_from_kabu(base_url, token, code)
mid: float = (board["best_bid"] + board["best_ask"]) / 2
board["mid_price"] = round(mid, 1)
records.append(board)
except Exception as e:
print(f"板情報取得エラー ({code}): {e}")
return records
コードの処理フロー解説
上記のコードは、以下の3ステップで構成されています。
* get_kabu_token関数でkabuステーションAPIのローカルサーバーに認証リクエストを送り、セッショントークンを取得する
* fetch_board_from_kabu関数でトークンを使い、指定銘柄の板情報（最良気配・数量・前日終値）をJSON形式で取得する
* fetch_multi_board関数で複数銘柄をループ処理し、メインコードのfetch_quote_dataと同じデータ構造で返す
kabuステーションAPIを使用するには、auカブコム証券の口座開設とkabuステーションアプリのインストールが必要です。APIの利用自体は無料で、1秒あたり10リクエストまでの制限があります。
よくあるエラーと対処法
ConnectionRefusedError（kabuステーションAPI接続時）
kabuステーションアプリが起動していない、またはAPI設定が有効化されていないことが原因です。APIはローカルホスト（localhost:18080）で動作するため、アプリが起動していなければ接続できません。
以下を試してください。
* kabuステーションアプリを起動し、ログイン済みであることを確認する
* アプリの設定画面で「API」が有効になっているか確認する
* ファイアウォールがポート18080をブロックしていないか確認する
インバランス値が常に0付近で変化しない
取得している気配値データの買い数量・売り数量が正しく取得されていない可能性があります。API側が数量ではなく金額を返しているケースや、データ型の不一致が原因として考えられます。
以下を試してください。
* APIのレスポンスをprint()で出力し、BidQtyとAskQtyのフィールドに実際の値が入っているか確認する
* レスポンスの型がfloatではなくstrで返されている場合はint()で明示的に変換する
* 市場の注文受付時間外（8:00以前・15:30以降）に実行していないか確認する
KeyError: ‘previous_close’（yfinance）
📘 外部参考：yfinance 公式GitHubリポジトリ ／ PyPIページ
yfinanceのバージョンアップにより、属性名が変更されることがあります。fast_infoオブジェクトのキー名がバージョンによって異なるのが原因です。
以下を試してください。
* pip install --upgrade yfinanceで最新版に更新する
* print(yf.Ticker("7203.T").fast_info)を実行し、利用可能なキー名の一覧を確認する
* fast_infoで取得できない場合はhistory(period="5d")で直近5日分の終値を取得し、最後から2番目の値を前日終値として使用する
まとめ
この記事では、寄り付き前の気配値をPythonで取得・数値化し、オーダーインバランスと前日比乖離率から当日の戦略シグナルを判定する方法を解説しました。
実際に使ってみた要点をまとめます。
* 気配値分析の核心は、買い数量と売り数量の偏りを示すオーダーインバランス比率（閾値±0.3が目安）である
* 前日比乖離率が1%を超えるギャップは、当日のトレンド方向を示す強いシグナルとなる
* スプレッド率0.5%以上は低流動性の警告であり、成行注文ではなく指値注文を使うべきである
* 実運用ではkabuステーションAPIやJ-Quants APIに接続し、デモ関数を差し替えて使用する
* 気配値は寄り付き直前まで変動するため、単一時点のスナップショットを過信してはいけない
また、本コードの出力をCSVに蓄積し、「寄り付き前のインバランスが+0.3以上だった日の寄り付き後30分リターン」を集計してみるといいかと思います。自分の監視銘柄でインバランスの予測精度を検証することで、シグナルの信頼度を定量的に評価できます。
さらに、また、日経225先物の夜間取引価格やドル円の変動幅と組み合わせると、外部要因も加味した分析に発展できます。
