my_metatrader5_mcp/mtlib/backtest.py

"""共享回测引擎 - 全项目唯一。

移植自旧 xauusd_scalp_backtest.py, 语义逐行对齐(冒烟测试锁定等价):
- 入场: 收盘价 ± spread/2 (多买入高半差/空卖出低半差)
- 出场: 触发优先 SL > TP > max_bars 时间止损(收盘价)
- 移动止损: 用入场 ATR 为基准, start/step 均以 ATR 倍数计, 每根K线收盘更新
- 每笔 PnL 扣满 spread: pnl = (exit - entry - spread) * volume * contract_size * rate
  rate = quote_to_account_rate (1 单位报价货币 = rate 账户货币):
    USD 报价对子恒 1.0 (与旧公式/MT5 实盘成交逐分一致)
    JPY 报价对子 (USDJPY) 按持仓期间收盘价折算, 与 MT5 tick_value 语义对齐
- 手数 = max(min_volume, round(balance*risk_per_trade/(sl_dist*contract_size*rate), 2))
"""

from .cli import position_sizing


def run_backtest(df, signals,
                 initial_balance=10000, risk_per_trade=0.01,
                 tp_atr_mult=1.2, sl_atr_mult=0.6,
                 trailing_start_atr=1.0, trailing_step_atr=0.2,
                 max_bars=40, spread=0.16, contract_size=100, min_volume=0.01,
                 quote_to_account_rate=None, instrument=None):
    """跑一遍事件循环回测。

    返回 (trades, equity_curve):
      trades: list[dict], 每笔含 entry_bar/exit_bar/direction/entry_price/
              exit_price/sl/tp/pnl/pnl_pips/bars_held/exit_reason/balance
      equity_curve: list[float], 逐K线权益曲线(含浮动盈亏)
    """
    balance = initial_balance
    equity = initial_balance
    equity_curve = []
    trades = []
    position = None

    atr = df['atr_14'].values
    close = df['close'].values
    high = df['high'].values
    low = df['low'].values

    # 折算率: None=USD报价对子恒等1.0; 标量=全程恒定; 序列=逐K线;
    # instrument 给定且为 JPY 报价对子时, 内部按当时收盘价推导 (分片安全)
    quote_to_account_rate_orig = quote_to_account_rate
    if instrument is not None:
        from .instruments import get_instrument, quote_to_account_series
        inst = get_instrument(instrument)
        series = quote_to_account_series(inst, df)
        if series is not None:
            quote_to_account_rate_orig = series
    if quote_to_account_rate_orig is None:
        rate_at = lambda i: 1.0
    elif isinstance(quote_to_account_rate_orig, (int, float)):
        _flat = float(quote_to_account_rate_orig)
        rate_at = lambda i: _flat
    else:
        _arr = [float(r) for r in quote_to_account_rate_orig]
        rate_at = lambda i: _arr[i]

    for i in range(len(df)):
        if position is not None:
            bars_held = i - position['entry_bar']
            if position['direction'] == 1:
                profit_so_far = close[i] - position['entry_price']
                if profit_so_far > trailing_start_atr * position['entry_atr']:
                    new_sl = close[i] - trailing_step_atr * position['entry_atr']
                    if new_sl > position['sl']:
                        position['sl'] = new_sl
            else:
                profit_so_far = position['entry_price'] - close[i]
                if profit_so_far > trailing_start_atr * position['entry_atr']:
                    new_sl = close[i] + trailing_step_atr * position['entry_atr']
                    if new_sl < position['sl']:
                        position['sl'] = new_sl

            if position['direction'] == 1:
                sl_hit = low[i] <= position['sl']
                tp_hit = high[i] >= position['tp']
            else:
                sl_hit = high[i] >= position['sl']
                tp_hit = low[i] <= position['tp']

            exit_price = None
            exit_reason = None
            if sl_hit:
                exit_price = position['sl']
                exit_reason = 'SL'
            elif tp_hit:
                exit_price = position['tp']
                exit_reason = 'TP'
            elif bars_held >= max_bars:
                exit_price = close[i]
                exit_reason = 'TIME'

            if exit_price is not None:
                # 历史精确折算: JPY 对子按平仓当根K线收盘价换算为账户货币
                rate_exit = rate_at(i)
                if position['direction'] == 1:
                    pnl_price = exit_price - position['entry_price'] - spread
                else:
                    pnl_price = position['entry_price'] - exit_price - spread
                pnl = pnl_price * position['volume'] * contract_size * rate_exit
                balance += pnl
                trades.append({
                    'entry_bar': position['entry_bar'],
                    'exit_bar': i,
                    'direction': position['direction'],
                    'entry_price': position['entry_price'],
                    'exit_price': exit_price,
                    'sl': position['sl'],
                    'tp': position['tp'],
                    'pnl': pnl,
                    'pnl_pips': pnl_price * 100,
                    'bars_held': bars_held,
                    'exit_reason': exit_reason,
                    'balance': balance
                })
                position = None

        if position is None and signals.iloc[i] != 0:
            direction = signals.iloc[i]
            entry_price = close[i] + (spread / 2 if direction == 1 else -spread / 2)
            atr_val = atr[i]
            sl_dist = sl_atr_mult * atr_val
            tp_dist = tp_atr_mult * atr_val
            if direction == 1:
                sl = entry_price - sl_dist
                tp = entry_price + tp_dist
            else:
                sl = entry_price + sl_dist
                tp = entry_price - tp_dist
            # 手数: 单源 position_sizing, 实盘调用同一函数 (折算率取入场当根)
            volume = position_sizing(
                balance, risk_per_trade, sl_dist, contract_size, min_volume,
                quote_to_account_rate=rate_at(i))
            position = {
                'entry_bar': i,
                'direction': direction,
                'entry_price': entry_price,
                'sl': sl,
                'tp': tp,
                'volume': volume,
                'entry_atr': atr_val
            }

        if position is not None:
            rate_now = rate_at(i)
            if position['direction'] == 1:
                unrealized = (close[i] - position['entry_price'] - spread / 2) * position['volume'] * contract_size * rate_now
            else:
                unrealized = (position['entry_price'] - close[i] - spread / 2) * position['volume'] * contract_size * rate_now
            equity = balance + unrealized
        else:
            equity = balance
        equity_curve.append(equity)

    return trades, equity_curve