DataFrame¶

@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 pd.DataFrame([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.
        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.
        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.copy_rates_from_as_df(
        symbol=symbol,
        timeframe=timeframe,
        date_from=date_from,
        count=count,
        skip_to_datetime=skip_to_datetime,
        index_keys=None,
    ).to_dict(orient="records")

@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.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,
    ).to_dict(orient="records")

@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.
        date_to: End date.
        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.
        date_to: End date.
        skip_to_datetime: Whether to skip converting time to datetime.

    Returns:
        List of dictionaries with OHLCV data.
    """
    return 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,
    ).to_dict(orient="records")

@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.
        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.
        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.copy_ticks_from_as_df(
        symbol=symbol,
        date_from=date_from,
        count=count,
        flags=flags,
        skip_to_datetime=skip_to_datetime,
        index_keys=None,
    ).to_dict(orient="records")  # type: ignore[reportReturnType]

@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.
        date_to: End date.
        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.
        date_to: End date.
        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.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,
    ).to_dict(orient="records")  # type: ignore[reportReturnType]

@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 (required if not using ticket/position).
        date_to: End date (required if not using ticket/position).
        group: Optional group filter.
        symbol: Optional symbol filter.
        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 (required if not using ticket/position).
        date_to: End date (required if not using ticket/position).
        group: Optional group filter.
        symbol: Optional symbol filter.
        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,
    )
    return [
        d._asdict()
        for d in self.history_deals_get(
            date_from=date_from,
            date_to=date_to,
            group=(f"*{symbol}*" if symbol else group),
            ticket=ticket,
            position=position,
        )
    ]

@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 (required if not using ticket/position).
        date_to: End date (required if not using ticket/position).
        group: Optional group filter.
        symbol: Optional symbol filter.
        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 (required if not using ticket/position).
        date_to: End date (required if not using ticket/position).
        group: Optional group filter.
        symbol: Optional symbol filter.
        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,
    )
    return [
        o._asdict()
        for o in self.history_orders_get(
            date_from=date_from,
            date_to=date_to,
            group=(f"*{symbol}*" if symbol else group),
            ticket=ticket,
            position=position,
        )
    ]

def initialize_and_login_mt5(
    self,
    path: str | None = None,
    login: int | None = None,
    password: str | None = None,
    server: str | None = None,
    timeout: int | None = None,
) -> None:
    """Initialize MetaTrader5 connection with retry logic.

    This method overrides the base class to add retry logic and use config values.

    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 fails after retries.
    """
    path = path or self.config.path
    login_kwargs = {
        "login": login or self.config.login,
        "password": password or self.config.password,
        "server": server or self.config.server,
        "timeout": timeout or self.config.timeout,
    }
    for i in range(1 + max(0, self.retry_count)):
        if i:
            self.logger.warning(
                "Retrying MT5 initialization (%d/%d)...",
                i,
                self.retry_count,
            )
            time.sleep(i)
        if self.initialize(path=path, **login_kwargs) and (
            (not login_kwargs["login"]) or self.login(**login_kwargs)
        ):
            return
    error_message = (
        f"MT5 initialize and login failed after {self.retry_count} retries:"
        f" {self.last_error()}"
    )
    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 pd.DataFrame([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 [b._asdict() for b in 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 pd.DataFrame([
        self._flatten_dict_to_one_level(
            dictionary=self.order_check_as_dict(request=request),
        )
    ])

def order_check_as_dict(self, request: dict[str, Any]) -> dict[str, Any]:
    """Check funds sufficiency for performing a requested trading operation as a dictionary.

    Args:
        request: Order request parameters.

    Returns:
        Dictionary with order check results.
    """  # noqa: E501
    return {
        k: (v._asdict() if k == "request" else v)
        for k, v in self.order_check(request=request)._asdict().items()
    }

@set_index_if_possible(index_parameters="index_keys")
def order_send_as_df(
    self,
    request: dict[str, Any],
    index_keys: str | None = None,  # noqa: ARG002
) -> pd.DataFrame:
    """Send a request to perform a trading operation from the terminal to the trade server as a data frame.

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

    Returns:
        DataFrame with order send results.
    """  # noqa: E501
    return pd.DataFrame([
        self._flatten_dict_to_one_level(
            dictionary=self.order_send_as_dict(request=request),
        )
    ])