CLI Module¶

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

@app.command(rich_help_panel="Execution")
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,
    deviation: Annotated[
        int | None,
        typer.Option(help="Optional slippage/deviation for each close request."),
    ] = None,
    comment: Annotated[
        str | None,
        typer.Option(help="Optional comment attached to each close request."),
    ] = None,
    magic: Annotated[
        int | None,
        typer.Option(
            help="Optional magic tag for close requests and position filtering.",
        ),
    ] = 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,
            deviation=deviation,
            comment=comment,
            magic=magic,
            dry_run=dry_run,
        )
    finally:
        client.shutdown()
    df = _execution_results_to_df(results)
    _execute_export(ctx, lambda: df)

@app.command(rich_help_panel="Collection")
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 rates, history-orders, history-deals."
                " Ticks are opt-in: pass --dataset ticks to include them."
            ),
        ),
    ] = 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``, ``history_orders``,
    ``history_deals`` by default. ``ticks`` are opt-in: pass
    ``--dataset ticks`` to include them (tick data grows the database quickly).
    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 is not None else None
    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(rich_help_panel="Collection")
def grafana_schema(
    ctx: typer.Context,
    publish_copy: Annotated[
        Path | None,
        typer.Option(
            "--publish-copy",
            help=(
                "Publish a Grafana-ready SQLite copy to this path"
                " after schema creation."
            ),
        ),
    ] = None,
) -> None:
    """Create or refresh Grafana-ready views and indexes in a SQLite database.

    Idempotent — safe to run repeatedly on the same database. Requires SQLite
    output. Does not connect to MetaTrader 5.

    Raises:
        typer.BadParameter: If the output format is not SQLite3.
    """
    import sqlite3 as _sqlite3  # noqa: PLC0415

    from .grafana import (  # noqa: PLC0415
        create_snapshot_tables,
        ensure_grafana_schema,
        publish_grafana_copy,
    )

    export_ctx = _get_export_context(ctx)
    if export_ctx.output_format != "sqlite3":
        msg = (
            "grafana-schema requires SQLite3 output."
            " Use a .db/.sqlite/.sqlite3 extension or --format sqlite3."
        )
        raise typer.BadParameter(msg)
    with _sqlite3.connect(export_ctx.output) as conn:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA synchronous=NORMAL")
        create_snapshot_tables(conn)
        ensure_grafana_schema(conn)
    logger.info("Grafana schema applied to %s", export_ctx.output)
    if publish_copy is not None:
        publish_grafana_copy(export_ctx.output, publish_copy)
        logger.info("Grafana copy published to %s", publish_copy)

@app.command(rich_help_panel="Data / Export")
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("history-gaps", rich_help_panel="Collection")
def history_gaps(
    ctx: typer.Context,
    sqlite3_path: Annotated[
        Path,
        typer.Option(
            "--sqlite3",
            help="Source SQLite history database to analyze.",
        ),
    ],
    table: Annotated[
        list[str] | None,
        typer.Option(
            "--table",
            help="Rate table or compatibility view to inspect (repeat for multiple).",
        ),
    ] = None,
    granularity_seconds: Annotated[
        int | None,
        typer.Option(help="Explicit bar interval in seconds for custom tables/views."),
    ] = None,
    min_gap_intervals: Annotated[
        int,
        typer.Option(help="Minimum missing-bar count required to emit a gap row."),
    ] = 1,
) -> None:
    """Export SQLite rate gaps without connecting to MT5.

    Raises:
        typer.BadParameter: If no compatible rate view is available and no
            explicit table is provided, or if granularity inference fails.
    """
    with sqlite3.connect(sqlite3_path) as conn:
        tables = list(table) if table else _default_gap_tables(conn)
        if not tables:
            msg = (
                "No managed rate compatibility views found; pass --table for a rate "
                "table or view."
            )
            raise typer.BadParameter(msg, param_hint="--table")
        frames: list[pd.DataFrame] = []
        for table_name in tables:
            interval_seconds = (
                granularity_seconds or _infer_gap_table_granularity_seconds(table_name)
            )
            if interval_seconds is None:
                msg = (
                    f"Could not infer granularity for {table_name!r}; pass "
                    "--granularity-seconds."
                )
                raise typer.BadParameter(msg, param_hint="--granularity-seconds")
            frames.append(
                report_rate_gaps(
                    conn,
                    table_name,
                    granularity_seconds=interval_seconds,
                    min_gap_intervals=min_gap_intervals,
                )
            )
    df = (
        pd.concat(frames, ignore_index=True)
        if frames
        else pd.DataFrame(columns=["table"])
    )
    _execute_export(ctx, lambda: df)

@app.command(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
def last_error(ctx: typer.Context) -> None:
    """Export the last error information."""
    _export_command(ctx, lambda client: client.last_error())

@app.command(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Execution")
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 raw trade request to the trade server (expert path, live execution).

    Passes the request JSON directly to MT5 ``order_send``. This is the
    low-level expert path — it places real trades on the connected account
    with no additional validation beyond what MT5 itself performs. Use
    ``order-check`` first to validate funds sufficiency. Prefer
    ``close-positions`` for closing open positions. ``--yes`` is required.

    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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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(rich_help_panel="Collection")
def snapshot(
    ctx: typer.Context,
    symbol: Annotated[
        list[str] | None,
        typer.Option(
            "--symbol",
            "-s",
            help="Symbol filter for positions/orders (repeat for multiple).",
        ),
    ] = None,
    with_account: Annotated[
        bool,
        typer.Option("--with-account/--no-account", help="Snapshot account info."),
    ] = True,
    with_positions: Annotated[
        bool,
        typer.Option(
            "--with-positions/--no-positions", help="Snapshot open positions."
        ),
    ] = True,
    with_orders: Annotated[
        bool,
        typer.Option("--with-orders/--no-orders", help="Snapshot active orders."),
    ] = True,
    with_terminal: Annotated[
        bool,
        typer.Option("--with-terminal/--no-terminal", help="Snapshot terminal info."),
    ] = True,
    with_grafana_schema: Annotated[
        bool,
        typer.Option(
            "--with-grafana-schema/--no-grafana-schema",
            help="Ensure Grafana views and indexes exist.",
        ),
    ] = False,
    publish_copy: Annotated[
        Path | None,
        typer.Option(
            "--publish-copy",
            help=("Publish a Grafana-ready SQLite copy to this path after snapshot."),
        ),
    ] = None,
) -> None:
    """Snapshot current account, position, order, and terminal state into SQLite.

    Appends a timestamped snapshot row for each data type. Never places
    orders or modifies trading state.

    Raises:
        typer.BadParameter: If the output format is not SQLite3.
    """
    export_ctx = _get_export_context(ctx)
    if export_ctx.output_format != "sqlite3":
        msg = (
            "snapshot requires SQLite3 output."
            " Use a .db/.sqlite/.sqlite3 extension or --format sqlite3."
        )
        raise typer.BadParameter(msg)
    sdk.update_observability_with_config(
        output=export_ctx.output,
        config=export_ctx.config,
        symbols=list(symbol) if symbol else None,
        include_account=with_account,
        include_positions=with_positions,
        include_orders=with_orders,
        include_terminal=with_terminal,
        with_grafana_schema=with_grafana_schema,
    )
    logger.info("Snapshot written to %s", export_ctx.output)
    if publish_copy is not None:
        from .grafana import publish_grafana_copy  # noqa: PLC0415

        publish_grafana_copy(export_ctx.output, publish_copy)
        logger.info("Grafana copy published to %s", publish_copy)

@app.command(rich_help_panel="Data / Export")
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(rich_help_panel="Data / Export")
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))