筆者の備忘録としてソースコードを置いています。使い方・トレードロジックなどはLLMにコピペして聞いてください。（Gemini、chatGPT、Claude、Kimiなど）
目次
Stop-Grid Traderの無料公開ソースコード
Stop-Grid Trader（CLI版）：ストラドル＋ピラミッディング
Chisikiさんの公開したコードを、筆者がCLI向けに書き直したもの。
"""
Stop-Grid Trader CLI – version 3.2
------------------------------------------------------------
MT5用 対称ストップグリッド・トレーディング・ボット (CLI版)
・起動中のMT5に自動接続
・Ctrl+Cで全ポジション決済＆予約注文削除して終了
"""
import time
import sys
import MetaTrader5 as mt5
from datetime import datetime
# =================================================================
# SETTING (パラメータ設定)
# =================================================================
SYMBOL = "XAUUSD" # 取扱通貨ペア
PRICE_DIGITS = 2 # 価格の小数点桁数
BASE_LOT = 0.02 # ベースロット (0.02以上の偶数推奨)
ORDERS_PER_SIDE = 10 # 片側の注文数
GRID_MULTIPLIER = 2.0 # スプレッドに対するグリッド幅の倍率
LOOP_COUNT = 100 # 全決済後の再開回数 (0なら1回のみ実行)
# 固定定数
DEVIATION = 100
MAGIC_NUMBER = 10 # マジックナンバー
GRID_TAG = "basic grid"
CHECK_INTERVAL = 1.0 # 監視間隔（秒）
# =================================================================
class StopGridTraderCLI:
def __init__(self):
self.symbol = SYMBOL
self.digits = PRICE_DIGITS
self.lot = BASE_LOT
self.side = ORDERS_PER_SIDE
self.mult = GRID_MULTIPLIER
self.loopN = LOOP_COUNT
self.done = 0
self.mid = None
self.step_pts = None
self.tp_high = None
self.tp_low = None
self.running = False
self.aborting = False # 手動終了フラグ
def log(self, message):
now = datetime.now().strftime("%H:%M:%S")
print(f"[{now}] {message}")
def _mt5_init(self):
if not mt5.initialize():
print("MT5初期化失敗: ターミナルが起動しているか確認してください。")
sys.exit()
if not mt5.symbol_select(self.symbol, True):
self.log(f"エラー: 通貨ペア {self.symbol} が見つかりません。")
mt5.shutdown()
sys.exit()
acc = mt5.account_info()
self.log(f"MT5接続成功: Account={acc.login}, Server={acc.server}")
def _norm_vol(self, vol: float) -> float:
info = mt5.symbol_info(self.symbol)
step = info.volume_step or 0.01
return round(max(info.volume_min, min(vol, info.volume_max)) / step) * step
def _pend(self, ord_type, price, sl, tp=0.0, vol=None, tag=GRID_TAG):
if vol is None: vol = self.lot
request = {
"action": mt5.TRADE_ACTION_PENDING,
"symbol": self.symbol,
"volume": self._norm_vol(vol),
"type": ord_type,
"price": round(price, self.digits),
"sl": round(sl, self.digits),
"tp": round(tp, self.digits) if tp > 0 else 0.0,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": tag,
"type_time": mt5.ORDER_TIME_GTC,
}
mt5.order_send(request)
def _build_grid(self):
tick = mt5.symbol_info_tick(self.symbol)
info = mt5.symbol_info(self.symbol)
if tick is None or info is None: return
self.mid = round((tick.bid + tick.ask) / 2, self.digits)
raw_spd_pts = int(round((tick.ask - tick.bid) / info.point))
self.step_pts = int(raw_spd_pts * self.mult)
pt = info.point
self.log(f"グリッド構築: Mid={self.mid}, Step={self.step_pts}pts")
for i in range(1, self.side + 1):
buy = self.mid + i * self.step_pts * pt
sell = self.mid - i * self.step_pts * pt
tp_b = (buy + self.step_pts * pt) if i == self.side else 0.0
tp_s = (sell - self.step_pts * pt) if i == self.side else 0.0
if i == self.side:
self.tp_high, self.tp_low = tp_b, tp_s
self._pend(mt5.ORDER_TYPE_BUY_STOP, buy, self.mid, tp=tp_b)
self._pend(mt5.ORDER_TYPE_SELL_STOP, sell, self.mid, tp=tp_s)
self.log("グリッド配置完了。監視中... (中止は Ctrl+C)")
def _full_close(self):
"""全ポジション決済と全予約注文削除"""
self.log("全ポジション決済および予約注文の削除を実行中...")
# 1. 予約注文の削除
orders = mt5.orders_get(symbol=self.symbol) or []
for o in orders:
mt5.order_send({"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket})
# 2. ポジションの決済
positions = mt5.positions_get(symbol=self.symbol) or []
for p in positions:
tick = mt5.symbol_info_tick(self.symbol)
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = tick.bid if p.type == mt5.POSITION_TYPE_BUY else tick.ask
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL, "symbol": self.symbol,
"position": p.ticket, "volume": p.volume, "type": side,
"price": price, "deviation": DEVIATION, "magic": MAGIC_NUMBER, "comment": "grid exit"
})
self.log(f"クリーンアップ完了: 決済={len(positions)}件, 削除={len(orders)}件")
# 手動終了（abortingフラグ）でなければループ継続を確認
if not self.aborting:
self.done += 1
if self.done <= self.loopN:
self.log(f"ループ継続 ({self.done}/{self.loopN})。再起動します。")
time.sleep(2)
self._build_grid()
else:
self.log("全ループ完了。終了します。")
self.running = False
else:
self.running = False
def monitor(self):
self.running = True
try:
while self.running:
time.sleep(CHECK_INTERVAL)
tick = mt5.symbol_info_tick(self.symbol)
if not tick: continue
mid_now = (tick.bid + tick.ask) / 2
pt = mt5.symbol_info(self.symbol).point
# 利確ライン到達チェック
if (self.tp_high and mid_now >= self.tp_high) or \
(self.tp_low and mid_now <= self.tp_low):
self.log("ターゲット到達。")
self._full_close()
continue
# 部分利確監視
positions = mt5.positions_get(symbol=self.symbol) or []
for pos in positions:
if abs(pos.volume - self.lot) > 1e-6: continue
trg = (pos.price_open + self.step_pts * pt if pos.type == mt5.POSITION_TYPE_BUY
else pos.price_open - self.step_pts * pt)
hit = (pos.type == mt5.POSITION_TYPE_BUY and tick.bid >= trg) or \
(pos.type == mt5.POSITION_TYPE_SELL and tick.ask <= trg)
if hit:
self._close_partial(pos, self.lot / 2, "partial TP")
time.sleep(0.3)
self._handle_partial(pos)
except KeyboardInterrupt:
self.log("\n--- [中断検知] プログラムを安全に終了します ---")
self.aborting = True
self._full_close()
finally:
mt5.shutdown()
self.log("MT5から切断されました。終了します。")
def _close_partial(self, pos, vol, comment):
tick = mt5.symbol_info_tick(self.symbol)
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = tick.bid if pos.type == mt5.POSITION_TYPE_BUY else tick.ask
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL, "symbol": self.symbol,
"position": pos.ticket, "volume": self._norm_vol(vol),
"type": side, "price": price, "deviation": DEVIATION, "magic": MAGIC_NUMBER, "comment": comment
})
def _handle_partial(self, pos):
be_price = round(pos.price_open, self.digits)
mt5.order_send({
"action": mt5.TRADE_ACTION_SLTP, "symbol": self.symbol,
"position": pos.ticket, "sl": be_price, "tp": 0.0
})
# BE-REV注文の配置
info = mt5.symbol_info(self.symbol)
pt = info.point
if pos.type == mt5.POSITION_TYPE_BUY:
otype, sl = mt5.ORDER_TYPE_SELL_STOP, be_price + self.step_pts * pt
else:
otype, sl = mt5.ORDER_TYPE_BUY_STOP, be_price - self.step_pts * pt
self._pend(otype, be_price, sl, vol=self.lot, tag="BE-REV")
if __name__ == "__main__":
print("=======================================")
print(" Stop-Grid Trader CLI v3.2")
print(" 終了するには [Ctrl + C] を押してください")
print("=======================================")
trader = StopGridTraderCLI()
trader._mt5_init()
trader._build_grid()
trader.monitor()リセンター（再構築）ロジック追加バージョン
オーダー数が片側の半分消化された時に、グリッドを現在の価格で作り直す機能が追加されています。
"""
Stop-Grid Trader CLI – version 3.4
------------------------------------------------------------
MT5用 対称ストップグリッド・トレーディング・ボット (CLI版)
・リセンター機能搭載（GUI v3.1完全準拠モデル）
・ノーポジ時に注文の25%（片側の半分）が消化されたら即座に再構築
"""
import time
import sys
import MetaTrader5 as mt5
from datetime import datetime
# =================================================================
# SETTING (パラメータ設定)
# =================================================================
SYMBOL = "XAUUSD" # 取扱通貨ペア
PRICE_DIGITS = 2 # 価格の小数点桁数
BASE_LOT = 0.02 # ベースロット (0.02以上の偶数推奨)
ORDERS_PER_SIDE = 10 # 片側の注文数
GRID_MULTIPLIER = 2.0 # スプレッドに対するグリッド幅の倍率
LOOP_COUNT = 100 # 全決済後の再開回数 (0なら1回のみ実行)
# 固定定数
DEVIATION = 100
MAGIC_NUMBER = 10 # マジックナンバー
GRID_TAG = "basic grid"
CHECK_INTERVAL = 1.0 # 監視間隔（秒）
# =================================================================
class StopGridTraderCLI:
def __init__(self):
self.symbol = SYMBOL
self.digits = PRICE_DIGITS
self.lot = BASE_LOT
self.side = ORDERS_PER_SIDE
self.mult = GRID_MULTIPLIER
self.loopN = LOOP_COUNT
self.done = 0
self.mid = None
self.step_pts = None
self.tp_high = None
self.tp_low = None
self.running = False
self.aborting = False # 手動終了フラグ
def log(self, message):
now = datetime.now().strftime("%H:%M:%S")
print(f"[{now}] {message}")
def _mt5_init(self):
if not mt5.initialize():
print("MT5初期化失敗: ターミナルが起動しているか確認してください。")
sys.exit()
if not mt5.symbol_select(self.symbol, True):
self.log(f"エラー: 通貨ペア {self.symbol} が見つかりません。")
mt5.shutdown()
sys.exit()
acc = mt5.account_info()
self.log(f"MT5接続成功: Account={acc.login}, Server={acc.server}")
def _norm_vol(self, vol: float) -> float:
info = mt5.symbol_info(self.symbol)
step = info.volume_step or 0.01
return round(max(info.volume_min, min(vol, info.volume_max)) / step) * step
def _pend(self, ord_type, price, sl, tp=0.0, vol=None, tag=GRID_TAG):
if vol is None: vol = self.lot
request = {
"action": mt5.TRADE_ACTION_PENDING,
"symbol": self.symbol,
"volume": self._norm_vol(vol),
"type": ord_type,
"price": round(price, self.digits),
"sl": round(sl, self.digits),
"tp": round(tp, self.digits) if tp > 0 else 0.0,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": tag,
"type_time": mt5.ORDER_TIME_GTC,
}
mt5.order_send(request)
def _build_grid(self):
tick = mt5.symbol_info_tick(self.symbol)
info = mt5.symbol_info(self.symbol)
if tick is None or info is None: return
self.mid = round((tick.bid + tick.ask) / 2, self.digits)
raw_spd_pts = int(round((tick.ask - tick.bid) / info.point))
self.step_pts = int(raw_spd_pts * self.mult)
pt = info.point
self.log(f"グリッド構築: Mid={self.mid}, Step={self.step_pts}pts")
for i in range(1, self.side + 1):
buy = self.mid + i * self.step_pts * pt
sell = self.mid - i * self.step_pts * pt
tp_b = (buy + self.step_pts * pt) if i == self.side else 0.0
tp_s = (sell - self.step_pts * pt) if i == self.side else 0.0
if i == self.side:
self.tp_high, self.tp_low = tp_b, tp_s
self._pend(mt5.ORDER_TYPE_BUY_STOP, buy, self.mid, tp=tp_b)
self._pend(mt5.ORDER_TYPE_SELL_STOP, sell, self.mid, tp=tp_s)
self.log("グリッド配置完了。監視中... (中止は Ctrl+C)")
def _full_close(self):
"""全ポジション決済と全予約注文削除"""
self.log("全ポジション決済および予約注文の削除を実行中...")
# 1. 予約注文の削除
orders = mt5.orders_get(symbol=self.symbol) or []
for o in orders:
mt5.order_send({"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket})
# 2. ポジションの決済
positions = mt5.positions_get(symbol=self.symbol) or []
for p in positions:
tick = mt5.symbol_info_tick(self.symbol)
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = tick.bid if p.type == mt5.POSITION_TYPE_BUY else tick.ask
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL, "symbol": self.symbol,
"position": p.ticket, "volume": p.volume, "type": side,
"price": price, "deviation": DEVIATION, "magic": MAGIC_NUMBER, "comment": "grid exit"
})
self.log(f"クリーンアップ完了: 決済={len(positions)}件, 削除={len(orders)}件")
if not self.aborting:
self.done += 1
if self.done <= self.loopN:
self.log(f"ループ継続 ({self.done}/{self.loopN})。再起動します。")
time.sleep(2)
self._build_grid()
else:
self.log("全ループ完了。終了します。")
self.running = False
else:
self.running = False
def monitor(self):
self.running = True
try:
while self.running:
time.sleep(CHECK_INTERVAL)
tick = mt5.symbol_info_tick(self.symbol)
if not tick: continue
mid_now = (tick.bid + tick.ask) / 2
pt = mt5.symbol_info(self.symbol).point
# --- 1. 利確ライン到達チェック ---
if (self.tp_high and mid_now >= self.tp_high) or \
(self.tp_low and mid_now <= self.tp_low):
self.log("ターゲット到達。")
self._full_close()
continue
# --- 2. リセンター（再構築）ロジック ---
# GUI版(v3.1)と同じ「片側の本数の半分が消化された時」にリセットする設定
positions = mt5.positions_get(symbol=self.symbol) or []
if not positions:
all_orders = mt5.orders_get(symbol=self.symbol) or []
grid_orders = [o for o in all_orders if o.comment == GRID_TAG]
executed_count = (self.side * 2) - len(grid_orders)
# 判定基準を GUI版と同じ (side // 2) に設定
if executed_count >= (self.side // 2) and len(grid_orders) > 0:
self.log(f"リセンター発動: 注文の1/4以上({executed_count}本)が消化されました。再構築します。")
for o in grid_orders:
mt5.order_send({"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket})
time.sleep(1)
self._build_grid()
continue
# --- 3. 部分利確監視 ---
for pos in positions:
if abs(pos.volume - self.lot) > 1e-6: continue
trg = (pos.price_open + self.step_pts * pt if pos.type == mt5.POSITION_TYPE_BUY
else pos.price_open - self.step_pts * pt)
hit = (pos.type == mt5.POSITION_TYPE_BUY and tick.bid >= trg) or \
(pos.type == mt5.POSITION_TYPE_SELL and tick.ask <= trg)
if hit:
self._close_partial(pos, self.lot / 2, "partial TP")
time.sleep(0.3)
self._handle_partial(pos)
except KeyboardInterrupt:
self.log("\n--- [中断検知] プログラムを安全に終了します ---")
self.aborting = True
self._full_close()
finally:
mt5.shutdown()
self.log("MT5から切断されました。終了します。")
def _close_partial(self, pos, vol, comment):
tick = mt5.symbol_info_tick(self.symbol)
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = tick.bid if pos.type == mt5.POSITION_TYPE_BUY else tick.ask
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL, "symbol": self.symbol,
"position": pos.ticket, "volume": self._norm_vol(vol),
"type": side, "price": price, "deviation": DEVIATION, "magic": MAGIC_NUMBER, "comment": comment
})
def _handle_partial(self, pos):
be_price = round(pos.price_open, self.digits)
mt5.order_send({
"action": mt5.TRADE_ACTION_SLTP, "symbol": self.symbol,
"position": pos.ticket, "sl": be_price, "tp": 0.0
})
# BE-REV注文の配置
info = mt5.symbol_info(self.symbol)
pt = info.point
if pos.type == mt5.POSITION_TYPE_BUY:
otype, sl = mt5.ORDER_TYPE_SELL_STOP, be_price + self.step_pts * pt
else:
otype, sl = mt5.ORDER_TYPE_BUY_STOP, be_price - self.step_pts * pt
self._pend(otype, be_price, sl, vol=self.lot, tag="BE-REV")
if __name__ == "__main__":
print("=======================================")
print(" Stop-Grid Trader CLI v3.4")
print(" リセンター機能(GUI準拠)搭載モデル")
print(" 終了するには [Ctrl + C] を押してください")
print("=======================================")
trader = StopGridTraderCLI()
trader._mt5_init()
trader._build_grid()
trader.monitor()Stop-Grid VER：VERでエントリーを厳選してピラミッディング
VERが一定以上になったら、グリッドを展開するバージョン。ゴールドのボラは一度広がれば、しばらく継続しやすい。
# stop_grid_VER.py
import MetaTrader5 as mt5
import numpy as np
import pandas as pd
import time
import datetime
import pytz
import signal
import sys
# ==========================================
# 1. Configuration Parameters
# ==========================================
SYMBOL = "XAUUSD"
MAGIC_NUMBER = 100000
BASE_LOT = 0.02
# --- Strategy Parameters ---
ORDERS_PER_SIDE = 12 # 片側12本（合計24本）
GRID_MULTIPLIER = 6.0
VOL_RATIO_THRESHOLD = 1.4
SPREAD_LIMIT = 45
SHORT_VOL_PERIOD = 5
LONG_VOL_PERIOD = 250
# --- Time Control (Server Time: EET) ---
SERVER_TZ = pytz.timezone('EET')
START_HOUR_SERVER = 2
END_ENTRY_HOUR_SERVER = 23
EXIT_MINUTE_MON_THU = 45
EXIT_MINUTE_FRI = 15
DEVIATION = 20
CHECK_INTERVAL = 0.1
GRID_TAG = "basic grid"
BEREV_TAG = "BE-REV"
class VolatilityStopGridBot:
def __init__(self):
self.is_active = False
self.last_log_time = 0
self.tp_high = None
self.tp_low = None
self.mid_at_start = None
self.step_pts = None
self.filling_mode = None
self.setup_signal_handler()
self.initialize_mt5()
self.recover_state()
def setup_signal_handler(self):
signal.signal(signal.SIGINT, self.handle_exit)
signal.signal(signal.SIGTERM, self.handle_exit)
def handle_exit(self, sig, frame):
print(f"\n[{datetime.datetime.now()}] Exit signal detected!")
try:
if not mt5.terminal_info(): mt5.initialize()
self.close_all_positions_and_orders()
finally:
mt5.shutdown()
sys.exit(0)
def initialize_mt5(self):
if not mt5.initialize():
print("MT5 initialization failed"); sys.exit(1)
if not mt5.symbol_select(SYMBOL, True):
print(f"Failed to select {SYMBOL}"); sys.exit(1)
info = mt5.symbol_info(SYMBOL)
if info.filling_mode & 1:
self.filling_mode = mt5.ORDER_FILLING_FOK
elif info.filling_mode & 2:
self.filling_mode = mt5.ORDER_FILLING_IOC
else:
self.filling_mode = mt5.ORDER_FILLING_RETURN
print(f"MT5 connected: {SYMBOL} / FillingMode:{self.filling_mode} / Magic:{MAGIC_NUMBER}")
def recover_state(self):
"""再起動時に既存の状態を復元"""
positions = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
if (positions and len(positions) > 0) or (orders and len(orders) > 0):
print("\n[Recovery] Active trades found. Restoring parameters...")
self.is_active = True
info = mt5.symbol_info(SYMBOL)
if orders and len(orders) >= 2:
prices = sorted([o.price_open for o in orders])
diffs = [round(prices[i+1] - prices[i], info.digits) for i in range(len(prices)-1)]
if diffs:
self.step_pts = max(set(diffs), key=diffs.count)
if orders:
tps_buy = [o.tp for o in orders if o.type == mt5.ORDER_TYPE_BUY_STOP and o.tp > 0]
tps_sell = [o.tp for o in orders if o.type == mt5.ORDER_TYPE_SELL_STOP and o.tp > 0]
if tps_buy: self.tp_high = max(tps_buy)
if tps_sell: self.tp_low = min(tps_sell)
def _round_volume(self, vol):
info = mt5.symbol_info(SYMBOL)
step = info.volume_step
rounded_vol = round(round(vol / step) * step, 2)
return max(min(rounded_vol, info.volume_max), info.volume_min)
def check_connection(self):
return mt5.terminal_info() is not None and mt5.terminal_info().connected
def reset_state_vars(self):
self.is_active = False
self.tp_high, self.tp_low, self.mid_at_start, self.step_pts = None, None, None, None
def calculate_volatility_expansion_ratio(self) -> float:
rates = mt5.copy_rates_from_pos(SYMBOL, mt5.TIMEFRAME_M1, 0, LONG_VOL_PERIOD + 1)
if rates is None or len(rates) < LONG_VOL_PERIOD: return 0.0
df = pd.DataFrame(rates)
df['range'] = df['high'] - df['low']
return df['range'].iloc[-SHORT_VOL_PERIOD:].mean() / df['range'].iloc[-LONG_VOL_PERIOD:].mean()
def _send_order(self, req):
if "type_filling" not in req: req["type_filling"] = self.filling_mode
res = mt5.order_send(req)
if res is None: return None
if res.retcode != mt5.TRADE_RETCODE_DONE:
print(f"Order failed: {res.comment} (Code: {res.retcode})")
return res
def deploy_grid(self, tick, info):
self.mid_at_start = round((tick.ask + tick.bid) / 2, info.digits)
self.step_pts = round((tick.ask - tick.bid) * GRID_MULTIPLIER, info.digits)
print(f"\n[{datetime.datetime.now()}] ★★★ Deploying Grid Mid:{self.mid_at_start} ★★★")
for i in range(1, ORDERS_PER_SIDE + 1):
buy_p, sell_p = self.mid_at_start + (i * self.step_pts), self.mid_at_start - (i * self.step_pts)
tp_b = (buy_p + self.step_pts) if i == ORDERS_PER_SIDE else 0.0
tp_s = (sell_p - self.step_pts) if i == ORDERS_PER_SIDE else 0.0
if i == ORDERS_PER_SIDE: self.tp_high, self.tp_low = tp_b, tp_s
self._pend(mt5.ORDER_TYPE_BUY_STOP, buy_p, self.mid_at_start, tp_b, info.digits)
self._pend(mt5.ORDER_TYPE_SELL_STOP, sell_p, self.mid_at_start, tp_s, info.digits)
self.is_active = True
def _pend(self, otype, price, sl, tp, digits, vol=BASE_LOT, tag=GRID_TAG):
vol = self._round_volume(vol)
req = {
"action": mt5.TRADE_ACTION_PENDING, "symbol": SYMBOL, "volume": float(vol),
"type": otype, "price": round(price, digits), "sl": round(sl, digits),
"tp": round(tp, digits) if tp > 0 else 0.0, "magic": MAGIC_NUMBER,
"comment": tag, "type_time": mt5.ORDER_TIME_GTC
}
return self._send_order(req)
def manage_grid(self, tick, info):
mid_now = (tick.bid + tick.ask) / 2
positions = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
# 目標到達確認 (外側TP)
if (self.tp_high and mid_now >= self.tp_high) or (self.tp_low and mid_now <= self.tp_low):
print("\nTarget reached (Outer TP).")
self.close_all_positions_and_orders()
return
# ==========================================
# ★リセンター判定: 片側の本数の半分が消化された時
# ==========================================
if self.is_active and not positions:
grid_orders = [o for o in orders if o.comment == GRID_TAG]
# 消化された注文数 = 全初期本数 - 現在の待機本数
executed_count = (ORDERS_PER_SIDE * 2) - len(grid_orders)
# 閾値: 片側の本数の半分 (side // 2)
# 例: 片側12本なら、6本消化された時点で発動
recenter_threshold = ORDERS_PER_SIDE // 2
if executed_count >= recenter_threshold and len(grid_orders) > 0:
print(f"\n[Recenter] {executed_count} orders filled/missed (Threshold: {recenter_threshold}). Rebuilding grid.")
self.close_all_positions_and_orders()
return
for p in positions:
if self.step_pts is None: continue
if abs(p.volume - BASE_LOT) < 1e-7:
trg = (p.price_open + self.step_pts) if p.type == mt5.POSITION_TYPE_BUY else (p.price_open - self.step_pts)
if (p.type == mt5.POSITION_TYPE_BUY and tick.bid >= trg) or (p.type == mt5.POSITION_TYPE_SELL and tick.ask <= trg):
self._handle_partial_tp(p, info)
else:
self._update_trailing_stop(p, tick, info)
def _handle_partial_tp(self, pos, info):
half_vol = self._round_volume(pos.volume / 2)
tick = mt5.symbol_info_tick(SYMBOL)
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
self._send_order({
"action": mt5.TRADE_ACTION_DEAL, "symbol": SYMBOL, "volume": half_vol,
"type": side, "position": pos.ticket, "price": (tick.bid if side==mt5.ORDER_TYPE_SELL else tick.ask),
"magic": MAGIC_NUMBER, "comment": "partial TP", "deviation": DEVIATION
})
be_price = round(pos.price_open, info.digits)
self._send_order({"action": mt5.TRADE_ACTION_SLTP, "symbol": SYMBOL, "position": pos.ticket, "sl": be_price, "tp": 0.0})
otype, sl = (mt5.ORDER_TYPE_SELL_STOP, be_price + self.step_pts) if pos.type == mt5.POSITION_TYPE_BUY else (mt5.ORDER_TYPE_BUY_STOP, be_price - self.step_pts)
if (tick.bid > be_price if otype == mt5.ORDER_TYPE_SELL_STOP else tick.ask < be_price):
self._pend(otype, be_price, sl, 0.0, info.digits, vol=BASE_LOT, tag=BEREV_TAG)
def _update_trailing_stop(self, pos, tick, info):
if self.step_pts is None: return
new_sl = None
if pos.type == mt5.POSITION_TYPE_BUY:
level = int((tick.bid - pos.price_open) / self.step_pts)
if level >= 1:
calc_sl = round(pos.price_open + ((level - 1) * self.step_pts), info.digits)
if calc_sl > pos.sl + (info.point * 0.1): new_sl = calc_sl
else:
level = int((pos.price_open - tick.ask) / self.step_pts)
if level >= 1:
calc_sl = round(pos.price_open - ((level - 1) * self.step_pts), info.digits)
if pos.sl == 0.0 or calc_sl < pos.sl - (info.point * 0.1): new_sl = calc_sl
if new_sl: self._send_order({"action": mt5.TRADE_ACTION_SLTP, "symbol": SYMBOL, "position": pos.ticket, "sl": new_sl, "tp": pos.tp})
def close_all_positions_and_orders(self):
for o in (mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []):
mt5.order_send({"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket, "type_filling": self.filling_mode})
for p in (mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []):
tick = mt5.symbol_info_tick(SYMBOL)
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
self._send_order({
"action": mt5.TRADE_ACTION_DEAL, "symbol": SYMBOL, "volume": p.volume,
"type": side, "position": p.ticket, "price": (tick.bid if side==mt5.ORDER_TYPE_SELL else tick.ask),
"magic": MAGIC_NUMBER, "comment": "clean exit", "deviation": DEVIATION
})
self.reset_state_vars()
def log_status(self, vol_ratio, spread_pts, dt_server):
now = time.time()
if now - self.last_log_time >= 1.0:
status = "ACTIVE" if self.is_active else "IDLE"
time_str = dt_server.strftime('%H:%M:%S')
pos_count = len(mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or [])
print(f"[{status}] SrvTime:{time_str} | Vol:{vol_ratio:.2f} | Spd:{spread_pts:.1f} | Pos:{pos_count}", end='\r')
self.last_log_time = now
def main_loop(self):
info = mt5.symbol_info(SYMBOL)
while True:
try:
time.sleep(CHECK_INTERVAL)
if not self.check_connection(): continue
tick = mt5.symbol_info_tick(SYMBOL)
if not tick: continue
dt_server = datetime.datetime.now(datetime.UTC).astimezone(SERVER_TZ)
h, m, weekday = dt_server.hour, dt_server.minute, dt_server.weekday()
is_friday = (weekday == 4)
exit_m = EXIT_MINUTE_FRI if is_friday else EXIT_MINUTE_MON_THU
if h == 23 and m >= exit_m:
if self.is_active or mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER):
self.close_all_positions_and_orders()
is_entry_window = (START_HOUR_SERVER <= h < END_ENTRY_HOUR_SERVER)
vol_ratio = self.calculate_volatility_expansion_ratio()
spread_pts = (tick.ask - tick.bid) / info.point
self.log_status(vol_ratio, spread_pts, dt_server)
if self.is_active: self.manage_grid(tick, info)
else:
if is_entry_window and spread_pts <= SPREAD_LIMIT and vol_ratio > VOL_RATIO_THRESHOLD:
self.deploy_grid(tick, info)
except Exception as e:
print(f"Error: {e}"); time.sleep(1)
if __name__ == "__main__":
bot = VolatilityStopGridBot()
bot.main_loop()Stop-Grid VERのWFA分析結果
全体20日、最適化9日、テスト3日でのWFA分析結果はこちら。
==================================================
BEST PARAMETERS (Train Period):
grid_multiplier: 3.354992148973733
vol_ratio_threshold: 1.2717006774335127
orders_per_side: 8
short_vol_period: 4
long_vol_period: 309
--------------------------------------------------
WFA TEST RESULT (Last 3 days):
Net Profit: $1077.85
Max Drawdown: $4805.48
Recovery Factor: 0.22
Expectancy: $0.15
Min Margin Level: 2897.02%
Trade Count: 7258
Status: PASSED
==================================================リカバリーファクター0.22はかなり酷い。
TSLをこまめに動かしているため、うまく利益を伸ばせるように見えるが、実際には細かいノイズで損切りされてしまう。ノイズ対策として、バスケット決済などを取り入れるべきか。
Dual_Grid：ストップ・リミットの両建てグリッド
import MetaTrader5 as mt5
import time
import datetime
import pytz
import signal
import sys
from concurrent.futures import ThreadPoolExecutor
import threading
# ==========================================
# 1. FINAL PARAMETERS
# ==========================================
SYMBOL = "XAUUSD"
MAGIC_NUMBER = 999000
FIXED_LOT = 0.01
GRID_STEP_PTS = 105
ORDERS_PER_SIDE = 40
RECENTER_THRESHOLD = 0.85
TRAIL_TRIGGER = 48.0
TRAIL_MAX_WIDTH = 8.0
TRAIL_MIN_WIDTH = 1.5
TIGHTEN_FACTOR = 0.34
GLOBAL_TP_DOLLARS = 135.0
MIN_MARGIN_LEVEL = 250.0
SPREAD_LIMIT = 50
DEVIATION = 30
CHECK_INTERVAL = 0.01
JST = pytz.timezone('Asia/Tokyo')
class UltimateGoldSystem:
def __init__(self):
self.is_active = False
self.is_processing_exit = False
self.initial_order_count = 0
self.lock = threading.Lock() # For thread-safe printing
self.peak_buy_profit = -999999.0
self.is_buy_trailing = False
self.peak_sell_profit = -999999.0
self.is_sell_trailing = False
self.executor = ThreadPoolExecutor(max_workers=20)
self.ensure_mt5_connection()
signal.signal(signal.SIGINT, self.handle_signal)
signal.signal(signal.SIGTERM, self.handle_signal)
def safe_print(self, msg):
"""Prevents log garbling by clearing the current line before printing"""
with self.lock:
sys.stdout.write("\033[K") # Clear line
print(msg)
def handle_signal(self, sig, frame):
self.safe_print(f"[{datetime.datetime.now()}] STOP SIGNAL RECEIVED. PURGING...")
self.close_all_and_exit()
def ensure_mt5_connection(self):
terminal_info = mt5.terminal_info()
if terminal_info is None or not terminal_info.connected:
self.is_active = False
self.safe_print("MT5 Connection lost. Reconnecting...")
while True:
if mt5.initialize():
if mt5.symbol_select(SYMBOL, True):
return True
time.sleep(5.0)
return True
def _send_order(self, req):
# 10015対策: 指値・逆指値が現在価格の正しい側にあるか最終チェック
tick = mt5.symbol_info_tick(SYMBOL)
if not tick: return None
if req['action'] == mt5.TRADE_ACTION_PENDING:
p = req['price']
t = req['type']
# 価格の上下関係が不正なら注文を送らない
if t == mt5.ORDER_TYPE_BUY_STOP and p <= tick.ask: return None
if t == mt5.ORDER_TYPE_SELL_LIMIT and p <= tick.bid: return None
if t == mt5.ORDER_TYPE_BUY_LIMIT and p >= tick.ask: return None
if t == mt5.ORDER_TYPE_SELL_STOP and p >= tick.bid: return None
res = mt5.order_send(req)
if res and res.retcode != mt5.TRADE_RETCODE_DONE:
if res.retcode not in [10025, 10017]:
self.safe_print(f"[DEBUG] Order Rejected: {res.comment} ({res.retcode}) at {req.get('price')}")
return res
def is_within_trade_window(self, tick):
dt = datetime.datetime.fromtimestamp(tick.time)
weekday = dt.weekday()
time_val = dt.hour * 100 + dt.minute
if weekday == 0: return 135 <= time_val <= 2355
elif 1 <= weekday <= 3: return 110 <= time_val <= 2355
elif weekday == 4: return 110 <= time_val <= 2325
return False
def deploy_grid(self, tick, info):
if self.is_processing_exit: return
center = round((tick.ask + tick.bid) / 2, info.digits)
pt = info.point
# ストップレベル（最低離隔距離）を取得して1段目を補正
stops_level = info.trade_stops_level * pt
self.safe_print(f"⚡ Deploying Grid Center: {center}")
reqs = []
for i in range(1, ORDERS_PER_SIDE + 1):
offset = max(i * GRID_STEP_PTS * pt, stops_level + pt) # ストップレベルを考慮
up_p = round(center + offset, info.digits)
low_p = round(center - offset, info.digits)
for p, otype in [(up_p, mt5.ORDER_TYPE_SELL_LIMIT), (up_p, mt5.ORDER_TYPE_BUY_STOP),
(low_p, mt5.ORDER_TYPE_BUY_LIMIT), (low_p, mt5.ORDER_TYPE_SELL_STOP)]:
reqs.append({
"action": mt5.TRADE_ACTION_PENDING, "symbol": SYMBOL, "volume": float(FIXED_LOT),
"type": otype, "price": p, "magic": MAGIC_NUMBER, "comment": f"LV_{p}",
"type_filling": mt5.ORDER_FILLING_RETURN, # 10013対策: 注文タイプを明示
"type_time": mt5.ORDER_TIME_GTC
})
list(self.executor.map(self._send_order, reqs))
time.sleep(0.5)
self.initial_order_count = len(mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or [])
self.is_active = True
def manage_logic(self, pos):
if not pos or self.is_processing_exit: return
buys = [p for p in pos if p.type == mt5.POSITION_TYPE_BUY]
sells = [p for p in pos if p.type == mt5.POSITION_TYPE_SELL]
b_p = sum(p.profit + p.swap for p in buys)
s_p = sum(p.profit + p.swap for p in sells)
if (b_p + s_p) >= GLOBAL_TP_DOLLARS:
self.safe_print(f"🏆 GLOBAL TP: ${b_p + s_p:.2f}")
self.close_all_positions_and_orders(); return
# Dynamic Trailing
for side, profit, peak, is_trailing, p_type in [('buy', b_p, self.peak_buy_profit, self.is_buy_trailing, mt5.POSITION_TYPE_BUY),
('sell', s_p, self.peak_sell_profit, self.is_sell_trailing, mt5.POSITION_TYPE_SELL)]:
if profit >= TRAIL_TRIGGER:
if side == 'buy': self.is_buy_trailing = True
else: self.is_sell_trailing = True
if profit > peak:
if side == 'buy': self.peak_buy_profit = profit
else: self.peak_sell_profit = profit
is_active_trail = self.is_buy_trailing if side == 'buy' else self.is_sell_trailing
current_peak = self.peak_buy_profit if side == 'buy' else self.peak_sell_profit
if is_active_trail:
cur_w = max(TRAIL_MIN_WIDTH, TRAIL_MAX_WIDTH - (profit - TRAIL_TRIGGER) * TIGHTEN_FACTOR)
if profit < current_peak - cur_w:
self.safe_print(f"📈 {side.upper()} TRAIL EXIT: ${profit:.2f}")
self.close_side(p_type)
self._reset_trail_state(side)
# CloseBy
if buys and sells:
buy_map = {b.comment: b for b in buys if b.comment.startswith("LV_")}
for s in sells:
if s.comment in buy_map:
b = buy_map[s.comment]
if (s.profit + b.profit) > (FIXED_LOT * 200):
req = {"action": mt5.TRADE_ACTION_CLOSE_BY, "position": b.ticket, "position_by": s.ticket, "symbol": SYMBOL}
self._send_order(req)
def close_side(self, p_type):
all_p = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
tick = mt5.symbol_info_tick(SYMBOL)
reqs = []
for p in all_p:
if p.type == p_type:
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = tick.bid if p.type == mt5.POSITION_TYPE_BUY else tick.ask
reqs.append({"action": mt5.TRADE_ACTION_DEAL, "symbol": SYMBOL, "volume": p.volume,
"type": side, "position": p.ticket, "price": price, "magic": MAGIC_NUMBER, "deviation": DEVIATION})
list(self.executor.map(self._send_order, reqs))
def cancel_all_orders(self):
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
reqs = [{"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket} for o in orders]
list(self.executor.map(self._send_order, reqs))
def close_all_positions_and_orders(self):
self.is_processing_exit = True
self.cancel_all_orders()
all_p = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
if all_p:
tick = mt5.symbol_info_tick(SYMBOL)
reqs = []
for p in all_p:
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = tick.bid if p.type == mt5.POSITION_TYPE_BUY else tick.ask
reqs.append({"action": mt5.TRADE_ACTION_DEAL, "symbol": SYMBOL, "volume": p.volume,
"type": side, "position": p.ticket, "price": price, "magic": MAGIC_NUMBER, "deviation": DEVIATION})
list(self.executor.map(self._send_order, reqs))
self.is_active = False; self.is_processing_exit = False; self._reset_trail_state('both')
def _reset_trail_state(self, side):
if side in ['buy', 'both']: self.is_buy_trailing = False; self.peak_buy_profit = -999999.0
if side in ['sell', 'both']: self.is_sell_trailing = False; self.peak_sell_profit = -999999.0
def close_all_and_exit(self):
self.close_all_positions_and_orders()
self.executor.shutdown(wait=True); mt5.shutdown(); sys.exit(0)
def main_loop(self):
info = mt5.symbol_info(SYMBOL)
last_log = 0
while True:
try:
self.ensure_mt5_connection()
time.sleep(CHECK_INTERVAL)
tick = mt5.symbol_info_tick(SYMBOL)
if not tick: continue
pos = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
if len(pos) == 0 and len(orders) == 0: self.is_active = False
in_window = self.is_within_trade_window(tick)
if not in_window:
if len(pos) > 0 or len(orders) > 0:
self.safe_print("Market closing. Purging.")
self.close_all_positions_and_orders()
continue
if not self.is_active and not self.is_processing_exit:
if (tick.ask - tick.bid) / info.point <= SPREAD_LIMIT:
self.deploy_grid(tick, info)
elif self.is_active:
self.manage_logic(pos)
if not self.is_processing_exit and len(orders) < (self.initial_order_count * RECENTER_THRESHOLD):
acc = mt5.account_info()
if acc and acc.margin_level >= MIN_MARGIN_LEVEL:
self.cancel_all_orders(); self.deploy_grid(tick, info)
if time.time() - last_log > 1.0:
acc = mt5.account_info()
b_p = sum(p.profit + p.swap for p in pos if p.type==0)
s_p = sum(p.profit + p.swap for p in pos if p.type==1)
ml = acc.margin_level if acc and acc.margin_level > 0 else 0
sys.stdout.write(f"\r[LIVE] Pos:{len(pos)} | Buy_PL:${b_p:.1f} | Sell_PL:${s_p:.1f} | Margin:{ml:.0f}% \033[K")
sys.stdout.flush()
last_log = time.time()
except Exception as e:
self.safe_print(f"[ERROR] {e}"); time.sleep(1.0)
if __name__ == "__main__":
UltimateGoldSystem().main_loop()Dual_GridのWFA分析結果
全体20日、最適化12日、テスト8日間のテスト結果は以下の通り。（利益は8日間のもの）
=============================================
FINAL RF-OPTIMIZED RESULT（8日間）
=============================================
grid_step : 300
ops : 41
trigger : 68.72
max_w : 19.32
min_w : 1.4
factor : 0.68
gtp : 105.16
thr : 0.64
---------------------------------------------
Net Profit (Test) : $6704.96
Recovery Factor (Test) : 1.5047
Min Margin Level (Test) : 287.93%
=============================================WFAでパラを厳選しても、RFは1.5と低め。ポジションを大量に抱えるため、維持率も圧迫されやすい。
おそらく2/12のゴールド20,000pips変動には耐えられない。ゴールドの安易なナンピンはNG。
stop_grid_ATR：ATRを基準にグリッド展開
分析スクリプトで検証したところ、stop_grid TraderはATRが高い時にエントリーすると、勝率が高くなることがわかった。このコードはATRが一定の閾値に達した時にエントリー＋リセンターするようになっている。
# stop_grid_ATR_Filter_v4_mod.py
import MetaTrader5 as mt5
import numpy as np
import pandas as pd
import time
import datetime
import pytz
import signal
import sys
# ==========================================
# 1. Configuration Parameters
# ==========================================
SYMBOL = "XAUUSD"
MAGIC_NUMBER = 100000
BASE_LOT = 0.02
# --- Strategy Parameters ---
ORDERS_PER_SIDE = 8 # 片側8本（合計16本）
GRID_MULTIPLIER = 3.0 # スプレッドに対するグリッド幅倍率
# ★★★ NEW: ATR Filter Settings (Based on Backtest) ★★★
ATR_PERIOD = 10 # ATR計算期間
ATR_THRESHOLD = 1.5 # エントリー閾値 (検証結果: 1.5以上で高勝率)
SPREAD_LIMIT = 45 # 最大許容スプレッド (Points)
# --- Time Control (Server Time: EET) ---
SERVER_TZ = pytz.timezone('EET')
START_HOUR_SERVER = 2
END_ENTRY_HOUR_SERVER = 23
EXIT_MINUTE_MON_THU = 45
EXIT_MINUTE_FRI = 15
DEVIATION = 20
CHECK_INTERVAL = 0.1
GRID_TAG = "basic grid"
BEREV_TAG = "BE-REV"
class VolatilityStopGridBot:
def __init__(self):
self.is_active = False
self.last_log_time = 0
self.tp_high = None
self.tp_low = None
self.mid_at_start = None
self.step_pts = None
self.filling_mode = None
self.setup_signal_handler()
self.initialize_mt5()
self.recover_state()
def setup_signal_handler(self):
signal.signal(signal.SIGINT, self.handle_exit)
signal.signal(signal.SIGTERM, self.handle_exit)
def handle_exit(self, sig, frame):
print(f"\n[{datetime.datetime.now()}] Exit signal detected!")
try:
if not mt5.terminal_info(): mt5.initialize()
self.close_all_positions_and_orders()
finally:
mt5.shutdown()
sys.exit(0)
def initialize_mt5(self):
if not mt5.initialize():
print("MT5 initialization failed"); sys.exit(1)
if not mt5.symbol_select(SYMBOL, True):
print(f"Failed to select {SYMBOL}"); sys.exit(1)
info = mt5.symbol_info(SYMBOL)
if info.filling_mode & 1:
self.filling_mode = mt5.ORDER_FILLING_FOK
elif info.filling_mode & 2:
self.filling_mode = mt5.ORDER_FILLING_IOC
else:
self.filling_mode = mt5.ORDER_FILLING_RETURN
print(f"MT5 connected: {SYMBOL} / FillingMode:{self.filling_mode} / Magic:{MAGIC_NUMBER}")
print(f"Strategy: ATR Filter (Threshold > {ATR_THRESHOLD})")
def recover_state(self):
positions = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
if (positions and len(positions) > 0) or (orders and len(orders) > 0):
print("\n[Recovery] Active trades found. Restoring parameters...")
self.is_active = True
info = mt5.symbol_info(SYMBOL)
if orders and len(orders) >= 2:
prices = sorted([o.price_open for o in orders])
diffs = [round(prices[i+1] - prices[i], info.digits) for i in range(len(prices)-1)]
if diffs:
self.step_pts = max(set(diffs), key=diffs.count)
if orders:
tps_buy = [o.tp for o in orders if o.type == mt5.ORDER_TYPE_BUY_STOP and o.tp > 0]
tps_sell = [o.tp for o in orders if o.type == mt5.ORDER_TYPE_SELL_STOP and o.tp > 0]
if tps_buy: self.tp_high = max(tps_buy)
if tps_sell: self.tp_low = min(tps_sell)
def _round_volume(self, vol):
info = mt5.symbol_info(SYMBOL)
step = info.volume_step
rounded_vol = round(round(vol / step) * step, 2)
return max(min(rounded_vol, info.volume_max), info.volume_min)
def check_connection(self):
return mt5.terminal_info() is not None and mt5.terminal_info().connected
def reset_state_vars(self):
self.is_active = False
self.tp_high, self.tp_low, self.mid_at_start, self.step_pts = None, None, None, None
# ==========================================
# ATR Calculation Logic
# ==========================================
def calculate_current_atr(self) -> float:
"""検証結果に基づき、直近のATR(14)を計算して返す"""
rates = mt5.copy_rates_from_pos(SYMBOL, mt5.TIMEFRAME_M1, 0, ATR_PERIOD + 2)
if rates is None or len(rates) < ATR_PERIOD + 1:
return 0.0
df = pd.DataFrame(rates)
# TR計算: Max(H-L, |H-Cp|, |L-Cp|)
df['h-l'] = df['high'] - df['low']
df['h-cp'] = np.abs(df['high'] - df['close'].shift(1))
df['l-cp'] = np.abs(df['low'] - df['close'].shift(1))
df['tr'] = df[['h-l', 'h-cp', 'l-cp']].max(axis=1)
# ATR計算 (直近の値を取得)
atr = df['tr'].rolling(window=ATR_PERIOD).mean().iloc[-1]
return atr if not np.isnan(atr) else 0.0
def _send_order(self, req):
if "type_filling" not in req: req["type_filling"] = self.filling_mode
res = mt5.order_send(req)
if res is None: return None
if res.retcode != mt5.TRADE_RETCODE_DONE:
print(f"Order failed: {res.comment} (Code: {res.retcode})")
return res
def deploy_grid(self, tick, info):
self.mid_at_start = round((tick.ask + tick.bid) / 2, info.digits)
self.step_pts = round((tick.ask - tick.bid) * GRID_MULTIPLIER, info.digits)
print(f"\n[{datetime.datetime.now()}] ★★★ Deploying Grid Mid:{self.mid_at_start} (ATR High) ★★★")
for i in range(1, ORDERS_PER_SIDE + 1):
buy_p, sell_p = self.mid_at_start + (i * self.step_pts), self.mid_at_start - (i * self.step_pts)
tp_b = (buy_p + self.step_pts) if i == ORDERS_PER_SIDE else 0.0
tp_s = (sell_p - self.step_pts) if i == ORDERS_PER_SIDE else 0.0
if i == ORDERS_PER_SIDE: self.tp_high, self.tp_low = tp_b, tp_s
self._pend(mt5.ORDER_TYPE_BUY_STOP, buy_p, self.mid_at_start, tp_b, info.digits)
self._pend(mt5.ORDER_TYPE_SELL_STOP, sell_p, self.mid_at_start, tp_s, info.digits)
self.is_active = True
def _pend(self, otype, price, sl, tp, digits, vol=BASE_LOT, tag=GRID_TAG):
vol = self._round_volume(vol)
req = {
"action": mt5.TRADE_ACTION_PENDING, "symbol": SYMBOL, "volume": float(vol),
"type": otype, "price": round(price, digits), "sl": round(sl, digits),
"tp": round(tp, digits) if tp > 0 else 0.0, "magic": MAGIC_NUMBER,
"comment": tag, "type_time": mt5.ORDER_TIME_GTC
}
return self._send_order(req)
def manage_grid(self, tick, info):
mid_now = (tick.bid + tick.ask) / 2
positions = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []
# 外側TP到達チェック
if (self.tp_high and mid_now >= self.tp_high) or (self.tp_low and mid_now <= self.tp_low):
print("\nTarget reached (Outer TP).")
self.close_all_positions_and_orders()
return
# リセンター判定 (片側半分消化で再構築)
# 現在ポジションがなく、かつ片側の注文残数が半分以下になった場合
if self.is_active and not positions:
buy_orders = [o for o in orders if o.comment == GRID_TAG and o.type == mt5.ORDER_TYPE_BUY_STOP]
sell_orders = [o for o in orders if o.comment == GRID_TAG and o.type == mt5.ORDER_TYPE_SELL_STOP]
threshold_side = ORDERS_PER_SIDE / 2 # 12 / 2 = 6本
# BuyまたはSellのどちらかの残数が閾値以下になり、かつ完全にゼロではない（全決済直後を除く）場合
if (len(buy_orders) <= threshold_side or len(sell_orders) <= threshold_side) and (buy_orders or sell_orders):
print(f"\n[Recenter] One side 50% consumed. BuyRem:{len(buy_orders)} SellRem:{len(sell_orders)}. Rebuilding...")
self.close_all_positions_and_orders()
return
# ポジション管理（部分利確とBE-REV）
for p in positions:
if self.step_pts is None: continue
# フルサイズポジション -> 利確判定
if abs(p.volume - BASE_LOT) < 1e-7:
trg = (p.price_open + self.step_pts) if p.type == mt5.POSITION_TYPE_BUY else (p.price_open - self.step_pts)
hit = (p.type == mt5.POSITION_TYPE_BUY and tick.bid >= trg) or (p.type == mt5.POSITION_TYPE_SELL and tick.ask <= trg)
if hit:
self._handle_partial_tp(p, info)
# 部分利確済みポジション -> トレーリングストップ
else:
self._update_trailing_stop(p, tick, info)
def _handle_partial_tp(self, pos, info):
half_vol = self._round_volume(pos.volume / 2)
tick = mt5.symbol_info_tick(SYMBOL)
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
# 1. 半分決済
self._send_order({
"action": mt5.TRADE_ACTION_DEAL, "symbol": SYMBOL, "volume": half_vol,
"type": side, "position": pos.ticket, "price": (tick.bid if side==mt5.ORDER_TYPE_SELL else tick.ask),
"magic": MAGIC_NUMBER, "comment": "partial TP", "deviation": DEVIATION
})
# 2. SLを建値へ
be_price = round(pos.price_open, info.digits)
self._send_order({"action": mt5.TRADE_ACTION_SLTP, "symbol": SYMBOL, "position": pos.ticket, "sl": be_price, "tp": 0.0})
# 3. 逆サイドにBE-REV注文配置
otype, sl = (mt5.ORDER_TYPE_SELL_STOP, be_price + self.step_pts) if pos.type == mt5.POSITION_TYPE_BUY else (mt5.ORDER_TYPE_BUY_STOP, be_price - self.step_pts)
if (tick.bid > be_price if otype == mt5.ORDER_TYPE_SELL_STOP else tick.ask < be_price):
self._pend(otype, be_price, sl, 0.0, info.digits, vol=BASE_LOT, tag=BEREV_TAG)
def _update_trailing_stop(self, pos, tick, info):
if self.step_pts is None: return
new_sl = None
if pos.type == mt5.POSITION_TYPE_BUY:
level = int((tick.bid - pos.price_open) / self.step_pts)
if level >= 1:
calc_sl = round(pos.price_open + ((level - 1) * self.step_pts), info.digits)
if calc_sl > pos.sl + (info.point * 0.1): new_sl = calc_sl
else:
level = int((pos.price_open - tick.ask) / self.step_pts)
if level >= 1:
calc_sl = round(pos.price_open - ((level - 1) * self.step_pts), info.digits)
if pos.sl == 0.0 or calc_sl < pos.sl - (info.point * 0.1): new_sl = calc_sl
if new_sl: self._send_order({"action": mt5.TRADE_ACTION_SLTP, "symbol": SYMBOL, "position": pos.ticket, "sl": new_sl, "tp": pos.tp})
def close_all_positions_and_orders(self):
# 注文全削除
for o in (mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []):
mt5.order_send({"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket, "type_filling": self.filling_mode})
# ポジション全決済
for p in (mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or []):
tick = mt5.symbol_info_tick(SYMBOL)
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
self._send_order({
"action": mt5.TRADE_ACTION_DEAL, "symbol": SYMBOL, "volume": p.volume,
"type": side, "position": p.ticket, "price": (tick.bid if side==mt5.ORDER_TYPE_SELL else tick.ask),
"magic": MAGIC_NUMBER, "comment": "clean exit", "deviation": DEVIATION
})
self.reset_state_vars()
def log_status(self, atr_val, spread_pts, dt_server):
now = time.time()
if now - self.last_log_time >= 1.0:
status = "ACTIVE" if self.is_active else "SCANNING"
time_str = dt_server.strftime('%H:%M:%S')
pos_count = len(mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER) or [])
print(f"[{status}] SrvTime:{time_str} | ATR:{atr_val:.3f} (Thresh:{ATR_THRESHOLD}) | Spd:{spread_pts:.1f} | Pos:{pos_count}", end='\r')
self.last_log_time = now
def main_loop(self):
info = mt5.symbol_info(SYMBOL)
print("Starting monitoring... Press Ctrl+C to stop.")
while True:
try:
time.sleep(CHECK_INTERVAL)
if not self.check_connection(): continue
tick = mt5.symbol_info_tick(SYMBOL)
if not tick: continue
dt_server = datetime.datetime.now(datetime.UTC).astimezone(SERVER_TZ)
# --- 時間管理 (週末/日次 強制決済) ---
h, m, weekday = dt_server.hour, dt_server.minute, dt_server.weekday()
is_friday = (weekday == 4)
exit_m = EXIT_MINUTE_FRI if is_friday else EXIT_MINUTE_MON_THU
if h == 23 and m >= exit_m:
if self.is_active or mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER):
print(f"\n[TimeLimit] Force closing at {h}:{m}")
self.close_all_positions_and_orders()
continue # 時間外はスキップ
# --- データ計算 ---
atr_val = self.calculate_current_atr()
spread_pts = (tick.ask - tick.bid) / info.point
is_entry_window = (START_HOUR_SERVER <= h < END_ENTRY_HOUR_SERVER)
self.log_status(atr_val, spread_pts, dt_server)
if self.is_active:
self.manage_grid(tick, info)
else:
# エントリー条件: 時間内 + スプレッド正常 + ATRが閾値超え
if is_entry_window and spread_pts <= SPREAD_LIMIT and atr_val >= ATR_THRESHOLD:
self.deploy_grid(tick, info)
except Exception as e:
print(f"Error: {e}"); time.sleep(1)
if __name__ == "__main__":
bot = VolatilityStopGridBot()
bot.main_loop()stop_grid_ATRのWFA分析結果
=== Starting WFA (Jobs: -1) ===
Processing Window 1
Train: 2026-02-02 ~ 2026-02-11
Test : 2026-02-12 ~ 2026-02-13
Best Params: {'GRID_MULTIPLIER': 3.0, 'ORDERS_PER_SIDE': 8, 'ATR_PERIOD': 10, 'ATR_THRESHOLD': 1.5} (Score: 122.
01)
Test Result: $3948.67 | DD: $357.67 | Margin: 2052.9%
Processing Window 2
Train: 2026-02-04 ~ 2026-02-13
Test : 2026-02-16 ~ 2026-02-17
Best Params: {'GRID_MULTIPLIER': 3.0, 'ORDERS_PER_SIDE': 6, 'ATR_PERIOD': 30, 'ATR_THRESHOLD': 0.850000000000000
1} (Score: 45.84)
Test Result: $157.97 | DD: $319.84 | Margin: 2192.9%
=================================================
WFA REPORT (High Performance)
=================================================
2026-02-12~2026-02-13 | P: 3948.67 | DD: 357.67 | M:2052.9%
Opt: {'GRID_MULTIPLIER': 3.0, 'ORDERS_PER_SIDE': 8, 'ATR_PERIOD': 10, 'ATR_THRESHOLD': 1.5}
2026-02-16~2026-02-17 | P: 157.97 | DD: 319.84 | M:2192.9%
Opt: {'GRID_MULTIPLIER': 3.0, 'ORDERS_PER_SIDE': 6, 'ATR_PERIOD': 30, 'ATR_THRESHOLD': 0.8500000000000001}
Total Profit: $4106.64リスクリワード的にかなり優秀。今後はATRエントリーをベースに改良していきたい。
取引bot以外のスクリプト
bot_analyzer：マジックナンバーごとの損益計算
任意のマジックナンバーを入力して稼働すれば、そのbotの過去1日の収益・RF・最大DDなどを計算してグラフ化してくれる。botのフォワード運用検証に便利。
import MetaTrader5 as mt5
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from datetime import datetime, timedelta, timezone
# ==========================================
# 設定エリア
# ==========================================
MAGIC_NUMBERS = [700000]
DAYS_BACK = 1 # サーバー時刻から何日前まで遡るか
FIG_SIZE = (12, 7)
# ==========================================
def get_server_time():
"""
MT5サーバーの現在時刻を取得する
"""
# 主要通貨ペアの最新ティックから時刻を取得（最も確実なサーバー時刻取得法）
symbol = "EURUSD"
tick = mt5.symbol_info_tick(symbol)
if tick is None:
# ティックが取れない場合は、直近の注文履歴などから推測（予備）
return datetime.now()
return datetime.fromtimestamp(tick.time)
def get_performance_metrics(df):
"""損益データから主要メトリクスを計算"""
if df.empty:
return None
# 純損益（損益 + 手数料 + スワップ）
net_profit_series = df['profit'] + df['commission'] + df['swap']
cum_profit = net_profit_series.cumsum()
total_profit = net_profit_series.sum()
trade_count = len(df)
win_rate = (net_profit_series > 0).sum() / trade_count * 100 if trade_count > 0 else 0
avg_trade = total_profit / trade_count if trade_count > 0 else 0
peak = cum_profit.cummax()
drawdown = peak - cum_profit
max_dd = drawdown.max()
recovery_factor = total_profit / max_dd if max_dd > 0 else float('inf')
return {
"Total Profit": total_profit,
"Max DD": max_dd,
"Recovery Factor": recovery_factor,
"Trade Count": trade_count,
"Win Rate (%)": win_rate,
"Avg Trade": avg_trade,
"Cum Profit": cum_profit,
"Times": df['time']
}
def main():
# 1. MT5接続
if not mt5.initialize():
print("MT5の初期化に失敗しました")
return
# 2. サーバー時刻の取得と分析期間の設定
server_now = get_server_time()
start_date = server_now - timedelta(days=DAYS_BACK)
print(f"--- 時刻同期設定 ---")
print(f"PCローカル時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"MT5サーバー時刻: {server_now.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"分析対象期間: {start_date} ～ {server_now} (過去{DAYS_BACK}日間)")
# 3. 履歴データの取得
# history_deals_get はサーバー時刻のdatetimeを受け付ける
deals = mt5.history_deals_get(start_date, server_now)
if deals is None or len(deals) == 0:
print("指定期間内に取引履歴が見つかりませんでした")
mt5.shutdown()
return
df_all = pd.DataFrame(list(deals), columns=deals[0]._asdict().keys())
df_all['time'] = pd.to_datetime(df_all['time'], unit='s')
results = []
plt.figure(figsize=FIG_SIZE)
# 4. マジックナンバーごとに分析
for magic in MAGIC_NUMBERS:
df_magic = df_all[(df_all['magic'] == magic) & (df_all['entry'] != 0)].copy()
metrics = get_performance_metrics(df_magic)
if metrics:
results.append({
"Magic": magic,
"Profit": metrics["Total Profit"],
"MaxDD": metrics["Max DD"],
"RF": metrics["Recovery Factor"],
"Trades": metrics["Trade Count"],
"WinRate": metrics["Win Rate (%)"],
"AvgTrade": metrics["Avg Trade"]
})
plt.plot(metrics["Times"], metrics["Cum Profit"],
marker='o', markersize=4, label=f"Magic: {magic}")
# 5. 結果表示
if not results:
print(f"対象マジックナンバーの決済データがありません")
else:
summary_df = pd.DataFrame(results).set_index("Magic")
print("\n--- パフォーマンス比較表 ---")
print(summary_df.round(2).to_string())
plt.title(f"Backtest: Past {DAYS_BACK} Days (Server Time Basis)")
plt.xlabel("Server Time")
plt.ylabel("Cumulative Profit")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
mt5.shutdown()
if __name__ == "__main__":
main()Sequence Lot Manager：ロット調整ロジック
Chisikiさんの新しいロット調整ロジック。単一ポジション向け（ソース）
"""
Sequence Lot Manager
====================
各トレードのロットサイズを決定するためのスタンドアロンなロジック。
「Decomposition（分解）法」に基づいたベッティングロジックに、
グッドマン法の連勝マルチプライヤーを組み合わせています。
アルゴリズムの概要:
- 数値のリスト（sequence）を保持する。
- 次のロットは (リストの最初 + 最後) * base_lot で計算される。
- 勝利時：数列の両端を削除（または中央値を分割）。
- 敗北時：賭けたロット相当の値が数列の末尾に追加され、数列が伸びる。
- グッドマン法：初期状態([0, 1])での連勝時に倍率(1→2→3→5)をかける。
- ストック機能：連勝で得た過剰利益を「蓄積」し、負けた際に数列の値を
相殺（減らす）するために使用する。これにより、サイクルの早期完了を目指す。
"""
from __future__ import annotations
import json
import os
from typing import Optional
class SequenceLotManager:
"""Decomposition Monte Carlo + Goodman method を用いたロット管理クラス"""
def __init__(
self,
base_lot: float = 0.01,
min_lot: float = 0.01,
max_lot: float = 100.0,
lot_step: float = 0.01,
state_file: Optional[str] = None,
):
# パラメータ設定
self.base_lot = base_lot
self.min_lot = min_lot
self.max_lot = max_lot
self.lot_step = lot_step
self.state_file = state_file
# 内部状態
self.sequence: list[int] = []
self.stock: float = 0.0
self.win_streak: int = 0
# 保存された状態があれば読み込み、なければ初期化
if not self.load_state():
self._init_fresh()
def get_next_lot(self) -> float:
"""現在の数列状態から次のロットサイズを計算する（読み取り専用）"""
if len(self.sequence) == 0:
return self._normalize_lot(self.base_lot)
left_val = self.sequence[0]
right_val = self.sequence[-1]
bet_val = left_val + right_val
# ベースとなるロット計算
current_bet_amount = bet_val * self.base_lot
# グッドマン法の連勝マルチプライヤー
# 1連勝: x1, 2連勝: x2, 3連勝: x3, 4連勝以上: x5
multiplier = 1
if self.win_streak == 2:
multiplier = 2
elif self.win_streak == 3:
multiplier = 3
elif self.win_streak >= 4:
multiplier = 5
final_bet = current_bet_amount * multiplier
# 安全策：0以下にならないようにする
if final_bet <= 0:
final_bet = self.base_lot
return self._normalize_lot(final_bet)
def process_result(self, is_win: bool) -> None:
"""トレード結果を記録し、数列を更新する"""
if len(self.sequence) < 2:
self.sequence = [0, 1]
left_val = self.sequence[0]
right_val = self.sequence[-1]
if is_win:
self._process_win(left_val, right_val)
else:
# 負けた場合は、連勝倍率を含めた実際の値を計算して処理
multiplier = 1
if self.win_streak == 2: multiplier = 2
elif self.win_streak == 3: multiplier = 3
elif self.win_streak >= 4: multiplier = 5
actual_bet = (left_val + right_val) * multiplier
self._process_lose(left_val, right_val, actual_bet)
self.save_state()
def reset(self) -> None:
"""初期状態 [0, 1] にリセットする"""
self._init_fresh()
self.save_state()
def get_state_summary(self) -> dict:
"""デバッグ/ログ用の現在の状態を返す"""
return {
"sequence": list(self.sequence),
"stock": self.stock,
"win_streak": self.win_streak,
"next_lot": self.get_next_lot(),
}
# === 内部ロジック ===
def _process_win(self, left_val: int, right_val: int) -> None:
"""勝利時の処理：数列の消化と連勝カウントアップ"""
# 数列が初期状態[0, 1]の間だけ連勝をカウント
if len(self.sequence) == 2 and self.sequence[0] == 0 and self.sequence[1] == 1:
self.win_streak += 1
if len(self.sequence) == 2:
# サイクル完了 -> リセット
self.sequence = [0, 1]
elif len(self.sequence) == 3:
# 残り3つの場合、真ん中の値を半分に割る
remaining_val = self.sequence[1]
half_val = remaining_val // 2
mod_check = remaining_val % 2
if mod_check == 0:
self.sequence = [half_val, half_val]
else:
self.sequence = [half_val, half_val + 1]
else:
# 長さ4以上：最初と最後を削除
self.sequence = self.sequence[1:-1]
self._equalize()
def _process_lose(self, left_val: int, right_val: int, actual_bet: int = 0) -> None:
"""敗北時の処理：ストックの積み増し、数列への追加、平準化、ストックでの相殺"""
# 1. 5連勝以上の過剰利益をストックに回す
if self.win_streak >= 5:
streak_profit = (self.win_streak - 2) * 5 - 8 # 簡易的な利益計算ロジック
self.stock += streak_profit
self.win_streak = 0
# 2. 負けた分（実際のベース値）を数列の末尾に追加
add_val = int(actual_bet) if actual_bet > 0 else (left_val + right_val)
self.sequence.append(add_val)
# 3. 数列の平準化
self._equalize()
# 4. ストックを使った左端の消去
if self.stock >= self.sequence[0] and self.sequence[0] > 0:
used_stock = self.sequence[0]
self.stock -= used_stock
self.sequence[0] = 0
# 5. ゼロの再分配（左端が0なら他の要素へ分散して数列を短くする）
if len(self.sequence) > 1 and self.sequence[0] <= 1:
redist_val = self.sequence[0]
self.sequence[0] = 0
redist_count = len(self.sequence) - 1
if redist_val >= 1:
total_val = sum(self.sequence) + redist_val
dist_val = total_val // redist_count
mod_val = total_val % redist_count
if redist_val < redist_count:
# 分散させるには小さすぎる場合は2番目に足す
self.sequence[1] += redist_val
else:
# 全体に均等に分配
for k in range(1, len(self.sequence)):
self.sequence[k] = dist_val
self.sequence[1] += mod_val
# 0になった要素を削除
self.sequence = self.sequence[1:]
def _equalize(self) -> None:
"""数列の値を平準化し、極端な偏りを防ぐ"""
seq = self.sequence
count = len(seq)
if count <= 1: return
total = sum(seq)
left_val = seq[0]
if left_val == 0 and count > 1:
# 左端が0なら残りで分配
dist_val = total // (count - 1)
mod_a = total % (count - 1)
self.sequence = [0] + [dist_val] * (count - 1)
self.sequence[1] += mod_a
elif left_val != 0:
# 全体で均等に分配
dist_b = total // count
mod_b = total % count
self.sequence = [dist_b] * count
if count >= 2:
self.sequence[1] += mod_b
else:
self.sequence[0] += mod_b
def _normalize_lot(self, lot: float) -> float:
"""ロットサイズをブローカーの制限（最小/最大/ステップ）に合わせる"""
if self.lot_step > 0:
lot = int(lot / self.lot_step) * self.lot_step
lot = max(self.min_lot, min(self.max_lot, lot))
return round(lot, 2)
def _init_fresh(self) -> None:
self.sequence = [0, 1]
self.stock = 0
self.win_streak = 0
def save_state(self) -> None:
if self.state_file is None: return
try:
data = {
"sequence": self.sequence,
"stock": self.stock,
"win_streak": self.win_streak,
}
with open(self.state_file, "w", encoding="utf-8") as f:
json.dump(data, f)
except Exception as e:
print(f"Error saving state: {e}")
def load_state(self) -> bool:
if self.state_file is None or not os.path.exists(self.state_file):
return False
try:
with open(self.state_file, "r", encoding="utf-8") as f:
data = json.load(f)
self.sequence = data.get("sequence", [0, 1])
self.stock = data.get("stock", 0)
self.win_streak = data.get("win_streak", 0)
return True
except:
return Falseソースコード改造におすすめのLLM
コーディングにおすすめのLLMはこちら。基本無料で利用できます。
この中で一番おすすめなのはGemini。パソコンの「Enter」が「改行」であり、「送信」ではないため、複数行でのチャット入力・編集がしやすいです。
Geminiでは上位モデルの「Pro」と軽量モデルの「Flash」が無料で使えますが、ProはAPIレート制限になりやすいです。原案を出すのはPro、コード実装・修正はFlashと使い分けましょう。
GPTは無駄なお世辞が多いので、プロンプトに「出力は最低限で」などを加えるといいでしょう。
Claudeは余計な機能を実装しがちなので、プロンプトに「実装は最低限で」などを加えてもいいでしょう。
LLMは無料アカウントだとAPI制限に引っかかりやすいです。API制限を避けたいなら、1つのモデルを使い続けるよりも、複数のLLM・モデルをこまめに切り替えた方がいいです。
プログラム言語をPythonからMQL5に変えるべき理由
オリジナルのstop_grid_traderがPythonだったので、なんとなく言語はPythonを使い続けていましたが、以下の理由からMQL5に変更すべきです。
- バックテストの精度がより正確になる
- 複数のチャートで同時運用できる（マジックナンバーは変える必要あり）
- 複数のMT5で同時運用できる
- VS code・ターミナルなどを使わないため、不具合が生じにくい
- 余計なライブラリを入れる必要がない（基本的にMQL5で完結するため）
- 約定速度が速くなる（HFT・レイテンシーでは優先度が上がる）
Python mt5 botの最大のデメリットは、バックテストの精度が低いこと。収益性の低いbotが過大評価されて、フォワード運用で大損を出してしまうことがあります。
LLMでPythonのバックテスト・WFAツールを作らせても、どこかに抜けがあって、フォワード成績と乖離しやすいです。逆にMQL5は正確なティックデータで高精度にバックテストするため、成績は信用度の高いものになります。
MQL5はPythonより記述の難しい言語ですが、LLMを使えば簡単にコードを出力できます。コンパイルエラーが発生しても、結果をスクショしてLLMに読ませれば、すぐに修正してくれます。
stop_grid_traderのロジックはやや複雑ですが、MQL5でも十分に表現可能です。エントリーロジックがシンプルなものなら、PythonよりもMQL5を使いましょう。
