forex-scalping.com

Code: Select all

pip install MetaTrader5 pandas numpy

Code: Select all

import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional

import MetaTrader5 as mt5
import numpy as np
import pandas as pd


@dataclass
class ScalperConfig:
    symbol: str = "EURUSD"
    timeframe: int = mt5.TIMEFRAME_M1
    magic_number: int = 884210
    risk_per_trade: float = 0.005       # 0.5% account equity risk per trade
    max_spread_pips: float = 0.8        # Hard spread ceiling in pips
    max_slippage_points: int = 10       # Max acceptable slippage (1.0 pip on 5-digit broker)
    atr_period: int = 14
    sl_atr_mult: float = 1.5
    tp_atr_mult: float = 2.0            # 1 : 1.33 Risk-to-Reward baseline
    ema_fast: int = 20
    ema_slow: int = 50
    rsi_period: int = 7
    session_start_utc: int = 7          # 07:00 UTC (London Open)
    session_end_utc: int = 16           # 16:00 UTC (NY Overlap Close)
    max_open_positions: int = 1
    poll_interval_sec: float = 1.0


def initialize_mt5() -> bool:
    """Initializes connection to the active MetaTrader 5 terminal."""
    if not mt5.initialize():
        print(f"[ERROR] MT5 initialization failed: {mt5.last_error()}")
        return False
    account = mt5.account_info()
    if account is None:
        print("[ERROR] Could not retrieve account info.")
        return False
    print(f"[INFO] Connected to MT5 | Account: {account.login} | Balance: {account.balance:.2f} {account.currency}")
    return True


def is_within_session(cfg: ScalperConfig) -> bool:
    """Checks if current UTC hour falls within active liquidity hours."""
    now_utc = datetime.now(timezone.utc)
    if now_utc.weekday() >= 5:  # Skip Saturday/Sunday
        return False
    return cfg.session_start_utc <= now_utc.hour < cfg.session_end_utc


def get_pip_size(symbol_info: mt5.SymbolInfo) -> float:
    """Returns standard pip value (0.0001 for 5-digit FX, 0.01 for 3-digit JPY pairs)."""
    return symbol_info.point * 10 if symbol_info.digits in (3, 5) else symbol_info.point


def get_current_spread_pips(symbol: str, pip_size: float) -> Optional[float]:
    """Fetches current tick and returns live spread in pips."""
    tick = mt5.symbol_info_tick(symbol)
    if tick is None:
        return None
    return (tick.ask - tick.bid) / pip_size


def count_bot_positions(symbol: str, magic: int) -> int:
    """Counts open positions managed by this specific bot instance."""
    positions = mt5.positions_get(symbol=symbol)
    if positions is None:
        return 0
    return sum(1 for pos in positions if pos.magic == magic)


def fetch_indicators(cfg: ScalperConfig) -> Optional[pd.DataFrame]:
    """Pulls recent M1 bars and computes EMA, RSI, and ATR using vectorized pandas."""
    bars_needed = max(cfg.ema_slow, cfg.atr_period, cfg.rsi_period) + 20
    rates = mt5.copy_rates_from_pos(cfg.symbol, cfg.timeframe, 0, bars_needed)
    if rates is None or len(rates) < bars_needed:
        return None

    df = pd.DataFrame(rates)
    df["time"] = pd.to_datetime(df["time"], unit="s", utc=True)

    # Exponential Moving Averages for micro-trend regime
    df["ema_fast"] = df["close"].ewm(span=cfg.ema_fast, adjust=False).mean()
    df["ema_slow"] = df["close"].ewm(span=cfg.ema_slow, adjust=False).mean()

    # Wilder's RSI(7) for short-term pullback exhaustion
    delta = df["close"].diff()
    gain = delta.clip(lower=0).ewm(alpha=1 / cfg.rsi_period, adjust=False).mean()
    loss = (-delta.clip(upper=0)).ewm(alpha=1 / cfg.rsi_period, adjust=False).mean()
    rs = gain / loss.replace(0, np.nan)
    df["rsi"] = 100 - (100 / (1 + rs))

    # True Range & ATR(14)
    prev_close = df["close"].shift(1)
    tr = pd.concat([
        df["high"] - df["low"],
        (df["high"] - prev_close).abs(),
        (df["low"] - prev_close).abs()
    ], axis=1).max(axis=1)
    df["atr"] = tr.ewm(alpha=1 / cfg.atr_period, adjust=False).mean()

    return df


def evaluate_signal(df: pd.DataFrame) -> Optional[str]:
    """
    Evaluates the last completed candle (index -2) to avoid intra-bar repainting.
    Strategy: Micro-trend pullback continuation.
    - BUY: Fast EMA > Slow EMA, RSI(7) dips below 35 and hooks back up above 35.
    - SELL: Fast EMA < Slow EMA, RSI(7) spikes above 65 and hooks back down below 65.
    """
    prev_bar = df.iloc[-3]
    last_closed = df.iloc[-2]

    uptrend = last_closed["ema_fast"] > last_closed["ema_slow"]
    downtrend = last_closed["ema_fast"] < last_closed["ema_slow"]

    if uptrend and prev_bar["rsi"] <= 35.0 and last_closed["rsi"] > 35.0:
        return "BUY"
    if downtrend and prev_bar["rsi"] >= 65.0 and last_closed["rsi"] < 65.0:
        return "SELL"
    return None


def calculate_lot_size(symbol_info: mt5.SymbolInfo, sl_distance_price: float, risk_pct: float) -> float:
    """Computes risk-adjusted lot volume normalized to broker step and min/max limits."""
    account = mt5.account_info()
    if account is None or sl_distance_price <= 0:
        return symbol_info.volume_min

    risk_amount = account.equity * risk_pct
    tick_size = symbol_info.trade_tick_size
    tick_value = symbol_info.trade_tick_value

    ticks_at_risk = sl_distance_price / tick_size
    loss_per_lot = ticks_at_risk * tick_value
    if loss_per_lot <= 0:
        return symbol_info.volume_min

    raw_lots = risk_amount / loss_per_lot
    step = symbol_info.volume_step
    normalized_lots = np.floor(raw_lots / step) * step
    return float(np.clip(normalized_lots, symbol_info.volume_min, symbol_info.volume_max))


def send_market_order(cfg: ScalperConfig, direction: str, atr_value: float) -> None:
    """Constructs and routes an Immediate-or-Cancel (IOC) market order with SL/TP."""
    symbol_info = mt5.symbol_info(cfg.symbol)
    tick = mt5.symbol_info_tick(cfg.symbol)
    if symbol_info is None or tick is None:
        return

    sl_dist = atr_value * cfg.sl_atr_mult
    tp_dist = atr_value * cfg.tp_atr_mult
    digits = symbol_info.digits

    if direction == "BUY":
        order_type = mt5.ORDER_TYPE_BUY
        price = tick.ask
        sl = round(price - sl_dist, digits)
        tp = round(price + tp_dist, digits)
    else:
        order_type = mt5.ORDER_TYPE_SELL
        price = tick.bid
        sl = round(price + sl_dist, digits)
        tp = round(price - tp_dist, digits)

    volume = calculate_lot_size(symbol_info, sl_dist, cfg.risk_per_trade)

    request = {
        "action": mt5.TRADE_ACTION_DEAL,
        "symbol": cfg.symbol,
        "volume": volume,
        "type": order_type,
        "price": price,
        "sl": sl,
        "tp": tp,
        "deviation": cfg.max_slippage_points,
        "magic": cfg.magic_number,
        "comment": "PyScalp_M1",
        "type_time": mt5.ORDER_TIME_GTC,
        "type_filling": mt5.ORDER_FILLING_IOC,
    }

    result = mt5.order_send(request)
    if result is None:
        print(f"[ERROR] order_send returned None: {mt5.last_error()}")
    elif result.retcode != mt5.TRADE_RETCODE_DONE:
        print(f"[WARN] Order rejected | Retcode: {result.retcode} | Comment: {result.comment}")
    else:
        print(f"[EXEC] {direction} {volume} {cfg.symbol} @ {result.price} | SL: {sl} | TP: {tp}")


def run_scalper(cfg: ScalperConfig) -> None:
    if not initialize_mt5():
        return

    symbol_info = mt5.symbol_info(cfg.symbol)
    if symbol_info is None:
        print(f"[ERROR] Symbol {cfg.symbol} not found.")
        mt5.shutdown()
        return

    if not symbol_info.visible:
        mt5.symbol_select(cfg.symbol, True)

    pip_size = get_pip_size(symbol_info)
    last_processed_bar_time = None

    print(f"[INFO] Starting scalper loop on {cfg.symbol} (Max Spread: {cfg.max_spread_pips} pips)...")
    try:
        while True:
            if not is_within_session(cfg):
                time.sleep(10)
                continue

            if count_bot_positions(cfg.symbol, cfg.magic_number) >= cfg.max_open_positions:
                time.sleep(cfg.poll_interval_sec)
                continue

            spread_pips = get_current_spread_pips(cfg.symbol, pip_size)
            if spread_pips is None or spread_pips > cfg.max_spread_pips:
                time.sleep(cfg.poll_interval_sec)
                continue

            df = fetch_indicators(cfg)
            if df is None:
                time.sleep(cfg.poll_interval_sec)
                continue

            # Evaluate once per newly closed M1 candle
            closed_bar_time = df.iloc[-2]["time"]
            if closed_bar_time != last_processed_bar_time:
                last_processed_bar_time = closed_bar_time
                signal = evaluate_signal(df)
                if signal is not None:
                    current_atr = float(df.iloc[-2]["atr"])
                    send_market_order(cfg, signal, current_atr)

            time.sleep(cfg.poll_interval_sec)
    except KeyboardInterrupt:
        print("\n[INFO] Shutting down scalper cleanly.")
    finally:
        mt5.shutdown()


if __name__ == "__main__":
    config = ScalperConfig()
    run_scalper(config)

Code: Select all

# Realistic cost deduction per round-turn trade in price terms
spread_cost = 0.00003        # 0.3 pips average Raw ECN spread on EURUSD
commission_cost = 0.00007    # ~0.7 pips equivalent ($7 per lot round-turn)
slippage_buffer = 0.00002    # 0.2 pips conservative execution slippage

total_friction = spread_cost + commission_cost + slippage_buffer  # 1.2 pips total drag
net_trade_pnl_pips = gross_trade_pnl_pips - (total_friction / 0.0001)