Utilities

pdmt5.utils

Utility functions for the pdmt5 package.

detect_and_convert_time_to_datetime 

detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[..., Any]

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

Parameters:

Name Type Description Default
skip_toggle str | None

Name of the parameter to skip conversion if set to False.

 None

Returns:

Type Description
Callable[..., Any]

Decorator function.
Source code in pdmt5/utils.py 
14
15
16
17
18
19
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
def detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[..., Any]:
    """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

    Args:
        skip_toggle: Name of the parameter to skip conversion if set to False.

    Returns:
        Decorator function.
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
            result = func(*args, **kwargs)
            if skip_toggle and kwargs.get(skip_toggle):
                return result
            elif isinstance(result, dict):
                return _convert_time_values_in_dict(dictionary=result)
            elif isinstance(result, list):
                return [
                    (
                        _convert_time_values_in_dict(dictionary=d)
                        if isinstance(d, dict)
                        else d
                    )
                    for d in result
                ]
            elif isinstance(result, pd.DataFrame):
                return _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[..., Any]

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[..., Any]

Decorator function.
Source code in pdmt5/utils.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def set_index_if_possible(index_parameters: str | None = None) -> Callable[..., Any]:
    """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[..., Any]) -> Callable[..., Any]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
            result = func(*args, **kwargs)
            if not isinstance(result, pd.DataFrame):
                error_message = (
                    f"Function {func.__name__} returned non-DataFrame result: "
                    f"{type(result).__name__}. Expected DataFrame."
                )
                raise TypeError(error_message)
            elif index_parameters and kwargs.get(index_parameters) and not result.empty:
                return result.set_index(kwargs[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[..., Any]

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

Parameters:

Name Type Description Default
skip_toggle str | None

Name of the parameter to skip conversion if set to False.

 None

Returns:

Type Description
Callable[..., Any]

Decorator function.
Source code in pdmt5/utils.py 
14
15
16
17
18
19
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
def detect_and_convert_time_to_datetime(
    skip_toggle: str | None = None,
) -> Callable[..., Any]:
    """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

    Args:
        skip_toggle: Name of the parameter to skip conversion if set to False.

    Returns:
        Decorator function.
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
            result = func(*args, **kwargs)
            if skip_toggle and kwargs.get(skip_toggle):
                return result
            elif isinstance(result, dict):
                return _convert_time_values_in_dict(dictionary=result)
            elif isinstance(result, list):
                return [
                    (
                        _convert_time_values_in_dict(dictionary=d)
                        if isinstance(d, dict)
                        else d
                    )
                    for d in result
                ]
            elif isinstance(result, pd.DataFrame):
                return _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[..., Any]

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[..., Any]

Decorator function.
Source code in pdmt5/utils.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def set_index_if_possible(index_parameters: str | None = None) -> Callable[..., Any]:
    """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[..., Any]) -> Callable[..., Any]:
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
            result = func(*args, **kwargs)
            if not isinstance(result, pd.DataFrame):
                error_message = (
                    f"Function {func.__name__} returned non-DataFrame result: "
                    f"{type(result).__name__}. Expected DataFrame."
                )
                raise TypeError(error_message)
            elif index_parameters and kwargs.get(index_parameters) and not result.empty:
                return result.set_index(kwargs[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
@detect_and_convert_time_to_datetime(skip_toggle="convert_time")
@set_index_if_possible(index_parameters="index_keys")
def some_method(self, convert_time: bool = True, index_keys: str | None = None):
    # Method implementation
    pass

Time Conversion Rules

The time conversion follows these rules:

  1. Millisecond Timestamps: Fields 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

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.get_symbols_as_df(convert_time=True)
# 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 only happen when requested
  • Dictionary operations use shallow copies
  • DataFrame operations use efficient pandas methods
  • Decorators add minimal overhead