SDK Module¶

def __enter__(self) -> Self:
    """Open a persistent MT5 connection for multiple calls.

    Returns:
        This client instance.
    """
    if self._client is not None:
        return self
    client = Mt5DataClient(config=self._config)
    try:
        client.initialize_and_login_mt5()
    except Exception:
        client.shutdown()
        raise
    self._client = client
    self._owns_client = True  # only set when this method created the client
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: object,
) -> None:
    """Shut down the persistent MT5 connection."""
    if self._client is not None and self._owns_client:
        self._client.shutdown()
        self._client = None

def account_info(self) -> pd.DataFrame:
    """Return account information."""
    return self._fetch(lambda c: c.account_info_as_df())

def collect_latest_rates(
    self,
    symbols: Sequence[str],
    timeframes: Sequence[int | str],
    *,
    count: int,
    start_pos: int = 0,
) -> dict[tuple[str, int], pd.DataFrame]:
    """Return latest rates for each symbol/timeframe pair.

    Returns:
        Mapping keyed by ``(symbol, timeframe_int)``.

    Raises:
        ValueError: If ``count`` is not positive or inputs are empty.
    """
    _require_positive(count, "count")
    if not symbols:
        msg = "At least one symbol is required."
        raise ValueError(msg)
    if not timeframes:
        msg = "At least one timeframe is required."
        raise ValueError(msg)
    resolved_timeframes = [_coerce_timeframe(timeframe) for timeframe in timeframes]
    return self._fetch_value(
        lambda c: {
            (symbol, timeframe): c.copy_rates_from_pos_as_df(
                symbol=symbol,
                timeframe=timeframe,
                start_pos=start_pos,
                count=count,
            )
            for symbol in symbols
            for timeframe in resolved_timeframes
        },
    )

def copy_rates_from(
    self,
    symbol: str,
    timeframe: int | str,
    date_from: datetime | str,
    count: int,
) -> pd.DataFrame:
    """Return rates starting from a date."""
    tf = _coerce_timeframe(timeframe)
    start = _require_datetime(date_from)
    return self._fetch(
        lambda c: c.copy_rates_from_as_df(
            symbol=symbol,
            timeframe=tf,
            date_from=start,
            count=count,
        ),
    )

def copy_rates_from_pos(
    self,
    symbol: str,
    timeframe: int | str,
    start_pos: int,
    count: int,
) -> pd.DataFrame:
    """Return rates starting from a bar position."""
    tf = _coerce_timeframe(timeframe)
    return self._fetch(
        lambda c: c.copy_rates_from_pos_as_df(
            symbol=symbol,
            timeframe=tf,
            start_pos=start_pos,
            count=count,
        ),
    )

def copy_rates_range(
    self,
    symbol: str,
    timeframe: int | str,
    date_from: datetime | str,
    date_to: datetime | str,
) -> pd.DataFrame:
    """Return rates for a date range."""
    tf = _coerce_timeframe(timeframe)
    start = _require_datetime(date_from)
    end = _require_datetime(date_to)
    return self._fetch(
        lambda c: c.copy_rates_range_as_df(
            symbol=symbol,
            timeframe=tf,
            date_from=start,
            date_to=end,
        ),
    )

def copy_ticks_from(
    self,
    symbol: str,
    date_from: datetime | str,
    count: int,
    flags: int | str,
) -> pd.DataFrame:
    """Return ticks starting from a date."""
    start = _require_datetime(date_from)
    tick_flags = _coerce_tick_flags(flags)
    return self._fetch(
        lambda c: c.copy_ticks_from_as_df(
            symbol=symbol,
            date_from=start,
            count=count,
            flags=tick_flags,
        ),
    )

def copy_ticks_range(
    self,
    symbol: str,
    date_from: datetime | str,
    date_to: datetime | str,
    flags: int | str,
) -> pd.DataFrame:
    """Return ticks for a date range."""
    start = _require_datetime(date_from)
    end = _require_datetime(date_to)
    tick_flags = _coerce_tick_flags(flags)
    return self._fetch(
        lambda c: c.copy_ticks_range_as_df(
            symbol=symbol,
            date_from=start,
            date_to=end,
            flags=tick_flags,
        ),
    )

@classmethod
def from_connected_client(cls, client: Mt5DataClient) -> Self:
    """Bind to an already-connected ``Mt5DataClient`` without owning it.

    The returned ``Mt5CliClient`` never initializes or shuts down the
    injected client, including when used as a context manager.

    Returns:
        Client wrapper bound to the injected connection.
    """
    return cls(client=client)

def history_deals(
    self,
    date_from: datetime | str | None = None,
    date_to: datetime | str | None = None,
    group: str | None = None,
    symbol: str | None = None,
    ticket: int | None = None,
    position: int | None = None,
) -> pd.DataFrame:
    """Return historical deals."""
    start = _coerce_datetime(date_from)
    end = _coerce_datetime(date_to)
    return self._fetch(
        lambda c: c.history_deals_get_as_df(
            date_from=start,
            date_to=end,
            group=group,
            symbol=symbol,
            ticket=ticket,
            position=position,
        ),
    )

def history_orders(
    self,
    date_from: datetime | str | None = None,
    date_to: datetime | str | None = None,
    group: str | None = None,
    symbol: str | None = None,
    ticket: int | None = None,
    position: int | None = None,
) -> pd.DataFrame:
    """Return historical orders."""
    start = _coerce_datetime(date_from)
    end = _coerce_datetime(date_to)
    return self._fetch(
        lambda c: c.history_orders_get_as_df(
            date_from=start,
            date_to=end,
            group=group,
            symbol=symbol,
            ticket=ticket,
            position=position,
        ),
    )

def last_error(self) -> pd.DataFrame:
    """Return the last error information."""
    return self._fetch(lambda c: c.last_error_as_df())

def latest_rates(
    self,
    symbol: str,
    timeframe: int | str,
    count: int,
    start_pos: int = 0,
) -> pd.DataFrame:
    """Return the latest rates from a bar position."""
    _require_positive(count, "count")
    return self.copy_rates_from_pos(symbol, timeframe, start_pos, count)

def market_book(self, symbol: str) -> pd.DataFrame:
    """Return market depth for a symbol."""
    return self._fetch(lambda c: c.market_book_get_as_df(symbol=symbol))

def minimum_margins(self, symbol: str) -> pd.DataFrame:
    """Return minimum-volume buy and sell margin requirements.

    Args:
        symbol: Symbol name.

    Returns:
        One-row DataFrame with columns ``symbol``, ``account_currency``,
        ``volume_min``, ``buy_margin``, and ``sell_margin``.
    """
    return self._fetch(lambda c: _fetch_minimum_margins(c, symbol))

def mt5_summary(self) -> dict[str, object]:
    """Return a compact terminal/account status summary."""

    def _summary(client: Mt5DataClient) -> dict[str, object]:
        return {
            "version": _plain_mt5_value(
                _call_required_client_method(client, "version"),
            ),
            "terminal_info": _plain_mt5_value(
                _call_required_client_method(client, "terminal_info"),
            ),
            "account_info": _plain_mt5_value(
                _call_required_client_method(client, "account_info"),
            ),
            "symbols_total": _plain_mt5_value(
                _call_required_client_method(client, "symbols_total"),
            ),
        }

    return self._fetch_value(_summary)

def mt5_summary_as_df(self) -> pd.DataFrame:
    """Return an export-safe one-row terminal/account summary DataFrame."""
    summary = self.mt5_summary()
    return pd.DataFrame(
        [
            {
                key: _mt5_summary_export_value(value)
                for key, value in summary.items()
            },
        ],
    )

def orders(
    self,
    symbol: str | None = None,
    group: str | None = None,
    ticket: int | None = None,
) -> pd.DataFrame:
    """Return active orders."""
    return self._fetch(
        lambda c: c.orders_get_as_df(
            symbol=symbol,
            group=group,
            ticket=ticket,
        ),
    )

def positions(
    self,
    symbol: str | None = None,
    group: str | None = None,
    ticket: int | None = None,
) -> pd.DataFrame:
    """Return open positions."""
    return self._fetch(
        lambda c: c.positions_get_as_df(
            symbol=symbol,
            group=group,
            ticket=ticket,
        ),
    )

def recent_history_deals(
    self,
    hours: float,
    date_to: datetime | str | None = None,
    group: str | None = None,
    symbol: str | None = None,
) -> pd.DataFrame:
    """Return historical deals from a recent trailing window."""
    _require_positive(hours, "hours")
    end = _require_datetime(date_to) if date_to is not None else datetime.now(UTC)
    start = end - timedelta(hours=hours)
    return self.history_deals(
        date_from=start,
        date_to=end,
        group=group,
        symbol=symbol,
    )

def recent_ticks(
    self,
    symbol: str,
    seconds: float,
    *,
    date_to: datetime | str | None = None,
    count: int = 10000,
    flags: int | str = "ALL",
) -> pd.DataFrame:
    """Return ticks from a recent time window.

    Args:
        symbol: Symbol name.
        seconds: Lookback window in seconds ending at ``date_to``.
        date_to: Window end time. When ``None``, uses the latest
            ``symbol_info_tick().time`` rather than wall-clock now.
        count: Maximum ticks to return. Values ``<= 0`` return the full
            window without trimming. Positive values keep the most recent
            ticks; when the window is sparse, ``copy_ticks_from`` avoids
            fetching the entire range.
        flags: Tick flags as ``ALL``, ``INFO``, ``TRADE``, or an integer.

    Returns:
        Tick DataFrame with MT5 tick columns such as ``time``, ``bid``,
        ``ask``, ``last``, and ``volume``.
    """
    tick_flags = _coerce_tick_flags(flags)
    end = _coerce_datetime(date_to)
    return self._fetch(
        lambda c: _fetch_recent_ticks(
            c,
            symbol,
            seconds,
            end,
            count,
            tick_flags,
        ),
    )

def symbol_info(self, symbol: str) -> pd.DataFrame:
    """Return details for one symbol."""
    return self._fetch(lambda c: c.symbol_info_as_df(symbol=symbol))

def symbol_info_tick(self, symbol: str) -> pd.DataFrame:
    """Return the last tick for a symbol."""
    return self._fetch(lambda c: c.symbol_info_tick_as_df(symbol=symbol))

def symbols(self, group: str | None = None) -> pd.DataFrame:
    """Return the symbol list."""
    return self._fetch(lambda c: c.symbols_get_as_df(group=group))

def terminal_info(self) -> pd.DataFrame:
    """Return terminal information."""
    return self._fetch(lambda c: c.terminal_info_as_df())

def version(self) -> pd.DataFrame:
    """Return MetaTrader5 version information."""
    return self._fetch(lambda c: c.version_as_df())