株自動売買プログラミング 楽天証券編【Python自作】

楽天証券で株自動売買をPython自作するなら、公式のマーケットスピード II RSSをExcelへ組み込み、PythonからExcelを制御する構成が現実的です。楽天証券はPython用REST APIを公開しているわけではありませんが、RSSには国内株のリアルタイム情報、保有株、余力、注文・約定情報、現物・信用の注文関数が公式に用意されています。

この記事では、Windows + MarketSpeed II + Microsoft Excel + Python（xlwings）で、株価取得、シグナル判定、注文数量計算、現物注文、注文状態の確認までをつなぎます。誤発注を防ぐため、Pythonが注文条件をセットした後に人がARMを入力しなければ発注トリガーを変更できない設計にします。

重要：マーケットスピード II RSSの注文は、Excelで条件を入力した時点ではなく楽天証券が注文を受け付けた時点で成立します。余力不足、通信切断、発注上限、重複した発注IDなどで拒否される場合があります。「関数を実行した」ではなく、注文一覧・約定一覧・建玉を照合して初めて処理完了です。

楽天証券編はPython単体ではなくExcelを注文ゲートにする
利用準備と安全設定を先に済ませる
Excelブックを注文ゲートとして設計する
PythonからExcelのリアルタイム値を読む
5分足ブレイクアウトとATR資金管理を実装する
注文条件をセットし、人のARM後に発注する
発注済みではなく約定・保有株まで照合する
停止条件をコードへ入れる
バックテストとペーパートレードの進め方
公式情報・参考資料

楽天証券編はPython単体ではなくExcelを注文ゲートにする

マーケットスピード II RSSは、Microsoft Excelに追加するアドインです。MarketSpeed IIへログインした状態で、ExcelのRSS関数が楽天証券クライアントを通じて情報取得・発注を行います。Pythonはxlwingsでセルを読み書きし、戦略計算と状態管理を担当します。

層	担当	停止時の影響
楽天証券	注文受付・約定・残高	市場取引を継続できない
MarketSpeed II	ログインとRSS接続	Excel関数が接続待ちになる
Excel + RSS	リアルタイム値・注文関数	価格更新と発注が止まる
Python	戦略・資金管理・監査ログ	新規シグナル処理を停止する
SQLite	発注ID・注文状態の永続化	再起動後の重複判定ができない

楽天証券の公式ヘルプでは、RSSから国内株の現物注文にRssStockOrder、信用新規にRssMarginOpenOrder、信用返済にRssMarginCloseOrder、訂正にRssModifyOrder、取消にRssCancelOrderを使います。注文一覧はRssOrderList、約定一覧はRssExecutionList、保有株はRssPositionListです。

この構成はWindows PCとExcelが常時動くことを前提にします。Mac、Googleスプレッドシート、LibreOfficeへ置き換えることはできません。クラウドVPSへPythonだけを置いてもRSS注文は動作しません。利用前にMarketSpeed II RSSの導入ガイドも確認してください。

利用準備と安全設定を先に済ませる

楽天証券のMarketSpeed IIをWindowsへインストールする
ExcelへMarketSpeed II RSSアドインを登録する
RSS確認書兼同意書を読み、内容に同意する
MarketSpeed IIの環境設定でRSS注文機能を有効にする
取引暗証番号はMarketSpeed II側へ入力し、Pythonへ保存しない
RSSの1回あたり発注上限金額を小さく設定する
注文確認画面を表示する設定のままテストする

RSSメニューが「接続中」かつ「発注可」で、MarketSpeed II側の注文利用設定が済み、注文金額が上限内、発注IDが未使用、入力内容が有効という条件を満たして初めて発注できます。開発中は「発注不可」にして、価格取得と待機中ステータスだけ確認します。

別のExcelファイルを開くと、意図しない発注を防ぐため「発注可」から「発注不可」へ自動変更される仕様があります。自動売買プログラムは、この状態変更をエラーとして扱い、勝手に再有効化してはいけません。人が画面を確認して再度許可します。

Excelブックを注文ゲートとして設計する

rakuten_bridge.xlsmに3つのシートを作ります。MarketはRSS株価、Orderは現物注文、Statusは注文・約定照会です。まずは公式サンプルシートで関数が動くことを確認してから、自作ブックへ移してください。

シート	主なセル	用途
Market	A2=銘柄コード、B2=現在値	RSS関数で市場情報を表示
Order	B2:B21=注文引数、E2=状態	Pythonが条件を書き、RSS関数が注文
Status	A1=RssOrderList()	注文番号と受付状態を確認
Control	B2=ARM、B3=停止理由	人による発注許可と障害表示

現物注文の公式入力形式は次の20引数です。

RssStockOrder(
  発注ID, 発注トリガー, 銘柄コード, 売買区分,
  注文区分, SOR区分, 注文数量, 価格区分,
  注文価格, 執行条件, 注文期限, 口座区分,
  逆指値条件価格, 逆指値条件区分,
  逆指値価格区分, 逆指値価格,
  セット注文区分, セット注文価格,
  セット注文執行条件, セット注文期限
)

Order!E2へ次の式を設定します。PythonはB列だけを書き換えます。B3の発注トリガーは必ず0から始め、すべての項目とステータスを確認した後に1へ変化させます。

=RssStockOrder(
  B2,B3,B4,B5,B6,B7,B8,B9,B10,B11,
  B12,B13,B14,B15,B16,B17,B18,B19,B20,B21
)

公式例では、東証の楽天グループ（4755）を100株、現物成行買い、本日中、特定口座で発注する式は=RssStockOrder(1,1,"4755.T",3,0,0,100,0,,1,,0)です。売買区分は1が売、3が買、注文区分は0が通常、1が逆指値付通常、2が逆指値、価格区分は0が成行、1が指値です。数値コードは独自に推測せず、必ず公式関数マニュアルを基準にします。

PythonからExcelのリアルタイム値を読む

Python環境を作り、xlwings、pandas、pydanticを入れます。ExcelブックはMarketSpeed IIへログインした後に手動で開いておきます。

python -m venv .venv
.venv\Scripts\activate
pip install xlwings pandas numpy pydantic

# excel_bridge.py
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
import math
import time
import xlwings as xw

BOOK_NAME = "rakuten_bridge.xlsm"

@dataclass(frozen=True)
class Quote:
  code: str
  price: float
  timestamp: datetime

class ExcelBridge:
  def __init__(self, book_name: str = BOOK_NAME) -> None:
    self.book = xw.books[book_name]
    self.market = self.book.sheets["Market"]
    self.order = self.book.sheets["Order"]
    self.control = self.book.sheets["Control"]

  def read_quote(self) -> Quote:
    code = str(self.market["A2"].value).strip()
    price = self.market["B2"].value
    updated = self.market["C2"].value

    if price is None or not math.isfinite(float(price)):
      raise RuntimeError("現在値が未取得です")
    if not isinstance(updated, datetime):
      raise RuntimeError("更新時刻が取得できません")
    age = (datetime.now() - updated).total_seconds()
    if age > 10:
      raise RuntimeError(f"株価が古すぎます: {age:.1f}秒")

    return Quote(code=code, price=float(price), timestamp=updated)

  def wait_status(self, timeout: float = 10.0) -> str:
    deadline = time.time() + timeout
    while time.time() < deadline:
      status = str(self.order["E2"].value or "")
      if status and status not in {"-", "None"}:
        return status
      time.sleep(0.2)
    raise TimeoutError("RSS注文ステータスが更新されません")

価格だけでなく更新時刻を検査するのが重要です。MarketSpeed IIとの接続が切れた後も、Excelセルには最後の価格が残る場合があります。古い値を最新値と誤認して発注しないよう、一定秒数を超えたら停止します。対象関数の更新時刻が取得できない場合は、Python側で値の変化時刻を記録し、長時間変化がない状態を警告します。

5分足ブレイクアウトとATR資金管理を実装する

リアルタイム価格をそのまま1回読むだけでは戦略になりません。ここではMarketシートから時刻・始値・高値・安値・終値・出来高をPythonへ取り込み、直近20本の高値突破をシグナルとします。説明を明確にするための例であり、利益を保証する戦略ではありません。

import numpy as np
import pandas as pd

def add_indicators(bars: pd.DataFrame) -> pd.DataFrame:
  out = bars.copy().sort_index()
  prev_close = out["close"].shift(1)
  tr = pd.concat([
    out["high"] - out["low"],
    (out["high"] - prev_close).abs(),
    (out["low"] - prev_close).abs()
  ], axis=1).max(axis=1)

  out["atr14"] = tr.ewm(
    alpha=1 / 14,
    adjust=False,
    min_periods=14
  ).mean()
  # 当該足を含めると、終値が高値を超えられないため1本ずらす
  out["previous_20_high"] = out["high"].rolling(20).max().shift(1)
  out["volume_median"] = out["volume"].rolling(20).median().shift(1)
  out["buy_signal"] = (
    (out["close"] > out["previous_20_high"])
    & (out["volume"] > out["volume_median"] * 1.5)
  )
  return out

def calculate_quantity(
  cash: float,
  entry: float,
  atr: float,
  risk_pct: float = 0.005,
  max_value_pct: float = 0.20
) -> tuple[int, float, float]:
  stop = entry - 2 * atr
  risk_per_share = entry - stop
  by_risk = int((cash * risk_pct) / risk_per_share)
  by_value = int((cash * max_value_pct) / entry)
  quantity = (min(by_risk, by_value) // 100) * 100
  if quantity < 100:
    raise ValueError("リスク上限内で100株を発注できません")
  target = entry + 2 * risk_per_share
  return quantity, round(stop, 1), round(target, 1)

5分足の終値が確定する前にシグナルを評価すると、足の途中で高値を超えた後に戻ったケースまで買ってしまいます。時刻が5分境界を越え、前のバーが確定したことを確認してから判定します。バックテストでも、シグナルが確定した次の足の始値で約定させ、同じ足の終値で買ったことにしないでください。

注文条件をセットし、人のARM後に発注する

以下のコードは現物の指値買いと、約定後の売り指値を予約するセット注文を作ります。損切りは別途逆指値注文として管理する必要があります。最初から完全自動にせず、注文確認画面を表示し、最小数量で検証してください。

# order_gateway.py
from __future__ import annotations

from dataclasses import dataclass
import sqlite3
import time

from excel_bridge import ExcelBridge

@dataclass(frozen=True)
class CashOrder:
  code: str
  quantity: int
  limit_price: float
  take_profit: float

class OrderGateway:
  def __init__(self, bridge: ExcelBridge, db_path: str = "orders.db") -> None:
    self.bridge = bridge
    self.db = sqlite3.connect(db_path)
    self.db.execute("""
      CREATE TABLE IF NOT EXISTS orders (
        order_id INTEGER PRIMARY KEY,
        code TEXT NOT NULL,
        quantity INTEGER NOT NULL,
        limit_price REAL NOT NULL,
        state TEXT NOT NULL,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP
      )
    """)

  def next_order_id(self) -> int:
    row = self.db.execute(
      "SELECT COALESCE(MAX(order_id), 100000) + 1 FROM orders"
    ).fetchone()
    return int(row[0])

  def stage_buy(self, order: CashOrder) -> int:
    if order.quantity % 100 != 0:
      raise ValueError("注文数量は100株単位にします")
    order_id = self.next_order_id()

    values = [
      order_id,             # B2 発注ID
      0,                    # B3 発注トリガー：必ず待機
      f"{order.code}.T",    # B4 銘柄コード
      "3",                  # B5 買
      "0",                  # B6 通常注文
      "0",                  # B7 通常市場
      order.quantity,       # B8 数量
      "1",                  # B9 指値
      order.limit_price,    # B10 注文価格
      "1",                  # B11 本日中
      None,                 # B12 期限
      "0",                  # B13 特定口座
      None, None, None, None,  # B14:B17 逆指値なし
      "1",                  # B18 セット注文あり
      order.take_profit,    # B19 売り指値
      "1",                  # B20 本日中
      None                  # B21 期限
    ]
    self.bridge.order["B2:B21"].options(transpose=True).value = values
    self.bridge.control["B2"].value = "REVIEW"
    self.bridge.book.app.calculate()
    status = self.bridge.wait_status()
    if "待機中" not in status:
      raise RuntimeError(f"注文条件が待機状態になりません: {status}")

    self.db.execute(
      "INSERT INTO orders(order_id,code,quantity,limit_price,state) "
      "VALUES(?,?,?,?,?)",
      (order_id, order.code, order.quantity, order.limit_price, "STAGED")
    )
    self.db.commit()
    return order_id

  def submit_after_human_arm(self, order_id: int) -> str:
    if str(self.bridge.control["B2"].value).strip().upper() != "ARM":
      raise PermissionError("Control!B2へARMを入力して確認してください")

    current_id = int(self.bridge.order["B2"].value)
    trigger = int(self.bridge.order["B3"].value)
    if current_id != order_id or trigger != 0:
      raise RuntimeError("発注IDまたはトリガー状態が不正です")

    # 公式仕様どおり0から1への変化で発注
    self.bridge.order["B3"].value = 1
    self.bridge.book.app.calculate()
    status = self.bridge.wait_status(timeout=20)
    self.db.execute(
      "UPDATE orders SET state=? WHERE order_id=?",
      (status, order_id)
    )
    self.db.commit()
    self.bridge.control["B2"].value = "LOCKED"
    return status

発注IDは1以上の未使用整数でなければならず、同じExcelセッションで再利用すると多重発注防止機能により拒否されます。SQLiteで採番し、プログラム再起動後も重複しないようにします。ただし、Excelを再起動すると発注IDの扱いが変わるため、ローカルIDだけでなく楽天証券の注文番号も取得して保存します。

実際の注文では、発注上限金額、値幅制限、SOR利用、注文期限、NISA区分なども考慮します。本記事のコードは特定口座・東証・現物買いの例です。信用取引は建日、建単価、建市場、保証金率など追加状態が必要で、損失が元本を超える可能性もあるため別システムとして設計してください。

発注済みではなく約定・保有株まで照合する

RSS関数が「発注済み」と返しても、指値注文が約定したとは限りません。注文一覧のRssOrderList、約定一覧のRssExecutionList、保有株のRssPositionListを別シートへ展開し、Pythonで読み取ります。

import pandas as pd
import xlwings as xw

def read_excel_table(sheet: xw.Sheet, top_left: str) -> pd.DataFrame:
  values = sheet[top_left].expand("table").value
  if not values or len(values) < 2:
    return pd.DataFrame()
  header = [str(v).strip() for v in values[0]]
  return pd.DataFrame(values[1:], columns=header)

def verify_broker_state(book: xw.Book, code: str) -> dict:
  orders = read_excel_table(book.sheets["Status"], "A1")
  executions = read_excel_table(book.sheets["Executions"], "A1")
  positions = read_excel_table(book.sheets["Positions"], "A1")

  # 列名はRSSの出力ヘッダーに合わせて設定する
  code_text = str(code).zfill(4)
  matched_orders = orders[
    orders["銘柄コード"].astype(str).str.startswith(code_text)
  ] if not orders.empty else orders
  matched_fills = executions[
    executions["銘柄コード"].astype(str).str.startswith(code_text)
  ] if not executions.empty else executions
  matched_positions = positions[
    positions["銘柄コード"].astype(str).str.startswith(code_text)
  ] if not positions.empty else positions

  return {
    "orders": len(matched_orders),
    "executions": len(matched_fills),
    "positions": len(matched_positions)
  }

列名や展開範囲はRSSのバージョンと関数引数で変わる可能性があります。最初にExcel上でヘッダーを確認し、その名前を設定ファイルへ保存します。部分約定の場合は「注文数量100、約定数量40、残数量60」のように状態を分け、100株すべてを保有した前提で損切り注文を出さないようにします。

停止条件をコードへ入れる

自動売買は、正常系より停止条件が重要です。次のいずれかに該当したら、新規発注を停止し、Controlシートへ理由を書きます。

RSSが接続中または発注可ではない
リアルタイム値の更新が10秒以上止まった
Excelの計算モードが手動になっている
注文関数が入力エラー、接続待ち、応答待ちのままタイムアウトした
同じ銘柄に未約定注文または保有株がある
1日の発注回数・損失額・売買金額が上限を超えた
SQLiteの状態とRSS注文一覧が一致しない
取引時間外、祝日、システムメンテナンス中である

from datetime import date

class DailyRiskGuard:
  def __init__(
    self,
    max_orders: int = 3,
    max_notional: float = 500_000,
    max_daily_loss: float = 15_000
  ) -> None:
    self.max_orders = max_orders
    self.max_notional = max_notional
    self.max_daily_loss = max_daily_loss

  def check(
    self,
    orders_today: int,
    notional_today: float,
    realized_pnl_today: float,
    new_notional: float
  ) -> None:
    if orders_today >= self.max_orders:
      raise RuntimeError("1日の発注回数上限です")
    if notional_today + new_notional > self.max_notional:
      raise RuntimeError("1日の売買金額上限です")
    if realized_pnl_today <= -self.max_daily_loss:
      raise RuntimeError("1日の損失上限です")

損失上限へ達した後に「取り返すための注文」を出さない仕組みが必要です。停止フラグはメモリだけでなくSQLiteへ保存し、プログラムを再起動しても当日中は解除されないようにします。解除は翌営業日か、人が理由を確認して管理画面から行います。

バックテストとペーパートレードの進め方

段階	期間の目安	確認するもの
バックテスト	複数相場局面	未来参照、コスト、最大DD、取引数
RSS接続テスト	数営業日	更新停止、再接続、セルエラー
発注不可テスト	1週間以上	注文条件と待機中ステータス
確認画面あり・最小単元	20取引以上	注文受付、滑り、部分約定
限定自動化	安定後	1銘柄・低金額・日次上限

5分足バックテストでは、シグナル足の終値で約定させず次の足の始値を使います。利益確定と損切りへ同じ足で到達した場合、OHLCだけでは順序が分からないため、保守的に損切りを先とするか、より細かいデータで確認します。スプレッドとスリッページを片道0.1〜0.3%としても優位性が残るか確認してください。

楽天証券編の利点は、国内株のデータ取得から注文・約定照会まで公式RSS関数の範囲で構築できることです。一方、Excelとデスクトップアプリへの依存、再起動、発注可状態、セル再計算など、REST APIにはない障害点があります。Pythonコードだけでなく、Excelブックと運用手順もシステムの一部としてバージョン管理します。

最初の完成形は、Pythonがシグナルと数量を計算し、Excelへ発注トリガー0で条件を配置し、人が確認後にARMする構成です。これで価格計算ミスや手入力の負担を減らしながら、誤発注を止める最後のゲートを残せます。運用ログが十分に蓄積してから、対象銘柄と上限金額を限定して自動化範囲を広げてください。

公式情報・参考資料

楽天証券：マーケットスピード II RSS — 機能、準備、サンプルシート
公式オンラインヘルプ：RSSとは — 情報取得・注文・照会機能
公式オンラインヘルプ：注文機能利用時の設定 — 発注利用設定と暗証番号
公式オンラインヘルプ：注文 — RssStockOrderの引数と入力例
公式オンラインヘルプ：注文約定 — 注文一覧・約定一覧・保有株
マーケットスピード II RSS確認書兼同意書 — 仕様・リスク・注文受付の扱い