Utils Module¶

def coerce_login(login: int | str | None) -> int | None:
    """Coerce a login value to int, treating empty strings as unset.

    Returns:
        Integer login, or None when unset or an empty string.
    """
    if login is None or isinstance(login, int):
        return login
    text = login.strip()
    if not text:
        return None
    return int(text)

def detect_format(
    output_path: Path,
    explicit_format: str | None = None,
) -> str:
    """Detect the output format from a file extension or explicit format string.

    Args:
        output_path: Path to the output file.
        explicit_format: Explicitly specified format, if any.

    Returns:
        The detected format string.

    Raises:
        ValueError: If the format cannot be determined.
    """
    if explicit_format is not None:
        return explicit_format
    suffix = output_path.suffix.lower()
    if suffix in _FORMAT_EXTENSIONS:
        return _FORMAT_EXTENSIONS[suffix]
    msg = (
        f"Cannot detect format from extension '{suffix}'."
        " Use --format to specify the output format."
    )
    raise ValueError(msg)

def export_dataframe(
    df: pd.DataFrame,
    output_path: Path,
    output_format: str,
    table_name: str = "data",
) -> None:
    """Export a pandas DataFrame to the specified file format.

    Args:
        df: DataFrame to export.
        output_path: Path to the output file.
        output_format: Output format (csv, json, parquet, or sqlite3).
        table_name: Table name for SQLite3 output.

    Raises:
        ValueError: If the output format is not supported.
    """
    if output_format == "csv":
        df.to_csv(output_path, index=False)
    elif output_format == "json":
        df.to_json(
            output_path,
            orient="records",
            date_format="iso",
            indent=2,
        )
    elif output_format == "parquet":
        df.to_parquet(output_path, index=False)
    elif output_format == "sqlite3":
        export_dataframe_to_sqlite(
            df,
            output_path,
            table_name,
            if_exists=IfExists.REPLACE,
            index=False,
        )
    else:
        msg = f"Unsupported output format: {output_format}"
        raise ValueError(msg)

def export_dataframe_to_sqlite(
    df: pd.DataFrame,
    output_path: Path,
    table_name: str = "data",
    *,
    if_exists: IfExists = IfExists.APPEND,
    index: bool = False,
    index_label: str | None = None,
    deduplicate_on: Sequence[str] | None = None,
) -> None:
    """Write a DataFrame to SQLite with configurable append and deduplication.

    Args:
        df: DataFrame to export.
        output_path: SQLite database path.
        table_name: Target table name.
        if_exists: Conflict behavior when the table already exists.
        index: Whether to write the DataFrame index as a column.
        index_label: Column name for the index when ``index=True``.
        deduplicate_on: Optional key columns to deduplicate after writing,
            keeping the latest ``ROWID`` per key group. Deduplication scans the
            full table, so repeated appends cost O(table size); index the key
            columns when appending frequently.
    """
    with sqlite3.connect(output_path) as conn:
        df.to_sql(  # type: ignore[reportUnknownMemberType]
            table_name,
            conn,
            if_exists=if_exists.value,
            index=index,
            index_label=index_label,
        )
        if deduplicate_on:
            from .history import drop_duplicates_in_table  # noqa: PLC0415

            drop_duplicates_in_table(
                conn.cursor(),
                table_name,
                list(deduplicate_on),
                keep="last",
            )
            conn.commit()

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_request(value: str) -> dict[str, Any]:
    """Parse a JSON-formatted order request string or file reference.

    Args:
        value: JSON object string, or '@path' to read JSON from a file.

    Returns:
        Parsed request dictionary.

    Raises:
        ValueError: If the request file cannot be read or the value is not a
            JSON object.
    """
    if value.startswith("@"):
        path = Path(value[1:])
        try:
            text = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError) as exc:
            msg = f"Failed to read JSON request file '{path}': {exc}"
            raise ValueError(msg) from exc
    else:
        text = value
    try:
        parsed: object = json.loads(text)
    except json.JSONDecodeError as exc:
        msg = f"Invalid JSON request: {exc}"
        raise ValueError(msg) from exc
    if not _is_request_dict(parsed):
        msg = "Order request must be a JSON object."
        raise ValueError(msg)
    return parsed

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