mymt5opp/strategies/gold_scalper_pro/scalper_engine.py

"""Python mirror of the GoldScalperPro EA (doc 03, doc 04 Rule 1).

A bar-by-bar fill simulator that reproduces the EA's trade lifecycle:

    new day → reset daily counters + snapshot equity
    each bar → manage open positions (BE / trailing) → daily breaker check
             → evaluate entry signal on the just-closed bar
             → if signal + all gates pass: open at next bar's open

Intra-bar model (doc 03 §2): the pessimistic 4-sub-tick order resolves a bar
that could touch both SL and TP in favour of the SL (the realistic worst
case). The EA's trailing stop is tick-sensitive; we approximate it bar-by-bar
using high/low (doc 03 §7 — the expected fidelity gap on a trailing-stop EA
in volatile history is wider than on a clean-directional setup).

Once this engine reproduces the EA's MT5 numbers within the expected gap
(doc 03 §8) it is FROZEN (doc 04 Rule 1). Fork — don't edit — to test ideas.

The engine consumes PRE-COMPUTED signal + SL/TP price arrays from the
caller (signals.py). It never decides *where* a stop goes; it only decides
whether price touched it. That seam is what makes it freezable.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Optional

import numpy as np
import pandas as pd

from shared.core.engine import Direction, Position, Result, SizingInputs, Trade
from shared.instruments.config import InstrumentConfig


@dataclass
class ScalperConfig:
    """Engine behaviour switches — mirror the EA's frozen inputs.

    These come from FROZEN_BASELINE (search_space.py) and are NOT optimized;
    they define *which* exit logic the EA runs. Tunable point values (BE
    trigger, trail step) arrive via the SL/TP/management arrays the caller
    passes to ``run``.
    """

    use_break_even: bool = True
    use_trailing: bool = True
    use_session: bool = False
    session_start_hour: int = 7
    session_end_hour: int = 20
    max_positions: int = 1
    max_trades_per_day: int = 6
    daily_loss_limit_pct: float = 5.0
    daily_profit_target_pct: float = 0.0       # 0 = off
    min_seconds_between: int = 60
    sizing_mode: int = 1                        # 1 = RISK_PERCENT (frozen)
    fixed_lots: float = 0.01                     # used only if sizing_mode == 0
    risk_percent: float = 1.0                   # used if sizing_mode == 1
    # BE / trailing point values — passed in from the tunable params so the
    # engine stays parametric without re-reading the EA inputs each bar.
    break_even_points: float = 150.0
    break_even_lock: float = 20.0
    trail_start_points: float = 200.0
    trail_step_points: float = 120.0


class ScalperEngine:
    """Bar-by-bar mirror of GoldScalperPro's trade lifecycle.

    Implements the ``Engine`` Protocol from shared.core.engine. Stateless
    across runs — all state lives inside ``run``. The engine is deliberately
    plain Python (no numba) until profiling shows a hot path worth compiling
    (doc 01 — numba is in the stack for that reason, not premature speed).
    """

    def run(
        self,
        bars: pd.DataFrame,
        signals_long: np.ndarray,
        signals_short: np.ndarray,
        sl_prices: np.ndarray,
        tp_prices: np.ndarray,
        instrument: InstrumentConfig,
        sizing: SizingInputs,
        initial_deposit: float,
        *,
        scalper_cfg: Optional[ScalperConfig] = None,
        m1_bars: Optional[pd.DataFrame] = None,
    ) -> Result:
        """Run the scalper over ``bars`` and return a :class:`Result`.

        ``signals_long`` / ``signals_short`` are edge-detected boolean arrays
        (True only on the transition bar). ``sl_prices`` / ``tp_prices`` are
        the per-bar SL/TP *prices* for an entry on that bar (NaN where none).
        The engine opens at the NEXT bar's open (look-ahead guard, doc 02 §3)
        so a signal computed from a closed bar executes on the following bar.

        If ``scalper_cfg`` is None it defaults to :class:`ScalperConfig` (the
        EA's frozen baseline switches). The optimizer wires the tunable BE /
        trailing point values via ``engine_kwargs`` on ObjectiveConfig.

        If ``m1_bars`` is provided, BE/trailing/SL/TP are simulated on the
        M1 tick sequence inside each M5 bar (4 synthetic ticks per M1 bar:
        open→(high|low)→(low|high)→close, direction-aware). This closes the
        bar-level optimism gap on trailing-stop strategies (doc 03 §7).
        """
        cfg = scalper_cfg or ScalperConfig()
        n = len(bars)
        ts = pd.to_datetime(bars["timestamp"].to_numpy())
        opens = bars["open"].to_numpy(dtype=float)
        highs = bars["high"].to_numpy(dtype=float)
        lows = bars["low"].to_numpy(dtype=float)
        closes = bars["close"].to_numpy(dtype=float)
        spreads = bars["spread"].to_numpy(dtype=float) if "spread" in bars else np.zeros(n)

        # ── M1 tick index: map each M5 bar i → slice [m1_lo, m1_hi) in m1 ──
        m1_ticks: Optional[list[np.ndarray]] = None
        if m1_bars is not None and len(m1_bars) > 0:
            m1_ticks = _build_m5_to_m1_index(ts, m1_bars)

        # ── Per-bar daily-state tracking ────────────────────────────────
        point = instrument.point
        trades: list[Trade] = []
        open_pos: Optional[Position] = None     # single-position strategy
        balance = float(initial_deposit)
        equity = float(initial_deposit)

        # Daily counters (mirror g_tradesToday / g_dayStartEquity / g_dayBlocked).
        cur_day = pd.Timestamp(0)
        day_start_equity = float(initial_deposit)
        trades_today = 0
        day_blocked = False
        last_trade_ts: Optional[pd.Timestamp] = None

        # Equity curve sampled at bar close (bounded; resample later if needed).
        eq_rows: list[tuple[pd.Timestamp, float, float]] = []

        for i in range(n):
            t = ts[i]
            day = t.normalize()

            # ── New trading day: reset counters + snapshot equity ───────
            if day != cur_day:
                cur_day = day
                trades_today = 0
                day_blocked = False
                day_start_equity = equity

            # ── 1. Manage open position (BE / trailing) + check exit ────
            if open_pos is not None:
                if m1_ticks is not None:
                    # Tick-level simulation: walk the M1 bars inside this M5 bar,
                    # updating BE/trailing and checking SL/TP on each synthetic tick.
                    exit_trade = self._simulate_m1_exits(
                        open_pos, t, m1_ticks[i], instrument, cfg,
                    )
                else:
                    # Bar-level approximation (original mode).
                    self._manage_position(
                        open_pos, t, opens[i], highs[i], lows[i], closes[i],
                        instrument, cfg, balance, equity,
                    )
                    exit_trade = self._check_exit(
                        open_pos, opens[i], highs[i], lows[i], closes[i],
                        instrument,
                    )
                if exit_trade is not None:
                    tr = self._close_trade(open_pos, exit_trade, t, instrument, balance)
                    balance += tr.pnl
                    equity = balance
                    trades.append(tr)
                    open_pos = None

            # ── 2. Daily circuit breaker ───────────────────────────────
            if not day_blocked and day_start_equity > 0:
                pct = (equity - day_start_equity) / day_start_equity * 100.0
                if cfg.daily_loss_limit_pct > 0 and pct <= -cfg.daily_loss_limit_pct:
                    day_blocked = True
                elif cfg.daily_profit_target_pct > 0 and pct >= cfg.daily_profit_target_pct:
                    day_blocked = True

            # ── 3. Evaluate entry on the just-closed bar; fill next bar ─
            # Look-ahead guard: signal at bar i → entry at bar i+1's open.
            if open_pos is None and i + 1 < n and not day_blocked:
                if self._entry_allowed(
                    cfg, t, trades_today, last_trade_ts, i,
                    signals_long, signals_short,
                ):
                    direction = Direction.LONG if signals_long[i] else Direction.SHORT
                    # Fill at next bar's open ± half spread (ask/bid).
                    spread_pts = instrument.spread_points(spreads[i + 1] if i + 1 < n else spreads[i])
                    spread_price = spread_pts * point
                    fill_price = opens[i + 1]
                    if direction is Direction.LONG:
                        fill_price += spread_price / 2.0     # buy at ask
                    else:
                        fill_price -= spread_price / 2.0     # sell at bid
                    fill_price = round(fill_price, instrument.digits)

                    sl = sl_prices[i] if not np.isnan(sl_prices[i]) else 0.0
                    tp = tp_prices[i] if not np.isnan(tp_prices[i]) else 0.0
                    # _calc_lots expects the SL *distance* (price units), not the
                    # SL price level. signals.py sets sl_prices[i] = close[i] ±
                    # sl_dist[i] (where sl_dist[i] = InpAtrSLMult × ATR[i]), so
                    # |close[i] - sl_prices[i]| recovers exactly sl_dist[i] — the
                    # same distance MT5 uses for sizing (it does NOT include the
                    # gap between signal-bar close and next-bar fill). Using
                    # fill_price here instead made sl_distance vary with the
                    # open-gap, sometimes bigger, sometimes smaller than MT5's
                    # value, which made lots inconsistent and the equity curve
                    # diverge exponentially from MT5 under risk% compounding.
                    sl_distance = abs(closes[i] - sl) if sl > 0 else 0.0
                    lots = self._calc_lots(cfg, instrument, sl_distance, equity)
                    if lots > 0:
                        open_pos = Position(
                            direction=direction,
                            entry_time=ts[i + 1],
                            entry_price=fill_price,
                            lots=lots,
                            open_swap=0.0,
                            sl=round(sl, instrument.digits),
                            tp=round(tp, instrument.digits),
                        )
                        trades_today += 1
                        last_trade_ts = ts[i + 1]

            # ── 4. Mark-to-market equity + sample curve ─────────────────
            if open_pos is not None:
                unreal = self._unrealized_pnl(open_pos, closes[i], instrument)
                # Accumulate swap on the open position daily.
                equity = balance + unreal
            else:
                equity = balance
            eq_rows.append((t, balance, equity))

        # ── End-of-data: close any still-open position at last close ──
        if open_pos is not None:
            exit_trade = ("end_of_data", closes[-1])
            tr = self._close_trade(open_pos, exit_trade, ts[-1], instrument, balance)
            balance += tr.pnl
            equity = balance
            trades.append(tr)
            open_pos = None
            eq_rows.append((ts[-1], balance, equity))

        eq_df = pd.DataFrame(eq_rows, columns=["timestamp", "balance", "equity"])
        return Result(
            trades=trades,
            equity_curve=eq_df,
            final_balance=balance,
            initial_deposit=float(initial_deposit),
            open_positions=[],
            diagnostics={},
        )

    # ────────────────────────────────────────────────────────────────────
    # Helpers — kept private; the public surface is just run().
    # ────────────────────────────────────────────────────────────────────

    def _simulate_m1_exits(
        self,
        pos: Position,
        m5_time: pd.Timestamp,
        m1_slice: np.ndarray,
        instrument: InstrumentConfig,
        cfg: ScalperConfig,
    ) -> Optional[tuple[str, float]]:
        """Tick-level BE/trailing + SL/TP check inside one M5 bar.

        ``m1_slice`` is an (M, 5) ndarray of [open, high, low, close, spread]
        for the M1 bars covered by this M5 bar. Each M1 bar yields 4 synthetic
        ticks in direction-aware order:

            LONG : open → low → high → close   (SL below, TP above — test SL first)
            SHORT: open → high → low → close   (SL above, TP below — test SL first)

        On each tick we (a) update BE/trailing using the tick price, then
        (b) test whether the CURRENT (possibly just-moved) SL or TP was hit.
        This is the critical difference from bar-level mode: the SL update and
        the SL trigger now happen on separate ticks, so a BE move can't fire
        and fill on the same bar's opposite extreme.

        Returns (reason, exit_price) on the first exit tick, else None.
        """
        point = instrument.point
        digits = instrument.digits
        is_long = pos.direction is Direction.LONG
        # Synthetic tick order per M1 bar (direction-aware).
        # Each tick is (price, is_high_extreme, is_low_extreme).
        ticks: list[tuple[float, bool, bool]] = []
        for row in m1_slice:
            o, h, l, c, _sp = row
            if is_long:
                ticks.append((o, False, False))
                ticks.append((l, False, True))
                ticks.append((h, True, False))
                ticks.append((c, False, False))
            else:
                ticks.append((o, False, False))
                ticks.append((h, True, False))
                ticks.append((l, False, True))
                ticks.append((c, False, False))

        sl = pos.sl
        tp = pos.tp
        for price, is_high, is_low in ticks:
            # ── (a) Update BE / trailing on this tick ────────────────────
            if is_long:
                profit_pts = (price - pos.entry_price) / point
                if cfg.use_break_even and profit_pts >= cfg.break_even_points:
                    be = round(pos.entry_price + cfg.break_even_lock * point, digits)
                    if be > sl:
                        sl = be
                if cfg.use_trailing and profit_pts >= cfg.trail_start_points:
                    trail = round(price - cfg.trail_step_points * point, digits)
                    if trail > sl:
                        sl = trail
                # Commit the new SL to the position so the next tick sees it.
                pos.sl = sl
            else:
                profit_pts = (pos.entry_price - price) / point
                if cfg.use_break_even and profit_pts >= cfg.break_even_points:
                    be = round(pos.entry_price - cfg.break_even_lock * point, digits)
                    if sl == 0.0 or be < sl:
                        sl = be
                if cfg.use_trailing and profit_pts >= cfg.trail_start_points:
                    trail = round(price + cfg.trail_step_points * point, digits)
                    if sl == 0.0 or trail < sl:
                        sl = trail
                pos.sl = sl

            # ── (b) Test SL / TP on this tick (pessimistic: SL first) ─────
            if sl > 0:
                if is_long and price <= sl:
                    return ("stop_loss", sl)
                if not is_long and price >= sl:
                    return ("stop_loss", sl)
            if tp > 0:
                if is_long and price >= tp:
                    return ("take_profit", tp)
                if not is_long and price <= tp:
                    return ("take_profit", tp)
        return None

    def _entry_allowed(
        self,
        cfg: ScalperConfig,
        t: pd.Timestamp,
        trades_today: int,
        last_trade_ts: Optional[pd.Timestamp],
        i: int,
        signals_long: np.ndarray,
        signals_short: np.ndarray,
    ) -> bool:
        """Gate stack mirroring EvaluateEntry's early returns (EA lines 254-263)."""
        if not (signals_long[i] or signals_short[i]):
            return False
        if cfg.use_session and not _in_session(t, cfg):
            return False
        if cfg.max_trades_per_day > 0 and trades_today >= cfg.max_trades_per_day:
            return False
        if last_trade_ts is not None and (t - last_trade_ts).total_seconds() < cfg.min_seconds_between:
            return False
        return True

    def _check_exit(
        self,
        pos: Position,
        o: float, h: float, l: float, c: float,
        instrument: InstrumentConfig,
    ) -> Optional[tuple[str, float]]:
        """Pessimistic 4-sub-tick SL/TP check (doc 03 §2).

        For a LONG (stop below, target above): OPEN → LOW → HIGH → CLOSE.
        For a SHORT (stop above, target below): OPEN → HIGH → LOW → CLOSE.
        Returns (reason, exit_price) or None if neither hit. Uses the position's
        CURRENT sl/tp (which BE/trailing may have moved this same bar).
        """
        if pos.direction is Direction.LONG:
            order = (("open", o), ("low", l), ("high", h), ("close", c))
        else:
            order = (("open", o), ("high", h), ("low", l), ("close", c))
        sl = pos.sl
        tp = pos.tp
        for label, price in order:
            if sl > 0 and (
                (pos.direction is Direction.LONG and price <= sl)
                or (pos.direction is Direction.SHORT and price >= sl)
            ):
                return ("stop_loss", sl)
            if tp > 0 and (
                (pos.direction is Direction.LONG and price >= tp)
                or (pos.direction is Direction.SHORT and price <= tp)
            ):
                return ("take_profit", tp)
        return None

    def _manage_position(
        self,
        pos: Position,
        t: pd.Timestamp,
        o: float, h: float, l: float, c: float,
        instrument: InstrumentConfig,
        cfg: ScalperConfig,
        balance: float,
        equity: float,
    ) -> None:
        """Break-even + trailing stop update (mirrors ManageOpenPositions).

        Uses the bar's high/low to approximate tick-level trailing (doc 03 §7).
        Mutates ``pos.sl`` in place; the subsequent _check_exit reads it.
        """
        point = instrument.point
        digits = instrument.digits
        new_sl = pos.sl

        if pos.direction is Direction.LONG:
            bid = h        # best case for trailing long = bar high
            profit_pts = (h - pos.entry_price) / point
            if cfg.use_break_even and profit_pts >= cfg.break_even_points:
                be = round(pos.entry_price + cfg.break_even_lock * point, digits)
                if be > new_sl:
                    new_sl = be
            if cfg.use_trailing and profit_pts >= cfg.trail_start_points:
                trail = round(bid - cfg.trail_step_points * point, digits)
                if trail > new_sl:
                    new_sl = trail
            if new_sl > pos.sl and new_sl < h:
                pos.sl = new_sl
        else:
            ask = l        # best case for trailing short = bar low
            profit_pts = (pos.entry_price - l) / point
            if cfg.use_break_even and profit_pts >= cfg.break_even_points:
                be = round(pos.entry_price - cfg.break_even_lock * point, digits)
                if pos.sl == 0.0 or be < new_sl:
                    new_sl = be
            if cfg.use_trailing and profit_pts >= cfg.trail_start_points:
                trail = round(ask + cfg.trail_step_points * point, digits)
                if pos.sl == 0.0 or trail < new_sl:
                    new_sl = trail
            if new_sl != pos.sl and (pos.sl == 0.0 or new_sl < pos.sl) and new_sl > l:
                pos.sl = new_sl

    def _calc_lots(
        self,
        cfg: ScalperConfig,
        instrument: InstrumentConfig,
        sl_distance: float,
        equity: float,
    ) -> float:
        """Mirror CalcLots: risk-percent sizing (mode 1) or fixed lot (mode 0).

        lots = riskMoney / (slDistance / tickSize × tickValue)
        Falls back to fixed lots if sizing mode is 0 or SL is zero.
        """
        if cfg.sizing_mode == 0 or sl_distance <= 0:
            return instrument.round_volume(cfg.fixed_lots)
        risk_money = equity * cfg.risk_percent / 100.0
        loss_per_lot = sl_distance / instrument.tick_size * instrument.tick_value
        if loss_per_lot <= 0:
            return instrument.round_volume(cfg.fixed_lots)
        lots = risk_money / loss_per_lot
        return instrument.round_volume(lots)

    def _unrealized_pnl(self, pos: Position, price: float, instrument: InstrumentConfig) -> float:
        """Mark-to-market PnL of an open position at ``price``."""
        direction_sign = 1.0 if pos.direction is Direction.LONG else -1.0
        price_diff = (price - pos.entry_price) * direction_sign
        ticks = price_diff / instrument.tick_size
        return ticks * instrument.tick_value * pos.lots

    def _close_trade(
        self,
        pos: Position,
        exit_info: tuple[str, float],
        exit_time: pd.Timestamp,
        instrument: InstrumentConfig,
        balance: float,
    ) -> Trade:
        """Build a closed Trade from a position + exit (reason, price)."""
        reason, exit_price = exit_info
        direction_sign = 1.0 if pos.direction is Direction.LONG else -1.0
        price_diff = (exit_price - pos.entry_price) * direction_sign
        ticks = price_diff / instrument.tick_size
        gross = ticks * instrument.tick_value * pos.lots
        # Swap: approximate with the daily rate × holding days.
        holding_days = max((exit_time - pos.entry_time).days, 0)
        swap_rate = instrument.swap_long if pos.direction is Direction.LONG else instrument.swap_short
        # Triple swap on the configured weekday (default Wed=3).
        swap = 0.0
        if holding_days > 0:
            swap = swap_rate * pos.lots * holding_days
            # Add triple-swap days crossed.
            for d in range(holding_days):
                day = (pos.entry_time + pd.Timedelta(days=d + 1))
                if day.weekday() == instrument.triple_swap_weekday:
                    swap += swap_rate * pos.lots * 2   # +2 extra (×3 total)
        return Trade(
            direction=pos.direction,
            entry_time=pos.entry_time,
            exit_time=exit_time,
            entry_price=pos.entry_price,
            exit_price=exit_price,
            lots=pos.lots,
            pnl=gross + swap,
            swap=swap,
            exit_reason=reason,
        )


def _in_session(t: pd.Timestamp, cfg: ScalperConfig) -> bool:
    """Mirror InSession(): wrap-aware hour window check."""
    hour = t.hour
    if cfg.session_start_hour == cfg.session_end_hour:
        return True
    if cfg.session_start_hour < cfg.session_end_hour:
        return cfg.session_start_hour <= hour < cfg.session_end_hour
    return hour >= cfg.session_start_hour or hour < cfg.session_end_hour


def _build_m5_to_m1_index(
    m5_ts: "pd.Series", m1_bars: pd.DataFrame
) -> list[np.ndarray]:
    """Map each M5 bar timestamp → (M, 5) ndarray of its M1 sub-bars.

    Uses ``searchsorted`` on the M1 timestamp column for O(N+M) alignment.
    Each entry is the [open, high, low, close, spread] rows of the M1 bars
    whose timestamp falls in [m5_ts, m5_ts + 5min). M5 bars with no M1
    coverage get an empty (0, 5) array — the simulator skips them safely.
    """
    m1_ts = pd.to_datetime(m1_bars["timestamp"].to_numpy())
    m1_ohlc = m1_bars[["open", "high", "low", "close", "spread"]].to_numpy(dtype=float)
    # For each M5 bar, find the M1 index range [lo, hi) with ts in [t, t+5min).
    m5_arr = np.asarray(m5_ts)
    lo = np.searchsorted(m1_ts.values, m5_arr, side="left")
    hi = np.searchsorted(m1_ts.values, m5_arr + pd.Timedelta(minutes=5), side="left")
    slices: list[np.ndarray] = []
    for a, b in zip(lo, hi):
        slices.append(m1_ohlc[a:b] if b > a else np.empty((0, 5), dtype=float))
    return slices


def config_from_params(params: dict) -> ScalperConfig:
    """Build a ScalperConfig from the merged params dict (frozen + sampled).

    Used as the ``build_engine_kwargs`` hook on ObjectiveConfig so the
    optimizer can pipe the tunable BE / trailing point values into the engine
    without the optimizer knowing about ScalperConfig.
    """
    return ScalperConfig(
        use_break_even=params["InpUseBreakEven"],
        use_trailing=params["InpUseTrailing"],
        use_session=params["InpUseSession"],
        session_start_hour=int(params["InpSessionStartHour"]),
        session_end_hour=int(params["InpSessionEndHour"]),
        max_positions=int(params["InpMaxPositions"]),
        max_trades_per_day=int(params["InpMaxTradesPerDay"]),
        daily_loss_limit_pct=float(params["InpDailyLossLimit"]),
        daily_profit_target_pct=float(params["InpDailyProfitTarget"]),
        min_seconds_between=int(params["InpMinSecondsBetween"]),
        sizing_mode=int(params["InpSizingMode"]),
        fixed_lots=float(params["InpFixedLots"]),
        risk_percent=float(params["InpRiskPercent"]),
        break_even_points=float(params["InpBreakEvenPoints"]),
        break_even_lock=float(params["InpBreakEvenLock"]),
        trail_start_points=float(params["InpTrailStartPoints"]),
        trail_step_points=float(params["InpTrailStepPoints"]),
    )


def engine_kwargs_from_params(params: dict) -> dict:
    """ObjectiveConfig.build_engine_kwargs hook: returns {"scalper_cfg": ...}."""
    return {"scalper_cfg": config_from_params(params)}