【MT5 API シストレ開発⑤】リアルタイムトレードを実装する
この記事は、システムトレードを自身のPCで実装・運用するためのシリーズ第5回です。
前回までは、Backtraderを用いた戦略のバックテスト方法や優位性を確かめるための評価方法を解説しました。
今回は、これまでに作成した戦略をMetaTrader5（MT5）に接続して、リアルタイムで動かす方法を解説します。
リアルタイムトレード
バックテストでは、過去データを使って「この戦略は良さそうか」を確認しました。
ただし、実際の取引ではプログラムが今この瞬間の相場を監視し、決められたタイミングで判断を続ける必要があります。
つまり、バックテストでやっていたことを、リアルタイムの世界で安全に繰り返す仕組みが必要です。
リアルタイムトレードを行うためのシステムは、どれくらい作り込むかによって、利便性を向上させたり管理しやすくしたりできますが、この記事では「実際に動くものをはじめて作る」という想定で、できるだけ最小設計のシステムを考えていきます。
相場を常時監視する仕組み
リアルタイムトレードでは、次の処理を繰り返します。
- 1.現在時刻を取得する
- 2.戦略を実行すべきタイミングか判定する
- 3.戦略を1回だけ実行する
- 4.少し待つ
この「1回実行して、待って、また実行」という形が基本となります。
無限に回るため、これをよく「常駐プログラム」と呼びます。
今回は、この常駐プログラムと、呼ばれたときだけ実行する「戦略プログラム」の2種類のプログラムを作成します。
Backtraderでは、以下のような形式の戦略を作成しました。
import backtrader as bt
class SmaCross(bt.Strategy):
params = dict(
fast=20,
slow=60,
size=100,
)
def __init__(self):
close = self.data.close
self.fast_ma = bt.ind.SMA(close, period=self.p.fast)
self.slow_ma = bt.ind.SMA(close, period=self.p.slow)
self.crossover = bt.ind.CrossOver(self.fast_ma, self.slow_ma)
def next(self):
if not self.position:
if self.crossover > 0:
self.buy(size=self.p.size)
else:
if self.crossover < 0:
self.close()
「params」は戦略のパラメーター、「__init__」は最初（初期化時）に一度だけ実行するプログラム、「next」は1期間進むごとに繰り返し実行するプログラムが書かれています。
この「1期間進むごとに繰り返し実行するプログラム」を「実際の時間が経過するたびに実行するプログラム」に変えて運用するイメージです。
戦略の拡張性
最小設計とは言いましたが、最低限の実用性を考え、今後システムトレード戦略を増やしていくことのできる設計にします。
常駐プログラム（main.py）と同じ場所に「strategies」という名前のフォルダを作り、その中に戦略（Pythonファイル）を入れます。
後から戦略を増やす場合は、strategiesにPythonファイルを配置することで、main.pyの変更は最小限に留めながら安全に戦略を追加できます。
機能の拡張性
ここまで説明した仕組みを作っておけば、後から別の機能を追加したり、さまざまなアップデートがやりやすくなります。
全ての機能を1つのプログラムに集約してしまうと、アップデートを行うたびに、既に正常に動作しているプログラムを誤って書き換えてしまうリスクがつきまといます。
機能ごとにプログラムを分割して作成することで、正常に動作している箇所は変更せずに、新たな機能の追加やコードの一部修正を行うことができます。
常駐プログラム「main.py」の作成
ここからは具体的にどのようなコードを書いていくのかを確認していきましょう。
なお、この記事内でのコードは、学習目的でできるだけシンプルに「実際に動くもの」を目指していることを前提にしており、実運用での安全性を保証するものではありません。
# main.py
import time
from datetime import datetime, timezone
import MetaTrader5 as mt5
from strategies.sma_cross_session import SmaCrossSession
def main():
# MT5 接続
if not mt5.initialize():
raise RuntimeError("MT5 initialize failed")
strategies = [
SmaCrossSession(),
]
last_exec_time = None
try:
while True:
now_utc = datetime.now(timezone.utc)
# ============================================
# 実行単位の設定
# --------------------------------------------
# ▼ 毎分実行する場合
# current_exec_time = now_utc.replace(second=0, microsecond=0)
#
# ▼ 毎時実行する場合
# current_exec_time = now_utc.replace(minute=0, second=0, microsecond=0)
# ============================================
current_exec_time = now_utc.replace(second=0, microsecond=0)
# 指定単位で1回だけ実行
if current_exec_time != last_exec_time:
for strategy in strategies:
strategy.run(mt5=mt5, now_utc=now_utc)
last_exec_time = current_exec_time
time.sleep(1)
finally:
mt5.shutdown()
if __name__ == "__main__":
main()
このプログラムは、1分に一度、作成した戦略を呼び出して実行し、毎分実行するために一度の処理で停止せず常駐するようになっています。
if __name__ == "__main__":
main()
この箇所では、作成した「main()」は停止操作が行われるまで無限ループします。
main()では、
- 1.MT5に接続
- 2.実行するタイミングかどうかを判定
- 3.戦略を実行
- 4.（終了時に）MT5の接続を解除
という流れをとっています。
MT5の接続の詳しい解説は、過去の記事を参照してください。
この例では、毎分実行するように動きますが、コード内のメモで毎時実行する場合の書き方も残しています。
2つ目以降の戦略を追加するときは？
from strategies.sma_cross_session import SmaCrossSession
この箇所で作成した戦略（この例ではsma_cross_session.py）を呼び出しています。
strategies = [
SmaCrossSession(),
]
importした戦略は、このコードで示すように配列で管理します。
戦略を後から加える場合は、strategiesフォルダに戦略ファイルを追加した上で、プログラムの該当箇所に新しい戦略を追記してください。
戦略作成
main.pyから呼び出される戦略ファイルがどのようになっているかを見ていきましょう。
ここでは、例として「2本のSMAがクロスしたら注文する」という単純な戦略を作成しています。
# sma_cross_session.py
import time
from datetime import timezone
import pytz
class SmaCrossSession:
"""
- 操作対象は「symbol × magic」のポジションだけ
- 逆ポジションを持たない（必要なら先に全決済→その後新規）
- 他戦略（別magic）のポジションには触らない
"""
def __init__(
self,
symbol: str = "USDJPY",
lot: float = 0.1,
tz: str = "UTC",
timeframe=None, # 例: mt5.TIMEFRAME_H1 を run() 側から渡してもOK
sma_fast: int = 5,
sma_slow: int = 20,
magic: int = 10001, # マジックナンバー
deviation: int = 20,
max_close_rounds: int = 8,
retries_per_pos: int = 2,
sleep_sec: float = 0.15,
):
self.symbol = symbol
self.lot = lot
self.tz = pytz.timezone(tz)
self.timeframe = timeframe
self.sma_fast = int(sma_fast)
self.sma_slow = int(sma_slow)
if self.sma_fast <= 0 or self.sma_slow <= 0:
raise ValueError("sma_fast / sma_slow must be positive integers")
if self.sma_fast >= self.sma_slow:
raise ValueError("sma_fast must be < sma_slow")
self.magic = int(magic)
self.deviation = int(deviation)
self.max_close_rounds = int(max_close_rounds)
self.retries_per_pos = int(retries_per_pos)
self.sleep_sec = float(sleep_sec)
# ==================================================
# Public API (main.pyから呼び出す箇所)
# ==================================================
def run(self, mt5, now_utc):
# --- シンボル有効化 ---
if not mt5.symbol_select(self.symbol, True):
return
# --- SMA Cross で desired position 決定 ---
desired_pos = self._calc_desired_pos(mt5)
# desired_pos: +1 long, 0 flat, -1 short
if desired_pos is None:
return
# --- 現在のポジション状況を取得 ---
managed = self._managed_positions(mt5)
current_pos = self._net_position_side(mt5, managed)
# --- 方向が同じなら何もしない ---
if current_pos == desired_pos:
return
# --- 既存 managed を全決済（できなければ新規しない） ---
if managed:
ok = self._close_all_managed(mt5)
if not ok:
return
# --- 新規ポジション ---
if desired_pos == 1:
self._open_position(mt5, mt5.ORDER_TYPE_BUY)
elif desired_pos == -1:
self._open_position(mt5, mt5.ORDER_TYPE_SELL)
else:
# flat：決済済みなら何もしない
return
# ==================================================
# シグナル
# ==================================================
def _calc_desired_pos(self, mt5):
"""
SMA(fast) > SMA(slow) => ロング
SMA(fast) < SMA(slow) => ショート
"""
timeframe = self.timeframe if self.timeframe is not None else mt5.TIMEFRAME_M1
bars_needed = self.sma_slow + 5 # 余裕
rates = mt5.copy_rates_from_pos(self.symbol, timeframe, 0, bars_needed)
if rates is None or len(rates) < self.sma_slow:
return None
closes = [r["close"] for r in rates]
fast = self._sma(closes, self.sma_fast)
slow = self._sma(closes, self.sma_slow)
if fast is None or slow is None:
return None
if fast > slow:
return 1
if fast < slow:
return -1
return 0
@staticmethod
def _sma(values, period: int):
if values is None or len(values) < period:
return None
s = 0.0
for v in values[-period:]:
s += float(v)
return s / float(period)
# ==================================================
# ポジション管理
# ==================================================
def _managed_positions(self, mt5):
ps = mt5.positions_get(symbol=self.symbol) or []
return [p for p in ps if getattr(p, "magic", None) == self.magic]
def _net_position_side(self, mt5, positions):
buy_vol = 0.0
sell_vol = 0.0
for p in positions or []:
if p.type == mt5.POSITION_TYPE_BUY:
buy_vol += float(p.volume)
elif p.type == mt5.POSITION_TYPE_SELL:
sell_vol += float(p.volume)
if buy_vol > sell_vol:
return 1
if sell_vol > buy_vol:
return -1
return 0
# ==================================================
# 注文
# ==================================================
def _open_position(self, mt5, order_type):
tick = mt5.symbol_info_tick(self.symbol)
if tick is None:
return
price = tick.ask if order_type == mt5.ORDER_TYPE_BUY else tick.bid
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": self.symbol,
"volume": self.lot,
"type": order_type,
"price": price,
"deviation": self.deviation,
"magic": self.magic,
"comment": "SMA Cross",
"type_time": mt5.ORDER_TIME_GTC,
}
result = mt5.order_send(request)
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
return
def _close_all_managed(self, mt5) -> bool:
"""
symbol×magic のポジションを全て閉じ切れたら True。
1つでも残る/閉じれないなら False（この場合は新規を出さない）。
"""
for _ in range(self.max_close_rounds):
ps = self._managed_positions(mt5)
if not ps:
return True
for p in ps:
closed = False
for _ in range(self.retries_per_pos):
if self._close_one(mt5, p):
closed = True
break
time.sleep(self.sleep_sec)
if not closed:
# この時点で「閉じれないポジがある」＝新規禁止
return False
time.sleep(self.sleep_sec)
# rounds 使い切りでも残っているなら失敗
return len(self._managed_positions(mt5)) == 0
def _close_one(self, mt5, position) -> bool:
tick = mt5.symbol_info_tick(self.symbol)
if tick is None:
return False
if position.type == mt5.POSITION_TYPE_BUY:
order_type = mt5.ORDER_TYPE_SELL
price = tick.bid
else:
order_type = mt5.ORDER_TYPE_BUY
price = tick.ask
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": self.symbol,
"volume": float(position.volume),
"type": order_type,
"position": position.ticket,
"price": price,
"deviation": self.deviation,
"magic": self.magic,
"comment": "SMA Cross close",
"type_time": mt5.ORDER_TIME_GTC,
}
result = mt5.order_send(request)
return (result is not None) and (result.retcode == mt5.TRADE_RETCODE_DONE)
パラメーター
「__init__」では戦略のパラメーターを書き込んでいます。
対象通貨ペア、ロット数、タイムフレーム、SMAの期間、マジックナンバーなど、システムトレード戦略に必要な設定項目を作っています。
マジックナンバーとは
システムトレードプログラムが出した注文・ポジションを識別するための「戦略ID」のことをマジックナンバーといいます。
マジックナンバーはパラメーターの「magic」で設定可能です。
複数の戦略を同時に運用する際に、戦略ごとに別々のマジックナンバーを振り分けておくことでポジション管理が楽になり、ある戦略が他の戦略のポジションを誤って決済するなどの事故を防げます。
プログラムの流れ
実際に呼び出されるのは「run()」です。
# --- SMA Cross で desired position 決定 ---
desired_pos = self._calc_desired_pos(mt5)
# desired_pos: +1 long, 0 flat, -1 short
if desired_pos is None:
return
「_calc_desired_pos()」にシグナル（SMAのクロス判定）を書きこみ、run()で判定結果を読み込みます。
戦略のロジックは「_calc_desired_pos()」の中に書き込むことになります。
「desired_pos」にシグナルの結果を格納し、この後の注文時に利用します。
# --- 現在のポジション状況を取得 ---
managed = self._managed_positions(mt5)
current_pos = self._net_position_side(mt5, managed)
「_managed_positions()」でマジックナンバーが一致するポジションを取得し、「_net_position_side()」で現在ロングしているか、ショートしているか、ノーポジションかを判断します。
# --- 方向が同じなら何もしない ---
if current_pos == desired_pos:
return
ロングポジションを持っているときにロングのシグナルが出るなど、シグナルと現在のポジションが一致している場合は、何もしないように制御します。
# --- 既存 managed を全決済（できなければ新規しない） ---
if managed:
ok = self._close_all_managed(mt5)
if not ok:
return
決済の必要があれば、この箇所で決済をします。
この後は新規注文のコードに続きますが、何かしらのトラブル（接続不良など）で決済されないときは、新規注文を行わないようにしています。
# --- 新規ポジション ---
if desired_pos == 1:
self._open_position(mt5, mt5.ORDER_TYPE_BUY)
elif desired_pos == -1:
self._open_position(mt5, mt5.ORDER_TYPE_SELL)
else:
# flat：決済済みなら何もしない
return
最後に、シグナルが発生していれば新規注文のコードを実行します。
「desired_pos」はシグナルを受け取る変数で、この中身に応じてロングするかショートするかを決定します。
動作確認
開発時には、少しのコードの書き間違いで、意図しないポジションを大量に持ったり、非常に大きなロット数で注文を出したりと、大きな損失につながる事故が起こる危険性があります。
動作確認は、デモ口座で十分に長い期間行い、意図した通りの動作を行えているか確認するようにしましょう。
MT5の左上で、現在デモ口座につながっているかどうかを確認することが可能です。
main.pyを動作させると、MT5上で実際にポジションを持つ様子が確認できるようになります。
MT5では、新規注文と決済注文がチャート上に描画され、ツールボックスで保有中のポジションや取引履歴も確認できます。
プログラムが正常に稼働し、意図した通りに動作しているかどうかを確認しましょう。
最後に
システムトレードを自身のPCで実装・運用するための本シリーズは、今回で最後になります。
この連載では、バックテスト方法、戦略の評価、リアルタイムトレードと、一連の流れを解説しました。
一連の流れの解説は本記事で終了となりますが、今後もシステムトレードの分析やプログラミングに関する記事は引き続き更新していく予定です。
【MT5 API シストレ開発】
- ・【MT5 API シストレ開発①】ルールベース戦略の概要
- ・【MT5 API シストレ開発②】シンプルな戦略をバックテストする方法
- ・【MT5 API シストレ開発➂】バックテスト応用編
- ・【MT5 API シストレ開発④】バックテストの評価方法
- ・【MT5 API シストレ開発⑤】リアルタイムトレードを実装
本記事の執筆者：藍崎@システムトレーダー
| 本記事の執筆者：藍崎@システムトレーダー | 経歴 |
|---|---|
 |
個人投資家としてEA開発＆システムトレード。 トレードに活かすためのデータサイエンス / 統計学 / 数理ファイナンス / 客観的なデータに基づくテクニカル分析 / 機械学習 / MQL5 / Python |
EA（自動売買）を学びたい方へオススメコンテンツ
OANDAではEA（自動売買）を稼働するプラットフォームMT4/MT5の基本的な使い方について、画像や動画付きで詳しく解説しています。MT4/MT5のインストールからEAの設定方法までを詳しく解説しているので、初心者の方でもスムーズにEA運用を始めることが可能です。またOANDAの口座をお持ちであれば、独自開発したオリジナルインジケーターを無料で利用することもできます。EA運用をお考えであれば、ぜひ口座開設をご検討ください。
本ホームページに掲載されている事項は、投資判断の参考となる情報の提供を目的としたものであり、投資の勧誘を目的としたものではありません。投資方針、投資タイミング等は、ご自身の責任において判断してください。本サービスの情報に基づいて行った取引のいかなる損失についても、当社は一切の責を負いかねますのでご了承ください。また、当社は、当該情報の正確性および完全性を保証または約束するものでなく、今後、予告なしに内容を変更または廃止する場合があります。なお、当該情報の欠落・誤謬等につきましてもその責を負いかねますのでご了承ください。
