この記事ではLLM + 海外FXで稼ぐ方法について解説していきます。
目次
LLM・VPS・口座の準備
手順1：LLMに登録しておく（無料プランでOK）
バイブコーディングで取引botを作りたいなら、まずはLLMに登録しましょう。アカウント登録すれば、無料で利用できます。
代表的なLLMサービスはこちら。
2026年4月現在だと、Codexが一番評判がいいです。ただWebブラウザ版のChatGPT（無料プラン）だと、出力制限によりbotをフルコードで出力できないことがあります。
LLMの世界は変化が激しく、突然スペックが大きく向上したり、大きく改悪されたりします。これら3つでもっとも評判がいいものを切り替えていくようにしましょう。
料金プランについては、無料プランでOKです。取引botの開発では、数時間ごとのログを取り、それをLLMに食わせて、改良していきます。数時間に2-3回のメッセージが使えれば十分です。
手順2：期待値の高いトレードロジックを把握する
次は期待値の高いトレードロジックを調べておきましょう。
おすすめのトレードロジックはこちら。
- A-bookブローカー：トレンドフォロー＋ピラミッディング
- B-bookブローカー：OFIリードラグ
A-bookブローカーを使っているなら、トレンドフォロー＋ピラミッディングがおすすめ。テクニカル系のトレードロジックの中ではトップクラスに期待値が高いです。
A-bookブローカーはトレーダーの注文を呑まずに取引所に流してくれるため、意図的なマイナススリッページなどは生じにくいですが、B-bookと比べると約定速度が遅く、価格もずれやすいです。HFTには使わず、普通のテクニカルトレードで使うべきです。
そのかわりA-bookはB-bookと違ってトレーダーと利益相反の関係にありません。稼げるトレードロジックを見つけたら、ロットを少しずつ増やしていくことで、収益を雪だるま式に増やしていくことができます。
取引コストの低いB-bookブローカーを使っているなら、OFIのリードラグがおすすめ。仮想通貨取引所のAPIでビットコインやゴールドなどの板情報を先読みし、価格反映の遅いMT5口座などで取引します。
板情報（特にOFI・マイクロプライス）は数ある先行指標の中でもトップクラスに予測精度が高いです。ただエッジの有効性は2秒から5秒と短く、価格も正確に約定させる必要があるため、執行速度に優れているB-bookを使うのが前提となります。
B-bookは顧客の注文を呑んでいるため約定速度は早いですが、トレーダーと利益相反の関係になるため、派手に荒稼ぎをすると利益没収や出金拒否などの被害にありやすいです。資金100,000円＋0.01ロットの運用で、なおかつ複数口座にて、低頻度で運用することをおすすめします。
手順3：Tradeviewで口座開設しておく（MT4/MT5/cTrader対応）
取引口座については、スプレッド・取引手数料が低いものを選びましょう。
おすすめはTradeviewで、取引コストの低さ、入出金経路の豊富さ、対応プラットフォームの種類（MT4/MT5/cTrader）などのバランスがいいです。
あわせて読みたい
Tradeview Markets – FX取引・低スプレッドのグローバルブローカー
TradeviewでFXや株式、先物を取引。低スプレッドと高速約定。今すぐ口座開設して始めましょう。
またTradeviewはA-bookの可能性が高いので、顧客と利益相反せず、マイナススリッページも生じにくいです。
TradeviewはMT4/MT5/cTraderの3種類のFXプラットフォームにも対応しており、会員登録なしでデモ口座を作れます。おすすめはMT5口座で、後述の無料Python botソースコードの練習台に使えます。
あわせて読みたい
Tradeview デモ口座の作り方・開設方法・有効期限
この記事ではTradeviewでのデモ口座の作り方・開設方法・使用ルールを解説していきます。 Tradeview デモ口座の詳細は、Tradeview「MT4デモ・MT5デモ・cTraderデモ」で…
またTradeviewは入出金経路も豊富で、国内銀行送金や海外銀行送金、bitwalletや仮想通貨などに幅広く対応しています。入出金方法の詳細は、Tradeviewの会員ページで確認できます。
あわせて読みたい
Tradeview 入金方法・入金手数料まとめ
この記事ではTradeviewの入金方法・入金手数料を見ていきます。 Tradeviewの入金方法は、Tradeview「会員ログイン→入金ページ」で確認できます。 Tradeviewの入金方法一…
あわせて読みたい
Tradeview 出金方法・出金手数料まとめ
この記事ではTradeviewの出金方法・出金手数料を解説していきます。 Tradeviewの出金方法は、Tradeview「会員ログイン→出金ページ」で確認できます。 Tradeviewの出金方…
手順4：取引口座の約定環境をチェックする
取引口座を開設したら、約定環境を調べておきましょう。特にHFTをするなら重要です。
- 取引サーバーの場所：VPSと同じ場所にすること
- ping：5-10ms以下なら合格
- 約定速度：150ms以下なら合格
MT5でアクセスするサーバーにはアクセスサーバーと取引サーバーの2種類があります。このうち約定速度・レイテンシーに影響するのは取引サーバーの方です。
取引サーバーの具体的な場所・IPアドレスは確認できないので、MT5「操作ログ」で表示されるMetaTraderのホスティングサービス宣伝メッセージで確認します。例えばニューヨークのVPSを勧められるようなら、取引サーバーはニューヨークにあると判断します。
pingついては、MT5取引口座にログインしたときに自動で行われます。「操作ログ」で確認しましょう。ここでのpingとはVPSと取引サーバーの通信速度のことで、5-10ms以下なら合格とします。
約定速度については、実際に取引してみないと分かりません。リアル口座で0.01ロットの成り行き注文をして、即座に決済しましょう。MT5「操作ログ」にてエントリー・決済の約定速度が確認できます。
約定速度は「ping＋取引にかかった時間」で構成されます。トータルの約定速度は100-150ms以下なら合格と判断します。
手順5：取引サーバーに対応したVPSを契約する
ブローカーの取引サーバーの場所が判明したら、それに対応したリージョンのVPSを契約します。VPS・取引サーバー間の距離が近いほど、約定速度が早くなり、bot取引で利益を出しやすくなります。
VPSはContaboがおすすめ。メモリ8GBの大容量プランが15ドル前後で使えます。
あわせて読みたい
Best Windows VPS Hosting: Fast & Scalable Solutions
Best Windows VPS hosting – affordable, secure, scalable. Instant setup, reliable performance, best value for smart users.
ただWindowsOSを使う場合は月額10ドルほど上乗せされるため、実質的な月額料金は25ドル前後になります。（OSはLinuxを使うことで、その月額10ドルを節約できる）
Contaboは数年前まではたまにサーバーが起きてしまうことがあり、評判が良くありませんでしたが、ここ1年は動作が安定しています。
EA/botの開発・改良
手順1：デモ口座でbot取引をしてみる（無料ソースコードコピペ可能）
まずは実際にデモ口座で取引してみましょう。すぐに取引を始められるよう、こちらでソースコードを用意しました。
以下のソースコードは、トレンドフォロー＋ピラミッディングのトレードロジックで、A-bookブローカーでの運用に適しています。（リアル口座での運用は自己責任で）
# Python_TrendFollow_Pyramid_v1.90 OFI廃止・マルチタイムフレームブレイクアウト+EMAフィルター+ピラミッディング(最大3ポジション)に全面改造
import sys
import time
import math
import threading
import collections
import MetaTrader5 as mt5
import logging
from datetime import datetime, timezone
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(encoding='utf-8')
if hasattr(sys.stderr, 'reconfigure'):
sys.stderr.reconfigure(encoding='utf-8')
SYMBOL_MT5 = "XAUUSD"
MAGIC_NUMBER = 100190
FIXED_LOT = 0.01
BASE_DEVIATION = 30
EXIT_DEVIATION = 50
SPREAD_FILTER_PT = 50
MAX_TICK_AGE_MS = 500
COOLDOWN = 5.0
ENTRY_BLACKOUT_SEC = 1.0
CLOSE_RETRY_INTERVAL = 0.05
CLOSE_MAX_RETRIES = 20
CLOSE_TIMEOUT_SEC = 10.0
# ATR settings
ATR_PERIOD = 14
ATR_TIMEFRAME = mt5.TIMEFRAME_M15
# Breakout settings (Donchian, M15)
BREAKOUT_PERIOD = 20
BREAKOUT_TIMEFRAME = mt5.TIMEFRAME_M15
# EMA filter (H1)
EMA_FAST = 50
EMA_SLOW = 200
EMA_TIMEFRAME = mt5.TIMEFRAME_H1
# ATR multipliers
ATR_SL_MULT = 2.0
ATR_TRAIL_START_MULT = 1.0
ATR_TRAIL_DIST_MULT = 1.5
ATR_PYRAMID2_MULT = 0.5
ATR_PYRAMID3_MULT = 1.0
ATR_COMMON_SL_MULT = 1.0
# Pyramiding
MAX_PYRAMID_LEVEL = 3
SL_UPDATE_MIN_PT = 5.0
MARKET_SESSIONS = {
0: (110, 2355), 1: (110, 2355), 2: (110, 2355),
3: (110, 2355), 4: (110, 2350), 5: None, 6: None,
}
logger = logging.getLogger("TrendFollow")
logger.setLevel(logging.INFO)
logger.propagate = False
_formatter = logging.Formatter('%(asctime)s.%(msecs)03d[%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
_file_handler = logging.FileHandler('trend_follow.log', encoding='utf-8')
_file_handler.setFormatter(_formatter)
_console_handler = logging.StreamHandler(sys.stdout)
_console_handler.setFormatter(_formatter)
logger.addHandler(_file_handler)
logger.addHandler(_console_handler)
def _is_market_open(server_timestamp: int) -> bool:
dt = datetime.fromtimestamp(server_timestamp, tz=timezone.utc)
session = MARKET_SESSIONS.get(dt.weekday())
if not session:
return False
return session[0] <= (dt.hour * 100 + dt.minute) <= session[1]
def _build_entry_req(symbol, side, price, sl_price, filling, digits):
return {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": float(FIXED_LOT),
"type": side,
"price": round(price, digits),
"sl": round(sl_price, digits),
"deviation": BASE_DEVIATION,
"magic": MAGIC_NUMBER,
"type_filling": filling,
}
def _build_close_req(symbol, ticket, volume, side, price, filling, digits, dev):
return {
"action": mt5.TRADE_ACTION_DEAL,
"position": ticket,
"symbol": symbol,
"volume": float(volume),
"type": side,
"price": round(price, digits),
"deviation": dev,
"magic": MAGIC_NUMBER,
"type_filling": filling,
}
def _get_exec_price(res, fallback_price):
if not res:
return fallback_price
if res.price != 0.0:
return res.price
if res.deal > 0:
deals = mt5.history_deals_get(ticket=res.deal)
if deals:
return deals[0].price
return fallback_price
class MT5TickFeed:
def __init__(self, symbol):
self.symbol = symbol
self.latest_tick = None
self.recv_perf = 0.0
self.lock = threading.Lock()
self.running = False
def start(self):
self.running = True
threading.Thread(target=self._update_loop, daemon=True).start()
def get(self):
with self.lock:
return self.latest_tick, self.recv_perf
def _update_loop(self):
p_t, p_b = 0, 0.0
while self.running:
t = mt5.symbol_info_tick(self.symbol)
if t and (t.time_msc != p_t or t.bid != p_b):
now = time.perf_counter()
with self.lock:
self.latest_tick, self.recv_perf = t, now
p_t, p_b = t.time_msc, t.bid
time.sleep(0.0001)
class TradeState:
def __init__(self):
self.lock = threading.Lock()
self.is_ordering = False
self.is_closing = False
self.pyramid_level = 0
self.direction = None
self.entry_price_first = 0.0
self.peak_price = 0.0
self.peak_unrealized_pt = 0.0
self.trail_active = False
self.last_sl_price = 0.0
self.last_close_perf = -COOLDOWN
self.last_entry_perf = -ENTRY_BLACKOUT_SEC
self.entry_count = 0
self.explicitly_closed_tickets = set()
def _reset_state(state):
state.is_closing = False
state.is_ordering = False
state.pyramid_level = 0
state.direction = None
state.entry_price_first = 0.0
state.peak_price = 0.0
state.peak_unrealized_pt = 0.0
state.trail_active = False
state.last_sl_price = 0.0
def calc_atr(symbol, timeframe, period):
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, period + 1)
if rates is None or len(rates) < period + 1:
return None
tr_list = []
for i in range(1, len(rates)):
high = rates[i]['high']
low = rates[i]['low']
prev_close = rates[i - 1]['close']
tr = max(high - low, abs(high - prev_close), abs(low - prev_close))
tr_list.append(tr)
atr = sum(tr_list) / len(tr_list)
return atr
def calc_ema(values, period):
if len(values) < period:
return None
k = 2.0 / (period + 1)
ema = sum(values[:period]) / period
for v in values[period:]:
ema = v * k + ema * (1 - k)
return ema
def get_ema_filter(symbol, timeframe, fast_period, slow_period):
needed = slow_period + 50
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, needed)
if rates is None or len(rates) < needed:
return None
closes = [r['close'] for r in rates]
ema_fast = calc_ema(closes, fast_period)
ema_slow = calc_ema(closes, slow_period)
if ema_fast is None or ema_slow is None:
return None
if ema_fast > ema_slow:
return 1
elif ema_fast < ema_slow:
return -1
return 0
def get_breakout_signal(symbol, timeframe, period):
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, period + 1)
if rates is None or len(rates) < period + 1:
return 0
prev_bars = rates[:-1]
current = rates[-1]
highest = max(r['high'] for r in prev_bars)
lowest = min(r['low'] for r in prev_bars)
if current['close'] > highest:
return 1
elif current['close'] < lowest:
return -1
return 0
def update_sl_all_positions(symbol, positions, new_sl, pt, digits, state, pos_type):
with state.lock:
last_sl = state.last_sl_price
if pos_type == mt5.ORDER_TYPE_BUY and new_sl <= last_sl:
return
if pos_type == mt5.ORDER_TYPE_SELL and new_sl >= last_sl:
return
if abs(new_sl - last_sl) / pt < SL_UPDATE_MIN_PT:
return
state.last_sl_price = new_sl
success = 0
for pos in positions:
req = {
"action": mt5.TRADE_ACTION_SLTP,
"position": pos.ticket,
"sl": round(new_sl, digits),
"tp": 0.0,
}
res = mt5.order_send(req)
if res and res.retcode == mt5.TRADE_RETCODE_DONE:
success += 1
else:
logger.warning(f"SL update failed ticket:{pos.ticket} retcode:{res.retcode if res else -1}")
if success > 0:
logger.info(f"SL updated: NewSL:{new_sl:.2f} ({success}/{len(positions)} positions)")
else:
with state.lock:
state.last_sl_price = last_sl
def execute_entry(symbol, side, filling, digits, atr, reason, state):
tick = mt5.symbol_info_tick(symbol)
if not tick:
with state.lock:
state.is_ordering = False
return
pt = mt5.symbol_info(symbol).point
order_price = tick.ask if side == mt5.ORDER_TYPE_BUY else tick.bid
sl_price = order_price - (ATR_SL_MULT * atr) if side == mt5.ORDER_TYPE_BUY else order_price + (ATR_SL_MULT * atr)
req = _build_entry_req(symbol, side, order_price, sl_price, filling, digits)
t0 = time.perf_counter()
try:
res = mt5.order_send(req)
send_ms = (time.perf_counter() - t0) * 1000.0
if res and res.retcode == mt5.TRADE_RETCODE_DONE:
exec_price = _get_exec_price(res, order_price)
slip_pt = (order_price - exec_price) / pt if side == mt5.ORDER_TYPE_BUY else (exec_price - order_price) / pt
with state.lock:
state.entry_count += 1
entry_no = state.entry_count
state.last_entry_perf = time.perf_counter()
if state.pyramid_level == 0:
state.entry_price_first = exec_price
state.last_sl_price = sl_price
state.direction = side
state.pyramid_level += 1
lvl = state.pyramid_level
logger.info(f"Entry[#{entry_no}][Lv{lvl}][{reason}]:{'BUY' if side == mt5.ORDER_TYPE_BUY else 'SELL'} ExecP:{exec_price:.2f} SL:{sl_price:.2f} ATR:{atr/pt:.1f}pt Slip:{slip_pt:.1f}pt Send:{send_ms:.1f}ms")
else:
logger.warning(f"Entry Failed [{reason}]: Retcode:{res.retcode if res else -1}")
finally:
with state.lock:
state.is_ordering = False
def execute_close_all(symbol, positions, filling, digits, reason, state):
pt = mt5.symbol_info(symbol).point
with state.lock:
state.is_closing = True
state.last_close_perf = time.perf_counter()
entry_no = state.entry_count
closed_count = 0
for pos in positions:
close_side = mt5.ORDER_TYPE_SELL if pos.type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY
closed = False
for attempt in range(CLOSE_MAX_RETRIES):
if not mt5.positions_get(ticket=pos.ticket):
closed = True
break
t = mt5.symbol_info_tick(symbol)
if not t:
time.sleep(CLOSE_RETRY_INTERVAL)
continue
price = t.bid if close_side == mt5.ORDER_TYPE_SELL else t.ask
req = _build_close_req(symbol, pos.ticket, pos.volume, close_side, price, filling, digits, EXIT_DEVIATION)
t0 = time.perf_counter()
res = mt5.order_send(req)
ms = (time.perf_counter() - t0) * 1000.0
if res and res.retcode in [mt5.TRADE_RETCODE_DONE, mt5.TRADE_RETCODE_DONE_PARTIAL]:
exec_p = _get_exec_price(res, price)
pft = (exec_p - pos.price_open) / pt if pos.type == mt5.ORDER_TYPE_BUY else (pos.price_open - exec_p) / pt
logger.info(f"Exit[#{entry_no}][{reason}] Ticket:{pos.ticket} Pft:{pft:.1f}pt ExecP:{exec_p:.2f} Send:{ms:.1f}ms")
with state.lock:
state.explicitly_closed_tickets.add(pos.ticket)
closed = True
closed_count += 1
break
time.sleep(CLOSE_RETRY_INTERVAL)
if not closed:
logger.warning(f"Exit[{reason}] Ticket:{pos.ticket} failed after {CLOSE_MAX_RETRIES} retries")
with state.lock:
_reset_state(state)
def _force_close_all(filling, digits):
for attempt in range(10):
positions = mt5.positions_get(magic=MAGIC_NUMBER)
if not positions:
return
for pos in positions:
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY
tick = mt5.symbol_info_tick(pos.symbol)
if not tick:
continue
price = tick.bid if side == mt5.ORDER_TYPE_SELL else tick.ask
req = _build_close_req(pos.symbol, pos.ticket, pos.volume, side, price, filling, digits, 200)
res = mt5.order_send(req)
if res and res.retcode in [mt5.TRADE_RETCODE_DONE, mt5.TRADE_RETCODE_DONE_PARTIAL]:
logger.info(f"ForceClose ticket:{pos.ticket} retcode:{res.retcode}")
else:
logger.warning(f"ForceClose failed ticket:{pos.ticket} retcode:{res.retcode if res else -1}")
time.sleep(0.2)
remaining = mt5.positions_get(magic=MAGIC_NUMBER)
if remaining:
logger.error(f"ForceClose: {len(remaining)} position(s) still open after all retries")
def main():
if not mt5.initialize():
return
info = mt5.symbol_info(SYMBOL_MT5)
if not info:
return
POINT = info.point
DIGITS = max(0, round(-math.log10(POINT)))
FILLING = mt5.ORDER_FILLING_FOK if info.filling_mode & 1 else mt5.ORDER_FILLING_RETURN
mt5_feed = MT5TickFeed(SYMBOL_MT5)
mt5_feed.start()
state = TradeState()
logger.info(
f"Bot v1.90 started: ATR_PERIOD={ATR_PERIOD} BREAKOUT_PERIOD={BREAKOUT_PERIOD} "
f"EMA={EMA_FAST}/{EMA_SLOW} SL_MULT={ATR_SL_MULT} TRAIL_START={ATR_TRAIL_START_MULT} "
f"TRAIL_DIST={ATR_TRAIL_DIST_MULT} PYRAMID_MAX={MAX_PYRAMID_LEVEL}"
)
prev_tickets = set()
prev_state_snapshot = {}
last_signal_check = 0.0
SIGNAL_CHECK_INTERVAL = 15.0
try:
while True:
now_perf = time.perf_counter()
now_wall = time.time()
tick, mt5_recv_p = mt5_feed.get()
if not tick:
time.sleep(0.001)
continue
m_age_ms = (now_perf - mt5_recv_p) * 1000.0
spread_pt = (tick.ask - tick.bid) / POINT
with state.lock:
is_ord = state.is_ordering
is_cls = state.is_closing
l_close = state.last_close_perf
l_entry = state.last_entry_perf
pyramid_level = state.pyramid_level
direction = state.direction
trail_active = state.trail_active
positions = mt5.positions_get(symbol=SYMBOL_MT5, magic=MAGIC_NUMBER)
curr_tickets = {p.ticket for p in positions} if positions else set()
if is_cls and (now_perf - l_close) > CLOSE_TIMEOUT_SEC:
if not curr_tickets:
with state.lock:
_reset_state(state)
is_cls = False
else:
logger.warning("CloseTimeout: triggering force close")
_force_close_all(FILLING, DIGITS)
vanished = prev_tickets - curr_tickets
for ticket in vanished:
with state.lock:
if ticket in state.explicitly_closed_tickets:
state.explicitly_closed_tickets.discard(ticket)
prev_state_snapshot.pop(ticket, None)
continue
snap = prev_state_snapshot.get(ticket, {})
snap_peak_pt = snap.get("peak_unrealized_pt", 0.0)
snap_entry_no = snap.get("entry_count", 0)
snap_trail = snap.get("trail_active", False)
sl_label = "SL_HIT(TRAIL)" if snap_trail else "SL_HIT(RAW)"
deals = mt5.history_deals_get(position=ticket)
if deals:
entry_deal = next((d for d in deals if d.entry == mt5.DEAL_ENTRY_IN), None)
exit_deal = next((d for d in deals if d.entry == mt5.DEAL_ENTRY_OUT), None)
if entry_deal and exit_deal:
pft_pt = (exit_deal.price - entry_deal.price) / POINT if entry_deal.type == mt5.ORDER_TYPE_BUY else (entry_deal.price - exit_deal.price) / POINT
hold = exit_deal.time - entry_deal.time
direction_str = "BUY" if entry_deal.type == mt5.ORDER_TYPE_BUY else "SELL"
logger.info(f"Exit[#{snap_entry_no}][{sl_label}] Ticket:{ticket} Dir:{direction_str} Pft:{pft_pt:.1f}pt Peak:{snap_peak_pt:.1f}pt Hold:{hold:.1f}s")
with state.lock:
state.last_close_perf = time.perf_counter()
if not state.is_closing:
_reset_state(state)
prev_state_snapshot.pop(ticket, None)
prev_tickets = curr_tickets
if positions and not is_cls:
pos0 = positions[0]
curr_p = tick.bid if pos0.type == mt5.ORDER_TYPE_BUY else tick.ask
with state.lock:
entry_price_first = state.entry_price_first
unrealized_pt = (curr_p - entry_price_first) / POINT if pos0.type == mt5.ORDER_TYPE_BUY else (entry_price_first - curr_p) / POINT
if not state.trail_active and unrealized_pt >= 0:
atr_now = calc_atr(SYMBOL_MT5, ATR_TIMEFRAME, ATR_PERIOD)
if atr_now and unrealized_pt >= (ATR_TRAIL_START_MULT * atr_now / POINT):
state.trail_active = True
state.peak_price = curr_p
state.peak_unrealized_pt = unrealized_pt
trail_active = True
logger.info(f"TrailSL activated at unrealized:{unrealized_pt:.1f}pt")
if state.trail_active:
if (pos0.type == mt5.ORDER_TYPE_BUY and curr_p > state.peak_price) or \
(pos0.type == mt5.ORDER_TYPE_SELL and curr_p < state.peak_price):
state.peak_price = curr_p
state.peak_unrealized_pt = unrealized_pt
for pos in positions:
prev_state_snapshot[pos.ticket] = {
"trail_active": state.trail_active,
"peak_unrealized_pt": state.peak_unrealized_pt,
"entry_count": state.entry_count,
}
if trail_active:
with state.lock:
peak_p = state.peak_price
atr_now = calc_atr(SYMBOL_MT5, ATR_TIMEFRAME, ATR_PERIOD)
if atr_now:
new_sl = peak_p - ATR_TRAIL_DIST_MULT * atr_now if pos0.type == mt5.ORDER_TYPE_BUY else peak_p + ATR_TRAIL_DIST_MULT * atr_now
threading.Thread(target=update_sl_all_positions, args=(SYMBOL_MT5, positions, new_sl, POINT, DIGITS, state, pos0.type), daemon=True).start()
if not _is_market_open(tick.time):
threading.Thread(target=execute_close_all, args=(SYMBOL_MT5, list(positions), FILLING, DIGITS, "MKT_CLOSE", state), daemon=True).start()
elif not is_ord and pyramid_level < MAX_PYRAMID_LEVEL and (now_perf - l_entry) > ENTRY_BLACKOUT_SEC:
if now_perf - last_signal_check >= SIGNAL_CHECK_INTERVAL:
last_signal_check = now_perf
atr_now = calc_atr(SYMBOL_MT5, ATR_TIMEFRAME, ATR_PERIOD)
if atr_now:
with state.lock:
entry_p_first = state.entry_price_first
dir_ = state.direction
curr_p2 = tick.bid if dir_ == mt5.ORDER_TYPE_BUY else tick.ask
profit_pt = (curr_p2 - entry_p_first) / POINT if dir_ == mt5.ORDER_TYPE_BUY else (entry_p_first - curr_p2) / POINT
atr_pt = atr_now / POINT
add_entry = False
if pyramid_level == 1 and profit_pt >= ATR_PYRAMID2_MULT * atr_pt:
add_entry = True
elif pyramid_level == 2 and profit_pt >= ATR_PYRAMID3_MULT * atr_pt:
add_entry = True
if add_entry and spread_pt <= SPREAD_FILTER_PT and m_age_ms < MAX_TICK_AGE_MS:
with state.lock:
if not state.is_ordering and not state.is_closing:
state.is_ordering = True
else:
add_entry = False
if add_entry:
new_sl = (curr_p2 - ATR_COMMON_SL_MULT * atr_now) if dir_ == mt5.ORDER_TYPE_BUY else (curr_p2 + ATR_COMMON_SL_MULT * atr_now)
threading.Thread(target=update_sl_all_positions, args=(SYMBOL_MT5, list(positions), new_sl, POINT, DIGITS, state, dir_), daemon=True).start()
threading.Thread(target=execute_entry, args=(SYMBOL_MT5, dir_, FILLING, DIGITS, atr_now, f"PYRAMID_Lv{pyramid_level+1}", state), daemon=True).start()
elif not positions and not is_ord and not is_cls:
if (now_perf - l_close) > COOLDOWN and (now_perf - l_entry) > ENTRY_BLACKOUT_SEC:
if _is_market_open(tick.time) and spread_pt <= SPREAD_FILTER_PT and m_age_ms < MAX_TICK_AGE_MS:
if now_perf - last_signal_check >= SIGNAL_CHECK_INTERVAL:
last_signal_check = now_perf
ema_dir = get_ema_filter(SYMBOL_MT5, EMA_TIMEFRAME, EMA_FAST, EMA_SLOW)
bo_sig = get_breakout_signal(SYMBOL_MT5, BREAKOUT_TIMEFRAME, BREAKOUT_PERIOD)
atr_now = calc_atr(SYMBOL_MT5, ATR_TIMEFRAME, ATR_PERIOD)
if ema_dir is not None and bo_sig != 0 and atr_now and ema_dir == bo_sig:
sig = bo_sig
with state.lock:
if state.is_ordering or state.is_closing:
sig = 0
else:
state.is_ordering = True
if sig != 0:
side = mt5.ORDER_TYPE_BUY if sig > 0 else mt5.ORDER_TYPE_SELL
threading.Thread(target=execute_entry, args=(SYMBOL_MT5, side, FILLING, DIGITS, atr_now, "BREAKOUT", state), daemon=True).start()
time.sleep(0.001)
except KeyboardInterrupt:
pass
finally:
mt5_feed.running = False
_force_close_all(FILLING, DIGITS)
mt5.shutdown()
logger.info("Bot shutdown complete")
if __name__ == "__main__":
main()約定速度が早く、取引コストの低いB-bookブローカーを使っているなら、板情報によるリードラグもおすすめです。こちらも同じくソースコードを用意してあります。
デフォルトはゴールドですが、ビットコインの方が乖離が大きく、利益を出しやすいです。
# Python_OFI_lead_lag_v1.89 TrailSL方向ガード追加・TRAIL_START=35/DIST=20に変更・エントリーブラックアウト0.5秒追加で多重エントリー根絶
import sys
import time
import math
import asyncio
import threading
import collections
import orjson
import MetaTrader5 as mt5
import websockets
import logging
from datetime import datetime, timezone
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(encoding='utf-8')
if hasattr(sys.stderr, 'reconfigure'):
sys.stderr.reconfigure(encoding='utf-8')
SYMBOL_MT5 = "XAUUSD"
SYMBOL_BINANCE = "xauusdt"
MAGIC_NUMBER = 100189
FIXED_LOT = 0.01
BASE_DEVIATION = 30
EXIT_DEVIATION = 50
SPREAD_FILTER_PT = 50
MAX_TICK_AGE_MS = 150
MAX_BINANCE_LATENCY_MS = 120
COOLDOWN = 1.0
CLOSE_RETRY_INTERVAL = 0.05
CLOSE_MAX_RETRIES = 20
CLOSE_TIMEOUT_SEC = 10.0
GAP_EMA_ALPHA = 0.08
MIN_DELTA_GAP_PT = 45.0
OFI_WINDOW_SIZE = 200
OFI_ACCUM_WINDOW = 20
Z_SCORE_THR = 2.0
SL_PT = 50.0
BE_TRIGGER_PT = 25.0
BE_PROFIT_PT = 5.0
TRAIL_START_PT = 35.0
TRAIL_DIST_PT = 20.0
SL_UPDATE_MIN_PT = 5.0
SLIP_FILTER_PT = 50.0
ENTRY_TICKET_RETRY = 10
ENTRY_TICKET_WAIT = 0.05
ENTRY_BLACKOUT_SEC = 0.5
WS_RECV_TIMEOUT = 20.0
WS_RECONNECT_DELAY = 2.0
MARKET_SESSIONS = {
0: (110, 2355), 1: (110, 2355), 2: (110, 2355),
3: (110, 2355), 4: (110, 2350), 5: None, 6: None,
}
logger = logging.getLogger("HFT_OFI")
logger.setLevel(logging.INFO)
logger.propagate = False
_formatter = logging.Formatter('%(asctime)s.%(msecs)03d[%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
_file_handler = logging.FileHandler('hft_ofi.log', encoding='utf-8')
_file_handler.setFormatter(_formatter)
_console_handler = logging.StreamHandler(sys.stdout)
_console_handler.setFormatter(_formatter)
logger.addHandler(_file_handler)
logger.addHandler(_console_handler)
def _is_market_open(server_timestamp: int) -> bool:
dt = datetime.fromtimestamp(server_timestamp, tz=timezone.utc)
session = MARKET_SESSIONS.get(dt.weekday())
if not session:
return False
return session[0] <= (dt.hour * 100 + dt.minute) <= session[1]
def _build_entry_req(symbol, side, price, sl_price, filling, digits):
return {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": float(FIXED_LOT),
"type": side,
"price": round(price, digits),
"sl": round(sl_price, digits),
"deviation": BASE_DEVIATION,
"magic": MAGIC_NUMBER,
"type_filling": filling,
}
def _build_close_req(symbol, ticket, volume, side, price, filling, digits, dev):
return {
"action": mt5.TRADE_ACTION_DEAL,
"position": ticket,
"symbol": symbol,
"volume": float(volume),
"type": side,
"price": round(price, digits),
"deviation": dev,
"magic": MAGIC_NUMBER,
"type_filling": filling,
}
class MT5TickFeed:
def __init__(self, symbol):
self.symbol = symbol
self.latest_tick = None
self.recv_perf = 0.0
self.lock = threading.Lock()
self.running = False
def start(self):
self.running = True
threading.Thread(target=self._update_loop, daemon=True).start()
def get(self):
with self.lock:
return self.latest_tick, self.recv_perf
def _update_loop(self):
p_t, p_b = 0, 0.0
while self.running:
t = mt5.symbol_info_tick(self.symbol)
if t and (t.time_msc != p_t or t.bid != p_b):
now = time.perf_counter()
with self.lock:
self.latest_tick, self.recv_perf = t, now
p_t, p_b = t.time_msc, t.bid
time.sleep(0.0001)
class TradeState:
def __init__(self):
self.lock = threading.Lock()
self.is_ordering = False
self.is_closing = False
self.gap_ema = None
self.last_b_etime = 0.0
self.peak_price = 0.0
self.peak_unrealized_pt = 0.0
self.last_close_perf = -COOLDOWN
self.last_entry_perf = -ENTRY_BLACKOUT_SEC
self.entry_perf = 0.0
self.trail_active = False
self.be_active = False
self.last_sl_price = 0.0
self.entry_count = 0
self.explicitly_closed_tickets = set()
class MarketMicrostructureState:
def __init__(self):
self.prev_bb = (0.0, 0.0)
self.prev_ba = (0.0, 0.0)
self.event_time = 0.0
self.micro_price = 0.0
self.raw_ofi_buf = collections.deque(maxlen=OFI_ACCUM_WINDOW)
self.ofi_history = collections.deque(maxlen=OFI_WINDOW_SIZE)
self.initialized = False
self.lock = threading.Lock()
def update_depth(self, bids, asks, etime):
with self.lock:
if not bids or not asks:
return
bb_p, bb_v = float(bids[0][0]), float(bids[0][1])
ba_p, ba_v = float(asks[0][0]), float(asks[0][1])
self.micro_price = ((bb_p * ba_v + ba_p * bb_v) / (bb_v + ba_v)) if (bb_v + ba_v) > 0 else (bb_p + ba_p) / 2.0
if not self.initialized:
self.prev_bb, self.prev_ba = (bb_p, bb_v), (ba_p, ba_v)
self.event_time, self.initialized = etime, True
return
p_bb_p, p_bb_v = self.prev_bb
p_ba_p, p_ba_v = self.prev_ba
ofi = 0.0
if bb_p > p_bb_p:
ofi += bb_v
elif bb_p == p_bb_p:
ofi += (bb_v - p_bb_v)
if ba_p < p_ba_p:
ofi -= ba_v
elif ba_p == p_ba_p:
ofi -= (ba_v - p_ba_v)
self.raw_ofi_buf.append(ofi)
if len(self.raw_ofi_buf) == OFI_ACCUM_WINDOW:
self.ofi_history.append(sum(self.raw_ofi_buf))
self.prev_bb, self.prev_ba = (bb_p, bb_v), (ba_p, ba_v)
self.event_time = etime
def get_signals(self):
with self.lock:
if len(self.ofi_history) < 2:
return 0.0, self.micro_price, self.event_time
avg = sum(self.ofi_history) / len(self.ofi_history)
std = math.sqrt(sum((x - avg)**2 for x in self.ofi_history) / len(self.ofi_history))
z = (self.ofi_history[-1] - avg) / std if std > 1e-9 else 0.0
return z, self.micro_price, self.event_time
class BinanceFeed:
def __init__(self, symbol):
self.symbol = symbol.lower()
self.ms = MarketMicrostructureState()
def start(self):
threading.Thread(target=lambda: asyncio.run(self._ws_loop()), daemon=True).start()
async def _ws_loop(self):
url = f"wss://fstream.binance.com/stream?streams={self.symbol}@depth5@100ms"
while True:
try:
async with websockets.connect(url, ping_interval=None) as ws:
while True:
raw = await asyncio.wait_for(ws.recv(), timeout=WS_RECV_TIMEOUT)
data = orjson.loads(raw).get("data", {})
self.ms.update_depth(data.get("b", []), data.get("a", []), data.get("E", 0) / 1000.0)
except Exception:
await asyncio.sleep(WS_RECONNECT_DELAY)
def _get_exec_price(res, fallback_price):
if not res:
return fallback_price
if res.price != 0.0:
return res.price
if res.deal > 0:
deals = mt5.history_deals_get(ticket=res.deal)
if deals:
return deals[0].price
return fallback_price
def _reset_state(state):
state.is_closing = False
state.peak_price = 0.0
state.peak_unrealized_pt = 0.0
state.entry_perf = 0.0
state.trail_active = False
state.be_active = False
state.last_sl_price = 0.0
def _sl_hit_label(be_active, trail_active):
if trail_active:
return "SL_HIT(TRAIL)"
if be_active:
return "SL_HIT(BE)"
return "SL_HIT(RAW)"
def update_sl(ticket, new_sl, pt, digits, state, pos_type, reason="SL", unrealized_pt=None):
with state.lock:
last_sl = state.last_sl_price
# 方向ガード: SLが既存より不利な方向への更新を禁止
if pos_type == mt5.ORDER_TYPE_BUY and new_sl <= last_sl:
return
if pos_type == mt5.ORDER_TYPE_SELL and new_sl >= last_sl:
return
if abs(new_sl - last_sl) / pt < SL_UPDATE_MIN_PT:
return
state.last_sl_price = new_sl
req = {
"action": mt5.TRADE_ACTION_SLTP,
"position": ticket,
"sl": round(new_sl, digits),
"tp": 0.0,
}
res = mt5.order_send(req)
if res and res.retcode == mt5.TRADE_RETCODE_DONE:
u_str = f" Unrealized:{unrealized_pt:.1f}pt" if unrealized_pt is not None else ""
logger.info(f"{reason} updated: NewSL:{new_sl:.2f}{u_str}")
else:
logger.warning(f"{reason} update failed: retcode:{res.retcode if res else -1} NewSL:{new_sl:.2f}")
with state.lock:
state.last_sl_price = last_sl
def _slip_filter_close(symbol, ticket, pos_type, exec_price, filling, digits, slip_pt, state):
close_side = mt5.ORDER_TYPE_SELL if pos_type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY
pt = mt5.symbol_info(symbol).point
for attempt in range(CLOSE_MAX_RETRIES):
if not mt5.positions_get(ticket=ticket):
with state.lock:
_reset_state(state)
return
tick = mt5.symbol_info_tick(symbol)
if not tick:
time.sleep(CLOSE_RETRY_INTERVAL)
continue
price = tick.bid if close_side == mt5.ORDER_TYPE_SELL else tick.ask
req = _build_close_req(symbol, ticket, FIXED_LOT, close_side, price, filling, digits, EXIT_DEVIATION)
res = mt5.order_send(req)
if res and res.retcode in [mt5.TRADE_RETCODE_DONE, mt5.TRADE_RETCODE_DONE_PARTIAL]:
exec_p = _get_exec_price(res, price)
pft = (exec_p - exec_price) / pt if pos_type == mt5.ORDER_TYPE_BUY else (exec_price - exec_p) / pt
logger.info(f"Exit[SlipFilter] Slip:{slip_pt:.1f}pt Pft:{pft:.1f}pt ExecP:{exec_p:.2f}")
with state.lock:
_reset_state(state)
return
time.sleep(CLOSE_RETRY_INTERVAL)
if mt5.positions_get(ticket=ticket):
logger.warning(f"Exit[SlipFilter] close failed after {CLOSE_MAX_RETRIES} retries, ticket:{ticket} - keeping is_closing=True")
else:
with state.lock:
_reset_state(state)
def execute_entry(symbol, side, filling, digits, z_score, curr_gap_entry, spread_pt, b_lat_ms, m_age_ms, state):
tick = mt5.symbol_info_tick(symbol)
if not tick:
with state.lock:
state.is_ordering = False
return
pt = mt5.symbol_info(symbol).point
order_price = tick.ask if side == mt5.ORDER_TYPE_BUY else tick.bid
sl_price = order_price - (SL_PT * pt) if side == mt5.ORDER_TYPE_BUY else order_price + (SL_PT * pt)
req = _build_entry_req(symbol, side, order_price, sl_price, filling, digits)
t0 = time.perf_counter()
try:
res = mt5.order_send(req)
send_ms = (time.perf_counter() - t0) * 1000.0
if res and res.retcode == mt5.TRADE_RETCODE_DONE:
exec_price = _get_exec_price(res, order_price)
slip_pt = (order_price - exec_price) / pt if side == mt5.ORDER_TYPE_BUY else (exec_price - order_price) / pt
with state.lock:
state.entry_count += 1
entry_no = state.entry_count
state.last_entry_perf = time.perf_counter()
logger.info(f"Entry[#{entry_no}]:{'BUY' if side == mt5.ORDER_TYPE_BUY else 'SELL'} Z:{z_score:.2f} D_Gap:{curr_gap_entry:.1f}pt ExecP:{exec_price:.2f} Slip:{slip_pt:.1f}pt Lat:{b_lat_ms:.1f}ms Age:{m_age_ms:.1f}ms Send:{send_ms:.1f}ms")
if slip_pt > SLIP_FILTER_PT:
ticket = None
for _ in range(ENTRY_TICKET_RETRY):
positions = mt5.positions_get(symbol=symbol)
matched = [p for p in positions if p.magic == MAGIC_NUMBER and p.type == side] if positions else []
if matched:
ticket = matched[0].ticket
break
time.sleep(ENTRY_TICKET_WAIT)
if ticket is not None:
with state.lock:
state.is_closing = True
state.last_close_perf = time.perf_counter()
_slip_filter_close(symbol, ticket, side, exec_price, filling, digits, slip_pt, state)
else:
logger.warning(f"SlipFilter: could not find position ticket after {ENTRY_TICKET_RETRY} retries, resetting state")
with state.lock:
_reset_state(state)
return
with state.lock:
state.peak_price = exec_price
state.peak_unrealized_pt = 0.0
state.entry_perf = time.perf_counter()
state.trail_active = False
state.be_active = False
state.last_sl_price = sl_price
else:
logger.warning(f"Entry Failed: Retcode:{res.retcode if res else -1}")
finally:
with state.lock:
state.is_ordering = False
def execute_close(symbol, ticket, pos_type, open_price, filling, digits, reason, state):
if not mt5.positions_get(ticket=ticket):
logger.info(f"Exit[{reason}] Ticket:{ticket} already closed, skipping")
with state.lock:
state.last_close_perf = time.perf_counter()
_reset_state(state)
return
close_side = mt5.ORDER_TYPE_SELL if pos_type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY
pt = mt5.symbol_info(symbol).point
closed = False
with state.lock:
hold_sec = time.perf_counter() - state.entry_perf
peak_unrealized_pt = state.peak_unrealized_pt
entry_no = state.entry_count
try:
for attempt in range(CLOSE_MAX_RETRIES):
pos_info = mt5.positions_get(ticket=ticket)
if not pos_info:
closed = True
break
rem_vol = pos_info[0].volume
t = mt5.symbol_info_tick(symbol)
if not t:
time.sleep(CLOSE_RETRY_INTERVAL)
continue
price = t.bid if close_side == mt5.ORDER_TYPE_SELL else t.ask
req = _build_close_req(symbol, ticket, rem_vol, close_side, price, filling, digits, EXIT_DEVIATION)
t0 = time.perf_counter()
res = mt5.order_send(req)
ms = (time.perf_counter() - t0) * 1000.0
if res and res.retcode in [mt5.TRADE_RETCODE_DONE, mt5.TRADE_RETCODE_DONE_PARTIAL]:
exec_p = _get_exec_price(res, price)
filled = res.volume if res.volume > 0 else rem_vol
pft = (exec_p - open_price) / pt if pos_type == mt5.ORDER_TYPE_BUY else (open_price - exec_p) / pt
logger.info(f"Exit[#{entry_no}][{reason}] Pft:{pft:.1f}pt Peak:{peak_unrealized_pt:.1f}pt ExecP:{exec_p:.2f} Send:{ms:.1f}ms Vol:{filled:.2f} Hold:{hold_sec:.1f}s")
closed = True
break
time.sleep(CLOSE_RETRY_INTERVAL)
if not closed:
if mt5.positions_get(ticket=ticket):
logger.warning(f"Exit[#{entry_no}][{reason}] Position {ticket} still open after {CLOSE_MAX_RETRIES} retries - keeping is_closing=True")
with state.lock:
state.last_close_perf = time.perf_counter()
return
else:
logger.warning(f"Exit[#{entry_no}][{reason}] Ticket:{ticket} not found after retries - assuming closed by SL")
closed = True
finally:
if closed:
with state.lock:
state.explicitly_closed_tickets.add(ticket)
state.last_close_perf = time.perf_counter()
_reset_state(state)
def _force_close_all(filling, digits):
for attempt in range(10):
positions = mt5.positions_get(magic=MAGIC_NUMBER)
if not positions:
return
for pos in positions:
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY
tick = mt5.symbol_info_tick(pos.symbol)
if not tick:
continue
price = tick.bid if side == mt5.ORDER_TYPE_SELL else tick.ask
req = _build_close_req(pos.symbol, pos.ticket, pos.volume, side, price, filling, digits, 200)
res = mt5.order_send(req)
if res and res.retcode in [mt5.TRADE_RETCODE_DONE, mt5.TRADE_RETCODE_DONE_PARTIAL]:
logger.info(f"ForceClose ticket:{pos.ticket} retcode:{res.retcode}")
else:
logger.warning(f"ForceClose failed ticket:{pos.ticket} retcode:{res.retcode if res else -1}")
time.sleep(0.2)
remaining = mt5.positions_get(magic=MAGIC_NUMBER)
if remaining:
logger.error(f"ForceClose: {len(remaining)} position(s) still open after all retries")
def main():
if not mt5.initialize():
return
info = mt5.symbol_info(SYMBOL_MT5)
if not info:
return
POINT = info.point
DIGITS = max(0, round(-math.log10(POINT)))
FILLING = mt5.ORDER_FILLING_FOK if info.filling_mode & 1 else mt5.ORDER_FILLING_RETURN
mt5_feed = MT5TickFeed(SYMBOL_MT5)
mt5_feed.start()
feed, state = BinanceFeed(SYMBOL_BINANCE), TradeState()
feed.start()
logger.info(f"Bot v1.89: SL_PT={SL_PT}, BE={BE_TRIGGER_PT}/{BE_PROFIT_PT}, TRAIL={TRAIL_START_PT}/{TRAIL_DIST_PT}, SL_UPDATE_MIN={SL_UPDATE_MIN_PT}, BLACKOUT={ENTRY_BLACKOUT_SEC}s")
prev_tickets = set()
prev_state_snapshot = {}
try:
while True:
now_perf, now_wall = time.perf_counter(), time.time()
tick, mt5_recv_p = mt5_feed.get()
z_score, micro_p, b_etime = feed.ms.get_signals()
if not tick:
time.sleep(0.0001)
continue
b_lat_ms = (now_wall - b_etime) * 1000.0
m_age_ms = (now_perf - mt5_recv_p) * 1000.0
curr_mid = (tick.bid + tick.ask) / 2.0
raw_gap = (micro_p - curr_mid) / POINT
spread_pt = (tick.ask - tick.bid) / POINT
with state.lock:
if b_etime != state.last_b_etime:
g_ema = state.gap_ema
g_ema = raw_gap if g_ema is None else g_ema * (1.0 - GAP_EMA_ALPHA) + raw_gap * GAP_EMA_ALPHA
state.gap_ema = g_ema
state.last_b_etime = b_etime
g_ema = state.gap_ema if state.gap_ema is not None else raw_gap
is_ord, is_cls, l_close = state.is_ordering, state.is_closing, state.last_close_perf
l_entry = state.last_entry_perf
positions = mt5.positions_get(symbol=SYMBOL_MT5, magic=MAGIC_NUMBER)
curr_tickets = {p.ticket for p in positions} if positions else set()
if is_cls and (now_perf - l_close) > CLOSE_TIMEOUT_SEC:
if not curr_tickets:
with state.lock:
_reset_state(state)
is_cls = False
else:
logger.warning("CloseTimeout: position still open, triggering force close")
_force_close_all(FILLING, DIGITS)
vanished = prev_tickets - curr_tickets
for ticket in vanished:
with state.lock:
if ticket in state.explicitly_closed_tickets:
state.explicitly_closed_tickets.discard(ticket)
prev_state_snapshot.pop(ticket, None)
continue
snap = prev_state_snapshot.get(ticket, {})
snap_be = snap.get("be_active", False)
snap_trail = snap.get("trail_active", False)
snap_peak_pt = snap.get("peak_unrealized_pt", 0.0)
snap_entry_no = snap.get("entry_count", 0)
sl_label = _sl_hit_label(snap_be, snap_trail)
deals = mt5.history_deals_get(position=ticket)
if deals:
entry_deal = next((d for d in deals if d.entry == mt5.DEAL_ENTRY_IN), None)
exit_deal = next((d for d in deals if d.entry == mt5.DEAL_ENTRY_OUT), None)
if entry_deal and exit_deal:
pft_pt = (exit_deal.price - entry_deal.price) / POINT if entry_deal.type == mt5.ORDER_TYPE_BUY else (entry_deal.price - exit_deal.price) / POINT
hold = exit_deal.time - entry_deal.time
direction = "BUY" if entry_deal.type == mt5.ORDER_TYPE_BUY else "SELL"
logger.info(f"Exit[#{snap_entry_no}][{sl_label}] Ticket:{ticket} Dir:{direction} Pft:{pft_pt:.1f}pt Peak:{snap_peak_pt:.1f}pt ExecP:{exit_deal.price:.2f} Hold:{hold:.1f}s")
with state.lock:
state.last_close_perf = time.perf_counter()
if not state.is_closing:
_reset_state(state)
prev_state_snapshot.pop(ticket, None)
prev_tickets = curr_tickets
if positions and not is_cls:
pos = positions[0]
curr_p = tick.bid if pos.type == mt5.ORDER_TYPE_BUY else tick.ask
is_exit, reason, peak_moved = False, "", False
with state.lock:
trail_active = state.trail_active
be_active = state.be_active
unrealized_pt = (curr_p - pos.price_open) / POINT if pos.type == mt5.ORDER_TYPE_BUY else (pos.price_open - curr_p) / POINT
if not trail_active and not be_active and unrealized_pt >= BE_TRIGGER_PT:
state.be_active = True
be_active = True
be_price = pos.price_open + BE_PROFIT_PT * POINT if pos.type == mt5.ORDER_TYPE_BUY else pos.price_open - BE_PROFIT_PT * POINT
threading.Thread(target=update_sl, args=(pos.ticket, be_price, POINT, DIGITS, state, pos.type, "BreakEven", unrealized_pt), daemon=True).start()
if not trail_active and unrealized_pt >= TRAIL_START_PT:
state.trail_active = True
state.peak_price = curr_p
state.peak_unrealized_pt = unrealized_pt
trail_active = True
if trail_active and not state.is_closing:
if (pos.type == mt5.ORDER_TYPE_BUY and curr_p > state.peak_price) or \
(pos.type == mt5.ORDER_TYPE_SELL and curr_p < state.peak_price):
state.peak_price = curr_p
state.peak_unrealized_pt = unrealized_pt
peak_moved = True
if not _is_market_open(tick.time):
is_exit, reason = True, "MKT_CLOSE"
if is_exit:
peak_moved = False
if not state.is_closing:
state.is_closing = True
state.last_close_perf = now_perf
prev_state_snapshot[pos.ticket] = {
"be_active": state.be_active,
"trail_active": state.trail_active,
"peak_unrealized_pt": state.peak_unrealized_pt,
"entry_count": state.entry_count,
}
if peak_moved:
new_sl = state.peak_price - TRAIL_DIST_PT * POINT if pos.type == mt5.ORDER_TYPE_BUY else state.peak_price + TRAIL_DIST_PT * POINT
threading.Thread(target=update_sl, args=(pos.ticket, new_sl, POINT, DIGITS, state, pos.type, "TrailSL", unrealized_pt), daemon=True).start()
if is_exit:
threading.Thread(target=execute_close, args=(SYMBOL_MT5, pos.ticket, pos.type, pos.price_open, FILLING, DIGITS, reason, state), daemon=True).start()
elif not positions and not is_ord and not is_cls:
curr_gap_entry = raw_gap - g_ema
if (now_perf - l_close) > COOLDOWN and (now_perf - l_entry) > ENTRY_BLACKOUT_SEC and _is_market_open(tick.time):
if b_lat_ms < MAX_BINANCE_LATENCY_MS and m_age_ms < MAX_TICK_AGE_MS and spread_pt <= SPREAD_FILTER_PT:
sig = 1 if (z_score > Z_SCORE_THR and curr_gap_entry > MIN_DELTA_GAP_PT) else \
-1 if (z_score < -Z_SCORE_THR and curr_gap_entry < -MIN_DELTA_GAP_PT) else 0
if sig != 0:
with state.lock:
if state.is_ordering or state.is_closing:
sig = 0
else:
state.is_ordering = True
if sig != 0:
threading.Thread(target=execute_entry, args=(SYMBOL_MT5, mt5.ORDER_TYPE_BUY if sig > 0 else mt5.ORDER_TYPE_SELL, FILLING, DIGITS, z_score, curr_gap_entry, spread_pt, b_lat_ms, m_age_ms, state), daemon=True).start()
time.sleep(0.0001)
except KeyboardInterrupt:
pass
finally:
mt5_feed.running = False
_force_close_all(FILLING, DIGITS)
mt5.shutdown()
logger.info("Bot shutdown complete")
if __name__ == "__main__":
main()ソースコードの動かし方などは、LLMにコードをコピペして質問すればわかりやすく解説してくれるはずので、こちらの記事では割愛します。
手順2：取引ログをLLMに食わせて、botを改良する
取引botをWindowsのコマンドプロンプトで運用していると、ログが出力されます。このログをLLMに食わせることで、コードの改善点などを発見しやすくなります。
LLMはインプットの質が高いほど、アウトプットの質も高くなります。雑に「収益性を改善して」と指示するよりも、詳細な取引ログとセットで指示した方が出力は改善します。
LLM指示書(Python bot)
ルール
・コードを記述する前に、原案を出すこと
・指示があるまで、コードは生成しないこと
・コードを修正したら、Verを0.01増やし＋修正内容をコード最上部に記述する（日本語で1行。前回のものに上書きする）
・コードを修正したら、マジックナンバーも1増やす
・最上部以外のコメントは最低限にする
・説明する時に図を使わないこと（文章メインで）
・コード生成する場合は、フルコードで
・初回起動時のログは、英語にする
・取引ログがあれば、チャット内でトレード1回あたりの平均獲得pips、平均ポジション保有時間にも言及する
やって欲しいこと
・取引ログを元に、ソースコードの収益性改善手順3：デモ口座で利益が出せたら、リアル口座に切り替える
デモ口座で安定して利益を出せるようになったら、リアル口座で0.01ロット取引から始めましょう。リアル口座はデモ口座と違って、約定に時間がかかるため、特にHFTだと利益を出しにくくなります。
botのトレードロックがデイトレードやスイングトレードなら問題ないですが、HFT・スキャならデモ口座では利益を出せても、リアル口座では赤字になってしまうケースがよくあります。
HFT botで利益を出したいなら、リアル口座のpingは10ms以下、約定時間は150ms以下になるよう取引環境を改善しましょう。
最初に使うブローカーはTradeviewがおすすめ。取引コストが優秀で、入出金経路も豊富で、約定力も優秀なので、海外FX上級者の間で評判がいいです。
あわせて読みたい
Tradeview Markets – FX取引・低スプレッドのグローバルブローカー
TradeviewでFXや株式、先物を取引。低スプレッドと高速約定。今すぐ口座開設して始めましょう。
他にも優れたブローカーを探したいなら、海外FX調査兵団のサイトとXアカウント（@wwfxinfo_tw）を参考にしましょう。
海外FX調査兵団
海外FX調査兵団
裁定取引(HFT)に基づく海外FX業者レビュー/詐欺アラート/ユーザー評価投稿サイト。口座開設数230超え 100社以上掲載中。忖度のない業者のクチコミ 評価お待ちしています。
ブローカー選びに関しては、以下の3つの記事が参考になります。
海外FX調査兵団
総合力の高い海外FX業者TOP5 | 海外FX調査兵団
世の中にあふれる海外FX業者ランキングのほとんどは、単なるアフィリエイト報酬順です。悪質とまでは言いませんが、実際に200社以上に口座開設をして得られた知見とは全く…
海外FX調査兵団
個人向けFIX API取引可能な海外FX業者TOP5 | 海外FX調査兵団
個人向けにFIX API取引を提供している業者をハードルの低さ、約定の安定性を重視し、完全A-Bookの業者の中から5社をランキングしました。
