Converters¶

def ensure_utc(value: datetime | str) -> datetime:
    """Return a timezone-aware UTC datetime.

    Args:
        value: Datetime instance or ISO 8601 string.

    Returns:
        UTC-aware datetime.
    """
    if isinstance(value, str):
        return parse_datetime(value)
    if value.tzinfo is None:
        return value.replace(tzinfo=UTC)
    return value.astimezone(UTC)

def granularity_name(timeframe: int | str) -> str:
    """Return a short granularity label for a timeframe integer or name.

    Args:
        timeframe: MT5 timeframe as integer or name (for example ``M1``).

    Returns:
        Short name such as ``M1`` or the stringified integer when unknown.
    """
    tf = parse_timeframe(timeframe)
    try:
        name = _get_timeframe_name(tf)
    except ValueError:
        return str(tf)
    return name.removeprefix("TIMEFRAME_")

def normalize_symbol(symbol: str) -> str:
    """Normalize a broker symbol name for MT5 API calls.

    Strips surrounding whitespace while preserving broker-specific casing and
    suffixes (for example ``XAUUSDm``, ``US500.cash``, or ``EURUSD.r``).

    Args:
        symbol: Raw symbol name.

    Returns:
        Normalized symbol string.

    Raises:
        ValueError: If the symbol is empty after normalization.
    """
    normalized = symbol.strip()
    if not normalized:
        msg = "Symbol must not be empty."
        raise ValueError(msg)
    return normalized

def normalize_symbols(symbols: Sequence[str]) -> list[str]:
    """Normalize a sequence of broker symbol names.

    Args:
        symbols: Raw symbol names.

    Returns:
        List of normalized, de-duplicated symbols preserving first-seen order.
    """
    seen: set[str] = set()
    resolved: list[str] = []
    for symbol in symbols:
        normalized = normalize_symbol(symbol)
        if normalized not in seen:
            seen.add(normalized)
            resolved.append(normalized)
    return resolved

def parse_date_range(
    date_from: datetime | str,
    date_to: datetime | str,
) -> tuple[datetime, datetime]:
    """Parse and validate an inclusive UTC date range.

    Args:
        date_from: Range start as datetime or ISO 8601 string.
        date_to: Range end as datetime or ISO 8601 string.

    Returns:
        Tuple of UTC-aware ``(start, end)`` datetimes.

    Raises:
        ValueError: If ``date_from`` is after ``date_to``.
    """
    start = ensure_utc(date_from)
    end = ensure_utc(date_to)
    if start > end:
        msg = (
            f"date_from ({start.isoformat()}) must not be after "
            f"date_to ({end.isoformat()})."
        )
        raise ValueError(msg)
    return start, end

def parse_datetime(value: str) -> datetime:
    """Parse an ISO 8601 datetime string to a timezone-aware datetime.

    Args:
        value: ISO 8601 datetime string (e.g., '2024-01-01' or
            '2024-01-01T12:00:00+00:00').

    Returns:
        Parsed datetime with UTC timezone if no timezone is specified.

    Raises:
        ValueError: If the string cannot be parsed.
    """
    try:
        dt = datetime.fromisoformat(value)
    except ValueError:
        msg = f"Invalid datetime format: '{value}'. Use ISO 8601 format."
        raise ValueError(msg) from None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt

def parse_tick_flags(value: object) -> int:
    """Parse tick flags string or integer value.

    Args:
        value: Tick flag name (ALL, INFO, TRADE, COPY_TICKS_*) or integer value.

    Returns:
        Integer tick flag value compatible with MetaTrader 5 ``COPY_TICKS_*``.

    Raises:
        ValueError: If the flag is invalid.
    """
    try:
        return _parse_copy_ticks(value)
    except ValueError:
        display = value if isinstance(value, str) else repr(value)
        valid = ", ".join(_TICK_FLAG_NAMES)
        msg = (
            f"Invalid tick flags: '{display}'. "
            f"Use one of: {valid}, or a supported integer."
        )
        raise ValueError(msg) from None

def parse_timeframe(value: object) -> int:
    """Parse a timeframe string or integer value.

    Args:
        value: Timeframe name (e.g., 'M1', 'H1', 'D1') or integer value.

    Returns:
        Integer timeframe value.

    Raises:
        ValueError: If the timeframe is invalid.
    """
    try:
        return _parse_timeframe(value)
    except ValueError:
        display = value if isinstance(value, str) else repr(value)
        valid = ", ".join(TIMEFRAME_NAMES)
        msg = (
            f"Invalid timeframe: '{display}'. "
            f"Use one of: {valid}, or a supported integer."
        )
        raise ValueError(msg) from None

def recent_window(
    *,
    hours: float | None = None,
    seconds: float | None = None,
    date_to: datetime | str | None = None,
) -> tuple[datetime, datetime]:
    """Build a trailing UTC window ending at ``date_to`` or now.

    Exactly one of ``hours`` or ``seconds`` must be provided.

    Args:
        hours: Trailing window length in hours.
        seconds: Trailing window length in seconds.
        date_to: Window end. Defaults to current UTC time.

    Returns:
        Tuple of UTC-aware ``(start, end)`` datetimes.

    Raises:
        ValueError: If neither or both window lengths are provided, or if a
            length is not positive.
    """
    if (hours is None) == (seconds is None):
        msg = "Provide exactly one of hours or seconds."
        raise ValueError(msg)
    if hours is not None:
        length = timedelta(hours=hours)
    else:
        length = timedelta(seconds=seconds if seconds is not None else 0)
    if length.total_seconds() <= 0:
        msg = "Window length must be positive."
        raise ValueError(msg)
    end = ensure_utc(date_to) if date_to is not None else datetime.now(UTC)
    return end - length, end