DataFrame¶

def __enter__(self) -> Self:
    """Context manager entry using config-aware initialization.

    Returns:
        The initialized client instance.
    """
    self.initialize_and_login_mt5()
    return self

@set_index_if_possible(index_parameters="index_keys")
def account_info_as_df(self, index_keys: str | None = None) -> pd.DataFrame:  # noqa: ARG002
    """Get info on the current account as a data frame.

    Args:
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with account information.
    """
    return self._as_single_row_df(self.account_info_as_dict())

def account_info_as_dict(self) -> dict[str, Any]:
    """Get info on the current account as a dictionary.

    Returns:
        Dictionary with account information.
    """
    return self.account_info()._asdict()

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def copy_rates_from_as_df(
    self,
    symbol: str,
    timeframe: int,
    date_from: datetime,
    count: int,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get bars for a specified date range as a data frame.

    Args:
        symbol: Symbol name.
        timeframe: Timeframe constant.
        date_from: Start date in trade-server time (not UTC).
        count: Number of rates to retrieve.
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with OHLCV data.
    """
    self._validate_positive_count(count=count)
    return pd.DataFrame(
        self.copy_rates_from(
            symbol=symbol,
            timeframe=timeframe,
            date_from=date_from,
            count=count,
        )
    )

def copy_rates_from_as_dicts(
    self,
    symbol: str,
    timeframe: int,
    date_from: datetime,
    count: int,
    skip_to_datetime: bool = False,
) -> list[dict[str, Any]]:
    """Get bars for a specified date range as a list of dictionaries.

    Args:
        symbol: Symbol name.
        timeframe: Timeframe constant.
        date_from: Start date in trade-server time (not UTC).
        count: Number of rates to retrieve.
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with OHLCV data.
    """
    return self._as_record_dicts(
        self.copy_rates_from_as_df(
            symbol=symbol,
            timeframe=timeframe,
            date_from=date_from,
            count=count,
            skip_to_datetime=skip_to_datetime,
            index_keys=None,
        )
    )

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def copy_rates_from_pos_as_df(
    self,
    symbol: str,
    timeframe: int,
    start_pos: int,
    count: int,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get bars from a specified position as a data frame.

    Args:
        symbol: Symbol name.
        timeframe: Timeframe constant.
        start_pos: Start position.
        count: Number of rates to retrieve.
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with OHLCV data.
    """
    self._validate_positive_count(count=count)
    self._validate_non_negative_position(position=start_pos)
    return pd.DataFrame(
        self.copy_rates_from_pos(
            symbol=symbol,
            timeframe=timeframe,
            start_pos=start_pos,
            count=count,
        )
    )

def copy_rates_from_pos_as_dicts(
    self,
    symbol: str,
    timeframe: int,
    start_pos: int,
    count: int,
    skip_to_datetime: bool = False,
) -> list[dict[str, Any]]:
    """Get bars from a specified position as a list of dictionaries.

    Args:
        symbol: Symbol name.
        timeframe: Timeframe constant.
        start_pos: Start position.
        count: Number of rates to retrieve.
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with OHLCV data.
    """
    return self._as_record_dicts(
        self.copy_rates_from_pos_as_df(
            symbol=symbol,
            timeframe=timeframe,
            start_pos=start_pos,
            count=count,
            skip_to_datetime=skip_to_datetime,
            index_keys=None,
        )
    )

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def copy_rates_range_as_df(
    self,
    symbol: str,
    timeframe: int,
    date_from: datetime,
    date_to: datetime,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get bars for a specified date range as a data frame.

    Args:
        symbol: Symbol name.
        timeframe: Timeframe constant.
        date_from: Start date in trade-server time (not UTC).
        date_to: End date in trade-server time (not UTC).
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with OHLCV data.
    """
    self._validate_date_range(date_from=date_from, date_to=date_to)
    return pd.DataFrame(
        self.copy_rates_range(
            symbol=symbol,
            timeframe=timeframe,
            date_from=date_from,
            date_to=date_to,
        )
    )

def copy_rates_range_as_dicts(
    self,
    symbol: str,
    timeframe: int,
    date_from: datetime,
    date_to: datetime,
    skip_to_datetime: bool = False,
) -> list[dict[str, Any]]:
    """Get bars for a specified date range as a list of dictionaries.

    Args:
        symbol: Symbol name.
        timeframe: Timeframe constant.
        date_from: Start date in trade-server time (not UTC).
        date_to: End date in trade-server time (not UTC).
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with OHLCV data.
    """
    return self._as_record_dicts(
        self.copy_rates_range_as_df(
            symbol=symbol,
            timeframe=timeframe,
            date_from=date_from,
            date_to=date_to,
            skip_to_datetime=skip_to_datetime,
            index_keys=None,
        )
    )

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def copy_ticks_from_as_df(
    self,
    symbol: str,
    date_from: datetime,
    count: int,
    flags: int,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get ticks from a specified date as a data frame.

    Args:
        symbol: Symbol name.
        date_from: Start date in trade-server time (not UTC).
        count: Number of ticks to retrieve.
        flags: Tick flags (use constants from MetaTrader5).
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with tick data.
    """
    self._validate_positive_count(count=count)
    return pd.DataFrame(
        self.copy_ticks_from(
            symbol=symbol,
            date_from=date_from,
            count=count,
            flags=flags,
        )
    )

def copy_ticks_from_as_dicts(
    self,
    symbol: str,
    date_from: datetime,
    count: int,
    flags: int,
    skip_to_datetime: bool = False,
) -> list[dict[str, Any]]:
    """Get ticks from a specified date as a list of dictionaries.

    Args:
        symbol: Symbol name.
        date_from: Start date in trade-server time (not UTC).
        count: Number of ticks to retrieve.
        flags: Tick flags (use constants from MetaTrader5).
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with tick data.
    """
    return self._as_record_dicts(
        self.copy_ticks_from_as_df(
            symbol=symbol,
            date_from=date_from,
            count=count,
            flags=flags,
            skip_to_datetime=skip_to_datetime,
            index_keys=None,
        )
    )

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def copy_ticks_range_as_df(
    self,
    symbol: str,
    date_from: datetime,
    date_to: datetime,
    flags: int,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get ticks for a specified date range as a data frame.

    Args:
        symbol: Symbol name.
        date_from: Start date in trade-server time (not UTC).
        date_to: End date in trade-server time (not UTC).
        flags: Tick flags (use constants from MetaTrader5).
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with tick data.
    """
    self._validate_date_range(date_from=date_from, date_to=date_to)
    return pd.DataFrame(
        self.copy_ticks_range(
            symbol=symbol,
            date_from=date_from,
            date_to=date_to,
            flags=flags,
        )
    )

def copy_ticks_range_as_dicts(
    self,
    symbol: str,
    date_from: datetime,
    date_to: datetime,
    flags: int,
    skip_to_datetime: bool = False,
) -> list[dict[str, Any]]:
    """Get ticks for a specified date range as a list of dictionaries.

    Args:
        symbol: Symbol name.
        date_from: Start date in trade-server time (not UTC).
        date_to: End date in trade-server time (not UTC).
        flags: Tick flags (use constants from MetaTrader5).
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with tick data.
    """
    return self._as_record_dicts(
        self.copy_ticks_range_as_df(
            symbol=symbol,
            date_from=date_from,
            date_to=date_to,
            flags=flags,
            skip_to_datetime=skip_to_datetime,
            index_keys=None,
        )
    )

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def history_deals_get_as_df(
    self,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    group: str | None = None,
    symbol: str | None = None,
    ticket: int | None = None,
    position: int | None = None,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get historical deals with optional filters as a data frame.

    Args:
        date_from: Start date in trade-server time, not UTC (required if
            not using ticket/position).
        date_to: End date in trade-server time, not UTC (required if not
            using ticket/position).
        group: Optional group filter. Mutually exclusive with symbol.
        symbol: Optional symbol filter matching the symbol name exactly.
            Mutually exclusive with group.
        ticket: Get deals by order ticket.
        position: Get deals by position ticket.
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with historical deal information.
    """
    return pd.DataFrame(
        self.history_deals_get_as_dicts(
            date_from=date_from,
            date_to=date_to,
            group=group,
            symbol=symbol,
            ticket=ticket,
            position=position,
            skip_to_datetime=True,
        )
    )

@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def history_deals_get_as_dicts(
    self,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    group: str | None = None,
    symbol: str | None = None,
    ticket: int | None = None,
    position: int | None = None,
    skip_to_datetime: bool = False,  # noqa: ARG002
) -> list[dict[str, Any]]:
    """Get historical deals with optional filters as a list of dictionaries.

    Args:
        date_from: Start date in trade-server time, not UTC (required if
            not using ticket/position).
        date_to: End date in trade-server time, not UTC (required if not
            using ticket/position).
        group: Optional group filter. Mutually exclusive with symbol.
        symbol: Optional symbol filter matching the symbol name exactly.
            Mutually exclusive with group.
        ticket: Get deals by order ticket.
        position: Get deals by position ticket.
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with historical deal information.
    """
    self._validate_history_input(
        date_from=date_from,
        date_to=date_to,
        ticket=ticket,
        position=position,
        group=group,
        symbol=symbol,
    )
    dicts = self._as_dicts(
        self.history_deals_get(
            date_from=date_from,
            date_to=date_to,
            group=(f"*{symbol}*" if symbol else group),
            ticket=ticket,
            position=position,
        )
    )
    return [d for d in dicts if d["symbol"] == symbol] if symbol else dicts

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def history_orders_get_as_df(
    self,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    group: str | None = None,
    symbol: str | None = None,
    ticket: int | None = None,
    position: int | None = None,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get historical orders with optional filters as a data frame.

    Args:
        date_from: Start date in trade-server time, not UTC (required if
            not using ticket/position).
        date_to: End date in trade-server time, not UTC (required if not
            using ticket/position).
        group: Optional group filter. Mutually exclusive with symbol.
        symbol: Optional symbol filter matching the symbol name exactly.
            Mutually exclusive with group.
        ticket: Get orders by ticket.
        position: Get orders by position.
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with historical order information.
    """
    return pd.DataFrame(
        self.history_orders_get_as_dicts(
            date_from=date_from,
            date_to=date_to,
            group=group,
            symbol=symbol,
            ticket=ticket,
            position=position,
            skip_to_datetime=True,
        )
    )

@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def history_orders_get_as_dicts(
    self,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    group: str | None = None,
    symbol: str | None = None,
    ticket: int | None = None,
    position: int | None = None,
    skip_to_datetime: bool = False,  # noqa: ARG002
) -> list[dict[str, Any]]:
    """Get historical orders with optional filters as a list of dictionaries.

    Args:
        date_from: Start date in trade-server time, not UTC (required if
            not using ticket/position).
        date_to: End date in trade-server time, not UTC (required if not
            using ticket/position).
        group: Optional group filter. Mutually exclusive with symbol.
        symbol: Optional symbol filter matching the symbol name exactly.
            Mutually exclusive with group.
        ticket: Get orders by ticket.
        position: Get orders by position.
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with historical order information.
    """
    self._validate_history_input(
        date_from=date_from,
        date_to=date_to,
        ticket=ticket,
        position=position,
        group=group,
        symbol=symbol,
    )
    dicts = self._as_dicts(
        self.history_orders_get(
            date_from=date_from,
            date_to=date_to,
            group=(f"*{symbol}*" if symbol else group),
            ticket=ticket,
            position=position,
        )
    )
    return [d for d in dicts if d["symbol"] == symbol] if symbol else dicts

def initialize_and_login_mt5(
    self,
    path: str | None = None,
    login: int | None = None,
    password: str | SecretStr | None = None,
    server: str | None = None,
    timeout: int | None = None,
) -> None:
    """Initialize MT5 and ensure the requested account is active.

    ``MetaTrader5.initialize()`` accepts account credentials and normally
    connects directly to the requested account. After initialization, this
    method verifies the active account and only calls ``login()`` when the
    terminal reports a different account, avoiding redundant authentication
    while preserving account-switch fallback behavior.

    Args:
        path: Path to terminal EXE file (overrides config).
        login: Account login (overrides config).
        password: Account password (overrides config).
        server: Server name (overrides config).
        timeout: Connection timeout (overrides config).

    Raises:
        Mt5RuntimeError: If initialization or account selection fails after retries.
    """
    path = path or self.config.path
    login_value = login if login is not None else self.config.login
    password_value = self._unwrap_password(
        password if password is not None else self.config.password
    )
    server_value = server if server is not None else self.config.server
    timeout_value = timeout if timeout is not None else self.config.timeout
    last_error_value: tuple[int, str] | None = None
    for i in range(1 + max(0, self.retry_count)):
        if i:
            logger.warning(
                "Retrying MT5 initialization (%d/%d)...",
                i,
                self.retry_count,
            )
            time.sleep(i)
        if not self.initialize(
            path=path,
            login=login_value,
            password=password_value,
            server=server_value,
            timeout=timeout_value,
        ):
            last_error_value = self.last_error()
            continue
        if login_value is None:
            return
        try:
            active_account = self.account_info()
        except Mt5RuntimeError:
            active_account = None
        if active_account is not None and (
            getattr(active_account, "login", None) == login_value
            and (
                server_value is None
                or getattr(active_account, "server", None) == server_value
            )
        ):
            return
        try:
            if self.login(
                login=login_value,
                password=password_value,
                server=server_value,
                timeout=timeout_value,
            ):
                return
        except Exception:
            self.shutdown()
            raise
        last_error_value = self.last_error()
        self.shutdown()
    error_message = (
        f"MT5 initialize and login failed after {self.retry_count} retries:"
        f" {last_error_value}"
    )
    raise Mt5RuntimeError(error_message)

@set_index_if_possible(index_parameters="index_keys")
def last_error_as_df(self, index_keys: str | None = None) -> pd.DataFrame:  # noqa: ARG002
    """Get the last error information as a data frame.

    Args:
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with last error information.
    """
    return self._as_single_row_df(self.last_error_as_dict())

def last_error_as_dict(self) -> dict[str, Any]:
    """Get the last error information as a dictionary.

    Returns:
        Dictionary with last error information.
    """
    response = self.last_error()
    return {"error_code": response[0], "error_description": response[1]}

@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def market_book_get_as_df(
    self,
    symbol: str,
    skip_to_datetime: bool = False,  # noqa: ARG002
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Get market depth for a specified symbol as a data frame.

    Args:
        symbol: Symbol name.
        skip_to_datetime: Whether to skip converting time to datetime.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with market depth data.
    """
    return pd.DataFrame(
        self.market_book_get_as_dicts(symbol=symbol, skip_to_datetime=True)
    )

@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def market_book_get_as_dicts(
    self,
    symbol: str,
    skip_to_datetime: bool = False,  # noqa: ARG002
) -> list[dict[str, Any]]:
    """Get market depth for a specified symbol as a list of dictionaries.

    Args:
        symbol: Symbol name.
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with market depth data.
    """
    return self._as_dicts(self.market_book_get(symbol=symbol))

@set_index_if_possible(index_parameters="index_keys")
def order_check_as_df(
    self,
    request: dict[str, Any],
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Check funds sufficiency for performing a requested trading operation as a data frame.

    Args:
        request: Order request parameters.
        index_keys: Column name to set as index if provided.

    Returns:
        DataFrame with order check results.
    """  # noqa: E501
    return self._as_single_row_df(
        self._flatten_dict_to_one_level(
            dictionary=self.order_check_as_dict(request=request),
        )
    )