「アルゴリズムトレード」は機関投資家のものというイメージがありますが、個人でもPythonで組めます。ただ、入門記事の多くは移動平均の計算とグラフ描画で終わっていて、そこから「実際に注文を出し続けるプログラム」までの距離が説明されていません。
この記事では、その距離を埋めます。シグナル生成・リスク管理・執行・状態管理を分けた、そのまま動くBOTの骨格を作ります。証券口座がなくても試せるように、ペーパートレード用の実装から始めて、実際の発注APIに差し替えられる形にします。
結論を先に書くと、自動売買で難しいのは戦略ではありません。プロセスが落ちても、二重に発注せず、正しい状態から再開できることです。ここを最初に作らないと、必ず事故ります。
自動売買を3つの部品に分ける
1つのスクリプトに全部書くと、必ず破綻します。役割で分けます。
| 部品 | 責務 | やってはいけないこと |
|---|---|---|
| Strategy | 売買シグナルを出す | 発注する、金額を決める |
| Risk | 数量と可否を決める | 相場観を持つ |
| Broker | 注文を出す・約定を返す | 売買判断をする |
| Store | ポジションと履歴を保存 | 売買判断をする |
この分離には実利があります。Strategyが発注を知らなければ、バックテストとライブで同じコードを使えます。「検証したものと違うロジックが動いていた」という最悪の事故が構造的に起きなくなります。
Broker：ペーパートレードから始める
まず注文を受け取るインターフェースを決めます。実装をペーパーと本番で差し替えられるようにしておきます。
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
import datetime as dt
@dataclass
class Order:
symbol: str
side: str # "BUY" or "SELL"
qty: int
client_id: str # 二重発注を防ぐための自前の識別子
@dataclass
class Fill:
order: Order
price: float
at: dt.datetime
fee: float
class Broker(ABC):
@abstractmethod
def last_price(self, symbol: str) -> float: ...
@abstractmethod
def send(self, order: Order) -> Fill | None: ...
@abstractmethod
def cash(self) -> float: ...
class PaperBroker(Broker):
"""実際には発注せず、終値＋スリッページで約定したことにする。"""
def __init__(self, cash: float = 1_000_000, fee_rate: float = 0.0005,
slippage: float = 0.001):
self._cash = cash
self.fee_rate = fee_rate
self.slippage = slippage
self.prices: dict[str, float] = {}
self.fills: list[Fill] = []
self._seen: set[str] = set()
def set_price(self, symbol: str, price: float) -> None:
self.prices[symbol] = price
def last_price(self, symbol: str) -> float:
return self.prices[symbol]
def cash(self) -> float:
return self._cash
def send(self, order: Order) -> Fill | None:
if order.client_id in self._seen: # 冪等性チェック
print(f"重複注文を無視: {order.client_id}")
return None
self._seen.add(order.client_id)
base = self.prices[order.symbol]
# 買いは不利な方向（高く）、売りも不利な方向（安く）に滑る
px = base * (1 + self.slippage) if order.side == "BUY" \
else base * (1 - self.slippage)
gross = px * order.qty
fee = gross * self.fee_rate
self._cash += -gross - fee if order.side == "BUY" else gross - fee
fill = Fill(order, px, dt.datetime.now(), fee)
self.fills.append(fill)
return fillスリッページを必ず不利な方向に入れるのがポイントです。終値ちょうどで約定する前提のペーパートレードは、実弾に移した瞬間に成績が落ちます。
client_idによる重複チェックも最初から入れておきます。実際の証券会社APIにはこの仕組みが無いことが多いので、自分側で持つ必要があります。
Strategy：シグナルだけを返す
戦略は「今日どうすべきか」だけを返します。金額も口座残高も知りません。
import numpy as np
import pandas as pd
import yfinance as yf
def load(symbol: str, period: str = "2y") -> pd.DataFrame:
df = yf.download(symbol, period=period, auto_adjust=True, progress=False)
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
return df.dropna()
class MaCrossStrategy:
"""短期MAが長期MAを上抜けたら買い、下抜けたら手仕舞い。"""
def __init__(self, short: int = 25, long: int = 75):
self.short, self.long = short, long
def signals(self, df: pd.DataFrame) -> pd.Series:
c = df["Close"]
ma_s = c.rolling(self.short).mean()
ma_l = c.rolling(self.long).mean()
want = (ma_s > ma_l).astype(int) # 保有したい状態（1 or 0）
# 当日の終値で判定したことを、翌日以降の行動に反映する
return want.shift(1).fillna(0).astype(int)
def decide(self, df: pd.DataFrame, holding: bool) -> str:
want = int(self.signals(df).iloc[-1])
if want == 1 and not holding:
return "BUY"
if want == 0 and holding:
return "SELL"
return "HOLD".shift(1)が命綱です。これが無いと、当日の終値を見て当日の終値で売買したことになります。バックテストの成績だけが良くなり、実運用でまったく再現しません。入門記事のサンプルコードで最も多い誤りがこれです。
そしてsignals()がSeries全体を返すようにしておくと、同じメソッドをバックテストでも使えます。ライブでは最終行だけを見ます。
Risk：数量と可否を決める
ここが戦略よりも成績を左右します。いくら賭けるかは、何を買うかより重要です。
@dataclass
class RiskConfig:
risk_per_trade: float = 0.01 # 1トレードで許容する損失（資産比）
stop_atr_mult: float = 2.0 # 損切り幅 = ATR × この倍率
max_position_ratio: float = 0.3 # 1銘柄への最大配分
max_daily_loss: float = 0.03 # 日次損失がこれを超えたら全停止
unit: int = 100 # 日本株の売買単位
def atr(df: pd.DataFrame, n: int = 14) -> float:
h, l, c = df["High"], df["Low"], df["Close"]
prev = c.shift(1)
tr = pd.concat([h - l, (h - prev).abs(), (l - prev).abs()], axis=1).max(axis=1)
return float(tr.rolling(n).mean().iloc[-1])
class RiskManager:
def __init__(self, cfg: RiskConfig):
self.cfg = cfg
def size(self, equity: float, price: float, df: pd.DataFrame) -> int:
"""損切り幅から逆算して、単元株単位の数量を返す。"""
stop_width = atr(df) * self.cfg.stop_atr_mult
if not np.isfinite(stop_width) or stop_width <= 0:
return 0
qty_by_risk = (equity * self.cfg.risk_per_trade) / stop_width
qty_by_cap = (equity * self.cfg.max_position_ratio) / price
qty = int(min(qty_by_risk, qty_by_cap))
return (qty // self.cfg.unit) * self.cfg.unit # 単元に丸める
def halted(self, equity: float, start_equity: float) -> bool:
return (equity / start_equity - 1) <= -self.cfg.max_daily_loss数量の決め方が肝です。「100株ずつ買う」ではなく、損切りにかかったときの損失額が資産の1%になる株数を計算します。値動きの荒い銘柄では自動的に少なくなり、穏やかな銘柄では多くなります。
日本株は単元株制度があるので、最後に100株単位へ切り捨てます。ここを忘れると発注エラーで止まります。1株の値段が高い銘柄では、計算結果が100株に満たずqty=0になることもあります。それも正しい挙動です（資金に対して大きすぎる）。
Store：落ちても再開できるようにする
変数にポジションを持つと、プロセスが落ちた瞬間にすべて失われます。次に起動したBOTは「ポジションなし」と誤認して、二重に買います。状態はファイルかDBに置きます。
import sqlite3
from pathlib import Path
class Store:
def __init__(self, path: str = "bot.db"):
self.con = sqlite3.connect(path)
self.con.executescript("""
CREATE TABLE IF NOT EXISTS positions(
symbol TEXT PRIMARY KEY,
qty INTEGER NOT NULL,
avg_price REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS fills(
client_id TEXT PRIMARY KEY,
symbol TEXT, side TEXT, qty INTEGER,
price REAL, fee REAL, at TEXT
);
""")
self.con.commit()
def position(self, symbol: str) -> tuple[int, float]:
row = self.con.execute(
"SELECT qty, avg_price FROM positions WHERE symbol=?", (symbol,)
).fetchone()
return (row[0], row[1]) if row else (0, 0.0)
def already_done(self, client_id: str) -> bool:
return self.con.execute(
"SELECT 1 FROM fills WHERE client_id=?", (client_id,)
).fetchone() is not None
def record(self, fill: Fill) -> None:
o = fill.order
qty, avg = self.position(o.symbol)
if o.side == "BUY":
new_qty = qty + o.qty
new_avg = (qty * avg + o.qty * fill.price) / new_qty
else:
new_qty = qty - o.qty
new_avg = avg if new_qty > 0 else 0.0
with self.con:
self.con.execute(
"INSERT OR REPLACE INTO positions VALUES (?,?,?)",
(o.symbol, new_qty, new_avg))
self.con.execute(
"INSERT OR IGNORE INTO fills VALUES (?,?,?,?,?,?,?)",
(o.client_id, o.symbol, o.side, o.qty,
fill.price, fill.fee, fill.at.isoformat()))client_idを主キーにしているのが要点です。同じIDの約定は物理的に二重登録できません。DBの制約でガードするのが、アプリのif文で頑張るより確実です。
メインループ：1日1回、冪等に
import logging
from zoneinfo import ZoneInfo
JST = ZoneInfo("Asia/Tokyo")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.FileHandler("bot.log", encoding="utf-8"),
logging.StreamHandler()],
)
log = logging.getLogger("bot")
class Bot:
def __init__(self, symbols: list[str], broker: Broker,
strategy: MaCrossStrategy, risk: RiskManager, store: Store):
self.symbols = symbols
self.broker, self.strategy = broker, strategy
self.risk, self.store = risk, store
def run_once(self) -> None:
today = dt.datetime.now(JST).date().isoformat()
for symbol in self.symbols:
try:
self._handle(symbol, today)
except Exception:
# 1銘柄の失敗で全体を止めない
log.exception("処理に失敗: %s", symbol)
def _handle(self, symbol: str, today: str) -> None:
df = load(symbol, "1y")
last_bar = df.index[-1].date().isoformat()
price = float(df["Close"].iloc[-1])
self.broker.set_price(symbol, price) if isinstance(
self.broker, PaperBroker) else None
qty_held, _ = self.store.position(symbol)
action = self.strategy.decide(df, holding=qty_held > 0)
log.info("%s price=%.1f held=%d action=%s", symbol, price,
qty_held, action)
if action == "HOLD":
return
# 「銘柄・日付・行動」で一意なIDを作る。同日に2回動かしても発注は1回
client_id = f"{symbol}-{last_bar}-{action}"
if self.store.already_done(client_id):
log.info("本日処理済みのためスキップ: %s", client_id)
return
if action == "BUY":
equity = self.broker.cash()
qty = self.risk.size(equity, price, df)
if qty <= 0:
log.warning("%s 数量が0のため見送り", symbol)
return
else:
qty = qty_held
fill = self.broker.send(Order(symbol, action, qty, client_id))
if fill:
self.store.record(fill)
log.info("約定 %s %s %d株 @%.1f", symbol, action, qty, fill.price)client_idを「銘柄＋足の日付＋行動」で組み立てているのが、この設計の中心です。手動で2回実行しても、cronが重複起動しても、発注は1回しか通りません。datetime.now()ではなくデータの最終日を使うことで、日付をまたいだ再実行にも耐えます。
また、try/exceptを銘柄ごとに置いています。1銘柄のデータ取得が失敗しただけで、他の銘柄の損切りが実行されないのは困ります。
動かしてみる
if __name__ == "__main__":
broker = PaperBroker(cash=1_000_000)
bot = Bot(
symbols=["7203.T", "6758.T", "8306.T"],
broker=broker,
strategy=MaCrossStrategy(25, 75),
risk=RiskManager(RiskConfig()),
store=Store("bot.db"),
)
bot.run_once()
print(f"\n現金: {broker.cash():,.0f}円")
for f in broker.fills:
print(f" {f.order.side} {f.order.symbol} {f.order.qty}株 "
f"@{f.price:,.1f} 手数料{f.fee:,.0f}円")2回続けて実行してみてください。2回目は「本日処理済み」でスキップされるはずです。これが確認できたら、土台はできています。
Windowsならタスクスケジューラ、LinuxならcronやsystemdタイマーでUTC 6:10（日本時間15:10）あたりに毎日実行します。大引け後に判断し、翌日の寄り付きで執行する運用が現実的です。
本番の証券会社APIに差し替える
Brokerを実装し直すだけで、他のコードは変わりません。日本株で個人が使える公式APIは、三菱UFJ eスマート証券（旧auカブコム証券）のkabuステーションAPIが現実的な選択肢です。
class KabuBroker(Broker):
def __init__(self, base: str, api_password: str, trade_password: str):
self.base, self.pw, self.trade_pw = base, api_password, trade_password
self._token = None
def send(self, order: Order) -> Fill | None:
# 実装の要点:
# 1) トークンは毎日失効するので自動再取得する
# 2) タイムアウトしても再送しない。注文照会で実在を確認する
# 3) 発注は秒間5件までなので間隔を空ける
...実APIでいちばん危険なのはタイムアウト時の再送です。応答が返らなくても注文は通っているかもしれません。素朴にリトライすると同じ成行注文が複数入ります。詳細はkabuステーションAPIの記事にまとめました。
実弾に移すまでの手順
- バックテスト。
Strategy.signals()を使って過去データで検証する。取引回数が数十回に満たないなら結論は出せない - ペーパートレードを1ヶ月。本番と同じスケジュールで動かし、落ちないことを確認する。ここで必ず何かが壊れる
- 最小単元で実弾。100株だけで1ヶ月。ペーパーとの成績差がスリッページの実測値になる
- 段階的に増やす。いきなり全額入れない
2番を飛ばしたくなりますが、飛ばすと必ず後悔します。祝日、決算またぎ、システムメンテナンス、ネットワーク断。コードのバグではなく環境のほうが先に壊れることを、身をもって知ることになります。
よくある失敗
| 失敗 | 結果 | 対策 |
|---|---|---|
シグナルをshiftしない | 検証だけ好成績 | 翌足で執行する前提を徹底 |
| 状態をメモリに持つ | 再起動で二重発注 | DBに永続化する |
| タイムアウトを再送 | 同じ注文が複数 | 照会してから判断 |
| 損切りがコード内だけ | BOT停止中に無防備 | 証券会社側に逆指値を置く |
| 固定株数で発注 | 銘柄ごとにリスクが不均一 | ATRから逆算する |
| 日次損失の上限がない | 悪い日に負け続ける | 閾値で全停止させる |
| ログを出さない | 原因が追えない | 判断と約定を全部残す |
まとめ
自動売買プログラムの本体は、売買ルールではありません。落ちても壊れない土台です。
戦略を差し替えるのはStrategyクラスを書き換えるだけで済みます。しかし、二重発注や状態消失は設計を直さないと解決しません。だから順序が大事で、先に土台を作り、あとから戦略を載せるのが正解です。
まずはこの記事のコードをそのまま動かして、2回実行してもスキップされることを確認してください。そこから自分の戦略をStrategyに書けば、あとは同じ枠組みで回せます。
