Trading Module¶

def calculate_account_projected_margin_ratio(
    client: Mt5TradingClient,
    *,
    symbol: str | None = None,
    new_position_side: OrderSide | None = None,
    new_position_volume: float = 0.0,
) -> float:
    """Return account-wide current plus optional new-position margin over equity.

    Current exposure comes from the broker account snapshot ``margin`` field so
    unrelated open positions remain in the baseline. Optional projected
    exposure is added via :func:`estimate_order_margin` only when a symbol, side,
    and positive volume are all supplied.

    """
    account = get_account_snapshot(client)
    equity = _required_account_number(account, "equity", allow_zero=False)
    margin = _required_account_number(account, "margin", allow_zero=True)
    if symbol is not None and new_position_side is not None and new_position_volume > 0:
        margin += estimate_order_margin(
            client,
            symbol,
            new_position_side,
            new_position_volume,
        )
    return margin / equity

def calculate_margin_and_volume(
    client: Mt5TradingClient,
    symbol: str,
    unit_margin_ratio: float,
    preserved_margin_ratio: float,
) -> MarginVolume:
    """Calculate tradable margin and volumes from account free margin.

    Applies ``preserved_margin_ratio`` to keep a reserve off ``margin_free``,
    then allocates ``unit_margin_ratio`` of the remainder as the margin budget
    for proportional volume sizing on both buy and sell sides. A
    ``unit_margin_ratio`` of ``0`` requests exactly one minimum valid unit per
    side when the post-reserve margin can afford it.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbol: Symbol used for minimum-lot margin and volume calculations.
        unit_margin_ratio: Fraction of post-reserve margin to allocate per unit.
        preserved_margin_ratio: Fraction of ``margin_free`` to preserve.

    Returns:
        Dictionary with ``margin_free``, ``available_margin``, ``trade_margin``,
        ``buy_volume``, and ``sell_volume``. Negative ``margin_free`` values are
        clamped to ``0.0`` before sizing.
    """
    _require_unit_ratio(unit_margin_ratio, "unit_margin_ratio")
    _require_unit_ratio(preserved_margin_ratio, "preserved_margin_ratio")

    account = client.account_info_as_dict()
    margin_free = max(0.0, float(account.get("margin_free") or 0.0))
    available_margin = margin_free * (1.0 - preserved_margin_ratio)
    trade_margin = available_margin * unit_margin_ratio
    if unit_margin_ratio == 0:
        buy_volume = _calculate_min_volume_if_affordable(
            client,
            symbol,
            available_margin,
            "BUY",
        )
        sell_volume = _calculate_min_volume_if_affordable(
            client,
            symbol,
            available_margin,
            "SELL",
        )
    else:
        buy_volume = calculate_volume_by_margin(client, symbol, trade_margin, "BUY")
        sell_volume = calculate_volume_by_margin(client, symbol, trade_margin, "SELL")
    try:
        symbol_info = get_symbol_snapshot(client, symbol)
        volume_min = float(symbol_info.get("volume_min") or 0.0)
        volume_max = float(symbol_info.get("volume_max") or 0.0)
        volume_step = float(symbol_info.get("volume_step") or 0.0)
    except AttributeError:
        volume_min = volume_max = volume_step = 0.0
    return {
        "margin_free": margin_free,
        "available_margin": available_margin,
        "trade_margin": trade_margin,
        "buy_volume": float(buy_volume),
        "sell_volume": float(sell_volume),
        "volume_min": volume_min,
        "volume_max": volume_max,
        "volume_step": volume_step,
    }

def calculate_new_position_margin_ratio(
    client: Mt5TradingClient,
    *,
    symbol: str,
    new_position_side: OrderSide | None = None,
    new_position_volume: float = 0.0,
) -> float:
    """Return total margin/equity ratio after an optional hypothetical position.

    Raises:
        Mt5TradingError: If equity or required tick data is invalid.
    """
    account = get_account_snapshot(client)
    equity = float(account.get("equity") or 0.0)
    if equity <= 0:
        msg = "Account equity must be positive to calculate margin ratio."
        raise Mt5TradingError(msg)
    margin = float(account.get("margin") or 0.0)
    if new_position_side is not None and new_position_volume > 0:
        side = _normalize_order_side(new_position_side)
        price = extract_tick_price(
            get_tick_snapshot(client, symbol), "ask" if side == "BUY" else "bid"
        )
        if price is None:
            msg = f"Tick price is unavailable for {symbol!r}."
            raise Mt5TradingError(msg)
        order_type = (
            client.mt5.ORDER_TYPE_BUY if side == "BUY" else client.mt5.ORDER_TYPE_SELL
        )
        margin += float(
            client.order_calc_margin(order_type, symbol, new_position_volume, price),
        )
    return margin / equity

def calculate_positions_margin(
    client: Mt5TradingClient,
    *,
    symbols: Sequence[str] | None = None,
) -> float:
    """Return the sum of estimated current margin for open positions.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbols: Optional symbol filter. When omitted, all open positions are
            included.

    Returns:
        Total estimated margin, or ``0.0`` when no matching positions exist.
    """
    frame = get_positions_frame(client)
    if frame.empty or "symbol" not in frame.columns:
        return 0.0
    if symbols is not None:
        frame = frame[frame["symbol"].isin(list(symbols))]
    if frame.empty:
        return 0.0
    grouped_volumes: dict[tuple[str, OrderSide], float] = {}
    for _, row in frame.iterrows():
        symbol = row.get("symbol")
        if not isinstance(symbol, str) or not symbol:
            continue
        volume = row.get("volume")
        if not _is_positive_finite_number(volume):
            continue
        order_side = _order_side_from_position_type(client, row.get("type"))
        if order_side is None:
            continue
        key = (symbol, order_side)
        finite_volume = float(cast("float | int", volume))
        grouped_volumes[key] = grouped_volumes.get(key, 0.0) + finite_volume
    total = 0.0
    for (symbol, order_side), volume in grouped_volumes.items():
        total += estimate_order_margin(client, symbol, order_side, volume)
    return total

def calculate_positions_margin_by_symbol(
    client: Mt5TradingClient,
    *,
    symbols: Sequence[str],
    suppress_errors: bool = True,
) -> dict[str, float]:
    """Return per-symbol estimated margin for open positions.

    Computes margin for each unique input symbol independently using the strict
    :func:`calculate_positions_margin` helper. Duplicates are deduplicated in
    first-seen order.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbols: Symbols to compute margin for.
        suppress_errors: When ``True``, log and skip symbols that raise
            ``Mt5TradingError``, ``Mt5RuntimeError``, or ``AttributeError``.
            When ``False``, re-raise the first failure.

    Returns:
        Mapping of symbol to margin total in first-seen unique-symbol order.
        Returns an empty dict when ``symbols`` is empty or all symbols fail
        with ``suppress_errors=True``.

    Raises:
        Mt5TradingError: When a symbol raises ``Mt5TradingError`` and
            ``suppress_errors=False``.
        Mt5RuntimeError: When a symbol raises ``Mt5RuntimeError`` and
            ``suppress_errors=False``.
        AttributeError: When a symbol raises ``AttributeError`` and
            ``suppress_errors=False``.
    """
    result: dict[str, float] = {}
    for symbol in dict.fromkeys(symbols):
        try:
            result[symbol] = calculate_positions_margin(client, symbols=[symbol])
        except (Mt5TradingError, Mt5RuntimeError, AttributeError) as exc:
            if not suppress_errors:
                raise
            _logger.warning("Skipping margin for %r: %s", symbol, exc)
    return result

def calculate_positions_margin_safe(
    client: Mt5TradingClient,
    *,
    symbols: Sequence[str],
) -> float:
    """Return the total estimated margin for open positions across symbols.

    Internally calls :func:`calculate_positions_margin_by_symbol` with
    ``suppress_errors=True``. Failed symbols are silently skipped.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbols: Symbols to include.

    Returns:
        Sum of per-symbol margins; ``0.0`` when no symbols or all fail.
    """
    return sum(
        calculate_positions_margin_by_symbol(client, symbols=symbols).values(),
        0.0,
    )

def calculate_projected_margin_ratio(
    client: Mt5TradingClient,
    *,
    symbol: str,
    new_position_side: OrderSide | None = None,
    new_position_volume: float = 0.0,
) -> float:
    """Return estimated current plus optional new-position margin over equity.

    Current exposure is estimated from open positions with
    :func:`calculate_positions_margin`. Optional projected exposure is added via
    :func:`estimate_order_margin`. Thresholds and guard actions are intentionally
    left to downstream applications.

    Account equity, position margin, and optional projected margin errors from
    the composed MT5 helpers propagate to the caller.
    """
    equity = _account_equity(client)
    margin = calculate_positions_margin(client, symbols=[symbol])
    if new_position_side is not None and new_position_volume > 0:
        margin += estimate_order_margin(
            client,
            symbol,
            new_position_side,
            new_position_volume,
        )
    return margin / equity

def calculate_spread_ratio(client: Mt5TradingClient, symbol: str) -> float:
    """Return ``(ask - bid) / ((ask + bid) / 2)`` for the latest tick.

    Raises:
        Mt5TradingError: If bid or ask is unavailable.
    """
    tick = get_tick_snapshot(client, symbol)
    bid = extract_tick_price(tick, "bid")
    ask = extract_tick_price(tick, "ask")
    if bid is None or ask is None:
        msg = f"Tick bid/ask is unavailable for {symbol!r}."
        raise Mt5TradingError(msg)
    return (ask - bid) / ((ask + bid) / 2.0)

def calculate_symbol_group_margin_ratio(
    client: Mt5TradingClient,
    *,
    symbols: Sequence[str],
    new_symbol: str | None = None,
    new_position_side: OrderSide | None = None,
    new_position_volume: float = 0.0,
    suppress_errors: bool = True,
) -> float:
    """Return estimated symbol-group margin over account equity.

    Per-symbol current exposure is summed with
    :func:`calculate_positions_margin_by_symbol`. When ``new_symbol`` is inside
    the input symbol group, optional projected order margin is added for that
    symbol. Invalid equity always raises to fail closed.

    Raises:
        AttributeError: When symbol margin lookup or projected margin lookup
            fails and ``suppress_errors`` is ``False``.
        Mt5RuntimeError: When symbol margin lookup or projected margin lookup
            fails and ``suppress_errors`` is ``False``.
        Mt5TradingError: When account equity is invalid, or when symbol margin
            lookup or projected margin lookup fails and ``suppress_errors`` is
            ``False``.
    """
    equity = _account_equity(client)
    unique_symbols = list(dict.fromkeys(symbols))
    margin = sum(
        calculate_positions_margin_by_symbol(
            client,
            symbols=unique_symbols,
            suppress_errors=suppress_errors,
        ).values(),
        0.0,
    )
    if (
        new_symbol in unique_symbols
        and new_position_side is not None
        and new_position_volume > 0
    ):
        try:
            margin += estimate_order_margin(
                client,
                new_symbol,
                new_position_side,
                new_position_volume,
            )
        except (Mt5TradingError, Mt5RuntimeError, AttributeError):
            if not suppress_errors:
                raise
            _logger.warning("Skipping projected margin for %r.", new_symbol)
    return margin / equity

def calculate_trailing_stop_updates(
    client: Mt5TradingClient,
    *,
    symbol: str,
    trailing_stop_ratio: float,
) -> dict[int, float]:
    """Return per-ticket trailing stop-loss updates for open symbol positions.

    Buy positions trail from bid using ``bid * (1 - trailing_stop_ratio)``.
    Sell positions trail from ask using ``ask * (1 + trailing_stop_ratio)``.
    Existing stop losses are preserved when they are already more favorable.
    Missing symbol metadata returns an empty update map. Positions with a
    missing side-specific tick price are skipped.
    """
    _require_protective_ratio(trailing_stop_ratio, "trailing_stop_ratio")
    positions = get_positions_frame(client, symbol=symbol)
    if positions.empty:
        return {}
    tick = get_tick_snapshot(client, symbol)
    bid = extract_tick_price(tick, "bid")
    ask = extract_tick_price(tick, "ask")
    digits = _symbol_digits(client, symbol)
    if digits is None:
        return {}

    updates: dict[int, float] = {}
    for row in positions.to_dict("records"):
        ticket = _position_ticket(row.get("ticket"))
        if ticket is None:
            continue
        next_sl = _trailing_stop_loss(
            client,
            position_type=row.get("type"),
            current_sl=_current_stop_loss(row.get("sl")),
            bid=bid,
            ask=ask,
            digits=digits,
            trailing_stop_ratio=trailing_stop_ratio,
        )
        if next_sl is None:
            continue
        updates[ticket] = next_sl
    return updates

def calculate_volume_by_margin(
    client: Mt5TradingClient,
    symbol: str,
    available_margin: float,
    order_side: OrderSide,
) -> float:
    """Calculate max normalized volume affordable for one side.

    Returns:
        Largest stepped volume whose actual margin (from ``order_calc_margin``)
        fits within ``available_margin``, rounded down to symbol volume
        constraints; ``0.0`` when no affordable step exists.

    Raises:
        Mt5TradingError: If symbol volume constraints or tick data are invalid.
    """
    if available_margin <= 0:
        return 0.0
    symbol_info = get_symbol_snapshot(client, symbol)
    volume_min = float(symbol_info.get("volume_min") or 0.0)
    volume_max = float(symbol_info.get("volume_max") or 0.0)
    volume_step = float(symbol_info.get("volume_step") or volume_min or 0.0)
    if volume_min <= 0 or volume_step <= 0:
        msg = f"Invalid volume constraints for {symbol!r}."
        raise Mt5TradingError(msg)
    side = _normalize_order_side(order_side)
    price = extract_tick_price(
        get_tick_snapshot(client, symbol), "ask" if side == "BUY" else "bid"
    )
    if price is None:
        msg = f"Tick price is unavailable for {symbol!r}."
        raise Mt5TradingError(msg)
    order_type = (
        client.mt5.ORDER_TYPE_BUY if side == "BUY" else client.mt5.ORDER_TYPE_SELL
    )
    min_margin = float(client.order_calc_margin(order_type, symbol, volume_min, price))
    if min_margin <= 0 or min_margin > available_margin:
        return 0.0
    lo = 0
    hi = int(
        max(
            0,
            floor(
                (
                    (
                        min(available_margin / min_margin * volume_min, volume_max)
                        if volume_max > 0
                        else available_margin / min_margin * volume_min
                    )
                    - volume_min
                )
                / volume_step
                + 1e-12
            ),
        )
    )
    best = -1

    while lo <= hi:
        mid = (lo + hi) // 2
        normalized = round(volume_min + mid * volume_step, 10)
        actual = float(client.order_calc_margin(order_type, symbol, normalized, price))

        if actual > 0 and actual <= available_margin:
            best = mid
            lo = mid + 1
        else:
            hi = mid - 1

    return round(volume_min + best * volume_step, 10) if best >= 0 else 0.0

def close_open_positions(
    client: Mt5TradingClient,
    *,
    symbols: str | list[str] | None = None,
    tickets: list[int] | None = None,
    dry_run: bool = False,
) -> list[OrderExecutionResult]:
    """Close matching open positions.

    Returns:
        Normalized execution results for matching positions.
    """
    positions = _filter_positions(
        get_positions_frame(client),
        symbols=symbols,
        tickets=tickets,
    )
    results: list[OrderExecutionResult] = []
    for row in positions.to_dict("records"):
        pos_type = row["type"]
        side: OrderSide = "SELL" if pos_type == client.mt5.POSITION_TYPE_BUY else "BUY"
        result = place_market_order(
            client,
            symbol=str(row["symbol"]),
            volume=float(row["volume"]),
            order_side=side,
            position=int(row["ticket"]),
            dry_run=dry_run,
        )
        results.append(result)
    return results

def create_trading_client(
    *,
    config: Mt5Config | None = None,
    login: int | str | None = None,
    password: str | None = None,
    server: str | None = None,
    path: str | None = None,
    timeout: int | None = None,
    retry_count: int = 0,
) -> Mt5TradingClient:
    """Return an initialized and logged-in trading client."""
    mt5_config = _resolve_config(
        config=config,
        login=login,
        password=password,
        server=server,
        path=path,
        timeout=timeout,
    )
    client = Mt5TradingClient(config=mt5_config, retry_count=retry_count)
    try:
        client.initialize_and_login_mt5()
    except Exception:
        client.shutdown()
        raise
    return client

def detect_position_side(
    client: Mt5TradingClient,
    symbol: str,
) -> PositionSide | None:
    """Detect the net open position side for a symbol.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbol: Symbol to inspect.

    Returns:
        ``"long"`` when there are buy positions and no sell positions,
        ``"short"`` when there are sell positions and no buy positions, or
        ``None`` when no positions or mixed exposure exists.
    """
    positions = get_positions_frame(client, symbol=symbol)
    if positions.empty:
        return None

    buy_type = client.mt5.POSITION_TYPE_BUY
    sell_type = client.mt5.POSITION_TYPE_SELL
    buy_volume = _sum_position_volume(positions, buy_type)
    sell_volume = _sum_position_volume(positions, sell_type)
    if buy_volume > 0 and sell_volume == 0:
        return "long"
    if sell_volume > 0 and buy_volume == 0:
        return "short"
    return None

def determine_order_limits(
    client: Mt5TradingClient,
    symbol: str,
    side: PositionSide | str,
    stop_loss_limit_ratio: float | None = None,
    take_profit_limit_ratio: float | None = None,
) -> OrderLimits:
    """Derive entry and protective order prices from current market quotes.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbol: Symbol used for the quote lookup.
        side: Position side as ``"long"``/``"short"`` (``"buy"``/``"sell"``
            aliases are accepted).
        stop_loss_limit_ratio: Relative distance from entry for stop loss in
            ``[0, 1)``. A value of ``0`` omits the stop loss.
        take_profit_limit_ratio: Relative distance from entry for take profit in
            ``[0, 1)``. A value of ``0`` omits the take profit.

    Returns:
        Dictionary with ``entry``, ``stop_loss``, and ``take_profit`` keys.
        Omitted protective levels are returned as ``None``.

    Raises:
        Mt5TradingError: If required tick data is invalid or computed SL/TP
            prices violate available ``trade_stops_level`` pre-validation.
    """
    stop_loss_ratio = stop_loss_limit_ratio or 0.0
    take_profit_ratio = take_profit_limit_ratio or 0.0
    _require_protective_ratio(stop_loss_ratio, "stop_loss_limit_ratio")
    _require_protective_ratio(take_profit_ratio, "take_profit_limit_ratio")
    normalized_side = _position_side_from_order_side(side)
    tick = get_tick_snapshot(client, symbol)
    entry_key = "ask" if normalized_side == "long" else "bid"
    entry = extract_tick_price(tick, entry_key)
    if entry is None:
        msg = f"Tick price is unavailable for {symbol!r}."
        raise Mt5TradingError(msg)
    try:
        symbol_info = get_symbol_snapshot(client, symbol)
    except (AttributeError, KeyError, TypeError, ValueError):
        symbol_info = {}
    try:
        digits = int(symbol_info.get("digits") or 8)
    except (TypeError, ValueError):
        digits = 8
    min_distance = _minimum_stop_distance(symbol_info)

    stop_loss: float | None = None
    if stop_loss_ratio > 0:
        if normalized_side == "long":
            stop_loss = entry * (1.0 - stop_loss_ratio)
        else:
            stop_loss = entry * (1.0 + stop_loss_ratio)
        stop_loss = round(stop_loss, digits)

    take_profit: float | None = None
    if take_profit_ratio > 0:
        if normalized_side == "long":
            take_profit = entry * (1.0 + take_profit_ratio)
        else:
            take_profit = entry * (1.0 - take_profit_ratio)
        take_profit = round(take_profit, digits)

    _validate_protective_prices(
        symbol=symbol,
        side=normalized_side,
        entry=entry,
        stop_loss=stop_loss,
        take_profit=take_profit,
        min_distance=min_distance,
    )

    return {
        "entry": entry,
        "stop_loss": stop_loss,
        "take_profit": take_profit,
    }

def ensure_symbol_selected(client: Mt5TradingClient, symbol: str) -> None:
    """Ensure a symbol is visible in Market Watch before sending orders.

    Args:
        client: Connected ``Mt5TradingClient`` instance.
        symbol: Symbol to select.

    Raises:
        Mt5TradingError: If the symbol cannot be selected in Market Watch or
            ``symbol_select`` is unavailable on the client.
    """
    snapshot = get_symbol_snapshot(client, symbol)
    if snapshot.get("visible"):
        return
    select = getattr(client, "symbol_select", None)
    if not callable(select):
        msg = "MT5 client is missing required method: symbol_select"
        raise Mt5TradingError(msg)
    if select(symbol, enable=True):
        return
    last_error = getattr(client, "last_error", None)
    detail = f" ({last_error()})" if callable(last_error) else ""
    msg = f"Failed to select symbol {symbol!r} in Market Watch{detail}."
    raise Mt5TradingError(msg)

def estimate_order_margin(
    client: Mt5TradingClient,
    symbol: str,
    order_side: OrderSide | str,
    volume: float,
) -> float:
    """Estimate required margin for one order at the current market price.

    Returns:
        Positive finite margin required for the order at the current quote.

    Raises:
        Mt5TradingError: If volume, tick data, or margin estimation is invalid.
    """
    if not _is_positive_finite_number(volume):
        msg = "Volume must be a positive finite number to estimate order margin."
        raise Mt5TradingError(msg)
    side = _normalize_order_side(order_side)
    tick = get_tick_snapshot(client, symbol)
    price = extract_tick_price(tick, "ask" if side == "BUY" else "bid")
    if price is None:
        msg = f"Tick price is unavailable for {symbol!r}."
        raise Mt5TradingError(msg)
    order_type = (
        client.mt5.ORDER_TYPE_BUY if side == "BUY" else client.mt5.ORDER_TYPE_SELL
    )
    raw_margin = client.order_calc_margin(order_type, symbol, volume, price)
    try:
        margin = float(raw_margin)
    except (TypeError, ValueError) as exc:
        msg = f"Margin estimate is invalid for {symbol!r}."
        raise Mt5TradingError(msg) from exc
    if margin <= 0 or not isfinite(margin):
        msg = f"Margin estimate is invalid for {symbol!r}."
        raise Mt5TradingError(msg)
    return margin

def extract_tick_price(tick: Mapping[str, object], key: str) -> float | None:
    """Return a positive finite float from tick[key], or None if invalid.

    Accepts int, float, or numeric string values. Returns None when the key is
    missing, the value is None, non-numeric, NaN, infinite, zero, or negative.
    Booleans are treated as non-numeric and return None.
    """
    value = tick.get(key)
    if value is None or isinstance(value, bool):
        return None
    if isinstance(value, int | float):
        price = float(value)
    elif isinstance(value, str):
        try:
            price = float(value)
        except ValueError:
            return None
    else:
        return None
    if not isfinite(price) or price <= 0:
        return None
    return price

def fetch_latest_closed_rates_for_trading_client(
    client: Mt5TradingClient,
    *,
    symbol: str,
    granularity: str,
    count: int,
) -> pd.DataFrame:
    """Fetch the latest closed bars from a connected trading client.

    Returns:
        Up to ``count`` closed bars ordered oldest to newest.

    Raises:
        ValueError: If ``count`` is not positive, rate data is empty or
            malformed, or the ``time`` column is missing.
        Mt5TradingError: If the trading client cannot fetch rate data.
    """
    if count <= 0:
        msg = "count must be positive."
        raise ValueError(msg)
    fetch_method = getattr(client, "fetch_latest_rates_as_df", None)
    if not callable(fetch_method):
        msg = "MT5 trading client cannot fetch rate data."
        raise Mt5TradingError(msg)
    fetched = fetch_method(symbol, granularity, count + 1)
    if not isinstance(fetched, pd.DataFrame):
        msg = (
            f"Malformed rate data for {symbol!r} at granularity {granularity!r}: "
            "expected a DataFrame."
        )
        raise ValueError(msg)  # noqa: TRY004
    frame = fetched
    frame = _ensure_rate_time_column(frame)
    if "time" not in frame.columns:
        msg = f"Rate data is missing a time column for {symbol!r}."
        raise ValueError(msg)
    closed = drop_forming_rate_bar(frame)
    if closed.empty:
        msg = (
            f"Rate data is empty for {symbol!r} at granularity {granularity!r} "
            f"with count {count}."
        )
        raise ValueError(msg)
    return closed.tail(count).reset_index(drop=True)