CLI Module¶

@app.command()
def account_info(ctx: typer.Context) -> None:
    """Export account information."""
    _export_command(ctx, lambda client: client.account_info())

@app.command()
def close_positions(
    ctx: typer.Context,
    symbol: Annotated[
        list[str] | None,
        typer.Option(
            "--symbol",
            "-s",
            help="Symbol to close (repeat for multiple symbols).",
        ),
    ] = None,
    ticket: Annotated[
        list[int] | None,
        typer.Option(
            "--ticket",
            "-t",
            help="Position ticket to close (repeat for multiple tickets).",
        ),
    ] = None,
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="Preview close orders without executing them."),
    ] = False,
    yes: Annotated[
        bool,
        typer.Option("--yes", help="Confirm live position closing."),
    ] = False,
) -> None:
    """Close open positions by symbol or ticket.

    Delegates to :func:`mt5cli.trading.close_open_positions`. At least one
    ``--symbol`` or ``--ticket`` must be provided to avoid accidentally closing
    all positions. Use ``--dry-run`` to preview without executing; ``--yes`` is
    required for live execution.

    ``order-send`` is the expert raw-request path. ``close-positions`` is the
    safer high-level helper that builds correct close requests automatically.

    Raises:
        typer.BadParameter: If neither ``--symbol`` nor ``--ticket`` is given,
            or if ``--yes`` is missing for a live (non-dry-run) run.
    """
    if not symbol and not ticket:
        msg = "Provide at least one --symbol or --ticket to close positions."
        raise typer.BadParameter(msg)
    if not dry_run and not yes:
        msg = "Pass --yes to close live positions."
        raise typer.BadParameter(msg, param_hint="--yes")
    export_ctx = _get_export_context(ctx)
    client = create_trading_client(config=export_ctx.config)
    try:
        results = close_open_positions(
            client,
            symbols=list(symbol) if symbol else None,
            tickets=list(ticket) if ticket else None,
            dry_run=dry_run,
        )
    finally:
        client.shutdown()
    df = _execution_results_to_df(results)
    _execute_export(ctx, lambda: df)

@app.command()
def collect_history(
    ctx: typer.Context,
    symbol: Annotated[
        list[str],
        typer.Option(
            "--symbol",
            "-s",
            help="Symbol to collect (repeat for multiple symbols).",
        ),
    ],
    date_from: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="Start date."),
    ],
    date_to: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="End date."),
    ],
    dataset: Annotated[
        list[Dataset] | None,
        typer.Option(
            "--dataset",
            help=(
                "Dataset to include (repeat for multiple)."
                " Defaults to all: rates, ticks, history-orders, history-deals."
            ),
        ),
    ] = None,
    timeframe: Annotated[
        int,
        typer.Option(
            click_type=TIMEFRAME_TYPE,
            help="Rates timeframe (e.g., M1, H1, D1).",
        ),
    ] = 1,
    flags: Annotated[
        int,
        typer.Option(
            click_type=TICK_FLAGS_TYPE,
            help="Tick copy flags (ALL, INFO, TRADE, or integer).",
        ),
    ] = "ALL",  # pyright: ignore[reportArgumentType]
    if_exists: Annotated[
        IfExists,
        typer.Option(
            "--if-exists",
            help="Behavior when a target table already exists.",
        ),
    ] = IfExists.FAIL,
    with_views: Annotated[
        bool,
        typer.Option(
            "--with-views",
            help=(
                "Add cash_events and positions_reconstructed SQLite views"
                " derived from history_deals."
            ),
        ),
    ] = False,
) -> None:
    """Collect historical datasets into a single SQLite database.

    Tables written depend on ``--dataset``: ``rates``, ``ticks``,
    ``history_orders``, ``history_deals``. History datasets are fetched per
    symbol and concatenated. Rates rows carry the requested ``timeframe`` so
    appended runs at different timeframes remain distinguishable.

    With ``--with-views`` (requires the ``history-deals`` dataset), optional
    views ``cash_events`` and ``positions_reconstructed`` are derived from
    ``history_deals`` when the required columns are present.

    Raises:
        typer.BadParameter: If the output format is not SQLite3.
    """
    export_ctx = _get_export_context(ctx)
    if export_ctx.output_format != "sqlite3":
        msg = (
            "collect-history requires SQLite3 output."
            " Use a .db/.sqlite/.sqlite3 extension or --format sqlite3."
        )
        raise typer.BadParameter(msg)
    datasets = set(dataset) if dataset else set(Dataset)
    sdk.collect_history(
        output=export_ctx.output,
        symbols=symbol,
        date_from=date_from,
        date_to=date_to,
        datasets=datasets,
        timeframe=timeframe,
        flags=flags,
        if_exists=if_exists,
        with_views=with_views,
        config=export_ctx.config,
    )

@app.command()
def history_deals(
    ctx: typer.Context,
    date_from: Annotated[
        datetime | None,
        typer.Option(click_type=DATETIME_TYPE, help="Start date."),
    ] = None,
    date_to: Annotated[
        datetime | None,
        typer.Option(click_type=DATETIME_TYPE, help="End date."),
    ] = None,
    group: Annotated[str | None, typer.Option(help="Group filter.")] = None,
    symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None,
    ticket: Annotated[int | None, typer.Option(help="Order ticket.")] = None,
    position: Annotated[int | None, typer.Option(help="Position ticket.")] = None,
) -> None:
    """Export historical deals."""
    _export_command(
        ctx,
        lambda client: client.history_deals(
            date_from=date_from,
            date_to=date_to,
            group=group,
            symbol=symbol,
            ticket=ticket,
            position=position,
        ),
    )

@app.command()
def history_orders(
    ctx: typer.Context,
    date_from: Annotated[
        datetime | None,
        typer.Option(click_type=DATETIME_TYPE, help="Start date."),
    ] = None,
    date_to: Annotated[
        datetime | None,
        typer.Option(click_type=DATETIME_TYPE, help="End date."),
    ] = None,
    group: Annotated[str | None, typer.Option(help="Group filter.")] = None,
    symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None,
    ticket: Annotated[int | None, typer.Option(help="Order ticket.")] = None,
    position: Annotated[int | None, typer.Option(help="Position ticket.")] = None,
) -> None:
    """Export historical orders."""
    _export_command(
        ctx,
        lambda client: client.history_orders(
            date_from=date_from,
            date_to=date_to,
            group=group,
            symbol=symbol,
            ticket=ticket,
            position=position,
        ),
    )

@app.command()
def last_error(ctx: typer.Context) -> None:
    """Export the last error information."""
    _export_command(ctx, lambda client: client.last_error())

@app.command()
def latest_rates(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    timeframe: Annotated[
        int,
        typer.Option(
            click_type=TIMEFRAME_TYPE,
            help="Timeframe.",
        ),
    ],
    count: Annotated[int, typer.Option(help="Number of records.")],
    start_pos: Annotated[
        int,
        typer.Option(help="Start position (0 = current bar)."),
    ] = 0,
) -> None:
    """Export latest rates from a start position."""
    _export_command(
        ctx,
        lambda client: client.latest_rates(
            symbol,
            timeframe,
            count,
            start_pos=start_pos,
        ),
    )

def main() -> None:
    """Run the mt5cli CLI."""
    app()

@app.command()
def market_book(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
) -> None:
    """Export market depth (order book) for a symbol."""
    _export_command(ctx, lambda client: client.market_book(symbol))

@app.command()
def minimum_margins(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
) -> None:
    """Export minimum-volume buy and sell margin requirements."""
    _export_command(ctx, lambda client: client.minimum_margins(symbol))

@app.command()
def mt5_summary(ctx: typer.Context) -> None:
    """Export a compact terminal/account status summary."""
    _export_command(ctx, lambda client: client.mt5_summary_as_df())

@app.command()
def order_check(
    ctx: typer.Context,
    request: Annotated[
        dict[str, Any],
        typer.Option(click_type=REQUEST_TYPE, help=_REQUEST_OPTION_HELP),
    ],
) -> None:
    """Check funds sufficiency for a trading operation."""
    _export_command(ctx, lambda client: client.order_check(request))

@app.command()
def order_send(
    ctx: typer.Context,
    request: Annotated[
        dict[str, Any],
        typer.Option(click_type=REQUEST_TYPE, help=_REQUEST_OPTION_HELP),
    ],
    yes: Annotated[
        bool,
        typer.Option("--yes", help="Confirm the live trade request."),
    ] = False,
) -> None:
    """Send a trading operation request to the trade server.

    Raises:
        typer.BadParameter: If --yes is not provided.
    """
    if not yes:
        msg = "Pass --yes to send a live trade request."
        raise typer.BadParameter(msg, param_hint="--yes")
    _export_command(ctx, lambda client: client.order_send(request))

@app.command()
def orders(
    ctx: typer.Context,
    symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None,
    group: Annotated[str | None, typer.Option(help="Group filter.")] = None,
    ticket: Annotated[int | None, typer.Option(help="Ticket filter.")] = None,
) -> None:
    """Export active orders."""
    _export_command(
        ctx,
        lambda client: client.orders(symbol=symbol, group=group, ticket=ticket),
    )

@app.command()
def positions(
    ctx: typer.Context,
    symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None,
    group: Annotated[str | None, typer.Option(help="Group filter.")] = None,
    ticket: Annotated[int | None, typer.Option(help="Ticket filter.")] = None,
) -> None:
    """Export open positions."""
    _export_command(
        ctx,
        lambda client: client.positions(symbol=symbol, group=group, ticket=ticket),
    )

@app.command()
def rates_from(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    timeframe: Annotated[
        int,
        typer.Option(
            click_type=TIMEFRAME_TYPE,
            help="Timeframe (e.g., M1, H1, D1, or integer).",
        ),
    ],
    date_from: Annotated[
        datetime,
        typer.Option(
            click_type=DATETIME_TYPE,
            help="Start date in ISO 8601 format.",
        ),
    ],
    count: Annotated[int, typer.Option(help="Number of records.")],
) -> None:
    """Export rates from a start date."""
    _export_command(
        ctx,
        lambda client: client.copy_rates_from(symbol, timeframe, date_from, count),
    )

@app.command()
def rates_from_pos(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    timeframe: Annotated[
        int,
        typer.Option(
            click_type=TIMEFRAME_TYPE,
            help="Timeframe.",
        ),
    ],
    start_pos: Annotated[int, typer.Option(help="Start position (0 = current bar).")],
    count: Annotated[int, typer.Option(help="Number of records.")],
) -> None:
    """Export rates from a start position."""
    _export_command(
        ctx,
        lambda client: client.copy_rates_from_pos(
            symbol,
            timeframe,
            start_pos,
            count,
        ),
    )

@app.command()
def rates_range(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    timeframe: Annotated[
        int,
        typer.Option(
            click_type=TIMEFRAME_TYPE,
            help="Timeframe.",
        ),
    ],
    date_from: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="Start date."),
    ],
    date_to: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="End date."),
    ],
) -> None:
    """Export rates for a date range."""
    _export_command(
        ctx,
        lambda client: client.copy_rates_range(symbol, timeframe, date_from, date_to),
    )

@app.command()
def recent_history_deals(
    ctx: typer.Context,
    hours: Annotated[float, typer.Option(help="Lookback window in hours.")],
    date_to: Annotated[
        datetime | None,
        typer.Option(click_type=DATETIME_TYPE, help="Window end date."),
    ] = None,
    group: Annotated[str | None, typer.Option(help="Group filter.")] = None,
    symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None,
) -> None:
    """Export historical deals from a recent trailing window."""
    _export_command(
        ctx,
        lambda client: client.recent_history_deals(
            hours,
            date_to=date_to,
            group=group,
            symbol=symbol,
        ),
    )

@app.command()
def symbol_info(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
) -> None:
    """Export symbol details."""
    _export_command(ctx, lambda client: client.symbol_info(symbol))

@app.command()
def symbol_info_tick(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
) -> None:
    """Export the last tick for a symbol."""
    _export_command(ctx, lambda client: client.symbol_info_tick(symbol))

@app.command()
def symbols(
    ctx: typer.Context,
    group: Annotated[
        str | None,
        typer.Option(help="Symbol group filter (e.g., *USD*)."),
    ] = None,
) -> None:
    """Export symbol list."""
    _export_command(ctx, lambda client: client.symbols(group=group))

@app.command()
def terminal_info(ctx: typer.Context) -> None:
    """Export terminal information."""
    _export_command(ctx, lambda client: client.terminal_info())

@app.command()
def ticks_from(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    date_from: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="Start date."),
    ],
    count: Annotated[int, typer.Option(help="Number of ticks.")],
    flags: Annotated[
        int,
        typer.Option(
            click_type=TICK_FLAGS_TYPE,
            help="Tick flags (ALL, INFO, TRADE, or integer).",
        ),
    ],
) -> None:
    """Export ticks from a start date."""
    _export_command(
        ctx,
        lambda client: client.copy_ticks_from(symbol, date_from, count, flags),
    )

@app.command()
def ticks_range(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    date_from: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="Start date."),
    ],
    date_to: Annotated[
        datetime,
        typer.Option(click_type=DATETIME_TYPE, help="End date."),
    ],
    flags: Annotated[
        int,
        typer.Option(click_type=TICK_FLAGS_TYPE, help="Tick flags."),
    ],
) -> None:
    """Export ticks for a date range."""
    _export_command(
        ctx,
        lambda client: client.copy_ticks_range(symbol, date_from, date_to, flags),
    )

@app.command()
def ticks_recent(
    ctx: typer.Context,
    symbol: Annotated[str, typer.Option(help="Symbol name.")],
    seconds: Annotated[
        float,
        typer.Option(help="Lookback window in seconds."),
    ],
    date_to: Annotated[
        datetime | None,
        typer.Option(click_type=DATETIME_TYPE, help="Window end date."),
    ] = None,
    count: Annotated[
        int,
        typer.Option(help="Maximum number of ticks to return."),
    ] = 10000,
    flags: Annotated[
        int,
        typer.Option(
            click_type=TICK_FLAGS_TYPE,
            help="Tick flags (ALL, INFO, TRADE, or integer).",
        ),
    ] = "ALL",  # pyright: ignore[reportArgumentType]
) -> None:
    """Export ticks from a recent time window."""
    _export_command(
        ctx,
        lambda client: client.recent_ticks(
            symbol,
            seconds,
            date_to=date_to,
            count=count,
            flags=flags,
        ),
    )

@app.command()
def version(ctx: typer.Context) -> None:
    """Export MetaTrader5 version information."""
    _export_command(ctx, lambda client: client.version())