Difference between revisions of "Робот"
(Created page with "<nowiki>#</nowiki> -*- coding: utf-8 -*- """ Scalping Bot – MetaTrader 5 • XAUUSD M1 Version : 2025-05-21 • rev-E4 + profit-guard + hedge-v7 + logging """ impor...") |
(No difference)
|
Latest revision as of 07:29, 29 August 2025
# -*- coding: utf-8 -*-
"""
Scalping Bot – MetaTrader 5 • XAUUSD M1
Version : 2025-05-21 • rev-E4 + profit-guard + hedge-v7 + logging
"""
import time
import logging
from datetime import datetime
from typing import Dict, Optional
import pandas as pd
import MetaTrader5 as mt5
# ───────────────────────── logging (file only) ─────────────────────────
LOGFILE = "scalper_debug.log"
logging.basicConfig(
filename=LOGFILE,
level=logging.DEBUG,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("ScalperBot")
# ───────────────────────── MT5 connection ─────────────────────────────
MT5_PATH = r"C:\Program Files\RoboForex-MT5-002\terminal64.exe"
LOGIN, PASSWORD, SERVER = 68222812, "o,elPj26", "RoboForex-Pro"
SYMBOL, TIMEFRAME = "XAUUSD", mt5.TIMEFRAME_M1
print("Connecting to MetaTrader 5…")
logger.info("Connecting to MT5")
if not mt5.initialize(MT5_PATH, login=LOGIN, password=PASSWORD, server=SERVER):
logger.critical(f"MT5 initialization failed: {mt5.last_error()}")
raise RuntimeError(f"MT5 initialization failed: {mt5.last_error()}")
else:
print("MT5 connected successfully.")
logger.info("MT5 connected")
# ───────────────────────── strategy constants ─────────────────────────
EMA_FAST, EMA_SLOW = 5, 21
TRAIL_START, TRAIL_STEP = 0.10, 0.05
MAX_LOOKAHEAD, LOT = 10, 0.01
BASE_MAGIC, MAX_CLONES = 20001, 1
MIN_POSITIVE_PROFIT_USD = 0.10
# profit-guard
TOTAL_PROFIT_EXIT_USD = 1.0
# momentum-hedge
ATR_PERIOD, MOM_WINDOW = 14, 8
MOM_FACTOR_BUY = 2.0
BREAK_WINDOW_BUY = 10
hedged_positions: Dict[int, int] = {} # {parent_ticket: hedge_ticket}
# ──────────────────────── order-send wrapper ─────────────────────────
def send_order(request: dict, tag: str = ""):
"""Log account metrics and every order request/result."""
acct = mt5.account_info()
if acct:
logger.info(
f"[{tag}] Account metrics → "
f"Balance={acct.balance:.2f}, "
f"Equity={acct.equity:.2f}, "
f"Margin={acct.margin:.2f}, "
f"FreeMargin={acct.margin_free:.2f}, "
f"MarginLevel={acct.margin_level:.2f}%"
)
logger.debug(f"ORDER_SEND {tag} request={request}")
res = mt5.order_send(request)
try:
logger.debug(f"ORDER_SEND {tag} result={res._asdict()}")
except AttributeError:
logger.debug(f"ORDER_SEND {tag} result={res}")
return res
# ─────────────────────── profit-guard helpers ───────────────────────
def total_account_profit() -> float:
return sum(p.profit for p in (mt5.positions_get(symbol=SYMBOL) or []))
def close_all_positions(reason: str):
positions = mt5.positions_get(symbol=SYMBOL)
if not positions:
return
print(f"[{datetime.now()}] {reason}: closing ALL positions (n={len(positions)}) …")
logger.info(f"{reason}: closing {len(positions)} positions")
for p in positions:
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = (
mt5.symbol_info_tick(SYMBOL).bid
if side == mt5.ORDER_TYPE_SELL
else mt5.symbol_info_tick(SYMBOL).ask
)
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": p.volume,
"type": side,
"position": p.ticket,
"price": price,
"deviation": 20,
"magic": p.magic,
"comment": reason,
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
send_order(req, reason)
# startup profit-guard
start_pl = total_account_profit()
if start_pl > 0:
positions = mt5.positions_get(symbol=SYMBOL)
if positions:
profitable = [p for p in positions if p.profit > MIN_POSITIVE_PROFIT_USD]
if profitable:
logger.info(f"Startup profit guard: closing {len(profitable)} profitable positions")
for p in profitable:
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = (
mt5.symbol_info_tick(SYMBOL).bid
if side == mt5.ORDER_TYPE_SELL
else mt5.symbol_info_tick(SYMBOL).ask
)
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": p.volume,
"type": side,
"position": p.ticket,
"price": price,
"deviation": 20,
"magic": p.magic,
"comment": "Startup positive P/L",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
send_order(req, "StartupPositivePL")
# ───────────────────── indicator / data helpers ─────────────────────
def calculate_indicators(df: pd.DataFrame) -> pd.DataFrame:
df['EMA_fast'] = df['close'].ewm(span=EMA_FAST).mean()
df['EMA_slow'] = df['close'].ewm(span=EMA_SLOW).mean()
df['momentum'] = df['close'].diff().rolling(MOM_WINDOW).sum()
tr = pd.concat([
df['high'] - df['low'],
(df['high'] - df['close']).abs(),
(df['low'] - df['close']).abs()
], axis=1).max(axis=1)
df['ATR'] = tr.rolling(ATR_PERIOD).mean()
df['don_high'] = df['high'].rolling(BREAK_WINDOW_BUY).max()
return df
def get_data(symbol, timeframe, n=100):
print(f"[{datetime.now()}] Fetching market data…")
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n)
df = pd.DataFrame(rates)
df['time'] = pd.to_datetime(df['time'], unit='s')
return calculate_indicators(df)
# ───────────────────── entry / clone logic ──────────────────────────
def check_entry(df):
row, prev = df.iloc[-1], df.iloc[-2]
signal = row['EMA_fast'] > row['EMA_slow'] and row['close'] > prev['high']
if signal:
print(f"[{datetime.now()}] Entry signal detected.")
return signal
def open_buy_clones(num_clones):
price = mt5.symbol_info_tick(SYMBOL).ask
print(f"[{datetime.now()}] Opening {num_clones} buy clones at price {price}…")
for i in range(num_clones):
magic = BASE_MAGIC + i
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": LOT,
"type": mt5.ORDER_TYPE_BUY,
"price": price,
"deviation": 20,
"magic": magic,
"comment": f"Clone {i+1}",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
res = send_order(req, f"Clone{i+1}")
if res.retcode == mt5.TRADE_RETCODE_DONE:
print(f" -> Clone {i+1} (magic {magic}) opened successfully.")
else:
print(f" -> Clone {i+1} (magic {magic}) FAILED: {res.retcode}")
def close_positions_by_magic(magic):
positions = mt5.positions_get(symbol=SYMBOL, magic=magic)
if positions:
for pos in positions:
if pos.profit <= MIN_POSITIVE_PROFIT_USD:
print(f" -> Skipping position (magic {magic}) with profit below threshold: {pos.profit:.2f}")
continue
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": pos.volume,
"type": mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY,
"position": pos.ticket,
"price": mt5.symbol_info_tick(SYMBOL).bid if pos.type == mt5.POSITION_TYPE_BUY else mt5.symbol_info_tick(SYMBOL).ask,
"deviation": 20,
"magic": magic,
"comment": "Scalp Exit",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
res = send_order(req, "ScalpExit")
if res.retcode == mt5.TRADE_RETCODE_DONE:
print(f" -> Closed position (ticket {pos.ticket}) with profit: {pos.profit:.2f}")
else:
print(f" -> Failed to close position (ticket {pos.ticket}): {res.retcode}")
def trailing_exit(entry_price, magic):
print(f"[{datetime.now()}] Starting trailing exit for magic {magic}…")
peak_price = entry_price
for step in range(MAX_LOOKAHEAD):
time.sleep(1)
tick = mt5.symbol_info_tick(SYMBOL)
if tick is None:
continue
current_price = tick.bid
print(f" [Step {step+1}] Current price: {current_price}")
if current_price > peak_price:
peak_price = current_price
print(f" -> New peak: {peak_price}")
trail_trigger = entry_price + TRAIL_START
if peak_price > trail_trigger:
exit_price = peak_price - TRAIL_STEP
if current_price <= exit_price:
close_positions_by_magic(magic)
break
positions = mt5.positions_get(symbol=SYMBOL, magic=magic)
if positions:
for pos in positions:
if pos.type == mt5.POSITION_TYPE_BUY and pos.profit > MIN_POSITIVE_PROFIT_USD:
print(f" -> Early exit with profit: {pos.profit:.2f}")
close_positions_by_magic(magic)
return
# ─────────────────────── momentum hedge helpers ─────────────────────
def largest_loser() -> Optional[mt5.TradePosition]:
losers = [p for p in (mt5.positions_get(symbol=SYMBOL) or []) if p.profit < 0]
return min(losers, key=lambda p: p.profit, default=None)
def _normalise_volume(vol: float) -> float:
"""Clamp and round to the broker's lot rules."""
info = mt5.symbol_info(SYMBOL)
if info is None:
return vol
vol = max(info.volume_min, min(vol, info.volume_max))
step = info.volume_step if info.volume_step else 0.01
return round(vol / step) * step
def hedge_on_momentum(df):
row, prev_mom = df.iloc[-1], df['momentum'].iloc[-2]
mom, atr = row['momentum'], row['ATR']
if pd.isna(atr) or (mom > 0) != (prev_mom > 0):
return
if mom > 0 and mom >= atr * MOM_FACTOR_BUY and row['close'] >= row['don_high']:
hedge_side, hedge_price = mt5.ORDER_TYPE_BUY, mt5.symbol_info_tick(SYMBOL).ask
else:
return
parent = largest_loser()
if parent is None or parent.ticket in hedged_positions:
return
volume = _normalise_volume(parent.volume)
magic = BASE_MAGIC + 4000 + parent.ticket % 1000
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": volume,
"type": hedge_side,
"price": hedge_price,
"deviation": 20,
"magic": magic,
"comment": f"MomHedge{parent.ticket}",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
res = send_order(req, "MomHedge")
if res.retcode == mt5.TRADE_RETCODE_DONE:
hedged_positions[parent.ticket] = res.order
print(f"[{datetime.now()}] Momentum hedge opened – parent {parent.ticket} ↔ hedge {res.order} (vol={volume})")
def try_close_pair(parent_ticket: int):
hedge_ticket = hedged_positions.get(parent_ticket)
if hedge_ticket is None:
return
p = mt5.positions_get(ticket=parent_ticket)
h = mt5.positions_get(ticket=hedge_ticket)
if not p or not h:
hedged_positions.pop(parent_ticket, None)
return
parent, hedge = p[0], h[0]
net_pl = parent.profit + hedge.profit
if net_pl < MIN_POSITIVE_PROFIT_USD:
return
for pos in (parent, hedge):
side = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = (
mt5.symbol_info_tick(SYMBOL).bid
if side == mt5.ORDER_TYPE_SELL
else mt5.symbol_info_tick(SYMBOL).ask
)
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": pos.volume,
"type": side,
"position": pos.ticket,
"price": price,
"deviation": 20,
"magic": pos.magic,
"comment": "PairExit",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
send_order(req, "PairExit")
print(f"[{datetime.now()}] ✅ Pair closed (parent {parent_ticket}, hedge {hedge_ticket}) – net +{net_pl:.2f} USD")
hedged_positions.pop(parent_ticket, None)
# ─────────────────────────── main loop ─────────────────────────────
print("Running scalping bot with clone scaling…")
logger.info("Bot started")
while True:
# live profit-guard
if total_account_profit() > TOTAL_PROFIT_EXIT_USD:
positions = mt5.positions_get(symbol=SYMBOL)
if positions:
profitable = [p for p in positions if p.profit > MIN_POSITIVE_PROFIT_USD]
if profitable:
print(f"[{datetime.now()}] Closing {len(profitable)} profitable positions due to total P/L > threshold.")
logger.info(f"Runtime profit guard: {len(profitable)} positions with profit > {MIN_POSITIVE_PROFIT_USD}")
for p in profitable:
side = mt5.ORDER_TYPE_SELL if p.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY
price = (
mt5.symbol_info_tick(SYMBOL).bid
if side == mt5.ORDER_TYPE_SELL
else mt5.symbol_info_tick(SYMBOL).ask
)
req = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": p.volume,
"type": side,
"position": p.ticket,
"price": price,
"deviation": 20,
"magic": p.magic,
"comment": "Runtime positive P/L",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling":mt5.ORDER_FILLING_IOC
}
send_order(req, "RuntimePositivePL")
time.sleep(1)
continue
try:
df = get_data(SYMBOL, TIMEFRAME, 100)
if check_entry(df):
bal = mt5.account_info().balance
clones = min(int(bal // 100), MAX_CLONES)
open_buy_clones(clones)
entry_px = mt5.symbol_info_tick(SYMBOL).ask
for i in range(clones):
trailing_exit(entry_px, BASE_MAGIC + i)
hedge_on_momentum(df)
for parent in list(hedged_positions.keys()):
try_close_pair(parent)
time.sleep(1)
except Exception as e:
print(f"[{datetime.now()}] ERROR: {e}")
logger.exception("Unhandled exception")
time.sleep(1)