kabuステーションAPIでPython自動売買｜二重発注を防ぐ実装

なぜkabuステーションAPIが現実解なのか

個人投資家が日本株を公式APIでリアルタイム自動売買できる選択肢は、いまも多くありません。SBI証券・楽天証券は個人向けの公開発注APIを提供していないため、現実的な本命が三菱UFJ eスマート証券（旧auカブコム証券）の「kabuステーションAPI」です。

ただし、ネット上の「最小構成サンプル」をそのまま実弾に回すと事故ります。トークンは毎日無効になりますし、板情報をGETしただけで銘柄登録の上限に達しますし、タイムアウト時に再送すると二重発注になります。この記事では最小構成を示したうえで、その先にある実務的な落とし穴を潰していきます。

📘 外部参考：kabuステーションAPI 公式リファレンス ／ よくあるご質問（流量制限・PUSH仕様）

利用条件と制約（先に確認すべきこと）

項目	内容
必要プラン	kabuステーション Professionalプラン以上
プラン条件	信用取引口座または先物OP口座を開設＋所定期間に約定1回以上
OS	Windowsのみ（kabuステーションがWindows専用）
接続元	kabuステーションと同一PC・同一IPからのみ
稼働時間	6:30〜翌6:15。毎日自動ログアウトされる
エンドポイント	本番 :18080 ／ 検証 :18081
流量制限	発注5件/秒、情報・余力・銘柄登録10件/秒
銘柄登録上限	REST/PUSH合計で50銘柄

設定はマイページの「らくらく電子契約」でAPI利用設定を行い、kabuステーション側の「システム設定→API」で有効化とAPIパスワード（英数6〜16桁）を登録します。同時にソフトリミット（ワンショット上限）も設定できます。BOTの発注金額に上限を掛ける安全装置なので、必ず現実的な値にしてください。

最小構成：トークン取得 → 板情報 → 成行発注

import json
import requests

BASE = "http://localhost:18081/kabusapi"   # 検証環境
API_PASSWORD = "YOUR_API_PASSWORD"

def get_token() -> str:
    r = requests.post(f"{BASE}/token",
                      json={"APIPassword": API_PASSWORD}, timeout=5)
    r.raise_for_status()
    return r.json()["Token"]

def get_board(token: str, symbol: str = "7203", exchange: int = 1) -> dict:
    r = requests.get(f"{BASE}/board/{symbol}@{exchange}",
                     headers={"X-API-KEY": token}, timeout=5)
    r.raise_for_status()
    return r.json()

tok = get_token()
print("現在値:", get_board(tok).get("CurrentPrice"))

ここまでは簡単です。問題はこの先にあります。

落とし穴1：トークンは毎日死ぬ

kabuステーションは毎日自動ログアウトします。夜間に起動しっぱなしにしたBOTは、翌朝の寄り付きで必ず401を返されます。トークンを起動時に1回取るだけの実装は、初日しか動きません。

また、トークンは発行のたびに新規発行され、古いトークンは無効化されます。複数プロセスから無自覚にget_token()を呼ぶと、互いのトークンを潰し合います。

import threading
import time

class Kabu:
    """トークンを自動で再取得するクライアント。"""

    def __init__(self, base: str, api_password: str):
        self.base = base.rstrip("/")
        self.pw = api_password
        self._token = None
        self._lock = threading.Lock()
        self._s = requests.Session()
        self._last_call = 0.0

    def _issue(self) -> str:
        r = self._s.post(f"{self.base}/token",
                         json={"APIPassword": self.pw}, timeout=5)
        r.raise_for_status()
        return r.json()["Token"]

    def token(self, force: bool = False) -> str:
        with self._lock:
            if force or self._token is None:
                self._token = self._issue()
            return self._token

    def _throttle(self, min_interval: float) -> None:
        wait = min_interval - (time.monotonic() - self._last_call)
        if wait > 0:
            time.sleep(wait)
        self._last_call = time.monotonic()

    def call(self, method: str, path: str, min_interval: float = 0.12,
             **kw) -> dict:
        """401なら1度だけトークンを取り直して再試行する。"""
        self._throttle(min_interval)
        url = f"{self.base}/{path.lstrip('/')}"
        for attempt in (0, 1):
            headers = {"X-API-KEY": self.token(force=bool(attempt)),
                       "Content-Type": "application/json"}
            r = self._s.request(method, url, headers=headers, timeout=5, **kw)
            if r.status_code == 401 and attempt == 0:
                continue
            r.raise_for_status()
            return r.json()
        raise RuntimeError("認証に失敗しました")

min_interval=0.12は情報系10件/秒に対する余裕です。発注系は5件/秒なので0.25秒以上空けます。制限を超えるとエラーが返るだけで、そこから復帰する処理を書いていないとBOTが止まります。

落とし穴2：GETしただけで銘柄登録される

これは仕様を知らないとまず原因が分かりません。/boardなどで情報を取得すると、その銘柄が自動的にAPI登録銘柄リストへ追加されます。リストの上限はREST/PUSH合計で50銘柄。スクリーニング目的で100銘柄をループで取得すると、51件目で「レジスト数エラー」になります。

def chunked(seq, n: int):
    for i in range(0, len(seq), n):
        yield seq[i:i + n]

def fetch_boards(api: Kabu, symbols: list[str], exchange: int = 1) -> list[dict]:
    """50銘柄ずつ、登録→取得→全解除 を繰り返す。"""
    out = []
    for group in chunked(symbols, 50):
        api.call("PUT", "/unregister/all")          # 先に必ず全解除
        api.call("PUT", "/register", data=json.dumps({
            "Symbols": [{"Symbol": s, "Exchange": exchange} for s in group]
        }))
        for s in group:
            try:
                out.append(api.call("GET", f"/board/{s}@{exchange}"))
            except requests.HTTPError as e:
                print(f"skip {s}: {e}")
    api.call("PUT", "/unregister/all")
    return out

処理の頭で/unregister/allを呼ぶのが要点です。前回の実行で登録が残っていると、次の実行が最初から上限に張り付きます。BOTの起動時にも一度呼んでおくと安定します。

落とし穴3：タイムアウト再送で二重発注する

いちばん怖いのがこれです。/sendorderを投げてタイムアウトしたとき、注文が通っていないのか、通ったのに応答が返っていないだけなのかはクライアント側からは区別できません。ここで素朴にリトライすると、同じ注文が2本入ります。

HTTPの世界の定石はべき等キーですが、kabuステーションAPIにその仕組みはありません。再送する前に注文一覧を照会して、自分が出した注文が既に存在しないかを確認します。

import datetime as dt

def recent_orders(api: Kabu, symbol: str, within_sec: int = 120) -> list[dict]:
    orders = api.call("GET", "/orders", params={"product": 1, "symbol": symbol})
    now = dt.datetime.now()
    fresh = []
    for o in orders:
        ts = dt.datetime.strptime(o["RecvTime"][:19], "%Y-%m-%dT%H:%M:%S")
        if (now - ts).total_seconds() <= within_sec:
            fresh.append(o)
    return fresh

def send_order_once(api: Kabu, payload: dict, trade_password: str) -> dict:
    """タイムアウト時は再送せず、注文照会で実在を確認する。"""
    body = dict(payload, Password=trade_password)
    symbol = payload["Symbol"]
    before = len(recent_orders(api, symbol))

    try:
        return api.call("POST", "/sendorder", min_interval=0.25,
                        data=json.dumps(body))
    except requests.Timeout:
        time.sleep(2)
        after = recent_orders(api, symbol)
        if len(after) > before:
            print("応答は落ちたが注文は通っていた:", after[-1]["ID"])
            return {"Result": 0, "OrderId": after[-1]["ID"], "recovered": True}
        raise RuntimeError("注文が通っていない可能性が高い。手動確認が必要")

タイムアウトを自動リトライしない。これだけは徹底してください。ネットワークが不安定な日に、同じ成行注文が5本入るのが最悪のシナリオです。

発注ペイロードの組み立て

def cash_buy_market(symbol: str, qty: int, exchange: int = 1) -> dict:
    """現物・特定口座・成行・当日限りの買い注文。"""
    return {
        "Symbol": symbol,
        "Exchange": exchange,      # 1=東証
        "SecurityType": 1,         # 1=株式
        "Side": "2",               # 1=売, 2=買
        "CashMargin": 1,           # 1=現物
        "DelivType": 2,            # 2=お預り金
        "AccountType": 4,          # 4=特定
        "Qty": qty,                # 単元株数の倍数
        "FrontOrderType": 10,      # 10=成行
        "Price": 0,
        "ExpireDay": 0,            # 0=当日
    }

def cash_sell_stop(symbol: str, qty: int, trigger: float,
                   exchange: int = 1) -> dict:
    """損切り用の逆指値（トリガー到達で成行売り）。"""
    return {
        "Symbol": symbol, "Exchange": exchange, "SecurityType": 1,
        "Side": "1", "CashMargin": 1, "DelivType": 2, "AccountType": 4,
        "Qty": qty,
        "FrontOrderType": 30,      # 30=逆指値
        "Price": 0, "ExpireDay": 0,
        "ReverseLimitOrder": {
            "TriggerSec": 1,           # 1=発注銘柄
            "TriggerPrice": trigger,
            "UnderOver": 1,            # 1=以下, 2=以上
            "AfterHitOrderType": 1,    # 1=成行
            "AfterHitPrice": 0,
        },
    }

DelivTypeとFrontOrderTypeの組み合わせは現物と信用で異なります。信用返済ではClosePositionsの指定も必要になるため、必ず公式リファレンスを確認してください。

そして、買い注文と同時に損切りの逆指値を入れるのが実務です。BOTが落ちても、証券会社側に置いた逆指値は生き残ります。

安全装置を先に書く

ロジックより先にガードを実装します。BOTのバグで損をするのは、たいてい戦略ではなく発注制御の側です。

class Guard:
    def __init__(self, max_notional: int = 300_000,
                 max_orders_per_day: int = 20,
                 max_loss_yen: int = 30_000):
        self.max_notional = max_notional
        self.max_orders = max_orders_per_day
        self.max_loss = max_loss_yen
        self.count = 0

    def check(self, api: Kabu, price: float, qty: int) -> None:
        notional = price * qty
        if notional > self.max_notional:
            raise RuntimeError(f"1回の発注額が上限超過: {notional:,.0f}円")
        if self.count >= self.max_orders:
            raise RuntimeError("本日の発注回数上限に到達")

        cash = api.call("GET", "/wallet/cash")
        if notional > float(cash["StockAccountWallet"]):
            raise RuntimeError("買付余力が不足")

        pnl = sum(float(p.get("ProfitLoss") or 0)
                  for p in api.call("GET", "/positions", params={"product": 1}))
        if pnl <= -self.max_loss:
            raise RuntimeError(f"日次損失が上限に到達: {pnl:,.0f}円 → 全停止")
        self.count += 1

日次損失の上限で全停止させる仕組みは、必ず入れてください。相場が想定外に動いたとき、これがなければBOTは淡々と負け続けます。

PUSH配信（WebSocket）でイベント駆動にする

ポーリングで板を叩き続けると流量制限に当たります。値動きに反応したいなら/registerで銘柄を登録し、WebSocketで受けます。間引き間隔は400ms、接続は1本のみです。

import websocket   # pip install websocket-client

def start_push(api: Kabu, symbols: list[str], on_tick, exchange: int = 1):
    api.call("PUT", "/unregister/all")
    api.call("PUT", "/register", data=json.dumps({
        "Symbols": [{"Symbol": s, "Exchange": exchange} for s in symbols]
    }))

    url = api.base.replace("http://", "ws://") + "/websocket"

    def _on_message(ws, message):
        try:
            on_tick(json.loads(message))
        except Exception as e:      # ハンドラの例外で接続を落とさない
            print("handler error:", e)

    ws = websocket.WebSocketApp(
        url,
        on_message=_on_message,
        on_error=lambda ws, e: print("ws error:", e),
        on_close=lambda ws, *a: print("ws closed"),
    )
    ws.run_forever(ping_interval=30, reconnect=5)

PUSHは値が更新されたときだけ配信されます。昼休みと引け後は何も来ません。「配信が止まった＝異常」と判定するロジックを書くと、毎日11時半に誤検知します。

本番投入前のチェックリスト

• 検証環境（:18081）で発注から取消まで全フローを通したか
• トークンの自動再取得を実装したか（毎日の自動ログアウト対策）
• 処理の先頭で/unregister/allを呼んでいるか
• 発注は0.25秒以上、情報取得は0.12秒以上の間隔を空けているか
• タイムアウト時に自動再送していないか
• 1回の発注額・日次発注回数・日次損失に上限を設けたか
• 買い注文と同時に逆指値の損切りを置いているか
• kabuステーション側のソフトリミットを設定したか
• 約定とエラーをSlack／Discordに通知しているか

まとめ

kabuステーションAPIは、日本株でAPI発注を行うための最も現実的なルートです。ただし「トークンを取って発注する」までのサンプルコードと、実際に毎日動かせるBOTのあいだには大きな距離があります。

その距離のほとんどは、戦略の巧拙ではなく仕様への対応です。毎日のログアウト、50銘柄の登録上限、秒間の流量制限、そして二重発注。この4つを潰したうえで、日次損失の上限による全停止を入れる。ここまでやって、ようやく実弾を入れられる状態になります。

まずは検証環境で「板取得→疑似発注→注文照会→取消」のループを1週間止めずに回してみてください。想像していないところで止まります。それを1つずつ潰していく作業が、そのまま本番の安定性になります。