gold-trading-monitor

#!/usr/bin/env python3
"""
gold-trading-monitor
====================

A complete, self-contained trading and monitoring script for XAUUSD (Gold) with a
recursive post-trade self-evaluation engine.

Strategy stack
--------------
1. EMA Alignment      : 10 > 50 > 100 > 200 (long) or 10 < 50 < 100 < 200 (short),
                        with price leading the stack.
2. Bollinger rejection: price pierces a band and closes back inside (20, 2).
3. RSI Divergence     : bullish divergence in oversold (<30) for longs,
                        bearish divergence in overbought (>70) for shorts (RSI 14).
4. 3-Candle Reversal  : the "3CR" exhaustion/entry trigger.
5. Key zones          : confluence at 4250 (S), 4350 (R), 4400 (R) and channel edges.

Self-evaluation engine
-----------------------
Every closed trade is journalled with its Maximum Adverse Excursion (MAE) and
Maximum Favorable Excursion (MFE). A recursive feedback loop reads that history and
auto-tunes the trigger thresholds - the 3CR exhaustion buffer and the Bollinger
rejection width - to minimise drawdown and avoid high-frequency stop-hunts. The
tuned parameters persist to disk and feed straight back into live signalling, so
the longer it runs the more it adapts to how XAUUSD has actually paid out.

Data source
------------
By default the script runs on a deterministic mock XAUUSD stream so the logic can be
tested and demonstrated anywhere. With --live (or --execute) it pulls candles from a
native MetaTrader5 terminal. Nothing here needs a paid API.

    pip install pandas numpy
    pip install MetaTrader5      # native MT5 feed + execution (Windows)

Native IC Markets MT5 execution
-------------------------------
Execution uses the official MetaTrader5 Python library directly - no bridge, no EA.
With --execute the script initialises and logs into an IC Markets MT5 terminal, then
maps qualifying strategy signals straight to mt5.order_send. The flow:

    Python strategy  -->  MetaTrader5 (mt5.order_send)  -->  IC Markets MT5 terminal

Only signals that clear EMA alignment AND the 3CR trigger are routed (Bollinger and RSI
divergence add confluence/sizing weight). For each routed signal the executor:
  * sizes volume from a risk-per-trade percentage and the live account balance,
  * derives Stop Loss / Take Profit from the adaptive stop and target multiple,
  * snaps SL/TP/volume to the symbol's tick size, stop level and lot step,
  * sends a market order with a deviation (slippage) cap and a magic number,
  * retries once on a requote, and journals the trade so the self-evaluation engine
    keeps adapting the thresholds.
A lightweight heartbeat (terminal_info / account_info) confirms the terminal is alive
before any order is sent. Credentials are read from the environment, never hard-coded:

    set ICM_LOGIN=12345678            # IC Markets MT5 account number
    set ICM_PASSWORD=your-password    # MT5 trading password (not the investor one)
    set ICM_SERVER=ICMarketsSC-Demo   # exact server name shown in the terminal
    set ICM_MT5_PATH=C:\\Program Files\\MetaTrader 5 IC Markets\\terminal64.exe

Run it
------
    python gold_trading_monitor.py                 # mock feed, single pass
    python gold_trading_monitor.py --live          # native MT5 price feed
    python gold_trading_monitor.py --loop 15       # poll every 15s
    python gold_trading_monitor.py --backtest      # walk-forward + self-tune demo
    python gold_trading_monitor.py --execute --risk 0.5 --loop 60   # live IC Markets MT5

This file is plain Python and the standard library + pandas/numpy. The MetaTrader5
package is only needed for --live / --execute; without it the monitor still signals and
self-evaluates on the mock stream.
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Optional

import numpy as np
import pandas as pd

# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #

SYMBOL = "XAUUSD"

EMA_PERIODS = (10, 50, 100, 200)
BB_PERIOD = 20
BB_STDDEV = 2.0
RSI_PERIOD = 14

RSI_OVERSOLD = 30.0
RSI_OVERBOUGHT = 70.0

# Confluence zones in USD. tol = how close (in $) price must be to count as "at" a zone.
KEY_ZONES = (
    {"name": "Support 4250", "price": 4250.0, "kind": "support", "tol": 6.0},
    {"name": "Resistance 4350", "price": 4350.0, "kind": "resistance", "tol": 6.0},
    {"name": "Resistance 4400", "price": 4400.0, "kind": "resistance", "tol": 6.0},
)

# How many EMA-200 bars of history we need before signals are trustworthy.
MIN_BARS = 220

# Divergence look-back: compare the two most recent swing lows / highs within this window.
DIVERGENCE_LOOKBACK = 40
SWING_STRENGTH = 3  # bars on each side that define a local pivot

# Trade management defaults used to journal outcomes (in USD per ounce).
DEFAULT_STOP = 12.0          # base protective stop distance
DEFAULT_TARGET = 24.0        # base take-profit distance (2R)
MAX_HOLD_BARS = 32           # close the trade if neither stop nor target hit

# Where the journal and tuned parameters live.
JOURNAL_PATH = os.environ.get("GTM_JOURNAL", "trade_journal.json")
PARAMS_PATH = os.environ.get("GTM_PARAMS", "adaptive_params.json")

# --- IC Markets native MT5 execution -----------------------------------------
# Credentials/paths read from the environment so they never live in source control.
ICM_LOGIN = os.environ.get("ICM_LOGIN", "")          # MT5 account number
ICM_PASSWORD = os.environ.get("ICM_PASSWORD", "")    # MT5 trading password
ICM_SERVER = os.environ.get("ICM_SERVER", "")        # e.g. ICMarketsSC-Demo
ICM_MT5_PATH = os.environ.get("ICM_MT5_PATH", "")    # path to terminal64.exe (optional)

# Risk + order parameters mapped onto mt5.order_send.
RISK_PER_TRADE_PCT = 0.5         # % of account equity risked per trade (default)
FALLBACK_EQUITY = 10_000.0       # used only if account_info() is unavailable
XAUUSD_CONTRACT_SIZE = 100.0     # ounces per 1.00 lot (overridden by symbol_info)
SLIPPAGE_POINTS = 20             # max price deviation passed to order_send
MAGIC_NUMBER = 4250_4400         # tags our orders so we only manage our own
DEFAULT_MIN_LOT, DEFAULT_MAX_LOT, DEFAULT_LOT_STEP = 0.01, 50.0, 0.01
TARGET_R_MULTIPLE = 2.0          # TP distance as a multiple of the stop distance
HEARTBEAT_EVERY_S = 20           # max seconds between terminal liveness checks
ORDER_RETRY_ON_REQUOTE = 1       # extra attempts when MT5 returns a requote

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-7s  %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("gold-monitor")


# --------------------------------------------------------------------------- #
# Adaptive parameters - the knobs the self-evaluation engine tunes
# --------------------------------------------------------------------------- #

@dataclass
class AdaptiveParams:
    """
    Tunable trigger thresholds. The self-evaluation engine adjusts these inside
    clamped bounds; the strategy reads them on every bar.

    threecr_buffer_pct  : how far past candle-1's high/low the 3CR close must break,
                          as a fraction of price. Higher = demands more exhaustion,
                          which filters stop-hunt fakeouts at the cost of frequency.
    bb_rejection_width  : multiple of the band's standard-deviation that the wick
                          must pierce before a close-back-inside counts as a real
                          rejection. Higher = only deep rejections qualify.
    min_confluence      : minimum confirmed conditions before a signal is actionable.
    cooldown_bars       : minimum bars between signals (anti-HFT throttle).
    stop_mult / tp_mult : scale the base stop / target distances.
    """
    threecr_buffer_pct: float = 0.0006   # 0.06% of price (~ $2.6 at 4300)
    bb_rejection_width: float = 0.0       # extra sigma beyond the band
    min_confluence: int = 3
    cooldown_bars: int = 4
    stop_mult: float = 1.0
    tp_mult: float = 1.0

    # Hard bounds so the recursive loop can never tune itself into nonsense.
    BOUNDS = {
        "threecr_buffer_pct": (0.0002, 0.0025),
        "bb_rejection_width": (0.0, 0.8),
        "min_confluence": (3, 5),
        "cooldown_bars": (2, 16),
        "stop_mult": (0.6, 2.0),
        "tp_mult": (0.8, 3.0),
    }

    def clamp(self) -> "AdaptiveParams":
        for fld, (lo, hi) in self.BOUNDS.items():
            val = getattr(self, fld)
            setattr(self, fld, type(val)(min(max(val, lo), hi)))
        return self

    def save(self, path: str = PARAMS_PATH) -> None:
        with open(path, "w") as fh:
            json.dump(asdict(self), fh, indent=2)

    @classmethod
    def load(cls, path: str = PARAMS_PATH) -> "AdaptiveParams":
        try:
            with open(path) as fh:
                data = json.load(fh)
            data.pop("BOUNDS", None)
            return cls(**data).clamp()
        except (FileNotFoundError, json.JSONDecodeError, TypeError):
            return cls()


# --------------------------------------------------------------------------- #
# Indicators (pandas / numpy only)
# --------------------------------------------------------------------------- #

def ema(series: pd.Series, period: int) -> pd.Series:
    """Exponential moving average."""
    return series.ewm(span=period, adjust=False).mean()


def bollinger_bands(
    close: pd.Series, period: int = BB_PERIOD, stddev: float = BB_STDDEV
) -> tuple[pd.Series, pd.Series, pd.Series]:
    """Return (middle, upper, lower) Bollinger Bands plus the rolling sigma."""
    middle = close.rolling(window=period).mean()
    sigma = close.rolling(window=period).std(ddof=0)
    upper = middle + stddev * sigma
    lower = middle - stddev * sigma
    return middle, upper, lower, sigma


def rsi(close: pd.Series, period: int = RSI_PERIOD) -> pd.Series:
    """Wilder's RSI."""
    delta = close.diff()
    gain = delta.clip(lower=0.0)
    loss = -delta.clip(upper=0.0)
    # Wilder smoothing == EMA with alpha = 1/period
    avg_gain = gain.ewm(alpha=1.0 / period, adjust=False).mean()
    avg_loss = loss.ewm(alpha=1.0 / period, adjust=False).mean()
    rs = avg_gain / avg_loss.replace(0.0, np.nan)
    out = 100.0 - (100.0 / (1.0 + rs))
    return out.fillna(50.0)


def add_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """Attach all indicator columns to an OHLC frame."""
    df = df.copy()
    for p in EMA_PERIODS:
        df[f"ema{p}"] = ema(df["close"], p)
    df["bb_mid"], df["bb_upper"], df["bb_lower"], df["bb_sigma"] = bollinger_bands(
        df["close"]
    )
    df["rsi"] = rsi(df["close"])
    return df


# --------------------------------------------------------------------------- #
# Signal building blocks (now parameterised by AdaptiveParams)
# --------------------------------------------------------------------------- #

def ema_alignment(row: pd.Series) -> Optional[str]:
    """
    Return 'long', 'short', or None based on full 10/50/100/200 stacking.
    Long  : close > ema10 > ema50 > ema100 > ema200
    Short : close < ema10 < ema50 < ema100 < ema200
    """
    c = row["close"]
    e10, e50, e100, e200 = (row[f"ema{p}"] for p in EMA_PERIODS)
    if c > e10 > e50 > e100 > e200:
        return "long"
    if c < e10 < e50 < e100 < e200:
        return "short"
    return None


def bollinger_rejection(
    df: pd.DataFrame, i: int, params: AdaptiveParams
) -> Optional[str]:
    """
    Detect a band rejection on bar i: the candle pierces a band by at least
    `bb_rejection_width` * sigma with its wick, but closes back inside the band.
    Returns 'long' (lower-band rejection), 'short' (upper-band rejection) or None.

    The adaptive width is what the engine widens after stop-hunt losses: shallow
    pokes stop counting as rejections, so we only act on convincing ones.
    """
    if i < 1:
        return None
    row = df.iloc[i]
    low, high, close = row["low"], row["high"], row["close"]
    lower, upper, sigma = row["bb_lower"], row["bb_upper"], row["bb_sigma"]
    if np.isnan(lower) or np.isnan(upper) or np.isnan(sigma):
        return None
    margin = params.bb_rejection_width * sigma
    # Lower-band rejection -> bullish (wick must pierce by `margin`)
    if low < lower - margin and close > lower:
        return "long"
    # Upper-band rejection -> bearish
    if high > upper + margin and close < upper:
        return "short"
    return None


def _pivots(values: np.ndarray, strength: int, kind: str) -> list[int]:
    """Indices of local minima ('low') or maxima ('high') with `strength` bars each side."""
    idx: list[int] = []
    n = len(values)
    for i in range(strength, n - strength):
        window = values[i - strength : i + strength + 1]
        center = values[i]
        if kind == "low" and center == window.min() and (window == center).sum() == 1:
            idx.append(i)
        elif kind == "high" and center == window.max() and (window == center).sum() == 1:
            idx.append(i)
    return idx


def rsi_divergence(df: pd.DataFrame, direction: str) -> Optional[str]:
    """
    Classic regular divergence on the most recent two qualifying swings.

    Bullish (for longs): price makes a lower low while RSI makes a higher low,
                         with the RSI trough in oversold territory (< 30).
    Bearish (for shorts): price makes a higher high while RSI makes a lower high,
                         with the RSI peak in overbought territory (> 70).

    Returns 'bullish', 'bearish' or None.
    """
    window = df.iloc[-DIVERGENCE_LOOKBACK:]
    if len(window) < SWING_STRENGTH * 2 + 2:
        return None

    if direction == "long":
        lows = window["low"].to_numpy()
        rsis = window["rsi"].to_numpy()
        piv = _pivots(lows, SWING_STRENGTH, "low")
        if len(piv) < 2:
            return None
        a, b = piv[-2], piv[-1]  # older, newer
        price_lower_low = lows[b] < lows[a]
        rsi_higher_low = rsis[b] > rsis[a]
        oversold = rsis[b] < RSI_OVERSOLD or rsis[a] < RSI_OVERSOLD
        if price_lower_low and rsi_higher_low and oversold:
            return "bullish"
        return None

    if direction == "short":
        highs = window["high"].to_numpy()
        rsis = window["rsi"].to_numpy()
        piv = _pivots(highs, SWING_STRENGTH, "high")
        if len(piv) < 2:
            return None
        a, b = piv[-2], piv[-1]
        price_higher_high = highs[b] > highs[a]
        rsi_lower_high = rsis[b] < rsis[a]
        overbought = rsis[b] > RSI_OVERBOUGHT or rsis[a] > RSI_OVERBOUGHT
        if price_higher_high and rsi_lower_high and overbought:
            return "bearish"
        return None

    return None


def three_candle_reversal(
    df: pd.DataFrame, i: int, direction: str, params: AdaptiveParams
) -> bool:
    """
    The 3-Candle Reversal (3CR) entry trigger, evaluated at bar i.

    Long  : after a down-leg, three candles where the 3rd closes above the
            HIGH of the 1st candle by `threecr_buffer_pct`, momentum turning up.
    Short : after an up-leg, three candles where the 3rd closes below the
            LOW of the 1st candle by `threecr_buffer_pct`, momentum turning down.

    `threecr_buffer_pct` is the exhaustion buffer the engine tunes: a bigger buffer
    demands a more decisive reversal close and rejects marginal stop-hunt wicks.
    """
    if i < 5:
        return False
    c1, c2, c3 = df.iloc[i - 2], df.iloc[i - 1], df.iloc[i]
    buf = params.threecr_buffer_pct * c3["close"]
    # prior leg = close of the candle before the pattern vs ~4 bars earlier
    prior_then = df.iloc[i - 5]["close"]
    prior_now = df.iloc[i - 3]["close"]

    if direction == "long":
        downtrend = prior_now < prior_then
        momentum_up = c3["close"] > c2["close"] and c2["close"] >= c1["close"] * 0.999
        breaks_first_high = c3["close"] > c1["high"] + buf
        return bool(downtrend and breaks_first_high and momentum_up)

    if direction == "short":
        uptrend = prior_now > prior_then
        momentum_dn = c3["close"] < c2["close"] and c2["close"] <= c1["close"] * 1.001
        breaks_first_low = c3["close"] < c1["low"] - buf
        return bool(uptrend and breaks_first_low and momentum_dn)

    return False


def zone_confluence(price: float, direction: str):
    """Return the matching key zone dict if price is within tolerance, else None."""
    for z in KEY_ZONES:
        if abs(price - z["price"]) <= z["tol"]:
            if direction == "long" and z["kind"] == "support":
                return z
            if direction == "short" and z["kind"] == "resistance":
                return z
            return z  # at a zone but counter to its bias -> still note it
    return None


# --------------------------------------------------------------------------- #
# Signal aggregation
# --------------------------------------------------------------------------- #

@dataclass
class Signal:
    direction: str                       # 'long' | 'short'
    price: float
    time: datetime
    confluence: int                      # count of confirmed conditions
    reasons: list[str] = field(default_factory=list)
    zone: Optional[dict] = None
    rsi: float = float("nan")
    bar_index: int = -1
    min_confluence: int = 3
    ema_aligned: bool = False     # full 10/50/100/200 stack confirmed
    has_3cr: bool = False         # 3-Candle Reversal trigger confirmed

    @property
    def is_actionable(self) -> bool:
        return self.confluence >= self.min_confluence

    @property
    def routable(self) -> bool:
        """Only EMA-aligned + 3CR-confirmed signals are routed to MT5 execution."""
        return self.is_actionable and self.ema_aligned and self.has_3cr

    def render(self) -> str:
        side = "LONG" if self.direction == "long" else "SHORT"
        where = f" at ${self.zone['price']:,.0f} ({self.zone['name']})" if self.zone else ""
        head = f"{side} SIGNAL{where}"
        body = " + ".join(self.reasons)
        return (
            f"{head}\n"
            f"    price   : ${self.price:,.2f}\n"
            f"    rsi(14) : {self.rsi:5.1f}\n"
            f"    confl.  : {self.confluence} conditions\n"
            f"    reasons : {body}"
        )


def evaluate(df: pd.DataFrame, params: AdaptiveParams) -> Optional[Signal]:
    """Run the full strategy on the latest closed bar using the tuned params."""
    if len(df) < MIN_BARS:
        log.warning("Only %d bars; need >= %d for reliable EMA200.", len(df), MIN_BARS)
    enriched = add_indicators(df)
    i = len(enriched) - 1
    row = enriched.iloc[i]

    direction = ema_alignment(row)
    if direction is None:
        return None  # no clean trend stack -> stand aside

    reasons = [f"EMA stack aligned ({direction})"]
    confluence = 1

    bb = bollinger_rejection(enriched, i, params)
    if bb == direction:
        reasons.append("Bollinger band rejection")
        confluence += 1

    div = rsi_divergence(enriched, direction)
    if (direction == "long" and div == "bullish") or (
        direction == "short" and div == "bearish"
    ):
        reasons.append(f"RSI {div} divergence")
        confluence += 1

    has_3cr = three_candle_reversal(enriched, i, direction, params)
    if has_3cr:
        reasons.append("3-Candle Reversal (3CR)")
        confluence += 1

    zone = zone_confluence(float(row["close"]), direction)
    if zone is not None:
        reasons.append(f"Key zone: {zone['name']}")
        confluence += 1

    return Signal(
        direction=direction,
        price=float(row["close"]),
        time=row.name if isinstance(row.name, datetime) else datetime.now(timezone.utc),
        confluence=confluence,
        reasons=reasons,
        zone=zone,
        rsi=float(row["rsi"]),
        bar_index=i,
        min_confluence=params.min_confluence,
        ema_aligned=True,          # evaluate() returns None unless the stack aligned
        has_3cr=has_3cr,
    )


# --------------------------------------------------------------------------- #
# Trade journal: outcome tracking with MAE / MFE
# --------------------------------------------------------------------------- #

@dataclass
class TradeRecord:
    """A closed trade, with the excursion stats the engine learns from."""
    time: str
    direction: str
    entry: float
    exit: float
    outcome: str                 # 'win' | 'loss' | 'timeout'
    pnl: float                   # in USD per ounce
    mae: float                   # Maximum Adverse Excursion (>= 0, worst drawdown)
    mfe: float                   # Maximum Favorable Excursion (>= 0, best run-up)
    bars_held: int
    confluence: int
    params: dict = field(default_factory=dict)   # snapshot of the params that fired it


class TradeJournal:
    """Append-only JSON journal of TradeRecords."""

    def __init__(self, path: str = JOURNAL_PATH):
        self.path = path
        self.records: list[TradeRecord] = self._load()

    def _load(self) -> list[TradeRecord]:
        try:
            with open(self.path) as fh:
                return [TradeRecord(**r) for r in json.load(fh)]
        except (FileNotFoundError, json.JSONDecodeError, TypeError):
            return []

    def add(self, record: TradeRecord) -> None:
        self.records.append(record)
        self._flush()

    def _flush(self) -> None:
        with open(self.path, "w") as fh:
            json.dump([asdict(r) for r in self.records], fh, indent=2)

    def recent(self, n: int) -> list[TradeRecord]:
        return self.records[-n:]


def simulate_trade_excursions(
    df: pd.DataFrame,
    entry_index: int,
    direction: str,
    entry_price: float,
    params: AdaptiveParams,
) -> Optional[TradeRecord]:
    """
    Walk a trade forward from `entry_index` and record its outcome, MAE and MFE.

    MAE = largest move against the position before it closed.
    MFE = largest move in favour of the position before it closed.
    A stop / target (scaled by params) or MAX_HOLD_BARS closes the trade.
    """
    enriched = df
    stop_dist = DEFAULT_STOP * params.stop_mult
    tp_dist = DEFAULT_TARGET * params.tp_mult
    sign = 1.0 if direction == "long" else -1.0

    mae = 0.0
    mfe = 0.0
    end = min(entry_index + MAX_HOLD_BARS, len(enriched) - 1)
    if end <= entry_index:
        return None

    outcome = "timeout"
    exit_price = float(enriched.iloc[end]["close"])
    bars_held = end - entry_index

    for k in range(entry_index + 1, end + 1):
        bar = enriched.iloc[k]
        # favourable / adverse extremes within the bar
        fav = sign * (bar["high"] - entry_price) if direction == "long" else sign * (
            entry_price - bar["low"]
        )
        adv = sign * (entry_price - bar["low"]) if direction == "long" else sign * (
            bar["high"] - entry_price
        )
        mfe = max(mfe, float(fav))
        mae = max(mae, float(adv))

        # stop hit first if both touched in the same bar (conservative)
        if mae >= stop_dist:
            outcome = "loss"
            exit_price = entry_price - sign * stop_dist
            bars_held = k - entry_index
            break
        if mfe >= tp_dist:
            outcome = "win"
            exit_price = entry_price + sign * tp_dist
            bars_held = k - entry_index
            break

    pnl = sign * (exit_price - entry_price)
    return TradeRecord(
        time=str(enriched.index[entry_index]),
        direction=direction,
        entry=round(entry_price, 2),
        exit=round(exit_price, 2),
        outcome=outcome,
        pnl=round(pnl, 2),
        mae=round(mae, 2),
        mfe=round(mfe, 2),
        bars_held=bars_held,
        confluence=0,
        params=asdict(params),
    )


# --------------------------------------------------------------------------- #
# Recursive self-evaluation engine
# --------------------------------------------------------------------------- #

class SelfEvaluationEngine:
    """
    Reads the trade journal and recursively tunes AdaptiveParams to:
      * minimise drawdown  -> widen stops / buffers when MAE dominates MFE,
      * avoid stop-hunts   -> raise the 3CR buffer and BB width after shallow-MAE
                              losses (price tagged the stop then reversed),
      * suppress HFT churn -> lengthen the cooldown and demand more confluence
                              when win-rate is poor or trades are too frequent.

    "Recursive" = each pass starts from the *current* tuned params and nudges them,
    so adjustments compound across evaluation cycles rather than resetting.
    """

    def __init__(self, journal: TradeJournal, window: int = 25, lr: float = 0.25):
        self.journal = journal
        self.window = window      # trades considered per evaluation
        self.lr = lr              # learning rate (how hard each pass nudges)

    def evaluate(self, params: AdaptiveParams) -> tuple[AdaptiveParams, dict]:
        trades = self.journal.recent(self.window)
        if len(trades) < 5:
            return params, {"status": "warming-up", "n": len(trades)}

        wins = [t for t in trades if t.outcome == "win"]
        losses = [t for t in trades if t.outcome == "loss"]
        n = len(trades)
        win_rate = len(wins) / n
        avg_mae = float(np.mean([t.mae for t in trades]))
        avg_mfe = float(np.mean([t.mfe for t in trades]))
        # Stop-hunt signature: a loss whose MFE shows price ran our way meaningfully
        # before the (shallow) adverse move stopped us out and reversed.
        stop_hunts = [t for t in losses if t.mfe >= 0.5 * DEFAULT_STOP]
        stop_hunt_rate = len(stop_hunts) / n
        # Efficiency: how much of the favourable excursion we actually captured.
        capture = float(np.mean([min(t.mfe, DEFAULT_TARGET) / DEFAULT_TARGET
                                 for t in trades]))

        new = AdaptiveParams(**{k: getattr(params, k)
                                for k in asdict(params)})  # copy current (recursive base)
        notes = []

        # 1) Stop-hunts -> demand more exhaustion (wider 3CR buffer + BB width)
        #    and give the stop a touch more room so noise can't tag it.
        if stop_hunt_rate > 0.2:
            new.threecr_buffer_pct *= 1 + self.lr * (stop_hunt_rate / 0.2)
            new.bb_rejection_width += self.lr * 0.25
            new.stop_mult *= 1 + self.lr * 0.5
            notes.append(f"stop-hunt rate {stop_hunt_rate:.0%} -> widen buffers/stop")

        # 2) Drawdown control -> if adverse excursion rivals favourable, tighten entries.
        if avg_mfe > 0 and avg_mae / avg_mfe > 0.8:
            new.threecr_buffer_pct *= 1 + self.lr * 0.5
            new.min_confluence = min(new.min_confluence + 1, 5)
            notes.append(f"MAE/MFE {avg_mae/avg_mfe:.2f} -> require more confluence")

        # 3) Poor win-rate or churn -> throttle frequency (anti-HFT).
        if win_rate < 0.45:
            new.cooldown_bars = int(round(new.cooldown_bars * (1 + self.lr)))
            notes.append(f"win-rate {win_rate:.0%} -> lengthen cooldown")

        # 4) Healthy & under-trading -> relax slightly so we don't over-filter.
        if win_rate > 0.6 and stop_hunt_rate < 0.1 and capture > 0.5:
            new.threecr_buffer_pct *= 1 - self.lr * 0.3
            new.bb_rejection_width *= 1 - self.lr * 0.3
            new.cooldown_bars = max(2, int(round(new.cooldown_bars * (1 - self.lr * 0.5))))
            notes.append("healthy edge -> relax filters to recover frequency")

        new.clamp()
        report = {
            "status": "tuned",
            "n": n,
            "win_rate": round(win_rate, 3),
            "avg_mae": round(avg_mae, 2),
            "avg_mfe": round(avg_mfe, 2),
            "stop_hunt_rate": round(stop_hunt_rate, 3),
            "capture": round(capture, 3),
            "notes": notes,
            "params_before": asdict(params),
            "params_after": asdict(new),
        }
        return new, report


# --------------------------------------------------------------------------- #
# Data sources
# --------------------------------------------------------------------------- #

def fetch_mt5(symbol: str = SYMBOL, bars: int = 500) -> Optional[pd.DataFrame]:
    """Pull recent M15 candles from MetaTrader5. Returns None if unavailable."""
    try:
        import MetaTrader5 as mt5  # type: ignore
    except Exception:
        return None

    if not mt5.initialize():
        log.warning("MT5 initialize() failed: %s", mt5.last_error())
        return None
    try:
        rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M15, 0, bars)
        if rates is None or len(rates) == 0:
            return None
        df = pd.DataFrame(rates)
        df["time"] = pd.to_datetime(df["time"], unit="s", utc=True)
        df = df.set_index("time")[["open", "high", "low", "close", "tick_volume"]]
        df = df.rename(columns={"tick_volume": "volume"})
        return df
    finally:
        mt5.shutdown()


def mock_stream(bars: int = 500, seed: int = 7) -> pd.DataFrame:
    """
    Deterministic synthetic XAUUSD candles around the 4,250-4,400 band so the
    key-zone, signal and self-evaluation logic can be demonstrated without a broker.
    """
    rng = np.random.default_rng(seed)
    start = datetime.now(timezone.utc) - timedelta(minutes=15 * bars)
    times = [start + timedelta(minutes=15 * k) for k in range(bars)]

    price = 4300.0
    drift = np.sin(np.linspace(0, 6 * np.pi, bars)) * 60.0  # roams the 4250-4400 band
    opens, highs, lows, closes = [], [], [], []
    for k in range(bars):
        target = 4300.0 + drift[k]
        price += (target - price) * 0.15 + rng.normal(0, 3.5)
        o = price + rng.normal(0, 1.5)
        c = price + rng.normal(0, 1.5)
        h = max(o, c) + abs(rng.normal(0, 2.5))
        lo = min(o, c) - abs(rng.normal(0, 2.5))
        opens.append(o); highs.append(h); lows.append(lo); closes.append(c)

    df = pd.DataFrame(
        {"open": opens, "high": highs, "low": lows, "close": closes,
         "volume": rng.integers(50, 500, bars)},
        index=pd.DatetimeIndex(times, name="time"),
    )
    return df


def get_data(use_live: bool) -> pd.DataFrame:
    if use_live:
        df = fetch_mt5()
        if df is not None and len(df):
            log.info("Using MetaTrader5 live feed (%d bars).", len(df))
            return df
        log.warning("MT5 unavailable - falling back to mock stream.")
    log.info("Using mock data stream.")
    return mock_stream()


# --------------------------------------------------------------------------- #
# Native IC Markets MT5 execution (official MetaTrader5 library, mt5.order_send)
# --------------------------------------------------------------------------- #

def compute_stop_target(
    signal: Signal, params: AdaptiveParams
) -> tuple[float, float]:
    """
    Translate a signal into absolute Stop Loss / Take Profit prices.

    The stop distance is the adaptive base stop (scaled by the self-evaluation
    engine's stop_mult); the target is TARGET_R_MULTIPLE times that distance, so the
    reward:risk that order_send executes tracks what the engine has been tuning.
    """
    stop_dist = DEFAULT_STOP * params.stop_mult
    tp_dist = stop_dist * TARGET_R_MULTIPLE * params.tp_mult
    if signal.direction == "long":
        sl = signal.price - stop_dist
        tp = signal.price + tp_dist
    else:
        sl = signal.price + stop_dist
        tp = signal.price - tp_dist
    return round(sl, 2), round(tp, 2)


def compute_lot_size(
    entry: float,
    stop: float,
    equity: float,
    risk_pct: float,
    contract_size: float = XAUUSD_CONTRACT_SIZE,
    min_lot: float = DEFAULT_MIN_LOT,
    max_lot: float = DEFAULT_MAX_LOT,
    lot_step: float = DEFAULT_LOT_STEP,
) -> float:
    """
    Risk-based position size in lots for XAUUSD.

    risk_amount  = equity * risk_pct%
    loss_per_lot = |entry - stop| * contract_size   (USD lost per 1.0 lot at the stop)
    lots         = risk_amount / loss_per_lot, snapped to the symbol's lot step + bounds.
    """
    stop_dist = abs(entry - stop)
    if stop_dist <= 0:
        return min_lot
    risk_amount = equity * (risk_pct / 100.0)
    loss_per_lot = stop_dist * contract_size
    raw = risk_amount / loss_per_lot
    snapped = round(raw / lot_step) * lot_step
    return float(min(max(snapped, min_lot), max_lot))


class MT5Executor:
    """
    Native execution against an IC Markets MetaTrader5 terminal.

    Uses the official MetaTrader5 library directly: mt5.initialize / mt5.login to
    connect, account_info / symbol_info for live equity and contract specs, and
    mt5.order_send for market entries with SL/TP, a deviation cap and a magic number.

    If the MetaTrader5 package or terminal is unavailable the executor stays
    `available = False` and the monitor keeps signalling without trading.
    """

    def __init__(self, risk_pct: float = RISK_PER_TRADE_PCT):
        self.risk_pct = risk_pct
        self.available = False
        self.equity = FALLBACK_EQUITY
        self._mt5 = None
        self._last_heartbeat = 0.0

    def connect(self) -> bool:
        try:
            import MetaTrader5 as mt5  # type: ignore
        except Exception:
            log.warning(
                "MetaTrader5 package not installed - execution disabled "
                "(pip install MetaTrader5). Monitoring and self-evaluation continue."
            )
            return False
        self._mt5 = mt5

        # Initialise the terminal (optionally at an explicit path) and log in.
        init_ok = (
            mt5.initialize(ICM_MT5_PATH) if ICM_MT5_PATH else mt5.initialize()
        )
        if not init_ok:
            log.warning("mt5.initialize() failed: %s", mt5.last_error())
            return False

        if ICM_LOGIN and ICM_PASSWORD and ICM_SERVER:
            if not mt5.login(int(ICM_LOGIN), password=ICM_PASSWORD, server=ICM_SERVER):
                log.warning("mt5.login() failed: %s", mt5.last_error())
                mt5.shutdown()
                return False
        else:
            log.info("No ICM_LOGIN/PASSWORD/SERVER set - using terminal's current login.")

        # Make sure our symbol is visible in Market Watch.
        if not mt5.symbol_select(SYMBOL, True):
            log.warning("Could not select %s in Market Watch.", SYMBOL)

        self.available = True
        log.info("MT5 connected to IC Markets terminal (symbol=%s).", SYMBOL)
        return self.heartbeat(force=True)

    def heartbeat(self, force: bool = False) -> bool:
        """
        Confirm the terminal is alive and connected, refresh cached equity. Throttled
        to HEARTBEAT_EVERY_S unless forced (always forced right before an order).
        """
        if not self.available or self._mt5 is None:
            return False
        now = time.time()
        if not force and now - self._last_heartbeat < HEARTBEAT_EVERY_S:
            return True
        mt5 = self._mt5
        term = mt5.terminal_info()
        if term is None or not getattr(term, "connected", False):
            log.warning("MT5 heartbeat: terminal not connected to broker.")
            self.available = bool(term is not None)
            return False
        acct = mt5.account_info()
        if acct is not None:
            self.equity = float(acct.equity)
        self._last_heartbeat = now
        log.info("MT5 heartbeat OK (equity=$%.2f).", self.equity)
        return True

    def _symbol_specs(self) -> dict:
        """Pull live contract size, lot step/bounds and tick size from MT5."""
        mt5 = self._mt5
        info = mt5.symbol_info(SYMBOL) if mt5 else None
        if info is None:
            return {
                "contract_size": XAUUSD_CONTRACT_SIZE,
                "min_lot": DEFAULT_MIN_LOT, "max_lot": DEFAULT_MAX_LOT,
                "lot_step": DEFAULT_LOT_STEP, "digits": 2, "point": 0.01,
                "stops_level": 0,
            }
        return {
            "contract_size": float(getattr(info, "trade_contract_size", XAUUSD_CONTRACT_SIZE)),
            "min_lot": float(getattr(info, "volume_min", DEFAULT_MIN_LOT)),
            "max_lot": float(getattr(info, "volume_max", DEFAULT_MAX_LOT)),
            "lot_step": float(getattr(info, "volume_step", DEFAULT_LOT_STEP)),
            "digits": int(getattr(info, "digits", 2)),
            "point": float(getattr(info, "point", 0.01)),
            "stops_level": int(getattr(info, "trade_stops_level", 0)),
        }

    def _enforce_stops_level(
        self, side: str, price: float, sl: float, tp: float, specs: dict
    ) -> tuple[float, float]:
        """Widen SL/TP if they sit inside the broker's minimum stop distance."""
        min_dist = specs["stops_level"] * specs["point"]
        if min_dist <= 0:
            return sl, tp
        if side == "BUY":
            sl = min(sl, price - min_dist)
            tp = max(tp, price + min_dist)
        else:
            sl = max(sl, price + min_dist)
            tp = min(tp, price - min_dist)
        d = specs["digits"]
        return round(sl, d), round(tp, d)

    def execute(self, signal: Signal, params: AdaptiveParams) -> bool:
        """
        Map a routable signal onto mt5.order_send.

        Routing rule: only EMA-aligned + 3CR-confirmed signals reach here (see
        Signal.routable). Bollinger / RSI divergence already raised the confluence
        that made the signal actionable. SL/TP come from the adaptive stop; volume
        from risk sizing against the live account balance.
        """
        if not signal.routable:
            return False
        if not self.available or self._mt5 is None:
            log.warning("MT5 unavailable - order NOT sent for %s signal.", signal.direction)
            return False
        if not self.heartbeat(force=True):
            log.error("Aborting order - heartbeat failed just before send.")
            return False

        mt5 = self._mt5
        specs = self._symbol_specs()
        tick = mt5.symbol_info_tick(SYMBOL)
        if tick is None:
            log.error("No tick for %s - cannot price the order.", SYMBOL)
            return False

        side = "BUY" if signal.direction == "long" else "SELL"
        price = tick.ask if side == "BUY" else tick.bid
        order_type = mt5.ORDER_TYPE_BUY if side == "BUY" else mt5.ORDER_TYPE_SELL

        sl, tp = compute_stop_target(signal, params)
        sl, tp = self._enforce_stops_level(side, price, sl, tp, specs)
        lots = compute_lot_size(
            price, sl, self.equity, self.risk_pct,
            contract_size=specs["contract_size"], min_lot=specs["min_lot"],
            max_lot=specs["max_lot"], lot_step=specs["lot_step"],
        )

        log.info(
            "Routing %s -> MT5: %.2f lots, entry ~%.2f, SL %.2f, TP %.2f (risk %.2f%%).",
            side, lots, price, sl, tp, self.risk_pct,
        )

        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": SYMBOL,
            "volume": lots,
            "type": order_type,
            "price": price,
            "sl": sl,
            "tp": tp,
            "deviation": SLIPPAGE_POINTS,
            "magic": MAGIC_NUMBER,
            "comment": "GTM 3CR+EMA",
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
        }

        for attempt in range(ORDER_RETRY_ON_REQUOTE + 1):
            result = mt5.order_send(request)
            if result is None:
                log.error("order_send returned None: %s", mt5.last_error())
                return False
            if result.retcode == mt5.TRADE_RETCODE_DONE:
                log.info(
                    "FILLED %s %.2f lots %s @ %.2f (ticket %s, SL %.2f / TP %.2f).",
                    side, lots, SYMBOL, result.price, result.order, sl, tp,
                )
                return True
            if result.retcode == mt5.TRADE_RETCODE_REQUOTE and attempt < ORDER_RETRY_ON_REQUOTE:
                tick = mt5.symbol_info_tick(SYMBOL)
                request["price"] = tick.ask if side == "BUY" else tick.bid
                log.warning("Requote - retrying at %.2f.", request["price"])
                continue
            log.error("order_send rejected: retcode=%s (%s).",
                      result.retcode, getattr(result, "comment", ""))
            return False
        return False

    def close(self) -> None:
        if self._mt5 is not None:
            try:
                self._mt5.shutdown()
            except Exception:
                pass


# --------------------------------------------------------------------------- #
# Monitoring / signalling
# --------------------------------------------------------------------------- #

def report(df: pd.DataFrame, params: AdaptiveParams) -> Optional[Signal]:
    enriched = add_indicators(df)
    last = enriched.iloc[-1]
    log.info(
        "XAUUSD  last=$%.2f  EMA10=%.1f EMA50=%.1f EMA100=%.1f EMA200=%.1f  RSI=%.1f",
        float(last["close"]), float(last["ema10"]), float(last["ema50"]),
        float(last["ema100"]), float(last["ema200"]), float(last["rsi"]),
    )
    signal = evaluate(df, params)
    if signal is None:
        log.info("No EMA alignment - standing aside.")
        return None
    if signal.is_actionable:
        for line in signal.render().splitlines():
            log.info(line)
    else:
        log.info(
            "Watch %s: %d/%d conditions (%s)",
            signal.direction.upper(), signal.confluence,
            params.min_confluence, ", ".join(signal.reasons),
        )
    return signal


# --------------------------------------------------------------------------- #
# Backtest + recursive self-tuning demonstration
# --------------------------------------------------------------------------- #

def backtest_and_tune(passes: int = 4) -> None:
    """
    Walk a fresh mock series, journal every actionable trade with MAE/MFE, then run
    the recursive self-evaluation engine. Repeat so the tuning compounds across passes.
    """
    journal = TradeJournal(path="backtest_journal.json")
    journal.records = []  # fresh demo run
    params = AdaptiveParams()

    for p in range(1, passes + 1):
        df = add_indicators(mock_stream(seed=10 + p))
        last_entry = -10_000
        n_trades = 0
        for i in range(MIN_BARS, len(df) - MAX_HOLD_BARS):
            if i - last_entry < params.cooldown_bars:
                continue
            window = df.iloc[: i + 1]
            sig = evaluate(window, params)
            if sig is None or not sig.is_actionable:
                continue
            rec = simulate_trade_excursions(df, i, sig.direction, sig.price, params)
            if rec is None:
                continue
            rec.confluence = sig.confluence
            journal.add(rec)
            last_entry = i
            n_trades += 1

        engine = SelfEvaluationEngine(journal)
        params, rep = engine.evaluate(params)
        params.save(path="backtest_params.json")

        if rep["status"] == "tuned":
            log.info(
                "Pass %d: %d trades | win=%.0f%% MAE=%.1f MFE=%.1f stop-hunts=%.0f%%",
                p, n_trades, rep["win_rate"] * 100, rep["avg_mae"],
                rep["avg_mfe"], rep["stop_hunt_rate"] * 100,
            )
            for note in rep["notes"]:
                log.info("   adapt: %s", note)
            log.info(
                "   params -> 3CR buffer=%.4f%% | BB width=%.2f sigma | "
                "min-confl=%d | cooldown=%d | stop x%.2f",
                params.threecr_buffer_pct * 100, params.bb_rejection_width,
                params.min_confluence, params.cooldown_bars, params.stop_mult,
            )
        else:
            log.info("Pass %d: %d trades (%s)", p, n_trades, rep["status"])

    # tidy demo artefacts
    for f in ("backtest_journal.json", "backtest_params.json"):
        try:
            os.remove(f)
        except OSError:
            pass


# --------------------------------------------------------------------------- #
# Main
# --------------------------------------------------------------------------- #

def main() -> int:
    parser = argparse.ArgumentParser(description="XAUUSD trading & monitoring")
    parser.add_argument("--live", action="store_true", help="use the native MT5 feed")
    parser.add_argument("--loop", type=int, default=0,
                        help="poll interval in seconds (0 = single pass)")
    parser.add_argument("--backtest", action="store_true",
                        help="run the walk-forward self-tuning demo and exit")
    parser.add_argument("--execute", action="store_true",
                        help="route EMA+3CR signals to IC Markets MT5 via order_send")
    parser.add_argument("--risk", type=float, default=RISK_PER_TRADE_PCT,
                        help="risk per trade as %% of equity (default %.2f)" % RISK_PER_TRADE_PCT)
    args = parser.parse_args()

    if args.backtest:
        log.info("Recursive self-evaluation backtest (XAUUSD)")
        backtest_and_tune()
        return 0

    params = AdaptiveParams.load()
    journal = TradeJournal()
    log.info(
        "gold-trading-monitor starting (symbol=%s, tuned params loaded: "
        "3CR buffer=%.4f%%, BB width=%.2f sigma, min-confl=%d, cooldown=%d)",
        SYMBOL, params.threecr_buffer_pct * 100, params.bb_rejection_width,
        params.min_confluence, params.cooldown_bars,
    )

    # --execute implies a live feed; connect the native MT5 executor.
    executor: Optional[MT5Executor] = None
    use_live = args.live or args.execute
    if args.execute:
        executor = MT5Executor(risk_pct=args.risk)
        if executor.connect():
            log.info("Native MT5 execution ARMED (risk %.2f%% per trade).", args.risk)
        else:
            log.warning("Execution requested but MT5 not ready - monitoring only.")

    last_signal_bar = -10_000
    try:
        while True:
            df = get_data(use_live)
            signal = report(df, params)

            # Journal + self-evaluate any actionable, non-throttled signal.
            if signal and signal.is_actionable:
                if signal.bar_index - last_signal_bar >= params.cooldown_bars:
                    enriched = add_indicators(df)
                    rec = simulate_trade_excursions(
                        enriched, signal.bar_index, signal.direction,
                        signal.price, params,
                    )
                    if rec is not None:
                        rec.confluence = signal.confluence
                        journal.add(rec)
                        last_signal_bar = signal.bar_index
                        log.info(
                            "Journalled %s trade: outcome=%s pnl=%.1f MAE=%.1f MFE=%.1f",
                            rec.direction, rec.outcome, rec.pnl, rec.mae, rec.mfe,
                        )
                        params, rep = SelfEvaluationEngine(journal).evaluate(params)
                        params.save()
                        if rep.get("status") == "tuned" and rep["notes"]:
                            log.info("Self-evaluation tuned params: %s",
                                     "; ".join(rep["notes"]))

                    # Route to MT5 only when EMA-aligned + 3CR confirmed.
                    if executor and executor.available:
                        if signal.routable:
                            executor.execute(signal, params)
                        else:
                            log.info(
                                "Signal not routed: needs EMA alignment + 3CR "
                                "(have EMA=%s, 3CR=%s).",
                                signal.ema_aligned, signal.has_3cr,
                            )
                else:
                    log.info("Cooldown active (anti-HFT) - signal not journalled.")

            if args.loop <= 0:
                break
            time.sleep(args.loop)
    except KeyboardInterrupt:
        log.info("Stopped by user.")
    finally:
        if executor:
            executor.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())