Utilities

pdmt5.utils

Utility functions for the pdmt5 package.

P module-attribute 

P = ParamSpec('P')

R module-attribute 

R = TypeVar('R')

__all__ module-attribute 

__all__: list[str] = []

detect_and_convert_time_to_datetime 

detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Decorator to convert time values/columns to datetime based on result type.

Automatically detects result type and applies appropriate time conversion: - dict: converts time values in the dictionary - list: converts time values in each dictionary item - DataFrame: converts time columns

MT5 epoch timestamps are trade-server wall-clock labels, so the converted datetimes are timezone-naive and represent server time, not UTC.

Parameters:

Name Type Description Default
skip_toggle str | None

Name of the bound parameter that skips conversion when its value is truthy.

 None

Returns:

Type Description
Callable[[Callable[P, R]], Callable[P, R]]

Decorator function.
Source code in pdmt5/utils.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator to convert time values/columns to datetime based on result type.

    Automatically detects result type and applies appropriate time conversion:
    - dict: converts time values in the dictionary
    - list: converts time values in each dictionary item
    - DataFrame: converts time columns

    MT5 epoch timestamps are trade-server wall-clock labels, so the converted
    datetimes are timezone-naive and represent server time, not UTC.

    Args:
        skip_toggle: Name of the bound parameter that skips conversion when
            its value is truthy.

    Returns:
        Decorator function.
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        signature = inspect.signature(func)

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            bound_arguments = signature.bind_partial(*args, **kwargs).arguments
            result = func(*args, **kwargs)
            if skip_toggle and bound_arguments.get(skip_toggle):
                return result
            elif isinstance(result, dict):
                return cast(
                    "R",
                    _convert_time_values_in_dict(
                        dictionary=cast("dict[str, Any]", result)
                    ),
                )
            elif isinstance(result, list):
                return [
                    (
                        _convert_time_values_in_dict(
                            dictionary=cast("dict[str, Any]", d)
                        )
                        if isinstance(d, dict)
                        else d
                    )
                    for d in cast("list[Any]", result)
                ]  # type: ignore[return-value]
            elif isinstance(result, pd.DataFrame):
                return cast("R", _convert_time_columns_in_df(result))
            else:
                return result

        return wrapper

    return decorator

set_index_if_possible 

set_index_if_possible(
    index_parameters: str | None = None,
) -> Callable[
    [Callable[P, DataFrame]], Callable[P, DataFrame]
]

Decorator to set index on DataFrame results if not empty.

Parameters:

Name Type Description Default
index_parameters str | None

Name of the parameter to use as index if provided.

 None

Returns:

Type Description
Callable[[Callable[P, DataFrame]], Callable[P, DataFrame]]

Decorator function.
Source code in pdmt5/utils.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def set_index_if_possible(
    index_parameters: str | None = None,
) -> Callable[[Callable[P, pd.DataFrame]], Callable[P, pd.DataFrame]]:
    """Decorator to set index on DataFrame results if not empty.

    Args:
        index_parameters: Name of the parameter to use as index if provided.

    Returns:
        Decorator function.
    """

    def decorator(
        func: Callable[P, pd.DataFrame],
    ) -> Callable[P, pd.DataFrame]:
        signature = inspect.signature(func)

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> pd.DataFrame:
            bound_arguments = signature.bind_partial(*args, **kwargs).arguments
            result = func(*args, **kwargs)
            if not isinstance(result, pd.DataFrame):  # type: ignore[reportUnnecessaryIsInstance]
                error_message = (
                    f"Function {func.__name__} returned non-DataFrame result: "
                    f"{type(result).__name__}. Expected DataFrame."
                )
                raise TypeError(error_message)
            elif (
                index_parameters
                and bound_arguments.get(index_parameters)
                and not result.empty
            ):
                return result.set_index(bound_arguments[index_parameters])
            else:
                return result

        return wrapper

    return decorator

Overview

The utils module provides internal utility functions and decorators used throughout the pdmt5 package for data transformation and DataFrame manipulation.

Functions

detect_and_convert_time_to_datetime

pdmt5.utils.detect_and_convert_time_to_datetime 

detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Decorator to convert time values/columns to datetime based on result type.

Automatically detects result type and applies appropriate time conversion: - dict: converts time values in the dictionary - list: converts time values in each dictionary item - DataFrame: converts time columns

MT5 epoch timestamps are trade-server wall-clock labels, so the converted datetimes are timezone-naive and represent server time, not UTC.

Parameters:

Name Type Description Default
skip_toggle str | None

Name of the bound parameter that skips conversion when its value is truthy.

 None

Returns:

Type Description
Callable[[Callable[P, R]], Callable[P, R]]

Decorator function.
Source code in pdmt5/utils.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator to convert time values/columns to datetime based on result type.

    Automatically detects result type and applies appropriate time conversion:
    - dict: converts time values in the dictionary
    - list: converts time values in each dictionary item
    - DataFrame: converts time columns

    MT5 epoch timestamps are trade-server wall-clock labels, so the converted
    datetimes are timezone-naive and represent server time, not UTC.

    Args:
        skip_toggle: Name of the bound parameter that skips conversion when
            its value is truthy.

    Returns:
        Decorator function.
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        signature = inspect.signature(func)

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            bound_arguments = signature.bind_partial(*args, **kwargs).arguments
            result = func(*args, **kwargs)
            if skip_toggle and bound_arguments.get(skip_toggle):
                return result
            elif isinstance(result, dict):
                return cast(
                    "R",
                    _convert_time_values_in_dict(
                        dictionary=cast("dict[str, Any]", result)
                    ),
                )
            elif isinstance(result, list):
                return [
                    (
                        _convert_time_values_in_dict(
                            dictionary=cast("dict[str, Any]", d)
                        )
                        if isinstance(d, dict)
                        else d
                    )
                    for d in cast("list[Any]", result)
                ]  # type: ignore[return-value]
            elif isinstance(result, pd.DataFrame):
                return cast("R", _convert_time_columns_in_df(result))
            else:
                return result

        return wrapper

    return decorator

options: show_bases: false

Decorator that automatically converts time values to datetime objects based on the result type.

set_index_if_possible

pdmt5.utils.set_index_if_possible 

set_index_if_possible(
    index_parameters: str | None = None,
) -> Callable[
    [Callable[P, DataFrame]], Callable[P, DataFrame]
]

Decorator to set index on DataFrame results if not empty.

Parameters:

Name Type Description Default
index_parameters str | None

Name of the parameter to use as index if provided.

 None

Returns:

Type Description
Callable[[Callable[P, DataFrame]], Callable[P, DataFrame]]

Decorator function.
Source code in pdmt5/utils.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def set_index_if_possible(
    index_parameters: str | None = None,
) -> Callable[[Callable[P, pd.DataFrame]], Callable[P, pd.DataFrame]]:
    """Decorator to set index on DataFrame results if not empty.

    Args:
        index_parameters: Name of the parameter to use as index if provided.

    Returns:
        Decorator function.
    """

    def decorator(
        func: Callable[P, pd.DataFrame],
    ) -> Callable[P, pd.DataFrame]:
        signature = inspect.signature(func)

        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> pd.DataFrame:
            bound_arguments = signature.bind_partial(*args, **kwargs).arguments
            result = func(*args, **kwargs)
            if not isinstance(result, pd.DataFrame):  # type: ignore[reportUnnecessaryIsInstance]
                error_message = (
                    f"Function {func.__name__} returned non-DataFrame result: "
                    f"{type(result).__name__}. Expected DataFrame."
                )
                raise TypeError(error_message)
            elif (
                index_parameters
                and bound_arguments.get(index_parameters)
                and not result.empty
            ):
                return result.set_index(bound_arguments[index_parameters])
            else:
                return result

        return wrapper

    return decorator

options: show_bases: false

Decorator that sets DataFrame index if specified and the DataFrame is not empty.

Internal Functions

These functions are used internally by the decorators:

_convert_time_values_in_dict

Converts Unix timestamp values in dictionaries to pandas datetime objects. Handles both seconds and milliseconds based on field naming conventions.

_convert_time_columns_in_df

Converts time columns in pandas DataFrames to datetime format. Automatically detects time columns by name patterns.

Usage

These utilities are primarily used internally by Mt5DataClient methods through decorators:

# Example of how decorators are applied internally
@set_index_if_possible(index_parameters="index_keys")
@detect_and_convert_time_to_datetime(skip_toggle="skip_to_datetime")
def some_method(
    self,
    skip_to_datetime: bool = False,
    index_keys: str | None = None,
):
    # Method implementation
    pass

set_index_if_possible must stay the outermost decorator: time conversion only touches DataFrame columns, so the index has to be set after the conversion has run. Reversing the order would move the raw epoch column into the index before conversion, leaving an unconverted integer index.

Time Conversion Rules

The time conversion follows these rules:

  1. Millisecond Timestamps: Fields starting with time_ and ending with _msc are converted using pd.to_datetime(value, unit="ms")
  2. Second Timestamps: Fields named time or starting with time_ are converted using pd.to_datetime(value, unit="s")
  3. Automatic Detection: Conversion happens automatically unless explicitly disabled
  4. Server Time, Not UTC: MT5 epochs are trade-server wall-clock labels (typically UTC+2 or UTC+3); the converted datetimes are timezone-naive and preserve server time — they are not UTC

DataFrame Index Setting

The index setting decorator:

  1. Only sets index if DataFrame is not empty
  2. Uses the specified column name as index
  3. Validates that the result is a DataFrame
  4. Preserves original DataFrame if no index column specified

Integration with Mt5DataClient

These utilities enable Mt5DataClient to provide user-friendly data:

# Time values are automatically converted
df = client.symbols_get_as_df(skip_to_datetime=False)
# Returns DataFrame with datetime objects instead of Unix timestamps

# DataFrames can have custom indexes
df = client.copy_rates_from_as_df(
    symbol="EURUSD",
    timeframe=mt5.TIMEFRAME_H1,
    date_from=datetime.now(),
    count=100,
    index_keys="time",  # Sets 'time' column as index
)

Design Philosophy

The utilities module follows these principles:

  1. Automatic Conversion: Time values should be human-readable by default
  2. Opt-out Capability: Users can disable conversion when needed
  3. Type Safety: Decorators validate input and output types
  4. Non-intrusive: Original data is never modified, only copies
  5. Consistent Behavior: Same conversion rules across all methods

Implementation Details

Time Detection Patterns

  • time: Base time field (seconds)
  • time_*: Time-prefixed fields (seconds)
  • time_*_msc: Millisecond timestamp fields

Error Handling

  • Invalid DataFrame types raise TypeError
  • Empty DataFrames are returned as-is
  • Non-time fields are ignored during conversion

Performance Considerations

  • Conversions run by default; opt out per call with skip_to_datetime=True
  • Dictionary operations use shallow copies
  • DataFrame operations use efficient pandas methods
  • Decorators add minimal overhead