使わなくなったコードを置いています。使い方・トレードロジックなどはLLMにコピペして聞いてください。（Gemini、chatGPT、Claude、Kimiなど）
目次
Stop-Grid Trader（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()Tick-V Trader：ティック加速度でグリッド展開
# tick_v_trader.py
# Tick-V Trader (CLI only)
# 2026-01 compatible / MT5 Python API
import time
import threading
import sys
import signal
import math
from datetime import datetime, timedelta, timezone
import MetaTrader5 as mt5
# ==============================
# 固定パラメータ（入力なし）
# ==============================
SYMBOL = "XAUUSD"
TIMEFRAME = mt5.TIMEFRAME_M1
PRICE_DIGITS = 2
BASE_LOT = 0.02
ORDERS_SIDE = 10
GRID_MULTIPLIER = 2.0
MAGIC_NUMBER = 555176
SPREAD_LIMIT = 30 # point
GRID_TIMEOUT = 120 # 秒
CHECK_INTERVAL = 1.0
# EWMA + Z-Score
EWMA_ALPHA = 0.30
Z_THRESHOLD = 2.6 # 緩め設定
# JST
JST = timezone(timedelta(hours=9))
DEVIATION = 100
GRID_TAG = "VOL_GRID"
running = True
# ==============================
# ロギング
# ==============================
def log(msg, level="INFO"):
ts = datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S")
print(f"[{ts}] {level:<5} {msg}", flush=True)
# ==============================
# MT5 初期化（自動接続）
# ==============================
def mt5_init():
if not mt5.initialize():
code, msg = mt5.last_error()
raise RuntimeError(f"MT5 init failed: {code} {msg}")
if not mt5.symbol_select(SYMBOL, True):
raise RuntimeError(f"Cannot select symbol {SYMBOL}")
log("MT5 connected")
# ==============================
# 終了時 強制クローズ
# ==============================
def full_close():
log("SCRIPT_ABORT_FULL_CLOSE", "WARN")
# pending削除
for o in mt5.orders_get(symbol=SYMBOL) or []:
if o.magic != MAGIC_NUMBER:
continue
mt5.order_send({
"action": mt5.TRADE_ACTION_REMOVE,
"order": o.ticket,
"symbol": o.symbol
})
tick = mt5.symbol_info_tick(SYMBOL)
# 全ポジション決済
for p in mt5.positions_get(symbol=SYMBOL) or []:
if p.magic != MAGIC_NUMBER:
continue
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"position": p.ticket,
"volume": p.volume,
"type": 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,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": "SCRIPT_EXIT"
})
log("FULL_CLOSE_DONE")
def signal_handler(sig, frame):
global running
running = False
full_close()
mt5.shutdown()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
# ==============================
# EWMA + Z-Score 管理
# ==============================
class AccelFilter:
def __init__(self):
self.prev_mid = None
self.prev_ret = 0.0
self.mean = 0.0
self.var = 1e-6
def update(self, mid):
if self.prev_mid is None:
self.prev_mid = mid
return 0.0, 0.0
ret = mid - self.prev_mid
accel = ret - self.prev_ret
# EWMA
self.mean = EWMA_ALPHA * accel + (1 - EWMA_ALPHA) * self.mean
diff = accel - self.mean
self.var = EWMA_ALPHA * (diff * diff) + (1 - EWMA_ALPHA) * self.var
z = accel / math.sqrt(self.var) if self.var > 0 else 0.0
self.prev_mid = mid
self.prev_ret = ret
return accel, z
# ==============================
# フィルター群（ログ出力なし）
# ==============================
def time_filter():
now = datetime.now(JST)
if 5 <= now.hour < 8:
full_close()
return False
return True
def spread_filter():
tick = mt5.symbol_info_tick(SYMBOL)
info = mt5.symbol_info(SYMBOL)
spread = int((tick.ask - tick.bid) / info.point)
if spread >= SPREAD_LIMIT:
return False
return True
# ==============================
# グリッド構築
# ==============================
def norm_vol(vol):
info = mt5.symbol_info(SYMBOL)
step = info.volume_step or 0.01
return round(max(info.volume_min, min(vol, info.volume_max)) / step) * step
def pend(ord_type, price, sl, tp=0.0, vol=None, tag=GRID_TAG):
if vol is None:
vol = BASE_LOT
mt5.order_send({
"action": mt5.TRADE_ACTION_PENDING,
"symbol": SYMBOL,
"volume": norm_vol(vol),
"type": ord_type,
"price": price,
"sl": sl,
"tp": tp,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": tag,
"type_time": mt5.ORDER_TIME_GTC
})
def build_grid():
tick = mt5.symbol_info_tick(SYMBOL)
info = mt5.symbol_info(SYMBOL)
mid = round((tick.bid + tick.ask) / 2, PRICE_DIGITS)
raw_spd_pts = int((tick.ask - tick.bid) / info.point)
step_pts = int(raw_spd_pts * GRID_MULTIPLIER)
pt = info.point
tp_high = tp_low = None
for i in range(1, ORDERS_SIDE + 1):
buy = mid + i * step_pts * pt
sell = mid - i * step_pts * pt
if i == ORDERS_SIDE:
tp_high = buy + step_pts * pt
tp_low = sell - step_pts * pt
tp_b, tp_s = tp_high, tp_low
else:
tp_b = tp_s = 0.0
pend(mt5.ORDER_TYPE_BUY_STOP , buy , mid, tp=tp_b)
pend(mt5.ORDER_TYPE_SELL_STOP, sell, mid, tp=tp_s)
log(f"GRID_START mid={mid} spread={raw_spd_pts}")
return time.time(), tp_high, tp_low, step_pts
# ==============================
# メインループ
# ==============================
def main():
mt5_init()
accel_filter = AccelFilter()
grid_active = False
grid_start = None
tp_high = tp_low = None
step_pts = None
while running:
time.sleep(CHECK_INTERVAL)
# 時間フィルター
if not time_filter():
continue
tick = mt5.symbol_info_tick(SYMBOL)
mid = (tick.bid + tick.ask) / 2
accel, z = accel_filter.update(mid)
# ボラ判定
vol_ok = abs(z) >= Z_THRESHOLD
if not grid_active:
if not vol_ok:
continue
# 時間フィルター再確認
time_ok = time_filter()
if not time_ok:
continue
# スプレッド判定（ログ用に値取得）
tick2 = mt5.symbol_info_tick(SYMBOL)
info2 = mt5.symbol_info(SYMBOL)
spread = int((tick2.ask - tick2.bid) / info2.point)
spread_ok = spread < SPREAD_LIMIT
if not spread_ok:
continue
# ===== エントリー時フィルター情報ログ（ここだけ出力）=====
log(
f"ENTRY_FILTER "
f"accel={accel:.6f} "
f"z={z:.2f} "
f"vol_ok={vol_ok} "
f"spread={spread} "
f"time_ok={time_ok}"
)
# グリッド開始
grid_start, tp_high, tp_low, step_pts = build_grid()
grid_active = True
continue
# ===== グリッド監視 =====
now = time.time()
# タイムアウト強制終了
if now - grid_start >= GRID_TIMEOUT:
log("GRID_TIMEOUT_EXIT", "WARN")
full_close()
grid_active = False
continue
mid_now = (tick.bid + tick.ask) / 2
# 外TP到達
if (tp_high and mid_now >= tp_high) or (tp_low and mid_now <= tp_low):
log("OUTER_TP_EXIT")
full_close()
grid_active = False
continue
# 部分利確チェック
info = mt5.symbol_info(SYMBOL)
pt = info.point
half = BASE_LOT / 2
for pos in mt5.positions_get(symbol=SYMBOL) or []:
if pos.magic != MAGIC_NUMBER:
continue
trg = (
pos.price_open + step_pts * pt
if pos.type == mt5.POSITION_TYPE_BUY
else pos.price_open - 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 and abs(pos.volume - BASE_LOT) < 1e-6:
mt5.order_send({
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"position": pos.ticket,
"volume": half,
"type": 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,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": "PARTIAL_TP"
})
if __name__ == "__main__":
try:
main()
finally:
full_close()
mt5.shutdown()Stop-Grid Basket：バスケット決済
import MetaTrader5 as mt5
import numpy as np
import pandas as pd
import time
import datetime
import pytz
import signal
import sys
import logging
import json
import os
# ==========================================
# 1. 設定パラメータ (Z-Score閾値を4.5に変更)
# ==========================================
SYMBOL = "XAUUSD"
MAGIC_NUMBER = 100003
BASE_LOT = 0.01
GRID_LEVELS = 50
GRID_MULTIPLIER = 3.0
SPREAD_LIMIT = 80
Z_SCORE_THRESHOLD = 1.5 # 非常に強いボラティリティのみを対象
BASE_TRAIL_PCT = 0.95
TICK_WINDOW = 500
MIN_PROFIT_FOR_TRAIL = 25.0
DEVIATION = 20
STOP_LOSS_WAIT_SEC = 600 # 損切り判定開始までの待機時間（秒）
JST = pytz.timezone('Asia/Tokyo')
STATE_FILE = "bot_state.json"
LOG_INTERVAL = 1 # 1秒ごとに更新
# ==========================================
# 2. ロギング設定
# ==========================================
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler("bot_log.log", encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class HighSpeedGridBot:
def __init__(self):
self.peak_equity = -999999.0
self.is_active = False
self.last_log_time = 0
self.grid_mid_price = None # グリッド展開時の中央価格
self.grid_start_time = None # グリッド展開開始時刻
self.setup_signal_handler()
self.load_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):
logger.warning(f"終了信号検知 (Signal:{sig})。状態を保存して終了します。")
self.save_state()
mt5.shutdown()
sys.exit(0)
# --- 状態管理 ---
def save_state(self):
state = {"peak_equity": self.peak_equity, "timestamp": datetime.datetime.now().isoformat()}
try:
with open(STATE_FILE, "w", encoding='utf-8') as f:
json.dump(state, f, indent=4)
except Exception as e:
logger.error(f"状態保存失敗: {e}")
def load_state(self):
if os.path.exists(STATE_FILE):
try:
with open(STATE_FILE, "r", encoding='utf-8') as f:
state = json.load(f)
self.peak_equity = state.get("peak_equity", -999999.0)
logger.info(f"以前の状態を復元: Peak Equity = {self.peak_equity:.2f}")
except Exception as e:
logger.error(f"状態復元失敗: {e}")
def reset_state(self):
self.peak_equity = -999999.0
self.grid_mid_price = None
self.grid_start_time = None
if os.path.exists(STATE_FILE):
try: os.remove(STATE_FILE)
except: pass
# --- MT5 操作 ---
def initialize_mt5(self):
if not mt5.initialize():
logger.error(f"MT5初期化失敗: {mt5.last_error()}")
sys.exit(1)
if not mt5.symbol_select(SYMBOL, True):
logger.error(f"シンボル {SYMBOL} が見つかりません")
sys.exit(1)
logger.info(f"MT5接続成功: {SYMBOL} / Magic:{MAGIC_NUMBER} / Threshold:{Z_SCORE_THRESHOLD}")
def check_connection(self):
terminal = mt5.terminal_info()
if terminal is None or not terminal.connected:
mt5.initialize()
return False
return True
# --- 計算・分析 ---
def calculate_acceleration_zscore(self) -> float:
# 最新ティックを確実に取得(+600秒)
ticks = mt5.copy_ticks_from(SYMBOL, int(time.time()) + 600, TICK_WINDOW, mt5.COPY_TICKS_ALL)
if ticks is None or len(ticks) < 100:
return 0.0
d_times = np.diff(ticks['time_msc'])
d_times = np.where(d_times <= 0, 1, d_times)
velocity = 1000.0 / d_times
acceleration = np.diff(velocity) / (d_times[1:] / 1000.0)
ewma_accel = pd.Series(acceleration).ewm(alpha=0.1, adjust=False).mean().values
current_val = ewma_accel[-1]
mean, std = np.mean(ewma_accel), np.std(ewma_accel)
return abs((current_val - mean) / std) if std > 1e-9 else 0.0
def calculate_dynamic_trail_pct(self, peak):
if peak <= MIN_PROFIT_FOR_TRAIL: return 0.50
scaling_factor = (MIN_PROFIT_FOR_TRAIL / peak) ** 0.5
return max(0.05, min(0.50, BASE_TRAIL_PCT * scaling_factor))
# --- トレード管理 ---
def sync_and_manage(self):
positions = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
self.is_active = (len(positions) > 0 or len(orders) > 0)
if not positions:
if not orders and self.peak_equity != -999999.0:
self.reset_state()
return
current_profit = sum(p.profit + p.swap for p in positions)
if self.peak_equity == -999999.0 or current_profit > self.peak_equity:
self.peak_equity = current_profit
self.save_state()
# トレーリング決済
if current_profit >= MIN_PROFIT_FOR_TRAIL:
threshold = self.peak_equity * (1.0 - self.calculate_dynamic_trail_pct(self.peak_equity))
if current_profit < threshold:
logger.info(f"トレーリング決済発動: Peak:{self.peak_equity:.2f} -> Curr:{current_profit:.2f}")
self.close_all_positions_and_orders()
return
# 損切り判定（時間フィルター付き）
if self.grid_mid_price is not None and self.grid_start_time is not None:
elapsed_time = time.time() - self.grid_start_time
# 待機時間経過後のみ損切り判定
if elapsed_time >= STOP_LOSS_WAIT_SEC:
tick = mt5.symbol_info_tick(SYMBOL)
if tick:
current_mid = (tick.ask + tick.bid) / 2
# 買いポジションと売りポジションを分けて判定
buy_positions = [p for p in positions if p.type == mt5.POSITION_TYPE_BUY]
sell_positions = [p for p in positions if p.type == mt5.POSITION_TYPE_SELL]
# 買いポジション: 現在価格がグリッド中央価格を下回った場合
if buy_positions and current_mid < self.grid_mid_price:
logger.warning(f"損切り発動(買い): Mid:{self.grid_mid_price:.2f} -> Curr:{current_mid:.2f} (経過:{elapsed_time:.0f}秒)")
self.close_all_positions_and_orders()
return
# 売りポジション: 現在価格がグリッド中央価格を上回った場合
if sell_positions and current_mid > self.grid_mid_price:
logger.warning(f"損切り発動(売り): Mid:{self.grid_mid_price:.2f} -> Curr:{current_mid:.2f} (経過:{elapsed_time:.0f}秒)")
self.close_all_positions_and_orders()
return
def deploy_grid(self, tick, info):
mid_price = (tick.ask + tick.bid) / 2
step_dist = (tick.ask - tick.bid) * GRID_MULTIPLIER
# グリッド展開情報を記録
self.grid_mid_price = mid_price
self.grid_start_time = time.time()
logger.info(f"★★★ グリッド展開開始 Mid:{mid_price:.2f} ★★★")
for i in range(1, GRID_LEVELS + 1):
self._send_order(mt5.ORDER_TYPE_BUY_STOP, round(mid_price + (i * step_dist), info.digits))
self._send_order(mt5.ORDER_TYPE_SELL_STOP, round(mid_price - (i * step_dist), info.digits))
self.peak_equity = -999999.0
self.save_state()
def _send_order(self, order_type, price):
req = {
"action": mt5.TRADE_ACTION_PENDING, "symbol": SYMBOL, "volume": float(BASE_LOT),
"type": order_type, "price": price, "magic": MAGIC_NUMBER,
"deviation": DEVIATION, "type_time": mt5.ORDER_TIME_GTC
}
return mt5.order_send(req)
def close_all_positions_and_orders(self):
"""全クリーンアップ実行（スリッページ計測付き）"""
orders = mt5.orders_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
if orders:
for o in orders:
mt5.order_send({"action": mt5.TRADE_ACTION_REMOVE, "order": o.ticket})
positions = mt5.positions_get(symbol=SYMBOL, magic=MAGIC_NUMBER)
if not positions:
self.is_active = False
self.reset_state()
return
info = mt5.symbol_info(SYMBOL)
point = info.point
total_slippage_points = 0
count = 0
for p in positions:
tick = mt5.symbol_info_tick(SYMBOL)
if p.type == mt5.POSITION_TYPE_BUY:
order_type = mt5.ORDER_TYPE_SELL
requested_price = tick.bid
else:
order_type = mt5.ORDER_TYPE_BUY
requested_price = tick.ask
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": p.volume,
"type": order_type,
"position": p.ticket,
"price": requested_price,
"magic": MAGIC_NUMBER,
"deviation": DEVIATION,
"type_filling": mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(req)
if result.retcode == mt5.TRADE_RETCODE_DONE:
actual_price = result.price
if p.type == mt5.POSITION_TYPE_BUY:
slippage = (requested_price - actual_price) / point
else:
slippage = (actual_price - requested_price) / point
msg = f"Pos:{p.ticket} 決済完了 | Slip: {slippage:.1f} pts"
if slippage > 5.0:
logger.warning(f"【警告】マイナススリップ大: {msg}")
else:
logger.info(msg)
total_slippage_points += slippage
count += 1
else:
logger.error(f"決済失敗 Pos:{p.ticket} Error:{result.comment}")
if count > 0:
avg_slip = total_slippage_points / count
logger.info(f"=== 決済サマリー: 平均スリッページ {avg_slip:.1f} pts ===")
self.is_active = False
self.reset_state()
logger.info("全クリーンアップ完了")
# --- 監視・ログ ---
def log_status(self, z_score, spread_pts, is_time_ok, now_jst):
now = time.time()
if now - self.last_log_time >= LOG_INTERVAL:
status = "ACTIVE" if self.is_active else "IDLE"
time_status = "OK" if is_time_ok else "Stop"
log_msg = (
f"[{status}] {now_jst.strftime('%H:%M:%S')} | "
f"Z-Score: {z_score:4.2f}/{Z_SCORE_THRESHOLD} | "
f"Spread: {spread_pts:3.0f}/{SPREAD_LIMIT} | "
f"Time: {time_status:4} | "
f"Peak: {self.peak_equity:8.2f}"
)
print(log_msg, end='\r', flush=True)
self.last_log_time = now
def main_loop(self):
self.initialize_mt5()
info = mt5.symbol_info(SYMBOL)
logger.info(f"監視開始: エントリー 09:00-05:00 / 強制決済 05:29 / Threshold: {Z_SCORE_THRESHOLD}")
while True:
try:
time.sleep(0.1)
if not self.check_connection(): continue
self.sync_and_manage()
tick = mt5.symbol_info_tick(SYMBOL)
if not tick: continue
now_jst = datetime.datetime.now(datetime.UTC).astimezone(JST)
spread_pts = (tick.ask - tick.bid) / info.point
z_score = self.calculate_acceleration_zscore()
# 強制決済(05:29)
if now_jst.hour == 5 and now_jst.minute == 29 and self.is_active:
logger.info("メンテナンス前強制クリーンアップ実行(05:29)")
self.close_all_positions_and_orders()
# エントリー許可(09:00 ～ 04:59)
is_time_ok = not (5 <= now_jst.hour < 9)
self.log_status(z_score, spread_pts, is_time_ok, now_jst)
# エントリー判定 (Z-Score 4.5以上のみ)
if not self.is_active and is_time_ok:
if spread_pts <= SPREAD_LIMIT and z_score > Z_SCORE_THRESHOLD:
self.deploy_grid(tick, info)
except Exception as e:
logger.error(f"ループ内エラー: {e}")
time.sleep(1)
if __name__ == "__main__":
bot = HighSpeedGridBot()
bot.main_loop()Scal_Basket_Nanpin：単発スキャ＋ナンピン
"""
Fair Value Basket Scalper v1.3 [Safety Fixed]
------------------------------------------------------------
修正内容:
- ポジション取得時のソート処理(Ticket順)を追加し、連打エントリーを完全に防止
- 各業者の注文仕様(Filling Mode)に対応した自動判定
- 買い・売りを個別にバスケット決済(合計ドル利益で利確)
"""
import time
import sys
import MetaTrader5 as mt5
from datetime import datetime
# =================================================================
# SETTING (パラメータ設定)
# =================================================================
SYMBOL = "XAUUSD" # 取扱通貨ペア
PRICE_DIGITS = 2 # 価格の小数点桁数
BASE_LOT = 0.01 # 1ポジションあたりのロット
GRID_STEP_PTS = 200 # グリッド間隔 (200pts = $2.00。安全のため広めを推奨)
TARGET_PROFIT_USD = 2.0 # バスケット決済目標利益 (ドル)
MAX_POS_SIDE = 26 # 片側最大ポジション数
CHECK_INTERVAL = 1.0 # 監視間隔（秒）
# 固定定数
DEVIATION = 100
MAGIC_NUMBER = 88888 # マジックナンバー
# =================================================================
class BasketScalperCLI:
def __init__(self):
self.symbol = SYMBOL
self.digits = PRICE_DIGITS
self.lot = BASE_LOT
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}")
self.log(f"監視設定: {self.symbol} / Step: {GRID_STEP_PTS}pts / Target: ${TARGET_PROFIT_USD}")
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 _get_filling_type(self):
"""取引サーバーに適した充填モードを自動選択(エラー回避版)"""
info = mt5.symbol_info(self.symbol)
if info is None: return mt5.ORDER_FILLING_FOK
fill_mode = info.filling_mode
# ビットフラグによる直接判定
if fill_mode & 1: return mt5.ORDER_FILLING_FOK
elif fill_mode & 2: return mt5.ORDER_FILLING_IOC
else: return mt5.ORDER_FILLING_RETURN
def _send_market_order(self, order_type):
"""成行注文の実行"""
tick = mt5.symbol_info_tick(self.symbol)
if tick is None: return None
price = tick.ask if order_type == mt5.ORDER_TYPE_BUY else tick.bid
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": self.symbol,
"volume": self._norm_vol(self.lot),
"type": order_type,
"price": price,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": "fv_basket_entry",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": self._get_filling_type(),
}
result = mt5.order_send(request)
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
msg = result.comment if result else "Unknown"
self.log(f"注文失敗: {msg}")
return result
def _get_basket_info(self, pos_type):
"""最新ポジションをチケット番号順に取得して解析（重要修正）"""
positions = mt5.positions_get(symbol=self.symbol, magic=MAGIC_NUMBER)
if not positions:
return 0.0, 0, 0.0
# 特定サイドのポジションを抽出
side_pos = [p for p in positions if p.type == pos_type]
if not side_pos:
return 0.0, 0, 0.0
# 【連打防止の要】チケット番号(ticket)で昇順ソートし、リストの最後を「絶対的な最新」にする
side_pos = sorted(side_pos, key=lambda x: x.ticket)
# 含み損益＋スワップで計算（手数料属性エラー回避）
total_profit = sum(p.profit + p.swap for p in side_pos)
count = len(side_pos)
last_price = side_pos[-1].price_open # これで正確に直近の建値と比較できる
return total_profit, count, last_price
def _close_basket(self, pos_type):
"""一括決済"""
side_str = "BUY" if pos_type == mt5.POSITION_TYPE_BUY else "SELL"
positions = mt5.positions_get(symbol=self.symbol, magic=MAGIC_NUMBER)
if not positions: return
self.log(f"[{side_str}] 合計利益目標達成。一括決済中...")
for p in positions:
if p.type == pos_type:
tick = mt5.symbol_info_tick(self.symbol)
type_opp = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price_opp = tick.bid if p.type == mt5.POSITION_TYPE_BUY else tick.ask
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": self.symbol,
"position": p.ticket,
"volume": p.volume,
"type": type_opp,
"price": price_opp,
"deviation": DEVIATION,
"magic": MAGIC_NUMBER,
"comment": "basket_exit",
"type_filling": self._get_filling_type(),
}
mt5.order_send(request)
def monitor(self):
self.running = True
self.log("高頻度バスケット監視を開始しました。 [終了: Ctrl+C]")
try:
while self.running:
time.sleep(CHECK_INTERVAL)
tick = mt5.symbol_info_tick(self.symbol)
if not tick or not mt5.symbol_select(self.symbol, True):
continue
for p_type in [mt5.POSITION_TYPE_BUY, mt5.POSITION_TYPE_SELL]:
profit, count, last_price = self._get_basket_info(p_type)
# 1. 初期エントリー
if count == 0:
m_type = mt5.ORDER_TYPE_BUY if p_type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_SELL
self._send_market_order(m_type)
self.log(f"初期エントリー完了 ({'BUY' if p_type == mt5.POSITION_TYPE_BUY else 'SELL'})")
continue
# 2. バスケット決済判定
if profit >= TARGET_PROFIT_USD:
self._close_basket(p_type)
continue
# 3. 追加判定 (最新の建値からの距離を確認)
if count < MAX_POS_SIDE:
# 買いならBid、売りならAskで建値との距離を測る
curr_price = tick.bid if p_type == mt5.POSITION_TYPE_BUY else tick.ask
dist = abs(curr_price - last_price)
pt = mt5.symbol_info(self.symbol).point
# 正しく設定されたグリッド幅(GRID_STEP_PTS)以上離れているか確認
if dist >= (GRID_STEP_PTS * pt):
m_type = mt5.ORDER_TYPE_BUY if p_type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_SELL
self._send_market_order(m_type)
self.log(f"グリッド追加: {count+1}個目 ({'BUY' if p_type == mt5.POSITION_TYPE_BUY else 'SELL'}) | Current Profit: ${profit:.2f}")
except KeyboardInterrupt:
self.log("\n--- [中断検知] ポジションを決済して終了します ---")
self.aborting = True
self._close_basket(mt5.POSITION_TYPE_BUY)
self._close_basket(mt5.POSITION_TYPE_SELL)
finally:
mt5.shutdown()
self.log("MT5切断完了。")
if __name__ == "__main__":
print("=======================================")
print(" Fair Value Basket Scalper v1.3")
print("=======================================")
trader = BasketScalperCLI()
trader._mt5_init()
trader.monitor()ソースコード改造におすすめのLLM
コーディングにおすすめのLLMはこちら。基本無料で利用できます。
この中で一番おすすめなのはGemini。パソコンの「Enter」が「改行」であり、「送信」ではないため、複数行でのチャット入力・編集がしやすいです。
Geminiでは上位モデルの「Pro」と軽量モデルの「Flash」が無料で使えますが、ProはAPIレート制限になりやすいです。原案を出すのはPro、コード実装・修正はFlashと使い分けましょう。
GPTは無駄なお世辞が多いので、プロンプトに「出力は最低限で」などを加えるといいでしょう。
Claudeは余計な機能を実装しがちなので、プロンプトに「実装は最低限で」などを加えてもいいでしょう。
LLMは無料アカウントだとAPI制限に引っかかりやすいです。API制限を避けたいなら、1つのモデルを使い続けるよりも、複数のLLM・モデルをこまめに切り替えた方がいいです。
