diff --git a/mt5cli/__init__.py b/mt5cli/__init__.py new file mode 100644 index 0000000..c4f174e --- /dev/null +++ b/mt5cli/__init__.py @@ -0,0 +1,12 @@ +"""mt5cli: Command-line tool for MetaTrader 5.""" + +from importlib.metadata import version + +from .cli import detect_format, export_dataframe + +__version__ = version(__package__) if __package__ else None + +__all__ = [ + "detect_format", + "export_dataframe", +] diff --git a/mt5cli/__main__.py b/mt5cli/__main__.py new file mode 100644 index 0000000..08b5116 --- /dev/null +++ b/mt5cli/__main__.py @@ -0,0 +1,5 @@ +"""Entry point for running mt5cli as a module via ``python -m mt5cli``.""" + +from mt5cli.cli import main + +main() diff --git a/mt5cli/cli.py b/mt5cli/cli.py new file mode 100644 index 0000000..340f6a7 --- /dev/null +++ b/mt5cli/cli.py @@ -0,0 +1,753 @@ +"""Command-line interface for MetaTrader 5 data export.""" + +from __future__ import annotations + +import logging +import sqlite3 +from dataclasses import dataclass +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path # noqa: TC003 +from typing import TYPE_CHECKING, Annotated, cast + +import click +import typer + +from pdmt5 import Mt5Config, Mt5DataClient + +if TYPE_CHECKING: + from collections.abc import Callable + + import pandas as pd + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +TIMEFRAME_MAP: dict[str, int] = { + "M1": 1, + "M2": 2, + "M3": 3, + "M4": 4, + "M5": 5, + "M6": 6, + "M10": 10, + "M12": 12, + "M15": 15, + "M20": 20, + "M30": 30, + "H1": 16385, + "H2": 16386, + "H3": 16387, + "H4": 16388, + "H6": 16390, + "H8": 16392, + "H12": 16396, + "D1": 16408, + "W1": 32769, + "MN1": 49153, +} + +TICK_FLAG_MAP: dict[str, int] = { + "ALL": 1, + "INFO": 2, + "TRADE": 4, +} + +_FORMAT_EXTENSIONS: dict[str, str] = { + ".csv": "csv", + ".json": "json", + ".parquet": "parquet", + ".pq": "parquet", + ".db": "sqlite3", + ".sqlite": "sqlite3", + ".sqlite3": "sqlite3", +} + +# --------------------------------------------------------------------------- +# Enums +# --------------------------------------------------------------------------- + + +class OutputFormat(StrEnum): + """Supported output file formats.""" + + csv = "csv" + json = "json" + parquet = "parquet" + sqlite3 = "sqlite3" + + +class LogLevel(StrEnum): + """Logging verbosity levels.""" + + DEBUG = "DEBUG" + INFO = "INFO" + WARNING = "WARNING" + ERROR = "ERROR" + + +# --------------------------------------------------------------------------- +# Click parameter types +# --------------------------------------------------------------------------- + + +class _DateTimeType(click.ParamType): + """Click parameter type for ISO 8601 datetime strings.""" + + name = "DATETIME" + + def convert( + self, + value: object, + param: click.Parameter | None, + ctx: click.Context | None, + ) -> datetime: + """Convert a string value to a timezone-aware datetime. + + Args: + value: Raw value from the command line. + param: Click parameter instance. + ctx: Click context. + + Returns: + Parsed datetime. + """ + if isinstance(value, datetime): + return value + try: + return parse_datetime(str(value)) + except ValueError as exc: + self.fail(str(exc), param, ctx) + + +class _TimeframeType(click.ParamType): + """Click parameter type for MT5 timeframe values.""" + + name = "TIMEFRAME" + + def convert( + self, + value: object, + param: click.Parameter | None, + ctx: click.Context | None, + ) -> int: + """Convert a string or integer value to a timeframe integer. + + Args: + value: Raw value from the command line. + param: Click parameter instance. + ctx: Click context. + + Returns: + Integer timeframe value. + """ + if isinstance(value, int): + return value + try: + return parse_timeframe(str(value)) + except ValueError as exc: + self.fail(str(exc), param, ctx) + + +class _TickFlagsType(click.ParamType): + """Click parameter type for MT5 tick copy flags.""" + + name = "FLAGS" + + def convert( + self, + value: object, + param: click.Parameter | None, + ctx: click.Context | None, + ) -> int: + """Convert a string or integer value to a tick flags integer. + + Args: + value: Raw value from the command line. + param: Click parameter instance. + ctx: Click context. + + Returns: + Integer tick flag value. + """ + if isinstance(value, int): + return value + try: + return parse_tick_flags(str(value)) + except ValueError as exc: + self.fail(str(exc), param, ctx) + + +DATETIME_TYPE = _DateTimeType() +TIMEFRAME_TYPE = _TimeframeType() +TICK_FLAGS_TYPE = _TickFlagsType() + +# --------------------------------------------------------------------------- +# Export context +# --------------------------------------------------------------------------- + + +@dataclass +class _ExportContext: + """Shared context data passed from the callback to each subcommand.""" + + output: Path + output_format: str + table: str + config: Mt5Config + + +# --------------------------------------------------------------------------- +# Public utility functions +# --------------------------------------------------------------------------- + + +def detect_format( + output_path: Path, + explicit_format: str | None = None, +) -> str: + """Detect the output format from a file extension or explicit format string. + + Args: + output_path: Path to the output file. + explicit_format: Explicitly specified format, if any. + + Returns: + The detected format string. + + Raises: + ValueError: If the format cannot be determined. + """ + if explicit_format is not None: + return explicit_format + suffix = output_path.suffix.lower() + if suffix in _FORMAT_EXTENSIONS: + return _FORMAT_EXTENSIONS[suffix] + msg = ( + f"Cannot detect format from extension '{suffix}'." + " Use --format to specify the output format." + ) + raise ValueError(msg) + + +def export_dataframe( + df: pd.DataFrame, + output_path: Path, + output_format: str, + table_name: str = "data", +) -> None: + """Export a pandas DataFrame to the specified file format. + + Args: + df: DataFrame to export. + output_path: Path to the output file. + output_format: Output format (csv, json, parquet, or sqlite3). + table_name: Table name for SQLite3 output. + + Raises: + ValueError: If the output format is not supported. + """ + if output_format == "csv": + df.to_csv(output_path, index=False) + elif output_format == "json": + df.to_json( + output_path, + orient="records", + date_format="iso", + indent=2, + ) + elif output_format == "parquet": + df.to_parquet(output_path, index=False) + elif output_format == "sqlite3": + with sqlite3.connect(output_path) as conn: + df.to_sql( # type: ignore[reportUnknownMemberType] + table_name, + conn, + if_exists="replace", + index=False, + ) + else: + msg = f"Unsupported output format: {output_format}" + raise ValueError(msg) + + +def parse_datetime(value: str) -> datetime: + """Parse an ISO 8601 datetime string to a timezone-aware datetime. + + Args: + value: ISO 8601 datetime string (e.g., '2024-01-01' or + '2024-01-01T12:00:00+00:00'). + + Returns: + Parsed datetime with UTC timezone if no timezone is specified. + + Raises: + ValueError: If the string cannot be parsed. + """ + try: + dt = datetime.fromisoformat(value) + except ValueError: + msg = f"Invalid datetime format: '{value}'. Use ISO 8601 format." + raise ValueError(msg) from None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=UTC) + return dt + + +def parse_timeframe(value: str) -> int: + """Parse a timeframe string or integer value. + + Args: + value: Timeframe name (e.g., 'M1', 'H1', 'D1') or integer value. + + Returns: + Integer timeframe value. + + Raises: + ValueError: If the timeframe is invalid. + """ + upper = value.upper() + if upper in TIMEFRAME_MAP: + return TIMEFRAME_MAP[upper] + try: + return int(value) + except ValueError: + valid = ", ".join(TIMEFRAME_MAP) + msg = f"Invalid timeframe: '{value}'. Use one of: {valid}, or an integer." + raise ValueError(msg) from None + + +def parse_tick_flags(value: str) -> int: + """Parse tick flags string or integer value. + + Args: + value: Tick flag name (ALL, INFO, TRADE) or integer value. + + Returns: + Integer tick flag value. + + Raises: + ValueError: If the flag is invalid. + """ + upper = value.upper() + if upper in TICK_FLAG_MAP: + return TICK_FLAG_MAP[upper] + try: + return int(value) + except ValueError: + valid = ", ".join(TICK_FLAG_MAP) + msg = f"Invalid tick flags: '{value}'. Use one of: {valid}, or an integer." + raise ValueError(msg) from None + + +# --------------------------------------------------------------------------- +# Typer application +# --------------------------------------------------------------------------- + +app = typer.Typer( + name="mt5cli", + help="Export MetaTrader5 data to CSV, JSON, Parquet, or SQLite3.", +) + + +def _get_export_context(ctx: typer.Context) -> _ExportContext: + return cast("_ExportContext", ctx.obj) + + +def _execute_export( + ctx: typer.Context, + fetch_fn: Callable[[Mt5DataClient], pd.DataFrame], +) -> None: + """Execute the common connect-fetch-export-shutdown workflow. + + Args: + ctx: Typer context carrying shared options. + fetch_fn: Callable that receives a connected client and returns a + DataFrame. + """ + export_ctx = _get_export_context(ctx) + client = Mt5DataClient(config=export_ctx.config) + client.initialize_and_login_mt5() + try: + df = fetch_fn(client) + export_dataframe( + df=df, + output_path=export_ctx.output, + output_format=export_ctx.output_format, + table_name=export_ctx.table, + ) + logger.info( + "Exported %d rows to %s (%s)", + len(df), + export_ctx.output, + export_ctx.output_format, + ) + finally: + client.shutdown() + + +@app.callback() +def _callback( # pyright: ignore[reportUnusedFunction] + ctx: typer.Context, + output: Annotated[ + Path, + typer.Option("--output", "-o", help="Output file path."), + ], + fmt: Annotated[ + OutputFormat | None, + typer.Option( + "--format", + "-f", + help="Output format (auto-detected from extension if omitted).", + ), + ] = None, + table: Annotated[ + str, + typer.Option(help="Table name for SQLite3 output."), + ] = "data", + login: Annotated[ + int | None, + typer.Option(help="Trading account login."), + ] = None, + password: Annotated[ + str | None, + typer.Option(help="Trading account password."), + ] = None, + server: Annotated[ + str | None, + typer.Option(help="Trading server name."), + ] = None, + path: Annotated[ + str | None, + typer.Option(help="Path to MetaTrader5 terminal EXE file."), + ] = None, + timeout: Annotated[ + int | None, + typer.Option(help="Connection timeout in milliseconds."), + ] = None, + log_level: Annotated[ + LogLevel, + typer.Option("--log-level", help="Logging level."), + ] = LogLevel.WARNING, +) -> None: + """Configure shared options for all export commands. + + Raises: + typer.BadParameter: If the output format cannot be determined. + """ + logging.basicConfig(level=getattr(logging, log_level.value)) + try: + output_format = detect_format( + output, + explicit_format=fmt.value if fmt is not None else None, + ) + except ValueError as exc: + raise typer.BadParameter(str(exc)) from exc + ctx.obj = _ExportContext( + output=output, + output_format=output_format, + table=table, + config=Mt5Config( + path=path, + login=login, + password=password, + server=server, + timeout=timeout, + ), + ) + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +@app.command() +def rates_from( + ctx: typer.Context, + symbol: Annotated[str, typer.Option(help="Symbol name.")], + timeframe: Annotated[ + int, + typer.Option( + click_type=TIMEFRAME_TYPE, + help="Timeframe (e.g., M1, H1, D1, or integer).", + ), + ], + date_from: Annotated[ + datetime, + typer.Option( + click_type=DATETIME_TYPE, + help="Start date in ISO 8601 format.", + ), + ], + count: Annotated[int, typer.Option(help="Number of records.")], +) -> None: + """Export rates from a start date.""" + _execute_export( + ctx, + lambda c: c.copy_rates_from_as_df( + symbol=symbol, + timeframe=timeframe, + date_from=date_from, + count=count, + ), + ) + + +@app.command() +def rates_from_pos( + ctx: typer.Context, + symbol: Annotated[str, typer.Option(help="Symbol name.")], + timeframe: Annotated[ + int, + typer.Option( + click_type=TIMEFRAME_TYPE, + help="Timeframe.", + ), + ], + start_pos: Annotated[int, typer.Option(help="Start position (0 = current bar).")], + count: Annotated[int, typer.Option(help="Number of records.")], +) -> None: + """Export rates from a start position.""" + _execute_export( + ctx, + lambda c: c.copy_rates_from_pos_as_df( + symbol=symbol, + timeframe=timeframe, + start_pos=start_pos, + count=count, + ), + ) + + +@app.command() +def rates_range( + ctx: typer.Context, + symbol: Annotated[str, typer.Option(help="Symbol name.")], + timeframe: Annotated[ + int, + typer.Option( + click_type=TIMEFRAME_TYPE, + help="Timeframe.", + ), + ], + date_from: Annotated[ + datetime, + typer.Option(click_type=DATETIME_TYPE, help="Start date."), + ], + date_to: Annotated[ + datetime, + typer.Option(click_type=DATETIME_TYPE, help="End date."), + ], +) -> None: + """Export rates for a date range.""" + _execute_export( + ctx, + lambda c: c.copy_rates_range_as_df( + symbol=symbol, + timeframe=timeframe, + date_from=date_from, + date_to=date_to, + ), + ) + + +@app.command() +def ticks_from( + ctx: typer.Context, + symbol: Annotated[str, typer.Option(help="Symbol name.")], + date_from: Annotated[ + datetime, + typer.Option(click_type=DATETIME_TYPE, help="Start date."), + ], + count: Annotated[int, typer.Option(help="Number of ticks.")], + flags: Annotated[ + int, + typer.Option( + click_type=TICK_FLAGS_TYPE, + help="Tick flags (ALL, INFO, TRADE, or integer).", + ), + ], +) -> None: + """Export ticks from a start date.""" + _execute_export( + ctx, + lambda c: c.copy_ticks_from_as_df( + symbol=symbol, + date_from=date_from, + count=count, + flags=flags, + ), + ) + + +@app.command() +def ticks_range( + ctx: typer.Context, + symbol: Annotated[str, typer.Option(help="Symbol name.")], + date_from: Annotated[ + datetime, + typer.Option(click_type=DATETIME_TYPE, help="Start date."), + ], + date_to: Annotated[ + datetime, + typer.Option(click_type=DATETIME_TYPE, help="End date."), + ], + flags: Annotated[ + int, + typer.Option(click_type=TICK_FLAGS_TYPE, help="Tick flags."), + ], +) -> None: + """Export ticks for a date range.""" + _execute_export( + ctx, + lambda c: c.copy_ticks_range_as_df( + symbol=symbol, + date_from=date_from, + date_to=date_to, + flags=flags, + ), + ) + + +@app.command() +def account_info(ctx: typer.Context) -> None: + """Export account information.""" + _execute_export(ctx, lambda c: c.account_info_as_df()) + + +@app.command() +def terminal_info(ctx: typer.Context) -> None: + """Export terminal information.""" + _execute_export(ctx, lambda c: c.terminal_info_as_df()) + + +@app.command() +def symbols( + ctx: typer.Context, + group: Annotated[ + str | None, + typer.Option(help="Symbol group filter (e.g., *USD*)."), + ] = None, +) -> None: + """Export symbol list.""" + _execute_export( + ctx, + lambda c: c.symbols_get_as_df(group=group), + ) + + +@app.command() +def symbol_info( + ctx: typer.Context, + symbol: Annotated[str, typer.Option(help="Symbol name.")], +) -> None: + """Export symbol details.""" + _execute_export( + ctx, + lambda c: c.symbol_info_as_df(symbol=symbol), + ) + + +@app.command() +def orders( + ctx: typer.Context, + symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None, + group: Annotated[str | None, typer.Option(help="Group filter.")] = None, + ticket: Annotated[int | None, typer.Option(help="Ticket filter.")] = None, +) -> None: + """Export active orders.""" + _execute_export( + ctx, + lambda c: c.orders_get_as_df( + symbol=symbol, + group=group, + ticket=ticket, + ), + ) + + +@app.command() +def positions( + ctx: typer.Context, + symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None, + group: Annotated[str | None, typer.Option(help="Group filter.")] = None, + ticket: Annotated[int | None, typer.Option(help="Ticket filter.")] = None, +) -> None: + """Export open positions.""" + _execute_export( + ctx, + lambda c: c.positions_get_as_df( + symbol=symbol, + group=group, + ticket=ticket, + ), + ) + + +@app.command() +def history_orders( + ctx: typer.Context, + date_from: Annotated[ + datetime | None, + typer.Option(click_type=DATETIME_TYPE, help="Start date."), + ] = None, + date_to: Annotated[ + datetime | None, + typer.Option(click_type=DATETIME_TYPE, help="End date."), + ] = None, + group: Annotated[str | None, typer.Option(help="Group filter.")] = None, + symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None, + ticket: Annotated[int | None, typer.Option(help="Order ticket.")] = None, + position: Annotated[int | None, typer.Option(help="Position ticket.")] = None, +) -> None: + """Export historical orders.""" + _execute_export( + ctx, + lambda c: c.history_orders_get_as_df( + date_from=date_from, + date_to=date_to, + group=group, + symbol=symbol, + ticket=ticket, + position=position, + ), + ) + + +@app.command() +def history_deals( + ctx: typer.Context, + date_from: Annotated[ + datetime | None, + typer.Option(click_type=DATETIME_TYPE, help="Start date."), + ] = None, + date_to: Annotated[ + datetime | None, + typer.Option(click_type=DATETIME_TYPE, help="End date."), + ] = None, + group: Annotated[str | None, typer.Option(help="Group filter.")] = None, + symbol: Annotated[str | None, typer.Option(help="Symbol filter.")] = None, + ticket: Annotated[int | None, typer.Option(help="Order ticket.")] = None, + position: Annotated[int | None, typer.Option(help="Position ticket.")] = None, +) -> None: + """Export historical deals.""" + _execute_export( + ctx, + lambda c: c.history_deals_get_as_df( + date_from=date_from, + date_to=date_to, + group=group, + symbol=symbol, + ticket=ticket, + position=position, + ), + ) + + +def main() -> None: + """Run the mt5cli CLI.""" + app() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c605061 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,179 @@ +[project] +name = "mt5cli" +version = "0.1.0" +description = "Command-line tool for MetaTrader 5" +authors = [{name = "dceoy", email = "dceoy@users.noreply.github.com"}] +maintainers = [{name = "dceoy", email = "dceoy@users.noreply.github.com"}] +license = "MIT" +license-files = ["LICENSE"] +readme = "README.md" +requires-python = ">= 3.11, < 3.14" +dependencies = [ + "pdmt5 >= 0.2.3", + "click >= 8.1.0", + "typer >= 0.15.0", +] +classifiers = [ + "Development Status :: 3 - Alpha", + "Environment :: Console", + "License :: OSI Approved :: MIT License", + "Operating System :: Microsoft :: Windows", + "Programming Language :: Python", + "Programming Language :: Python :: 3", + "Intended Audience :: Financial and Insurance Industry", + "Topic :: Office/Business :: Financial :: Investment", +] + +[project.scripts] +mt5cli = "mt5cli.cli:main" + +[project.urls] +Repository = "https://github.com/dceoy/mt5cli.git" + +[tool.uv] +required-environments = ["platform_system == 'Windows'"] + +[dependency-groups] +dev = [ + "ruff >= 0.11.0", + "pyright >= 1.1.407", + "pytest >= 8.0.0", + "pytest-mock >= 3.12.0", + "pytest-cov >= 5.0.0", + "pandas-stubs >= 2.2.3.250527", +] + +[tool.uv.build-backend] +source-include = ["mt5cli/**", "LICENSE"] +source-exclude = ["tests/**"] + +[tool.ruff] +line-length = 88 +exclude = ["build", ".venv"] +preview = true + +[tool.ruff.lint] +select = [ + "F", # Pyflakes (F) + "E", # pycodestyle error (E) + "W", # pycodestyle warning (W) + "C90", # mccabe (C90) + "I", # isort (I) + "N", # pep8-naming (N) + "D", # pydocstyle (D) + "UP", # pyupgrade (UP) + "S", # flake8-bandit (S) + "B", # flake8-bugbear (B) + "C4", # flake8-comprehensions (C4) + "SIM", # flake8-simplify (SIM) + "ARG", # flake8-unused-arguments (ARG) + "PD", # pandas-vet (PD) + "PLC", # Pylint convention (PLC) + "PLE", # Pylint error (PLE) + "PLR", # Pylint refactor (PLR) + "PLW", # Pylint warning (PLW) + "FLY", # flynt (FLY) + "NPY", # NumPy-specific rules (NPY) + "PERF", # Perflint (PERF) + "FURB", # refurb (FURB) + "RUF", # Ruff-specific rules (RUF) + "YTT", # flake8-2020 (YTT) + "ANN", # flake8-annotations (ANN) + "ASYNC", # flake8-async (ASYNC) + "BLE", # flake8-blind-except (BLE) + "FBT", # flake8-boolean-trap (FBT) + "A", # flake8-builtins (A) + "COM", # flake8-commas (COM) + "DTZ", # flake8-datetimez (DTZ) + "T10", # flake8-debugger (T10) + "DJ", # flake8-django (DJ) + "EM", # flake8-errmsg (EM) + "EXE", # flake8-executable (EXE) + "FA", # flake8-future-annotations (FA) + "ISC", # flake8-implicit-str-concat (ISC) + "ICN", # flake8-import-conventions (ICN) + "LOG", # flake8-logging (LOG) + "G", # flake8-logging-format (G) + "INP", # flake8-no-pep420 (INP) + "PIE", # flake8-pie (PIE) + "T20", # flake8-print (T20) + "PYI", # flake8-pyi (PYI) + "PT", # flake8-pytest-style (PT) + "Q", # flake8-quotes (Q) + "RSE", # flake8-raise (RSE) + "SLF", # flake8-self (SLF) + "SLOT", # flake8-slots (SLOT) + "TID", # flake8-tidy-imports (TID) + "TCH", # flake8-type-checking (TCH) + "INT", # flake8-gettext (INT) + "PTH", # flake8-use-pathlib (PTH) + "TD", # flake8-todos (TD) + "FIX", # flake8-fixme (FIX) + "ERA", # eradicate (ERA) + "PGH", # pygrep-hooks (PGH) + "TRY", # tryceratops (TRY) + "FAST", # FastAPI (FAST) + "AIR", # Airflow (AIR) + "DOC", # pydoclint (DOC) +] +ignore = [ + "COM812", # missing-trailing-comma + "FBT001", # boolean-type-hint-positional-argument + "FBT002", # boolean-default-value-positional-argument +] + +[tool.ruff.lint.per-file-ignores] +"tests/**/*.py" = [ + "DOC201", # Missing return documentation + "DOC501", # Raised exception missing from docstring + "PLC2701", # Private name import + "PLR0904", # Too many public methods + "PLR2004", # Magic value used in comparison + "PLR6301", # Method could be function/static/classmethod + "S101", # Use of assert (acceptable in tests) + "S106", # Possible hardcoded password + "SLF001", # Private member accessed +] + +[tool.ruff.lint.pydocstyle] +convention = "google" + +[tool.ruff.lint.pylint] +max-args = 10 +max-public-methods = 40 + +[tool.pyright] +exclude = ["build", ".venv"] +venvPath = "." +venv = ".venv" +typeCheckingMode = "strict" + +[tool.pytest.ini_options] +addopts = [ + "--cov=mt5cli", + "--cov-branch", + "--doctest-modules", + "--capture=no", +] +pythonpath = ["."] +testpaths = ["tests"] +python_files = ["test_*.py", "*_test.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +minversion = "6.0" + +[tool.coverage.run] +source = ["mt5cli"] +omit = [ + "**/__init__.py", + "tests/**", +] + +[tool.coverage.report] +show_missing = true +fail_under = 100 +exclude_lines = ["if TYPE_CHECKING:"] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..29e8d82 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,751 @@ +"""Tests for mt5cli.cli module.""" + +from __future__ import annotations + +import json +import sqlite3 +from datetime import UTC, datetime +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pandas as pd +import pytest +from pytest_mock import MockerFixture # noqa: TC002 +from typer.testing import CliRunner + +if TYPE_CHECKING: + from pathlib import Path + +from mt5cli.cli import ( + DATETIME_TYPE, + TICK_FLAG_MAP, + TICK_FLAGS_TYPE, + TIMEFRAME_MAP, + TIMEFRAME_TYPE, + _execute_export, # type: ignore[reportPrivateUsage] + _ExportContext, # type: ignore[reportPrivateUsage] + app, + detect_format, + export_dataframe, + main, + parse_datetime, + parse_tick_flags, + parse_timeframe, +) + +runner = CliRunner() + + +# --------------------------------------------------------------------------- +# detect_format +# --------------------------------------------------------------------------- + + +class TestDetectFormat: + """Tests for detect_format.""" + + def test_explicit_format_returned(self, tmp_path: Path) -> None: + """Test that explicit format overrides extension.""" + result = detect_format(tmp_path / "data.txt", explicit_format="csv") + assert result == "csv" + + @pytest.mark.parametrize( + ("filename", "expected"), + [ + ("data.csv", "csv"), + ("data.json", "json"), + ("data.parquet", "parquet"), + ("data.pq", "parquet"), + ("data.db", "sqlite3"), + ("data.sqlite", "sqlite3"), + ("data.sqlite3", "sqlite3"), + ("DATA.CSV", "csv"), + ("DATA.JSON", "json"), + ("DATA.PARQUET", "parquet"), + ], + ) + def test_auto_detect_from_extension( + self, + tmp_path: Path, + filename: str, + expected: str, + ) -> None: + """Test format auto-detection from file extension.""" + result = detect_format(tmp_path / filename) + assert result == expected + + def test_unknown_extension_raises(self, tmp_path: Path) -> None: + """Test that unknown extension raises ValueError.""" + with pytest.raises(ValueError, match="Cannot detect format"): + detect_format(tmp_path / "data.xyz") + + +# --------------------------------------------------------------------------- +# export_dataframe +# --------------------------------------------------------------------------- + + +class TestExportDataframe: + """Tests for export_dataframe.""" + + @pytest.fixture + def sample_df(self) -> pd.DataFrame: + """Create a sample DataFrame for testing.""" + return pd.DataFrame({"a": [1, 2, 3], "b": ["x", "y", "z"]}) + + def test_export_csv(self, tmp_path: Path, sample_df: pd.DataFrame) -> None: + """Test CSV export.""" + output = tmp_path / "out.csv" + export_dataframe(sample_df, output, "csv") + result = pd.read_csv(output) + pd.testing.assert_frame_equal(result, sample_df) + + def test_export_json(self, tmp_path: Path, sample_df: pd.DataFrame) -> None: + """Test JSON export.""" + output = tmp_path / "out.json" + export_dataframe(sample_df, output, "json") + with output.open() as f: + records = json.load(f) + assert len(records) == 3 + assert records[0]["a"] == 1 + + def test_export_parquet(self, tmp_path: Path, sample_df: pd.DataFrame) -> None: + """Test Parquet export.""" + output = tmp_path / "out.parquet" + export_dataframe(sample_df, output, "parquet") + result = pd.read_parquet(output) + pd.testing.assert_frame_equal(result, sample_df) + + def test_export_sqlite3(self, tmp_path: Path, sample_df: pd.DataFrame) -> None: + """Test SQLite3 export.""" + output = tmp_path / "out.db" + export_dataframe(sample_df, output, "sqlite3", table_name="test_table") + with sqlite3.connect(output) as conn: + result = pd.read_sql( # type: ignore[reportUnknownMemberType] + "SELECT * FROM test_table", + conn, + ) + pd.testing.assert_frame_equal(result, sample_df) + + def test_unsupported_format_raises( + self, + tmp_path: Path, + sample_df: pd.DataFrame, + ) -> None: + """Test that unsupported format raises ValueError.""" + with pytest.raises(ValueError, match="Unsupported output format"): + export_dataframe(sample_df, tmp_path / "out.txt", "xml") + + +# --------------------------------------------------------------------------- +# Parse helpers +# --------------------------------------------------------------------------- + + +class TestParseDatetime: + """Tests for parse_datetime.""" + + def test_valid_date(self) -> None: + """Test parsing a date string.""" + result = parse_datetime("2024-01-15") + assert result == datetime(2024, 1, 15, tzinfo=UTC) + + def test_valid_datetime_with_tz(self) -> None: + """Test parsing a datetime with timezone.""" + result = parse_datetime("2024-01-15T12:00:00+00:00") + assert result == datetime(2024, 1, 15, 12, 0, 0, tzinfo=UTC) + + def test_invalid_format_raises(self) -> None: + """Test that invalid format raises ValueError.""" + with pytest.raises(ValueError, match="Invalid datetime"): + parse_datetime("not-a-date") + + +class TestParseTimeframe: + """Tests for parse_timeframe.""" + + @pytest.mark.parametrize( + ("value", "expected"), + [("M1", 1), ("h1", 16385), ("D1", 16408), ("MN1", 49153)], + ) + def test_named_timeframe(self, value: str, expected: int) -> None: + """Test parsing named timeframes.""" + assert parse_timeframe(value) == expected + + def test_integer_timeframe(self) -> None: + """Test parsing integer timeframe.""" + assert parse_timeframe("42") == 42 + + def test_invalid_timeframe_raises(self) -> None: + """Test that invalid timeframe raises ValueError.""" + with pytest.raises(ValueError, match="Invalid timeframe"): + parse_timeframe("INVALID") + + +class TestParseTickFlags: + """Tests for parse_tick_flags.""" + + @pytest.mark.parametrize( + ("value", "expected"), + [("ALL", 1), ("info", 2), ("TRADE", 4)], + ) + def test_named_flag(self, value: str, expected: int) -> None: + """Test parsing named tick flags.""" + assert parse_tick_flags(value) == expected + + def test_integer_flag(self) -> None: + """Test parsing integer tick flag.""" + assert parse_tick_flags("7") == 7 + + def test_invalid_flag_raises(self) -> None: + """Test that invalid flag raises ValueError.""" + with pytest.raises(ValueError, match="Invalid tick flags"): + parse_tick_flags("INVALID") + + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + + +class TestConstants: + """Tests for module constants.""" + + def test_timeframe_map_has_expected_keys(self) -> None: + """Test that TIMEFRAME_MAP contains standard timeframes.""" + for key in ("M1", "M5", "M15", "M30", "H1", "H4", "D1", "W1", "MN1"): + assert key in TIMEFRAME_MAP + + def test_tick_flag_map_has_expected_keys(self) -> None: + """Test that TICK_FLAG_MAP contains standard flags.""" + assert set(TICK_FLAG_MAP) == {"ALL", "INFO", "TRADE"} + + +# --------------------------------------------------------------------------- +# Click ParamTypes +# --------------------------------------------------------------------------- + + +class TestDateTimeType: + """Tests for _DateTimeType.""" + + def test_convert_string(self) -> None: + """Test converting a string to datetime.""" + result = DATETIME_TYPE.convert("2024-06-15", None, None) + assert result == datetime(2024, 6, 15, tzinfo=UTC) + + def test_convert_datetime_passthrough(self) -> None: + """Test that datetime values pass through unchanged.""" + dt = datetime(2024, 1, 1, tzinfo=UTC) + assert DATETIME_TYPE.convert(dt, None, None) is dt + + def test_convert_invalid(self) -> None: + """Test that invalid values raise BadParameter.""" + with pytest.raises(Exception, match="Invalid datetime"): + DATETIME_TYPE.convert("bad", None, None) + + +class TestTimeframeType: + """Tests for _TimeframeType.""" + + def test_convert_string(self) -> None: + """Test converting a string to timeframe integer.""" + assert TIMEFRAME_TYPE.convert("H1", None, None) == 16385 + + def test_convert_int_passthrough(self) -> None: + """Test that integer values pass through unchanged.""" + assert TIMEFRAME_TYPE.convert(42, None, None) == 42 + + def test_convert_invalid(self) -> None: + """Test that invalid values raise BadParameter.""" + with pytest.raises(Exception, match="Invalid timeframe"): + TIMEFRAME_TYPE.convert("bad", None, None) + + +class TestTickFlagsType: + """Tests for _TickFlagsType.""" + + def test_convert_string(self) -> None: + """Test converting a string to tick flags integer.""" + assert TICK_FLAGS_TYPE.convert("ALL", None, None) == 1 + + def test_convert_int_passthrough(self) -> None: + """Test that integer values pass through unchanged.""" + assert TICK_FLAGS_TYPE.convert(7, None, None) == 7 + + def test_convert_invalid(self) -> None: + """Test that invalid values raise BadParameter.""" + with pytest.raises(Exception, match="Invalid tick flags"): + TICK_FLAGS_TYPE.convert("bad", None, None) + + +# --------------------------------------------------------------------------- +# _execute_export +# --------------------------------------------------------------------------- + + +class TestExecuteExport: + """Tests for _execute_export.""" + + def test_shutdown_on_error( + self, + tmp_path: Path, + mocker: MockerFixture, + ) -> None: + """Test that shutdown is called even when fetch raises.""" + mock_client = MagicMock() + mock_client.account_info_as_df.side_effect = RuntimeError("boom") + mocker.patch("mt5cli.cli.Mt5DataClient", return_value=mock_client) + ctx = MagicMock() + ctx.obj = _ExportContext( + output=tmp_path / "out.csv", + output_format="csv", + table="data", + config=MagicMock(), + ) + with pytest.raises(RuntimeError, match="boom"): + _execute_export(ctx, lambda c: c.account_info_as_df()) + mock_client.shutdown.assert_called_once() + + +# --------------------------------------------------------------------------- +# CLI commands via CliRunner +# --------------------------------------------------------------------------- + + +@pytest.fixture +def mock_client(mocker: MockerFixture) -> MagicMock: + """Create and patch a mock Mt5DataClient for CLI tests.""" + client = MagicMock() + sample_df = pd.DataFrame({"col": [1]}) + client.copy_rates_from_as_df.return_value = sample_df + client.copy_rates_from_pos_as_df.return_value = sample_df + client.copy_rates_range_as_df.return_value = sample_df + client.copy_ticks_from_as_df.return_value = sample_df + client.copy_ticks_range_as_df.return_value = sample_df + client.account_info_as_df.return_value = sample_df + client.terminal_info_as_df.return_value = sample_df + client.symbols_get_as_df.return_value = sample_df + client.symbol_info_as_df.return_value = sample_df + client.orders_get_as_df.return_value = sample_df + client.positions_get_as_df.return_value = sample_df + client.history_orders_get_as_df.return_value = sample_df + client.history_deals_get_as_df.return_value = sample_df + mocker.patch("mt5cli.cli.Mt5DataClient", return_value=client) + return client + + +class TestCommands: + """Tests for all CLI subcommands via CliRunner.""" + + def test_account_info( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test account-info command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + ["-o", str(output), "account-info"], + ) + assert result.exit_code == 0, result.output + mock_client.account_info_as_df.assert_called_once() + assert output.exists() + + def test_terminal_info( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test terminal-info command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + ["-o", str(output), "terminal-info"], + ) + assert result.exit_code == 0, result.output + mock_client.terminal_info_as_df.assert_called_once() + + def test_symbols( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test symbols command.""" + output = tmp_path / "out.json" + result = runner.invoke( + app, + ["-o", str(output), "symbols", "--group", "*USD*"], + ) + assert result.exit_code == 0, result.output + mock_client.symbols_get_as_df.assert_called_once_with( + group="*USD*", + ) + + def test_symbol_info( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test symbol-info command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + ["-o", str(output), "symbol-info", "--symbol", "EURUSD"], + ) + assert result.exit_code == 0, result.output + mock_client.symbol_info_as_df.assert_called_once_with( + symbol="EURUSD", + ) + + def test_rates_from( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test rates-from command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "rates-from", + "--symbol", + "EURUSD", + "--timeframe", + "M1", + "--date-from", + "2024-01-01", + "--count", + "100", + ], + ) + assert result.exit_code == 0, result.output + mock_client.copy_rates_from_as_df.assert_called_once_with( + symbol="EURUSD", + timeframe=1, + date_from=datetime(2024, 1, 1, tzinfo=UTC), + count=100, + ) + + def test_rates_from_pos( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test rates-from-pos command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "rates-from-pos", + "--symbol", + "GBPUSD", + "--timeframe", + "H1", + "--start-pos", + "0", + "--count", + "50", + ], + ) + assert result.exit_code == 0, result.output + mock_client.copy_rates_from_pos_as_df.assert_called_once_with( + symbol="GBPUSD", + timeframe=16385, + start_pos=0, + count=50, + ) + + def test_rates_range( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test rates-range command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "rates-range", + "--symbol", + "USDJPY", + "--timeframe", + "D1", + "--date-from", + "2024-01-01", + "--date-to", + "2024-02-01", + ], + ) + assert result.exit_code == 0, result.output + mock_client.copy_rates_range_as_df.assert_called_once_with( + symbol="USDJPY", + timeframe=16408, + date_from=datetime(2024, 1, 1, tzinfo=UTC), + date_to=datetime(2024, 2, 1, tzinfo=UTC), + ) + + def test_ticks_from( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test ticks-from command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "ticks-from", + "--symbol", + "EURUSD", + "--date-from", + "2024-01-01", + "--count", + "100", + "--flags", + "ALL", + ], + ) + assert result.exit_code == 0, result.output + mock_client.copy_ticks_from_as_df.assert_called_once_with( + symbol="EURUSD", + date_from=datetime(2024, 1, 1, tzinfo=UTC), + count=100, + flags=1, + ) + + def test_ticks_range( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test ticks-range command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "ticks-range", + "--symbol", + "EURUSD", + "--date-from", + "2024-01-01", + "--date-to", + "2024-02-01", + "--flags", + "INFO", + ], + ) + assert result.exit_code == 0, result.output + mock_client.copy_ticks_range_as_df.assert_called_once_with( + symbol="EURUSD", + date_from=datetime(2024, 1, 1, tzinfo=UTC), + date_to=datetime(2024, 2, 1, tzinfo=UTC), + flags=2, + ) + + def test_orders( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test orders command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "orders", + "--symbol", + "EURUSD", + ], + ) + assert result.exit_code == 0, result.output + mock_client.orders_get_as_df.assert_called_once() + + def test_positions( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test positions command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + ["-o", str(output), "positions"], + ) + assert result.exit_code == 0, result.output + mock_client.positions_get_as_df.assert_called_once() + + def test_history_orders( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test history-orders command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "history-orders", + "--date-from", + "2024-01-01", + "--date-to", + "2024-02-01", + ], + ) + assert result.exit_code == 0, result.output + mock_client.history_orders_get_as_df.assert_called_once() + + def test_history_deals( + self, + tmp_path: Path, + mock_client: MagicMock, + ) -> None: + """Test history-deals command.""" + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "-o", + str(output), + "history-deals", + "--ticket", + "12345", + ], + ) + assert result.exit_code == 0, result.output + mock_client.history_deals_get_as_df.assert_called_once() + + +# --------------------------------------------------------------------------- +# Callback / shared options +# --------------------------------------------------------------------------- + + +class TestCallback: + """Tests for callback (shared options).""" + + def test_format_detection_error(self, tmp_path: Path) -> None: + """Test that bad extension triggers a user-friendly error.""" + output = tmp_path / "out.xyz" + result = runner.invoke( + app, + ["-o", str(output), "account-info"], + ) + assert result.exit_code != 0 + assert "Cannot detect format" in result.output + + def test_connection_args_forwarded( + self, + tmp_path: Path, + mocker: MockerFixture, + ) -> None: + """Test that connection arguments reach Mt5Config.""" + mock_client = MagicMock() + mock_client.account_info_as_df.return_value = pd.DataFrame({"a": [1]}) + mocker.patch( + "mt5cli.cli.Mt5DataClient", + return_value=mock_client, + ) + mock_config = mocker.patch("mt5cli.cli.Mt5Config") + output = tmp_path / "out.csv" + result = runner.invoke( + app, + [ + "--login", + "123", + "--password", + "pw", + "--server", + "srv", + "-o", + str(output), + "account-info", + ], + ) + assert result.exit_code == 0, result.output + mock_config.assert_called_once_with( + path=None, + login=123, + password="pw", + server="srv", + timeout=None, + ) + + def test_explicit_format( + self, + tmp_path: Path, + mock_client: MagicMock, # noqa: ARG002 + ) -> None: + """Test explicit --format flag.""" + output = tmp_path / "out.txt" + result = runner.invoke( + app, + ["-o", str(output), "--format", "json", "account-info"], + ) + assert result.exit_code == 0, result.output + assert output.exists() + + def test_sqlite3_with_table( + self, + tmp_path: Path, + mocker: MockerFixture, + ) -> None: + """Test SQLite3 output with custom table name.""" + mock_client = MagicMock() + mock_client.symbols_get_as_df.return_value = pd.DataFrame( + {"s": ["EURUSD"]}, + ) + mocker.patch( + "mt5cli.cli.Mt5DataClient", + return_value=mock_client, + ) + output = tmp_path / "out.db" + result = runner.invoke( + app, + [ + "-o", + str(output), + "--table", + "symbols", + "symbols", + "--group", + "*USD*", + ], + ) + assert result.exit_code == 0, result.output + with sqlite3.connect(output) as conn: + result_df = pd.read_sql( # type: ignore[reportUnknownMemberType] + "SELECT * FROM symbols", + conn, + ) + assert len(result_df) == 1 + + +# --------------------------------------------------------------------------- +# main entry point +# --------------------------------------------------------------------------- + + +class TestMain: + """Tests for the main entry point.""" + + def test_main_invokes_app(self, mocker: MockerFixture) -> None: + """Test that main() calls the typer app.""" + mock_app = mocker.patch("mt5cli.cli.app") + main() + mock_app.assert_called_once()