| @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="Data / Export")
+def account_info(ctx: typer.Context) -> None:
+ """Export account information."""
+ _export_command(ctx, lambda client: client.account_info())
|
@@ -342,127 +360,161 @@ or if --yes is missing for a live (non-dry-run) run.
Source code in mt5cli/cli.py
- | @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,
- 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(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)
|
cash_events and positions_reconstructed are deri
Source code in mt5cli/cli.py
- | @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 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,
+ )
|
Source code in mt5cli/cli.py
- | @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="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)
|
Source code in mt5cli/cli.py
- | @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(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,
+ ),
+ )
+
|
+
+
+
+ history_gaps
+
+
+
+ history_gaps(
+ ctx: Context,
+ sqlite3_path: Annotated[
+ Path,
+ Option(
+ "--sqlite3",
+ help="Source SQLite history database to analyze.",
+ ),
+ ],
+ table: Annotated[
+ list[str] | None,
+ Option(
+ "--table",
+ help="Rate table or compatibility view to inspect (repeat for multiple).",
+ ),
+ ] = None,
+ granularity_seconds: Annotated[
+ int | None,
+ Option(
+ help="Explicit bar interval in seconds for custom tables/views."
+ ),
+ ] = None,
+ min_gap_intervals: Annotated[
+ int,
+ Option(
+ help="Minimum missing-bar count required to emit a gap row."
+ ),
+ ] = 1,
+) -> None
+
+
+
+
+ Export SQLite rate gaps without connecting to MT5.
+
+
+ Raises:
+
+
+
+ | Type |
+ Description |
+
+
+
+
+
+ BadParameter
+ |
+
+
+ If no compatible rate view is available and no
+explicit table is provided, or if granularity inference fails.
+ |
+
+
+
+
+
+
+ Source code in mt5cli/cli.py
+ | @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)
|
@@ -1115,61 +1372,61 @@ output. Does not connect to MetaTrader 5.
Source code in mt5cli/cli.py
- | @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 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,
+ ),
+ )
|
Source code in mt5cli/cli.py
- | @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 last_error(ctx: typer.Context) -> None:
+ """Export the last error information."""
+ _export_command(ctx, lambda client: client.last_error())
|
Source code in mt5cli/cli.py
- | @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,
- ),
- )
+ | @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,
+ ),
+ )
|
Source code in mt5cli/cli.py
- | def main() -> None:
- """Run the mt5cli CLI."""
- app()
+ | def main() -> None:
+ """Run the mt5cli CLI."""
+ app()
|
Source code in mt5cli/cli.py
- | @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 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))
|
Source code in mt5cli/cli.py
- | @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 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))
|
Source code in mt5cli/cli.py
- | @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 mt5_summary(ctx: typer.Context) -> None:
+ """Export a compact terminal/account status summary."""
+ _export_command(ctx, lambda client: client.mt5_summary_as_df())
|
Source code in mt5cli/cli.py
- | @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="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))
|
Source code in mt5cli/cli.py
- | @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="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))
|
Source code in mt5cli/cli.py
- | @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 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),
+ )
|
Source code in mt5cli/cli.py
- | @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 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),
+ )
|
Source code in mt5cli/cli.py
- | @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(
+ 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),
+ )
|
Source code in mt5cli/cli.py
- | @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_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,
+ ),
+ )
|
Source code in mt5cli/cli.py
- | @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 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),
+ )
|
Source code in mt5cli/cli.py
- | @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="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,
+ ),
+ )
|
Source code in mt5cli/cli.py
- | @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="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)
|
Source code in mt5cli/cli.py
- | @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(
+ ctx: typer.Context,
+ symbol: Annotated[str, typer.Option(help="Symbol name.")],
+) -> None:
+ """Export symbol details."""
+ _export_command(ctx, lambda client: client.symbol_info(symbol))
|
Source code in mt5cli/cli.py
- | @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))
+ | @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))
|
Source code in mt5cli/cli.py
- | @app.command(rich_help_panel="Data / Export")
-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(rich_help_panel="Data / Export")
+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))
|
Source code in mt5cli/cli.py
- | @app.command(rich_help_panel="Data / Export")
-def terminal_info(ctx: typer.Context) -> None:
- """Export terminal information."""
- _export_command(ctx, lambda client: client.terminal_info())
+ | @app.command(rich_help_panel="Data / Export")
+def terminal_info(ctx: typer.Context) -> None:
+ """Export terminal information."""
+ _export_command(ctx, lambda client: client.terminal_info())
|
Source code in mt5cli/cli.py
- | @app.command(rich_help_panel="Data / Export")
-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(rich_help_panel="Data / Export")
+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),
+ )
|
Source code in mt5cli/cli.py
- | @app.command(rich_help_panel="Data / Export")
-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(rich_help_panel="Data / Export")
+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),
+ )
|
Source code in mt5cli/cli.py
- | @app.command(rich_help_panel="Data / Export")
-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(rich_help_panel="Data / Export")
+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,
+ ),
+ )
|
Source code in mt5cli/cli.py
- | @app.command(rich_help_panel="Data / Export")
-def version(ctx: typer.Context) -> None:
- """Export MetaTrader5 version information."""
- _export_command(ctx, lambda client: client.version())
+ | @app.command(rich_help_panel="Data / Export")
+def version(ctx: typer.Context) -> None:
+ """Export MetaTrader5 version information."""
+ _export_command(ctx, lambda client: client.version())
|
Source code in mt5cli/sdk.py
- 427
-428
-429
+ | def __init__(
- self,
- *,
- path: str | None = None,
- login: int | None = None,
- password: str | None = None,
- server: str | None = None,
- timeout: int | None = None,
- retry_count: int = 3,
- config: Mt5Config | None = None,
- client: Mt5DataClient | None = None,
-) -> None:
- """Initialize the SDK client.
-
- Args:
- path: Path to MetaTrader5 terminal EXE file.
- login: Trading account login.
- password: Trading account password.
- server: Trading server name.
- timeout: Connection timeout in milliseconds.
- retry_count: Number of MT5 initialization retries for sessions
- opened by this client.
- config: Optional pre-built ``Mt5Config`` (overrides other args).
- client: Optional already-connected ``Mt5DataClient``. Injected
- clients are reused as-is and are not initialized or shut down.
- """
- self._config = config or build_config(
- path=path,
- login=login,
- password=password,
- server=server,
- timeout=timeout,
- )
- self._retry_count = retry_count
- self._client = client
- self._owns_client = client is None
+462
+463
+464
| def __init__(
+ self,
+ *,
+ path: str | None = None,
+ login: int | None = None,
+ password: str | None = None,
+ server: str | None = None,
+ timeout: int | None = None,
+ retry_count: int = 3,
+ config: Mt5Config | None = None,
+ client: Mt5DataClient | None = None,
+) -> None:
+ """Initialize the SDK client.
+
+ Args:
+ path: Path to MetaTrader5 terminal EXE file.
+ login: Trading account login.
+ password: Trading account password.
+ server: Trading server name.
+ timeout: Connection timeout in milliseconds.
+ retry_count: Number of MT5 initialization retries for sessions
+ opened by this client.
+ config: Optional pre-built ``Mt5Config`` (overrides other args).
+ client: Optional already-connected ``Mt5DataClient``. Injected
+ clients are reused as-is and are not initialized or shut down.
+ """
+ self._config = config or build_config(
+ path=path,
+ login=login,
+ password=password,
+ server=server,
+ timeout=timeout,
+ )
+ self._retry_count = retry_count
+ self._client = client
+ self._owns_client = client is None
|
path, login, password, and serve
Source code in mt5cli/sdk.py
- 319
-320
-321
+ | def build_config(
- *,
- path: str | None = None,
- login: int | str | None = None,
- password: str | None = None,
- server: str | None = None,
- timeout: int | None = None,
- allow_whole_dollar_env: bool = False,
-) -> Mt5Config:
- """Build an ``Mt5Config`` from optional connection parameters.
-
- Args:
- path: Optional terminal executable path.
- login: Optional trading account login. Integers are preserved. String
- values are coerced: empty or whitespace-only strings become
- ``None``; numeric strings such as ``"12345"`` are converted to
- ``int``; non-numeric strings raise ``ValueError``. When
- ``allow_whole_dollar_env=True``, ``$ENV_NAME`` and
- ``${ENV_NAME}`` placeholders are expanded before coercion.
- password: Optional trading account password.
- server: Optional trading server name.
- timeout: Optional connection timeout in milliseconds.
- allow_whole_dollar_env: When ``True``, string parameters that are
- exactly ``$ENV_NAME`` are expanded from the environment. Applies
- to ``path``, ``login``, ``password``, and ``server``. Default
- ``False`` preserves existing behavior.
-
- Returns:
- Configured ``Mt5Config`` instance.
- """
- if allow_whole_dollar_env:
- if path is not None:
- path = substitute_env_placeholders(path, allow_whole_dollar_env=True)
- if isinstance(login, str):
- login = substitute_env_placeholders(login, allow_whole_dollar_env=True)
- if password is not None:
- password = substitute_env_placeholders(
- password, allow_whole_dollar_env=True
- )
- if server is not None:
- server = substitute_env_placeholders(server, allow_whole_dollar_env=True)
- return Mt5Config(
- path=path,
- login=_coerce_login(login),
- password=password,
- server=server,
- timeout=timeout,
- )
+366
+367
+368
| def build_config(
+ *,
+ path: str | None = None,
+ login: int | str | None = None,
+ password: str | None = None,
+ server: str | None = None,
+ timeout: int | None = None,
+ allow_whole_dollar_env: bool = False,
+) -> Mt5Config:
+ """Build an ``Mt5Config`` from optional connection parameters.
+
+ Args:
+ path: Optional terminal executable path.
+ login: Optional trading account login. Integers are preserved. String
+ values are coerced: empty or whitespace-only strings become
+ ``None``; numeric strings such as ``"12345"`` are converted to
+ ``int``; non-numeric strings raise ``ValueError``. When
+ ``allow_whole_dollar_env=True``, ``$ENV_NAME`` and
+ ``${ENV_NAME}`` placeholders are expanded before coercion.
+ password: Optional trading account password.
+ server: Optional trading server name.
+ timeout: Optional connection timeout in milliseconds.
+ allow_whole_dollar_env: When ``True``, string parameters that are
+ exactly ``$ENV_NAME`` are expanded from the environment. Applies
+ to ``path``, ``login``, ``password``, and ``server``. Default
+ ``False`` preserves existing behavior.
+
+ Returns:
+ Configured ``Mt5Config`` instance.
+ """
+ if allow_whole_dollar_env:
+ if path is not None:
+ path = substitute_env_placeholders(path, allow_whole_dollar_env=True)
+ if isinstance(login, str):
+ login = substitute_env_placeholders(login, allow_whole_dollar_env=True)
+ if password is not None:
+ password = substitute_env_placeholders(
+ password, allow_whole_dollar_env=True
+ )
+ if server is not None:
+ server = substitute_env_placeholders(server, allow_whole_dollar_env=True)
+ return Mt5Config(
+ path=path,
+ login=_coerce_login(login),
+ password=password,
+ server=server,
+ timeout=timeout,
+ )
|
Source code in mt5cli/history.py
- | def __post_init__(self) -> None:
- """Normalize accepted timeframe aliases to the stored integer value."""
- if not isinstance(self.timeframe, int):
- object.__setattr__(self, "timeframe", parse_timeframe(self.timeframe))
+ | def __post_init__(self) -> None:
+ """Normalize accepted timeframe aliases to the stored integer value."""
+ if not isinstance(self.timeframe, int):
+ object.__setattr__(self, "timeframe", parse_timeframe(self.timeframe))
|
Source code in mt5cli/history.py
- | def append_dataframe(
- conn: sqlite3.Connection,
- frame: pd.DataFrame,
- table_name: str,
- if_exists: IfExists,
-) -> bool:
- """Append a DataFrame to SQLite when it has a schema.
-
- Returns:
- True if a table was written, False if the frame had no columns.
- """
- if len(frame.columns) == 0:
- logger.warning("Skipping %s: dataset returned no columns", table_name)
- return False
- writable = _canonicalize_sqlite_time_columns(frame)
- writable.to_sql( # type: ignore[reportUnknownMemberType]
- table_name,
- conn,
- if_exists=if_exists.value,
- index=False,
- chunksize=50_000,
- )
- return True
+ | def append_dataframe(
+ conn: sqlite3.Connection,
+ frame: pd.DataFrame,
+ table_name: str,
+ if_exists: IfExists,
+) -> bool:
+ """Append a DataFrame to SQLite when it has a schema.
+
+ Returns:
+ True if a table was written, False if the frame had no columns.
+ """
+ if len(frame.columns) == 0:
+ logger.warning("Skipping %s: dataset returned no columns", table_name)
+ return False
+ writable = _canonicalize_sqlite_time_columns(frame)
+ writable.to_sql( # type: ignore[reportUnknownMemberType]
+ table_name,
+ conn,
+ if_exists=if_exists.value,
+ index=False,
+ chunksize=50_000,
+ )
+ return True
|
Source code in mt5cli/history.py
- | def augment_written_columns_from_sqlite(
- conn: sqlite3.Connection,
- datasets: set[Dataset],
- written_columns: dict[Dataset, set[str]],
-) -> None:
- """Add existing table columns to the written column map."""
- for dataset in datasets:
- columns = get_table_columns(conn, dataset.table_name)
- if not columns:
- continue
- if dataset in written_columns:
- written_columns[dataset].update(columns)
- else:
- written_columns[dataset] = columns
+ | def augment_written_columns_from_sqlite(
+ conn: sqlite3.Connection,
+ datasets: set[Dataset],
+ written_columns: dict[Dataset, set[str]],
+) -> None:
+ """Add existing table columns to the written column map."""
+ for dataset in datasets:
+ columns = get_table_columns(conn, dataset.table_name)
+ if not columns:
+ continue
+ if dataset in written_columns:
+ written_columns[dataset].update(columns)
+ else:
+ written_columns[dataset] = columns
|
symbol=None for each timeframe instead of raising.
Source code in mt5cli/history.py
- | def build_rate_targets(
- symbols: Sequence[str],
- timeframes: Sequence[int | str],
- *,
- allow_missing_symbol: bool = False,
-) -> list[RateTarget]:
- """Build rate targets for every symbol and timeframe combination.
-
- Args:
- symbols: MT5 symbol names. May be empty when ``allow_missing_symbol``.
- timeframes: MT5 timeframes as integers or names (for example ``M1``).
- allow_missing_symbol: When True and ``symbols`` is empty, build targets
- with ``symbol=None`` for each timeframe instead of raising.
-
- Returns:
- Targets in row-major order: every timeframe for the first symbol, then
- every timeframe for the next symbol, and so on.
-
- Raises:
- ValueError: If ``timeframes`` is empty, or ``symbols`` is empty and
- ``allow_missing_symbol`` is False.
- """
- if not timeframes:
- msg = "At least one timeframe is required."
- raise ValueError(msg)
- if not symbols:
- if not allow_missing_symbol:
- msg = "At least one symbol is required."
- raise ValueError(msg)
- return [RateTarget(symbol=None, timeframe=tf) for tf in timeframes]
- return [
- RateTarget(symbol=symbol, timeframe=tf)
- for symbol in symbols
- for tf in timeframes
- ]
+ | def build_rate_targets(
+ symbols: Sequence[str],
+ timeframes: Sequence[int | str],
+ *,
+ allow_missing_symbol: bool = False,
+) -> list[RateTarget]:
+ """Build rate targets for every symbol and timeframe combination.
+
+ Args:
+ symbols: MT5 symbol names. May be empty when ``allow_missing_symbol``.
+ timeframes: MT5 timeframes as integers or names (for example ``M1``).
+ allow_missing_symbol: When True and ``symbols`` is empty, build targets
+ with ``symbol=None`` for each timeframe instead of raising.
+
+ Returns:
+ Targets in row-major order: every timeframe for the first symbol, then
+ every timeframe for the next symbol, and so on.
+
+ Raises:
+ ValueError: If ``timeframes`` is empty, or ``symbols`` is empty and
+ ``allow_missing_symbol`` is False.
+ """
+ if not timeframes:
+ msg = "At least one timeframe is required."
+ raise ValueError(msg)
+ if not symbols:
+ if not allow_missing_symbol:
+ msg = "At least one symbol is required."
+ raise ValueError(msg)
+ return [RateTarget(symbol=None, timeframe=tf) for tf in timeframes]
+ return [
+ RateTarget(symbol=symbol, timeframe=tf)
+ for symbol in symbols
+ for tf in timeframes
+ ]
|
EURUSD_M1 cannot collide with EURUSD
Source code in mt5cli/history.py
- | def build_rate_view_name(
- *,
- symbol: str,
- granularity: str,
- granularity_count: int,
- timeframe: int,
-) -> str:
- """Return a collision-free offline optimize view name.
-
- View names always include the timeframe integer after a ``__`` separator so
- a symbol such as ``EURUSD_M1`` cannot collide with ``EURUSD`` at timeframe
- ``M1``.
- """
- if granularity_count == 1:
- return f"rate_{symbol}__{timeframe}"
- return f"rate_{symbol}__{granularity}_{timeframe}"
+ | def build_rate_view_name(
+ *,
+ symbol: str,
+ granularity: str,
+ granularity_count: int,
+ timeframe: int,
+) -> str:
+ """Return a collision-free offline optimize view name.
+
+ View names always include the timeframe integer after a ``__`` separator so
+ a symbol such as ``EURUSD_M1`` cannot collide with ``EURUSD`` at timeframe
+ ``M1``.
+ """
+ if granularity_count == 1:
+ return f"rate_{symbol}__{timeframe}"
+ return f"rate_{symbol}__{granularity}_{timeframe}"
|
EURUSD_M1 cannot collide with EURUSD
Source code in mt5cli/history.py
- | def create_cash_events_view(
- conn: sqlite3.Connection,
- deals_columns: set[str],
-) -> bool:
- """Create the cash_events SQLite view derived from history_deals.
-
- Returns:
- True if the view was created, False if required columns are missing.
- """
- if "type" not in deals_columns:
- logger.warning("Skipping cash_events view: history_deals.type is missing")
- return False
- conn.execute("DROP VIEW IF EXISTS cash_events")
- conn.execute(
- "CREATE VIEW cash_events AS" # noqa: S608
- f" SELECT * FROM history_deals WHERE type NOT IN {_TRADE_DEAL_TYPES_SQL}",
- )
- return True
+ | def create_cash_events_view(
+ conn: sqlite3.Connection,
+ deals_columns: set[str],
+) -> bool:
+ """Create the cash_events SQLite view derived from history_deals.
+
+ Returns:
+ True if the view was created, False if required columns are missing.
+ """
+ if "type" not in deals_columns:
+ logger.warning("Skipping cash_events view: history_deals.type is missing")
+ return False
+ conn.execute("DROP VIEW IF EXISTS cash_events")
+ conn.execute(
+ "CREATE VIEW cash_events AS" # noqa: S608
+ f" SELECT * FROM history_deals WHERE type NOT IN {_TRADE_DEAL_TYPES_SQL}",
+ )
+ return True
|
EURUSD_M1 cannot collide with EURUSD
Source code in mt5cli/history.py
- | def create_history_indexes(
- conn: sqlite3.Connection,
- written_columns: dict[Dataset, set[str]],
-) -> None:
- """Create useful indexes for collected history tables when present."""
- if {"symbol", "timeframe", "time"}.issubset(
- written_columns.get(Dataset.rates, set()),
- ):
- conn.execute(
- "CREATE INDEX IF NOT EXISTS idx_rates_symbol_timeframe_time"
- " ON rates(symbol, timeframe, time)",
- )
- if {"symbol", "time"}.issubset(written_columns.get(Dataset.ticks, set())):
- conn.execute(
- "CREATE INDEX IF NOT EXISTS idx_ticks_symbol_time ON ticks(symbol, time)",
- )
- if {"position_id", "symbol"}.issubset(
- written_columns.get(Dataset.history_deals, set()),
- ):
- conn.execute(
- "CREATE INDEX IF NOT EXISTS idx_history_deals_position_symbol"
- " ON history_deals(position_id, symbol)",
- )
+ | def create_history_indexes(
+ conn: sqlite3.Connection,
+ written_columns: dict[Dataset, set[str]],
+) -> None:
+ """Create useful indexes for collected history tables when present."""
+ if {"symbol", "timeframe", "time"}.issubset(
+ written_columns.get(Dataset.rates, set()),
+ ):
+ conn.execute(
+ "CREATE INDEX IF NOT EXISTS idx_rates_symbol_timeframe_time"
+ " ON rates(symbol, timeframe, time)",
+ )
+ if {"symbol", "time"}.issubset(written_columns.get(Dataset.ticks, set())):
+ conn.execute(
+ "CREATE INDEX IF NOT EXISTS idx_ticks_symbol_time ON ticks(symbol, time)",
+ )
+ if {"position_id", "symbol"}.issubset(
+ written_columns.get(Dataset.history_deals, set()),
+ ):
+ conn.execute(
+ "CREATE INDEX IF NOT EXISTS idx_history_deals_position_symbol"
+ " ON history_deals(position_id, symbol)",
+ )
|
EURUSD_M1 cannot collide with EURUSD
Source code in mt5cli/history.py
- | def create_positions_reconstructed_view(
- conn: sqlite3.Connection,
- deals_columns: set[str],
-) -> bool:
- """Create the positions_reconstructed SQLite view derived from history_deals.
-
- Returns:
- True if the view was created, False if required columns are missing.
- """
- if not _POSITIONS_VIEW_REQUIRED_COLUMNS.issubset(deals_columns):
- missing = ", ".join(sorted(_POSITIONS_VIEW_REQUIRED_COLUMNS - deals_columns))
- logger.warning(
- "Skipping positions_reconstructed view: history_deals missing columns: %s",
- missing,
- )
- return False
- conn.execute("DROP VIEW IF EXISTS positions_reconstructed")
- conn.execute(
- "CREATE VIEW positions_reconstructed AS" # noqa: S608
- " SELECT"
- " position_id,"
- " symbol,"
- " MIN(CASE WHEN entry = 0 THEN time END) AS open_time,"
- " MAX(CASE WHEN entry IN (1, 2, 3) THEN time END) AS close_time,"
- " MIN(CASE WHEN entry = 0 THEN type END) AS direction,"
- " SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END) AS volume_open,"
- " SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END) AS volume_close,"
- " SUM(CASE WHEN entry = 2 THEN volume ELSE 0 END) AS volume_reversal,"
- " CASE"
- " WHEN SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END) > 0"
- " THEN SUM(CASE WHEN entry = 0 THEN price * volume ELSE 0 END)"
- " / SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END)"
- " END AS open_price,"
- " CASE"
- " WHEN SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END) > 0"
- " THEN SUM(CASE WHEN entry IN (1, 2, 3) THEN price * volume ELSE 0 END)"
- " / SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END)"
- " END AS close_price,"
- " SUM(profit) AS total_profit,"
- " SUM(CASE WHEN entry = 2 THEN 1 ELSE 0 END) AS reversal_count,"
- " COUNT(*) AS deals_count"
- " FROM history_deals"
- f" WHERE type IN {_TRADE_DEAL_TYPES_SQL} AND position_id != 0"
- " GROUP BY position_id, symbol"
- " HAVING SUM(CASE WHEN entry IN (1, 2, 3) THEN 1 ELSE 0 END) > 0",
- )
- return True
+ | def create_positions_reconstructed_view(
+ conn: sqlite3.Connection,
+ deals_columns: set[str],
+) -> bool:
+ """Create the positions_reconstructed SQLite view derived from history_deals.
+
+ Returns:
+ True if the view was created, False if required columns are missing.
+ """
+ if not _POSITIONS_VIEW_REQUIRED_COLUMNS.issubset(deals_columns):
+ missing = ", ".join(sorted(_POSITIONS_VIEW_REQUIRED_COLUMNS - deals_columns))
+ logger.warning(
+ "Skipping positions_reconstructed view: history_deals missing columns: %s",
+ missing,
+ )
+ return False
+ conn.execute("DROP VIEW IF EXISTS positions_reconstructed")
+ conn.execute(
+ "CREATE VIEW positions_reconstructed AS" # noqa: S608
+ " SELECT"
+ " position_id,"
+ " symbol,"
+ " MIN(CASE WHEN entry = 0 THEN time END) AS open_time,"
+ " MAX(CASE WHEN entry IN (1, 2, 3) THEN time END) AS close_time,"
+ " MIN(CASE WHEN entry = 0 THEN type END) AS direction,"
+ " SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END) AS volume_open,"
+ " SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END) AS volume_close,"
+ " SUM(CASE WHEN entry = 2 THEN volume ELSE 0 END) AS volume_reversal,"
+ " CASE"
+ " WHEN SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END) > 0"
+ " THEN SUM(CASE WHEN entry = 0 THEN price * volume ELSE 0 END)"
+ " / SUM(CASE WHEN entry = 0 THEN volume ELSE 0 END)"
+ " END AS open_price,"
+ " CASE"
+ " WHEN SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END) > 0"
+ " THEN SUM(CASE WHEN entry IN (1, 2, 3) THEN price * volume ELSE 0 END)"
+ " / SUM(CASE WHEN entry IN (1, 2, 3) THEN volume ELSE 0 END)"
+ " END AS close_price,"
+ " SUM(profit) AS total_profit,"
+ " SUM(CASE WHEN entry = 2 THEN 1 ELSE 0 END) AS reversal_count,"
+ " COUNT(*) AS deals_count"
+ " FROM history_deals"
+ f" WHERE type IN {_TRADE_DEAL_TYPES_SQL} AND position_id != 0"
+ " GROUP BY position_id, symbol"
+ " HAVING SUM(CASE WHEN entry IN (1, 2, 3) THEN 1 ELSE 0 END) > 0",
+ )
+ return True
|
EURUSD_M1 cannot collide with EURUSD
Source code in mt5cli/history.py
- | def create_rate_compatibility_views(conn: sqlite3.Connection) -> None:
- """Create rate compatibility views from the normalized rates table."""
- columns = get_table_columns(conn, Dataset.rates.table_name)
- if not {"symbol", "timeframe", "time"}.issubset(columns):
- return
- drop_rate_compatibility_views(conn)
- select_columns = sorted(columns - {"symbol", "timeframe"})
- quoted_columns = ", ".join(f'"{column}"' for column in select_columns)
- rows = conn.execute(
- "SELECT DISTINCT symbol, timeframe FROM rates ORDER BY symbol, timeframe",
- ).fetchall()
- timeframes_by_symbol: dict[str, list[int]] = {}
- for symbol, timeframe in rows:
- timeframes_by_symbol.setdefault(str(symbol), []).append(int(timeframe))
- for symbol, timeframes in timeframes_by_symbol.items():
- for timeframe in timeframes:
- granularity = resolve_granularity_name(timeframe)
- view_name = build_rate_view_name(
- symbol=symbol,
- granularity=granularity,
- granularity_count=len(timeframes),
- timeframe=timeframe,
- )
- quoted_view_name = quote_sqlite_identifier(view_name)
- escaped_symbol = symbol.replace("'", "''")
- conn.execute(
- f"CREATE VIEW {quoted_view_name} AS" # noqa: S608
- f" SELECT {quoted_columns} FROM rates"
- f" WHERE symbol = '{escaped_symbol}'"
- f" AND timeframe = {timeframe}",
- )
+ | def create_rate_compatibility_views(conn: sqlite3.Connection) -> None:
+ """Create rate compatibility views from the normalized rates table."""
+ columns = get_table_columns(conn, Dataset.rates.table_name)
+ if not {"symbol", "timeframe", "time"}.issubset(columns):
+ return
+ drop_rate_compatibility_views(conn)
+ select_columns = sorted(columns - {"symbol", "timeframe"})
+ quoted_columns = ", ".join(f'"{column}"' for column in select_columns)
+ rows = conn.execute(
+ "SELECT DISTINCT symbol, timeframe FROM rates ORDER BY symbol, timeframe",
+ ).fetchall()
+ timeframes_by_symbol: dict[str, list[int]] = {}
+ for symbol, timeframe in rows:
+ timeframes_by_symbol.setdefault(str(symbol), []).append(int(timeframe))
+ for symbol, timeframes in timeframes_by_symbol.items():
+ for timeframe in timeframes:
+ granularity = resolve_granularity_name(timeframe)
+ view_name = build_rate_view_name(
+ symbol=symbol,
+ granularity=granularity,
+ granularity_count=len(timeframes),
+ timeframe=timeframe,
+ )
+ quoted_view_name = quote_sqlite_identifier(view_name)
+ escaped_symbol = symbol.replace("'", "''")
+ conn.execute(
+ f"CREATE VIEW {quoted_view_name} AS" # noqa: S608
+ f" SELECT {quoted_columns} FROM rates"
+ f" WHERE symbol = '{escaped_symbol}'"
+ f" AND timeframe = {timeframe}",
+ )
|
Source code in mt5cli/history.py
- | def deduplicate_history_tables(
- conn: sqlite3.Connection,
- written_columns: dict[Dataset, set[str]],
- written_tables: set[Dataset],
- dedup_scopes: Mapping[Dataset, Sequence[DedupScope]] | None = None,
-) -> None:
- """Deduplicate appended history tables by stable identifiers.
-
- Scopes whose required columns are not present in the written table are
- skipped. If all scopes for a dataset are skipped, the table receives one
- unscoped deduplication pass instead.
- """
- cursor = conn.cursor()
- for dataset in written_tables:
- columns = written_columns.get(dataset, set())
- table = dataset.table_name
- keys = next(
- (
- candidate
- for candidate in _HISTORY_DEDUP_KEYS[dataset]
- if set(candidate).issubset(columns)
- ),
- None,
- )
- if keys is None:
- logger.warning(
- "Skipping %s deduplication: no supported key columns",
- table,
- )
- continue
- raw_scopes: Sequence[DedupScope] = (
- dedup_scopes.get(dataset, ()) if dedup_scopes else ()
- )
- scopes = [scope for scope in raw_scopes if scope.required_columns <= columns]
- if scopes:
- for scope in scopes:
- drop_duplicates_in_table(
- cursor,
- table,
- list(keys),
- keep="last",
- scope_where=scope.where,
- scope_params=scope.params,
- )
- continue
- drop_duplicates_in_table(cursor, table, list(keys), keep="last")
+ | def deduplicate_history_tables(
+ conn: sqlite3.Connection,
+ written_columns: dict[Dataset, set[str]],
+ written_tables: set[Dataset],
+ dedup_scopes: Mapping[Dataset, Sequence[DedupScope]] | None = None,
+) -> None:
+ """Deduplicate appended history tables by stable identifiers.
+
+ Scopes whose required columns are not present in the written table are
+ skipped. If all scopes for a dataset are skipped, the table receives one
+ unscoped deduplication pass instead.
+ """
+ cursor = conn.cursor()
+ for dataset in written_tables:
+ columns = written_columns.get(dataset, set())
+ table = dataset.table_name
+ keys = next(
+ (
+ candidate
+ for candidate in _HISTORY_DEDUP_KEYS[dataset]
+ if set(candidate).issubset(columns)
+ ),
+ None,
+ )
+ if keys is None:
+ logger.warning(
+ "Skipping %s deduplication: no supported key columns",
+ table,
+ )
+ continue
+ raw_scopes: Sequence[DedupScope] = (
+ dedup_scopes.get(dataset, ()) if dedup_scopes else ()
+ )
+ scopes = [scope for scope in raw_scopes if scope.required_columns <= columns]
+ if scopes:
+ for scope in scopes:
+ drop_duplicates_in_table(
+ cursor,
+ table,
+ list(keys),
+ keep="last",
+ scope_where=scope.where,
+ scope_params=scope.params,
+ )
+ continue
+ drop_duplicates_in_table(cursor, table, list(keys), keep="last")
|
Source code in mt5cli/history.py
- | def drop_duplicates_in_table(
- cursor: sqlite3.Cursor,
- table: str,
- ids: list[str],
- *,
- keep: Literal["first", "last"] = "last",
- scope_where: str | None = None,
- scope_params: tuple[object, ...] = (),
-) -> None:
- """Remove duplicate rows, keeping the first or last ROWID per key group.
-
- Raises:
- ValueError: If the table or column names are invalid.
- """
- if not table.isidentifier():
- msg = f"Invalid table name: {table}"
- raise ValueError(msg)
- if invalid := {column for column in ids if not column.isidentifier()}:
- msg = f"Invalid column names: {', '.join(sorted(invalid))}"
- raise ValueError(msg)
- ids_csv = ", ".join(_sqlite_dedup_key_expression(column) for column in ids)
- rowid_selector = "MIN" if keep == "first" else "MAX"
- prepared_scope_params = tuple(
- _require_serialized_sqlite_timestamp(value)
- if isinstance(value, datetime)
- else value
- for value in scope_params
- )
- if scope_where:
- delete_sql = (
- f"DELETE FROM {table} WHERE {scope_where} AND ROWID NOT IN" # noqa: S608
- f" (SELECT {rowid_selector}(ROWID) FROM {table} WHERE {scope_where}"
- f" GROUP BY {ids_csv})"
- )
- cursor.execute(delete_sql, prepared_scope_params + prepared_scope_params)
- return
- cursor.execute(
- f"DELETE FROM {table} WHERE ROWID NOT IN" # noqa: S608
- f" (SELECT {rowid_selector}(ROWID) FROM {table} GROUP BY {ids_csv})",
- )
+ | def drop_duplicates_in_table(
+ cursor: sqlite3.Cursor,
+ table: str,
+ ids: list[str],
+ *,
+ keep: Literal["first", "last"] = "last",
+ scope_where: str | None = None,
+ scope_params: tuple[object, ...] = (),
+) -> None:
+ """Remove duplicate rows, keeping the first or last ROWID per key group.
+
+ Raises:
+ ValueError: If the table or column names are invalid.
+ """
+ if not table.isidentifier():
+ msg = f"Invalid table name: {table}"
+ raise ValueError(msg)
+ if invalid := {column for column in ids if not column.isidentifier()}:
+ msg = f"Invalid column names: {', '.join(sorted(invalid))}"
+ raise ValueError(msg)
+ ids_csv = ", ".join(_sqlite_dedup_key_expression(column) for column in ids)
+ rowid_selector = "MIN" if keep == "first" else "MAX"
+ prepared_scope_params = tuple(
+ _require_serialized_sqlite_timestamp(value)
+ if isinstance(value, datetime)
+ else value
+ for value in scope_params
+ )
+ if scope_where:
+ delete_sql = (
+ f"DELETE FROM {table} WHERE {scope_where} AND ROWID NOT IN" # noqa: S608
+ f" (SELECT {rowid_selector}(ROWID) FROM {table} WHERE {scope_where}"
+ f" GROUP BY {ids_csv})"
+ )
+ cursor.execute(delete_sql, prepared_scope_params + prepared_scope_params)
+ return
+ cursor.execute(
+ f"DELETE FROM {table} WHERE ROWID NOT IN" # noqa: S608
+ f" (SELECT {rowid_selector}(ROWID) FROM {table} GROUP BY {ids_csv})",
+ )
|
Source code in mt5cli/history.py
- | def drop_forming_rate_bar(df_rate: pd.DataFrame) -> pd.DataFrame:
- """Return closed bars from chronologically ordered MT5 rate data.
-
- MetaTrader 5 ``copy_rates_from_pos(start_pos=0)`` includes the still-forming
- current bar as the last row. Slice it off so downstream logic only sees
- completed bars. Empty frames and single-row frames return empty results.
-
- Args:
- df_rate: Rate data ordered oldest-to-newest with the forming bar last.
-
- Returns:
- A new DataFrame with all rows except the last. Index and columns are
- preserved. The input frame is not modified.
- """
- return df_rate.iloc[:-1].copy()
+ | def drop_forming_rate_bar(df_rate: pd.DataFrame) -> pd.DataFrame:
+ """Return closed bars from chronologically ordered MT5 rate data.
+
+ MetaTrader 5 ``copy_rates_from_pos(start_pos=0)`` includes the still-forming
+ current bar as the last row. Slice it off so downstream logic only sees
+ completed bars. Empty frames and single-row frames return empty results.
+
+ Args:
+ df_rate: Rate data ordered oldest-to-newest with the forming bar last.
+
+ Returns:
+ A new DataFrame with all rows except the last. Index and columns are
+ preserved. The input frame is not modified.
+ """
+ return df_rate.iloc[:-1].copy()
|
Source code in mt5cli/history.py
- | def drop_rate_compatibility_views(conn: sqlite3.Connection) -> None:
- """Drop all mt5cli-managed ``rate_*`` compatibility views."""
- rows = conn.execute(
- "SELECT name FROM sqlite_master WHERE type = 'view' AND name GLOB 'rate_*'",
- ).fetchall()
- for (view_name,) in rows:
- quoted_view_name = quote_sqlite_identifier(str(view_name))
- conn.execute(f"DROP VIEW IF EXISTS {quoted_view_name}")
+ | def drop_rate_compatibility_views(conn: sqlite3.Connection) -> None:
+ """Drop all mt5cli-managed ``rate_*`` compatibility views."""
+ rows = conn.execute(
+ "SELECT name FROM sqlite_master WHERE type = 'view' AND name GLOB 'rate_*'",
+ ).fetchall()
+ for (view_name,) in rows:
+ quoted_view_name = quote_sqlite_identifier(str(view_name))
+ conn.execute(f"DROP VIEW IF EXISTS {quoted_view_name}")
|
Source code in mt5cli/history.py
- | def filter_incremental_history_deals_frame(
- frame: pd.DataFrame,
- symbols: Sequence[str],
- start_by_symbol: dict[str, datetime],
- account_event_start: datetime,
-) -> pd.DataFrame:
- """Filter incrementally fetched history_deals by symbol and event start times.
-
- Returns:
- Rows for selected symbols at or after each symbol start, plus account
- events at or after ``account_event_start``.
- """
- if frame.empty:
- return frame.copy()
- parsed_times = _frame_parsed_times(frame)
- time_valid = parsed_times.notna()
- account_event_mask = _history_deals_account_event_mask(frame)
- account_keep = account_event_mask & (parsed_times >= account_event_start)
- trade_keep = pd.Series(data=False, index=frame.index)
- if "symbol" in frame.columns:
- for symbol in symbols:
- trade_keep |= (
- (frame["symbol"] == symbol)
- & (parsed_times >= start_by_symbol[symbol])
- & ~account_event_mask
- )
- keep = (account_keep | trade_keep) & time_valid
- return frame.loc[keep].copy()
+ | def filter_incremental_history_deals_frame(
+ frame: pd.DataFrame,
+ symbols: Sequence[str],
+ start_by_symbol: dict[str, datetime],
+ account_event_start: datetime,
+) -> pd.DataFrame:
+ """Filter incrementally fetched history_deals by symbol and event start times.
+
+ Returns:
+ Rows for selected symbols at or after each symbol start, plus account
+ events at or after ``account_event_start``.
+ """
+ if frame.empty:
+ return frame.copy()
+ parsed_times = _frame_parsed_times(frame)
+ time_valid = parsed_times.notna()
+ account_event_mask = _history_deals_account_event_mask(frame)
+ account_keep = account_event_mask & (parsed_times >= account_event_start)
+ trade_keep = pd.Series(data=False, index=frame.index)
+ if "symbol" in frame.columns:
+ for symbol in symbols:
+ trade_keep |= (
+ (frame["symbol"] == symbol)
+ & (parsed_times >= start_by_symbol[symbol])
+ & ~account_event_mask
+ )
+ keep = (account_keep | trade_keep) & time_valid
+ return frame.loc[keep].copy()
|
Source code in mt5cli/history.py
- | def filter_trade_history_frame(
- frame: pd.DataFrame,
- symbols: Sequence[str],
- *,
- include_account_events: bool,
-) -> pd.DataFrame:
- """Filter trade history rows to selected symbols and account events.
-
- Returns:
- Filtered history rows.
- """
- if "symbol" not in frame.columns:
- return frame
- symbol_mask = frame["symbol"].isin(symbols)
- if not include_account_events:
- return frame.loc[symbol_mask].copy()
- account_event_mask = _history_deals_account_event_mask(frame)
- return frame.loc[symbol_mask | account_event_mask].copy()
+ | def filter_trade_history_frame(
+ frame: pd.DataFrame,
+ symbols: Sequence[str],
+ *,
+ include_account_events: bool,
+) -> pd.DataFrame:
+ """Filter trade history rows to selected symbols and account events.
+
+ Returns:
+ Filtered history rows.
+ """
+ if "symbol" not in frame.columns:
+ return frame
+ symbol_mask = frame["symbol"].isin(symbols)
+ if not include_account_events:
+ return frame.loc[symbol_mask].copy()
+ account_event_mask = _history_deals_account_event_mask(frame)
+ return frame.loc[symbol_mask | account_event_mask].copy()
|
Source code in mt5cli/history.py
- | def get_history_deals_account_event_start_datetime(
- conn: sqlite3.Connection,
- *,
- fallback_start: datetime,
-) -> datetime:
- """Return the next update start for account-level history_deals rows."""
- table = Dataset.history_deals.table_name
- columns = get_table_columns(conn, table)
- if "time" not in columns:
- return fallback_start
- if "type" in columns:
- where_clause = f"type NOT IN {_TRADE_DEAL_TYPES_SQL}"
- elif "symbol" in columns:
- where_clause = "symbol IS NULL OR symbol = ''"
- else:
- return fallback_start
- parsed = _load_latest_parseable_time(
- conn,
- table,
- where_clause=where_clause,
- )
- return parsed if parsed is not None else fallback_start
+ | def get_history_deals_account_event_start_datetime(
+ conn: sqlite3.Connection,
+ *,
+ fallback_start: datetime,
+) -> datetime:
+ """Return the next update start for account-level history_deals rows."""
+ table = Dataset.history_deals.table_name
+ columns = get_table_columns(conn, table)
+ if "time" not in columns:
+ return fallback_start
+ if "type" in columns:
+ where_clause = f"type NOT IN {_TRADE_DEAL_TYPES_SQL}"
+ elif "symbol" in columns:
+ where_clause = "symbol IS NULL OR symbol = ''"
+ else:
+ return fallback_start
+ parsed = _load_latest_parseable_time(
+ conn,
+ table,
+ where_clause=where_clause,
+ )
+ return parsed if parsed is not None else fallback_start
|
Source code in mt5cli/history.py
- | def get_incremental_start_datetime(
- conn: sqlite3.Connection,
- dataset: Dataset,
- *,
- symbol: str,
- timeframe: int | None,
- fallback_start: datetime,
-) -> datetime:
- """Return the next update start datetime from existing MAX(time)."""
- timeframes = [timeframe] if timeframe is not None else None
- starts = load_incremental_start_datetimes(
- conn,
- dataset,
- symbols=[symbol],
- timeframes=timeframes,
- fallback_start=fallback_start,
- )
- return starts[symbol, timeframe]
+
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|