From 5b1d54bfe9ebeea3033037a7c0ee4e7daa6a856f Mon Sep 17 00:00:00 2001 From: Daichi Narushima <1938249+dceoy@users.noreply.github.com> Date: Wed, 10 Jun 2026 00:15:07 +0900 Subject: [PATCH] Add resilient multi-account orchestration helpers (#22) * Add SDK orchestration helpers for resilient multi-account collection - collect_latest_rates_for_accounts_with_retries(): exponential-backoff retries around collect_latest_rates_for_accounts(), retrying only Mt5TradingError/Mt5RuntimeError and re-raising on exhaustion. - resolve_account_spec()/resolve_account_specs() and substitute_env_placeholders(): merge explicit overrides over AccountSpec fields and expand ${ENV_VAR} placeholders, raising ValueError on missing variables. - ThrottledHistoryUpdater: monotonic-clock throttled wrapper around update_history() with should_update()/update() and opt-in suppress_errors. - load_rate_series_by_granularity(): rate-series loader keyed by (symbol | None, granularity_name). - Export new APIs, add unit tests (100% coverage), and document in README and docs/api. * chore: bump version from 0.5.1 to 0.5.3 (#24) Co-authored-by: Cursor Agent Co-authored-by: Daichi Narushima * fix: resolve leftover merge conflict markers in version files Co-authored-by: Daichi Narushima * fix: address PR review feedback on SDK orchestration helpers - Use single-pass env substitution to avoid TOCTOU KeyError - Apply backoff_base to all retry delays (backoff_base ** (attempt + 1)) - Preserve integer logins in resolve_account_spec; hide login in repr - Fix docs examples (env ordering, while True loop, backoff comment) - Parametrize suppress_errors tests for MT5 and SQLite errors Co-authored-by: Daichi Narushima --------- Co-authored-by: Claude Co-authored-by: Cursor Agent Co-authored-by: Daichi Narushima --- README.md | 5 +- docs/api/history.md | 12 ++ docs/api/sdk.md | 80 +++++++++++ mt5cli/__init__.py | 12 ++ mt5cli/history.py | 49 +++++++ mt5cli/sdk.py | 325 +++++++++++++++++++++++++++++++++++++++++- pyproject.toml | 2 +- tests/test_history.py | 51 +++++++ tests/test_sdk.py | 312 ++++++++++++++++++++++++++++++++++++++++ uv.lock | 2 +- 10 files changed, 845 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 98de680..3bed10c 100644 --- a/README.md +++ b/README.md @@ -136,7 +136,10 @@ update_history_with_config( - **Rate view resolution**: use `resolve_rate_view_name()` / `resolve_rate_view_names()` to map symbols and granularities to existing SQLite compatibility views without creating databases. Both accept `None` (or a missing path) and return deterministic default names unless `require_existing=True`. - **Rate view loading**: use `load_rate_data()` / `load_rate_data_from_connection()` to load a SQLite rate table or view into a `DatetimeIndex` DataFrame. - **Multi-series rate loading**: use `build_rate_targets()` to build neutral `RateTarget(symbol, timeframe)` pairs, `resolve_rate_tables()` to map them to table/view names (pass `require_existing=True` for strict resolution), and `load_rate_series_from_sqlite()` to load them into a mapping keyed by `(symbol, integer timeframe)`. The loader requires existing managed views unless `explicit_tables` is supplied, and rejects duplicate `(symbol, timeframe)` targets. -- **Multi-account latest rates**: use `collect_latest_rates_for_accounts()` with `AccountSpec` to read the latest bars for several account groups, merged into a `(symbol, integer timeframe)` mapping. +- **Multi-account latest rates**: use `collect_latest_rates_for_accounts()` with `AccountSpec` to read the latest bars for several account groups, merged into a `(symbol, integer timeframe)` mapping. For long-running pollers, `collect_latest_rates_for_accounts_with_retries()` adds bounded exponential backoff that retries only `pdmt5.Mt5TradingError` / `pdmt5.Mt5RuntimeError` and re-raises once `retry_count` is exhausted. +- **Credential resolution**: use `resolve_account_spec()` / `resolve_account_specs()` to merge explicit override values over `AccountSpec` fields and expand `${ENV_VAR}` placeholders (via `substitute_env_placeholders()`), raising `ValueError` for missing variables. This keeps secrets out of plan/config files without coupling to any strategy code. +- **Throttled history updates**: use `ThrottledHistoryUpdater` to wrap `update_history()` with a minimum `interval_seconds` between successful runs (monotonic clock). Call `should_update()` / `update(client, symbols)` from an application loop; errors propagate by default, or pass `suppress_errors=True` to swallow recoverable `Mt5*Error`/`sqlite3.Error` and let the caller decide logging. +- **Granularity-keyed rate loading**: `load_rate_series_by_granularity()` builds targets with `build_rate_targets()`, loads them with `load_rate_series_from_sqlite()`, and returns a mapping keyed by `(symbol | None, granularity_name)` such as `("EURUSD", "M1")` to reduce downstream boilerplate. - **MT5 session helper**: use the `mt5_session()` context manager to attach to (or, when `Mt5Config.path` is set, launch) an MT5 terminal, log in, and yield a connected `Mt5CliClient` that shuts down on exit. - **SQLite export helpers**: use `export_dataframe_to_sqlite()` for append mode, optional index export, and post-write deduplication by key columns. - **Recent ticks and margins**: `recent_ticks()` and `minimum_margins()` SDK helpers (and matching CLI commands) cover common downstream read-only queries. diff --git a/docs/api/history.md b/docs/api/history.md index 25115ee..9173187 100644 --- a/docs/api/history.md +++ b/docs/api/history.md @@ -215,3 +215,15 @@ frame = series["EURUSD", 1] # keyed by (symbol, integer timeframe) requires existing managed `rate_*` compatibility views and raises `ValueError` when they are missing. Duplicate `(symbol, timeframe)` targets are rejected. +- `load_rate_series_by_granularity()` is a thin wrapper that builds the targets, + loads the series, and rekeys the result by granularity name to avoid + converting integer timeframes downstream: + + ```python + from mt5cli import load_rate_series_by_granularity + + series = load_rate_series_by_granularity( + "history.db", ["EURUSD"], ["M1", "H1"], count=1000 + ) + frame = series["EURUSD", "M1"] # keyed by (symbol | None, granularity_name) + ``` diff --git a/docs/api/sdk.md b/docs/api/sdk.md index a7af946..adb6629 100644 --- a/docs/api/sdk.md +++ b/docs/api/sdk.md @@ -1,3 +1,83 @@ # SDK Module ::: mt5cli.sdk + +## Resilient multi-account orchestration + +The SDK ships strategy-agnostic helpers for building long-running collectors on +top of the read-only client. None of them depend on a particular trading +application. + +### Retrying transient rate collection + +`collect_latest_rates_for_accounts_with_retries()` wraps +`collect_latest_rates_for_accounts()` with bounded exponential backoff. Only +`pdmt5.Mt5TradingError` and `pdmt5.Mt5RuntimeError` are retried; the final +failure is re-raised once `retry_count` is exhausted. + +```python +from mt5cli import AccountSpec, collect_latest_rates_for_accounts_with_retries + +accounts = [AccountSpec(symbols=["EURUSD"], login=12345)] +rates = collect_latest_rates_for_accounts_with_retries( + accounts, + ["M1", "H1"], + count=500, + retry_count=3, + backoff_base=2, # sleeps 2s, 4s, 8s between attempts +) +``` + +### Resolving credentials and `${ENV_VAR}` placeholders + +`resolve_account_spec()` / `resolve_account_specs()` merge explicit override +values over `AccountSpec` fields and expand `${ENV_VAR}` placeholders, keeping +secrets out of plan/config files. A missing environment variable raises +`ValueError`. + +```python +import os + +from mt5cli import AccountSpec, resolve_account_specs + +os.environ["MT5_LOGIN"] = "12345" +os.environ["MT5_PASSWORD"] = "secret" +accounts = [ + AccountSpec(symbols=["EURUSD"], login="${MT5_LOGIN}", password="${MT5_PASSWORD}") +] + +resolved = resolve_account_specs(accounts, server="Broker-Demo") +# resolved[0].login == "12345", resolved[0].server == "Broker-Demo" +``` + +### Throttled incremental history updates + +`ThrottledHistoryUpdater` wraps `update_history()` with a minimum interval +between successful runs (using a monotonic clock), so an application loop can +call it every iteration without over-fetching. + +```python +from pdmt5 import Mt5Config, Mt5DataClient + +from mt5cli import Dataset, ThrottledHistoryUpdater + +updater = ThrottledHistoryUpdater( + output="history.db", + datasets={Dataset.rates}, + timeframes=["M1"], + interval_seconds=60, # <= 0 updates on every call +) + +client = Mt5DataClient(config=Mt5Config(login=12345)) +client.initialize_and_login_mt5() +try: + while True: + updater.update(client, ["EURUSD", "GBPUSD"]) # no-op until 60s elapse + # ... do other work; break when shutting down ... +finally: + client.shutdown() +``` + +By default `Mt5TradingError`, `Mt5RuntimeError`, and `sqlite3.Error` propagate so +the caller controls logging; pass `suppress_errors=True` to swallow them and +return `False` without advancing the throttle. diff --git a/mt5cli/__init__.py b/mt5cli/__init__.py index 14336e5..4eb58fc 100644 --- a/mt5cli/__init__.py +++ b/mt5cli/__init__.py @@ -8,6 +8,7 @@ from .history import ( build_rate_view_name, load_rate_data, load_rate_data_from_connection, + load_rate_series_by_granularity, load_rate_series_from_sqlite, resolve_history_datasets, resolve_history_tick_flags, @@ -19,11 +20,13 @@ from .history import ( from .sdk import ( AccountSpec, Mt5CliClient, + ThrottledHistoryUpdater, account_info, build_config, collect_history, collect_latest_rates, collect_latest_rates_for_accounts, + collect_latest_rates_for_accounts_with_retries, copy_rates_from, copy_rates_from_pos, copy_rates_range, @@ -42,6 +45,9 @@ from .sdk import ( positions, recent_history_deals, recent_ticks, + resolve_account_spec, + resolve_account_specs, + substitute_env_placeholders, symbol_info, symbol_info_tick, symbols, @@ -75,6 +81,7 @@ __all__ = [ "IfExists", "Mt5CliClient", "RateTarget", + "ThrottledHistoryUpdater", "account_info", "build_config", "build_rate_targets", @@ -82,6 +89,7 @@ __all__ = [ "collect_history", "collect_latest_rates", "collect_latest_rates_for_accounts", + "collect_latest_rates_for_accounts_with_retries", "copy_rates_from", "copy_rates_from_pos", "copy_rates_range", @@ -96,6 +104,7 @@ __all__ = [ "latest_rates", "load_rate_data", "load_rate_data_from_connection", + "load_rate_series_by_granularity", "load_rate_series_from_sqlite", "market_book", "minimum_margins", @@ -110,12 +119,15 @@ __all__ = [ "positions", "recent_history_deals", "recent_ticks", + "resolve_account_spec", + "resolve_account_specs", "resolve_history_datasets", "resolve_history_tick_flags", "resolve_history_timeframes", "resolve_rate_tables", "resolve_rate_view_name", "resolve_rate_view_names", + "substitute_env_placeholders", "symbol_info", "symbol_info_tick", "symbols", diff --git a/mt5cli/history.py b/mt5cli/history.py index 0b73884..02e9665 100644 --- a/mt5cli/history.py +++ b/mt5cli/history.py @@ -706,6 +706,55 @@ def load_rate_series_from_sqlite( conn.close() +def load_rate_series_by_granularity( + conn_or_path: SqliteConnOrPath, + symbols: Sequence[str], + granularities: Sequence[int | str], + count: int, + *, + explicit_tables: Sequence[str] | None = None, + allow_missing_symbol: bool = False, +) -> dict[tuple[str | None, str], pd.DataFrame]: + """Load rate series keyed by symbol and string granularity name. + + Builds targets with :func:`build_rate_targets` and loads them with + :func:`load_rate_series_from_sqlite`, then rekeys the result by granularity + name (for example ``M1``) instead of the integer timeframe to reduce + downstream boilerplate. + + Args: + conn_or_path: SQLite database path or open connection. + symbols: MT5 symbol names. May be empty when ``allow_missing_symbol``. + granularities: MT5 timeframes as integers or names (for example ``M1``). + count: Number of most recent rows to load per series. + explicit_tables: Optional explicit table or view names matching the + built targets in row-major order. Required when symbols are omitted. + allow_missing_symbol: When True and ``symbols`` is empty, build targets + with ``symbol=None`` for each granularity instead of raising. + + Returns: + Mapping keyed by ``(symbol | None, granularity_name)`` to each rate + DataFrame. Propagates ``ValueError`` (via :func:`build_rate_targets` and + :func:`load_rate_series_from_sqlite`) when inputs are empty or invalid, + table resolution fails, or duplicate targets are present. + """ + targets = build_rate_targets( + symbols, + granularities, + allow_missing_symbol=allow_missing_symbol, + ) + series = load_rate_series_from_sqlite( + conn_or_path, + targets, + count, + explicit_tables=explicit_tables, + ) + return { + (symbol, resolve_granularity_name(timeframe)): frame + for (symbol, timeframe), frame in series.items() + } + + def get_table_columns(conn: sqlite3.Connection, table: str) -> set[str]: """Return existing SQLite columns for a table.""" quoted_table = quote_sqlite_identifier(table) diff --git a/mt5cli/sdk.py b/mt5cli/sdk.py index 10641d1..2a046d8 100644 --- a/mt5cli/sdk.py +++ b/mt5cli/sdk.py @@ -4,7 +4,10 @@ from __future__ import annotations import json import logging +import os +import re import sqlite3 +import time from contextlib import contextmanager from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta @@ -12,7 +15,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Self, TypeVar, cast import pandas as pd -from pdmt5 import Mt5Config, Mt5DataClient +from pdmt5 import Mt5Config, Mt5DataClient, Mt5RuntimeError, Mt5TradingError from .history import ( create_cash_events_view, @@ -42,11 +45,13 @@ logger = logging.getLogger(__name__) __all__ = [ "AccountSpec", "Mt5CliClient", + "ThrottledHistoryUpdater", "account_info", "build_config", "collect_history", "collect_latest_rates", "collect_latest_rates_for_accounts", + "collect_latest_rates_for_accounts_with_retries", "copy_rates_from", "copy_rates_from_pos", "copy_rates_range", @@ -65,6 +70,9 @@ __all__ = [ "positions", "recent_history_deals", "recent_ticks", + "resolve_account_spec", + "resolve_account_specs", + "substitute_env_placeholders", "symbol_info", "symbol_info_tick", "symbols", @@ -960,6 +968,115 @@ def update_history_with_config( # noqa: PLR0913 ) +class ThrottledHistoryUpdater: + """Throttled incremental SQLite history updater for long-running apps. + + Wraps :func:`update_history` with a minimum interval between successful + updates, so a tight application loop can call :meth:`update` every + iteration without re-fetching MT5 history more often than desired. Timing + uses a monotonic clock, so it is unaffected by wall-clock changes. + """ + + def __init__( + self, + *, + output: Path | str, + datasets: set[Dataset] | None = None, + timeframes: Sequence[int | str] | None = None, + flags: int | str = "ALL", + lookback_hours: float = 24.0, + with_views: bool = False, + include_account_events: bool = True, + interval_seconds: float = 0.0, + suppress_errors: bool = False, + ) -> None: + """Initialize the throttled updater. + + Args: + output: SQLite database path. + datasets: Datasets to include (defaults to all). + timeframes: Rate timeframes to update (defaults to all fixed MT5 + timeframes). + flags: Tick copy flags as integer or name (e.g. ``ALL``). + lookback_hours: First-run lookback when a table has no prior rows. + with_views: Create ``cash_events`` and ``positions_reconstructed`` + views. + include_account_events: Include account-level cash events. + interval_seconds: Minimum seconds between successful updates. Values + ``<= 0`` update on every call. + suppress_errors: When True, ``Mt5TradingError``, ``Mt5RuntimeError``, + and ``sqlite3.Error`` raised during an update are swallowed and + :meth:`update` returns False without advancing the throttle. When + False (default), such errors propagate so callers control logging. + """ + self.output = output + self.datasets = datasets + self.timeframes = timeframes + self.flags = flags + self.lookback_hours = lookback_hours + self.with_views = with_views + self.include_account_events = include_account_events + self.interval_seconds = interval_seconds + self.suppress_errors = suppress_errors + self._last_update_monotonic: float | None = None + + @property + def last_update_monotonic(self) -> float | None: + """Return the monotonic timestamp of the last successful update.""" + return self._last_update_monotonic + + def should_update(self) -> bool: + """Return whether enough time has elapsed to run another update. + + Returns: + True when ``interval_seconds <= 0``, when no update has succeeded + yet, or when at least ``interval_seconds`` have elapsed since the + last successful update. + """ + if self.interval_seconds <= 0 or self._last_update_monotonic is None: + return True + return (time.monotonic() - self._last_update_monotonic) >= self.interval_seconds + + def update(self, client: Mt5DataClient, symbols: Sequence[str]) -> bool: + """Run a throttled incremental history update. + + Args: + client: Connected MT5 data client. + symbols: Symbols to update. + + Returns: + True if an update ran successfully, False if it was throttled or + (when ``suppress_errors`` is True) failed with a recoverable error. + + Raises: + Mt5TradingError: If the update fails and ``suppress_errors`` is False. + Mt5RuntimeError: If the update fails and ``suppress_errors`` is False. + sqlite3.Error: If the SQLite write fails and ``suppress_errors`` is + False. + """ + if not self.should_update(): + return False + try: + update_history( + client=client, + output=self.output, + symbols=symbols, + datasets=self.datasets, + timeframes=self.timeframes, + flags=self.flags, + lookback_hours=self.lookback_hours, + with_views=self.with_views, + include_account_events=self.include_account_events, + ) + except (Mt5TradingError, Mt5RuntimeError, sqlite3.Error): + if self.suppress_errors: + logger.warning("Suppressed history update error", exc_info=True) + return False + raise + self._last_update_monotonic = time.monotonic() + return True + + def collect_history( output: Path, symbols: list[str], @@ -1113,13 +1230,155 @@ class AccountSpec: """ symbols: Sequence[str] - login: int | str | None = None + login: int | str | None = field(default=None, repr=False) password: str | None = field(default=None, repr=False) server: str | None = None path: str | None = None timeout: int | None = None +_ENV_PLACEHOLDER_PATTERN = re.compile(r"\$\{(?P[A-Za-z_][A-Za-z0-9_]*)\}") + + +def substitute_env_placeholders(value: str) -> str: + """Replace ``${ENV_VAR}`` placeholders in a string with environment values. + + Args: + value: String that may contain one or more ``${ENV_VAR}`` placeholders. + + Returns: + The string with every placeholder replaced by its environment value. + + Raises: + ValueError: If a referenced environment variable is not set. + """ + parts: list[str] = [] + last_end = 0 + for match in _ENV_PLACEHOLDER_PATTERN.finditer(value): + parts.append(value[last_end : match.start()]) + name = match.group("name") + if name not in os.environ: + msg = f"Environment variable {name!r} is not set." + raise ValueError(msg) + parts.append(os.environ[name]) + last_end = match.end() + parts.append(value[last_end:]) + return "".join(parts) + + +def _resolve_field(override: str | None, account_value: str | None) -> str | None: + """Resolve a string field from an override or account value with env subst. + + Returns: + The explicit override when provided, otherwise the account value, with + any ``${ENV_VAR}`` placeholders substituted. + """ + value = override if override is not None else account_value + if value is None: + return None + return substitute_env_placeholders(value) + + +def _resolve_login( + override: int | str | None, + account_login: int | str | None, +) -> int | str | None: + """Resolve a login from an override or account value with env substitution. + + Returns: + The explicit override when provided, otherwise the account login. + Integer values are preserved; string values have ``${ENV_VAR}`` + placeholders substituted. + """ + if override is not None: + if isinstance(override, int): + return override + return substitute_env_placeholders(override) + if account_login is None or isinstance(account_login, int): + return account_login + return substitute_env_placeholders(account_login) + + +def resolve_account_spec( + account: AccountSpec, + *, + login: int | str | None = None, + password: str | None = None, + server: str | None = None, + path: str | None = None, + timeout: int | None = None, +) -> AccountSpec: + """Resolve an account's credentials from overrides and ``${ENV_VAR}`` values. + + Explicit override arguments take precedence over the corresponding + :class:`AccountSpec` fields. The resolved string fields (``login``, + ``password``, ``server``, ``path``) have any ``${ENV_VAR}`` placeholders + substituted from the environment. + + Args: + account: Source account specification. + login: Optional explicit login override. + password: Optional explicit password override. + server: Optional explicit server override. + path: Optional explicit terminal path override. + timeout: Optional explicit connection timeout override. + + Returns: + A new :class:`AccountSpec` with resolved credentials and the original + symbols preserved. Raises ``ValueError`` (via + :func:`substitute_env_placeholders`) if a referenced environment + variable is not set. + """ + return AccountSpec( + symbols=account.symbols, + login=_resolve_login(login, account.login), + password=_resolve_field(password, account.password), + server=_resolve_field(server, account.server), + path=_resolve_field(path, account.path), + timeout=timeout if timeout is not None else account.timeout, + ) + + +def resolve_account_specs( + accounts: Sequence[AccountSpec], + *, + login: int | str | None = None, + password: str | None = None, + server: str | None = None, + path: str | None = None, + timeout: int | None = None, +) -> list[AccountSpec]: + """Resolve credentials for multiple accounts. + + Applies the same overrides and ``${ENV_VAR}`` substitution as + :func:`resolve_account_spec` to every account. + + Args: + accounts: Source account specifications. + login: Optional explicit login override applied to each account. + password: Optional explicit password override applied to each account. + server: Optional explicit server override applied to each account. + path: Optional explicit terminal path override applied to each account. + timeout: Optional explicit timeout override applied to each account. + + Returns: + Resolved account specifications in the original order. Raises + ``ValueError`` (via :func:`substitute_env_placeholders`) if a referenced + environment variable is not set. + """ + return [ + resolve_account_spec( + account, + login=login, + password=password, + server=server, + path=path, + timeout=timeout, + ) + for account in accounts + ] + + def _coerce_login(login: int | str | None) -> int | None: """Coerce a login value to int, treating empty strings as unset. @@ -1212,6 +1471,68 @@ def collect_latest_rates_for_accounts( return result +def collect_latest_rates_for_accounts_with_retries( + accounts: Sequence[AccountSpec], + timeframes: Sequence[int | str], + count: int, + *, + start_pos: int = 0, + base_config: Mt5Config | None = None, + retry_count: int = 0, + backoff_base: float = 2.0, +) -> dict[tuple[str, int], pd.DataFrame]: + """Collect latest rates across accounts, retrying transient MT5 failures. + + Wraps :func:`collect_latest_rates_for_accounts` with bounded exponential + backoff. Only ``pdmt5.Mt5TradingError`` and ``pdmt5.Mt5RuntimeError`` are + retried; other exceptions propagate immediately. The final failure is + re-raised once retries are exhausted. + + Args: + accounts: Account groups to read. Each must define at least one symbol. + timeframes: MT5 timeframes as integers or names (for example ``M1``). + count: Number of most recent bars to read per symbol/timeframe. + start_pos: Initial bar position offset. + base_config: Optional base configuration whose fields fill any value not + set on an individual account. + retry_count: Maximum number of retries after the first attempt. ``0`` + disables retries. + backoff_base: Base for exponential backoff. The delay before retry + attempt ``n`` (1-indexed) is ``backoff_base ** n`` seconds. + + Returns: + Mapping keyed by ``(symbol, timeframe_int)``. Propagates ``ValueError`` + for invalid inputs (see :func:`collect_latest_rates_for_accounts`) and + re-raises the last ``pdmt5.Mt5TradingError`` or ``pdmt5.Mt5RuntimeError`` + once retries are exhausted. + """ + attempts = max(retry_count, 0) + 1 + + def _collect() -> dict[tuple[str, int], pd.DataFrame]: + return collect_latest_rates_for_accounts( + accounts, + timeframes, + count, + start_pos=start_pos, + base_config=base_config, + ) + + for attempt in range(attempts - 1): + try: + return _collect() + except (Mt5TradingError, Mt5RuntimeError) as exc: + delay = backoff_base ** (attempt + 1) + logger.warning( + "Rate collection failed (attempt %d/%d): %s; retrying in %.1fs", + attempt + 1, + attempts, + exc, + delay, + ) + time.sleep(delay) + return _collect() + + def copy_rates_range( symbol: str, timeframe: int | str, diff --git a/pyproject.toml b/pyproject.toml index 5f03574..788b9c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mt5cli" -version = "0.5.2" +version = "0.5.3" 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"}] diff --git a/tests/test_history.py b/tests/test_history.py index e8b6356..4e9da81 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -38,6 +38,7 @@ from mt5cli.history import ( load_incremental_start_datetimes, load_rate_data, load_rate_data_from_connection, + load_rate_series_by_granularity, load_rate_series_from_sqlite, parse_sqlite_timestamp, quote_sqlite_identifier, @@ -2257,6 +2258,56 @@ class TestRateSourceHelpers: assert set(result) == {("EURUSD", 1)} assert len(result["EURUSD", 1]) == 2 + def test_load_rate_series_by_granularity(self, tmp_path: Path) -> None: + """Test loading rate series keyed by symbol and granularity name.""" + db_path = tmp_path / "granularity.db" + with sqlite3.connect(db_path) as conn: + conn.execute( + "CREATE TABLE rates(" + " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)", + ) + conn.executemany( + "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)", + [ + ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0), + ("EURUSD", 16385, "2024-01-01T00:00:00+00:00", 1.1), + ], + ) + create_rate_compatibility_views(conn) + + result = load_rate_series_by_granularity( + db_path, + ["EURUSD"], + ["M1", "H1"], + count=1, + ) + + assert set(result) == {("EURUSD", "M1"), ("EURUSD", "H1")} + + def test_load_rate_series_by_granularity_explicit_tables( + self, + tmp_path: Path, + ) -> None: + """Test explicit tables with None-symbol targets key by granularity.""" + db_path = tmp_path / "granularity-explicit.db" + with sqlite3.connect(db_path) as conn: + conn.execute("CREATE TABLE custom_view(time TEXT, close REAL)") + conn.execute( + "INSERT INTO custom_view(time, close) VALUES (?, ?)", + ("2024-01-01T00:00:00+00:00", 1.0), + ) + + result = load_rate_series_by_granularity( + db_path, + [], + ["M1"], + count=1, + explicit_tables=["custom_view"], + allow_missing_symbol=True, + ) + + assert set(result) == {(None, "M1")} + def test_load_rate_series_reuses_path_connection( self, tmp_path: Path, diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 131cc32..b971989 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -10,6 +10,7 @@ from unittest.mock import MagicMock, call import pandas as pd import pytest +from pdmt5 import Mt5RuntimeError, Mt5TradingError from pytest_mock import MockerFixture # noqa: TC002 if TYPE_CHECKING: @@ -22,11 +23,13 @@ from mt5cli.history import DEFAULT_HISTORY_TIMEFRAMES from mt5cli.sdk import ( AccountSpec, Mt5CliClient, + ThrottledHistoryUpdater, account_info, build_config, collect_history, collect_latest_rates, collect_latest_rates_for_accounts, + collect_latest_rates_for_accounts_with_retries, copy_rates_from, copy_rates_from_pos, copy_rates_range, @@ -45,6 +48,9 @@ from mt5cli.sdk import ( positions, recent_history_deals, recent_ticks, + resolve_account_spec, + resolve_account_specs, + substitute_env_placeholders, symbol_info, symbol_info_tick, symbols, @@ -1425,3 +1431,309 @@ class TestCollectLatestRatesForAccounts: collect_latest_rates_for_accounts(accounts, ["M1"], count=1) mt5_data_client.assert_not_called() + + +class TestCollectLatestRatesForAccountsWithRetries: + """Tests for collect_latest_rates_for_accounts_with_retries.""" + + def test_returns_result_on_first_success(self, mocker: MockerFixture) -> None: + """Test no retry happens when the first attempt succeeds.""" + expected = {("EURUSD", 1): pd.DataFrame()} + wrapped = mocker.patch( + "mt5cli.sdk.collect_latest_rates_for_accounts", + return_value=expected, + ) + sleep = mocker.patch("mt5cli.sdk.time.sleep") + accounts = [AccountSpec(symbols=["EURUSD"])] + + result = collect_latest_rates_for_accounts_with_retries( + accounts, + ["M1"], + count=1, + retry_count=3, + ) + + assert result is expected + assert wrapped.call_count == 1 + sleep.assert_not_called() + + def test_retries_then_succeeds(self, mocker: MockerFixture) -> None: + """Test transient MT5 errors are retried with exponential backoff.""" + expected = {("EURUSD", 1): pd.DataFrame()} + wrapped = mocker.patch( + "mt5cli.sdk.collect_latest_rates_for_accounts", + side_effect=[ + Mt5TradingError("boom"), + Mt5RuntimeError("boom"), + expected, + ], + ) + sleep = mocker.patch("mt5cli.sdk.time.sleep") + accounts = [AccountSpec(symbols=["EURUSD"])] + + result = collect_latest_rates_for_accounts_with_retries( + accounts, + ["M1"], + count=1, + retry_count=2, + backoff_base=2, + ) + + assert result is expected + assert wrapped.call_count == 3 + assert sleep.call_args_list == [call(2), call(4)] + + def test_reraises_after_exhausting_retries(self, mocker: MockerFixture) -> None: + """Test the final error is re-raised once retries are exhausted.""" + wrapped = mocker.patch( + "mt5cli.sdk.collect_latest_rates_for_accounts", + side_effect=Mt5RuntimeError("boom"), + ) + sleep = mocker.patch("mt5cli.sdk.time.sleep") + accounts = [AccountSpec(symbols=["EURUSD"])] + + with pytest.raises(Mt5RuntimeError, match="boom"): + collect_latest_rates_for_accounts_with_retries( + accounts, + ["M1"], + count=1, + retry_count=2, + ) + + assert wrapped.call_count == 3 + assert sleep.call_count == 2 + + def test_does_not_retry_unrelated_errors(self, mocker: MockerFixture) -> None: + """Test non-MT5 errors propagate without retrying.""" + wrapped = mocker.patch( + "mt5cli.sdk.collect_latest_rates_for_accounts", + side_effect=ValueError("bad input"), + ) + sleep = mocker.patch("mt5cli.sdk.time.sleep") + + with pytest.raises(ValueError, match="bad input"): + collect_latest_rates_for_accounts_with_retries( + [AccountSpec(symbols=["EURUSD"])], + ["M1"], + count=1, + retry_count=3, + ) + + assert wrapped.call_count == 1 + sleep.assert_not_called() + + +class TestSubstituteEnvPlaceholders: + """Tests for ${ENV_VAR} substitution.""" + + def test_substitutes_known_variables( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test placeholders are replaced with environment values.""" + monkeypatch.setenv("MT5_LOGIN", "12345") + monkeypatch.setenv("MT5_SERVER", "Broker-Demo") + + assert substitute_env_placeholders("${MT5_LOGIN}") == "12345" + assert substitute_env_placeholders("srv=${MT5_SERVER}!") == "srv=Broker-Demo!" + + def test_returns_plain_strings_unchanged(self) -> None: + """Test strings without placeholders are returned as-is.""" + assert substitute_env_placeholders("plain") == "plain" + + def test_raises_on_missing_variable( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test a missing environment variable raises a clear error.""" + monkeypatch.delenv("MT5_MISSING", raising=False) + + with pytest.raises(ValueError, match="'MT5_MISSING' is not set"): + substitute_env_placeholders("${MT5_MISSING}") + + +class TestResolveAccountSpec: + """Tests for resolve_account_spec and resolve_account_specs.""" + + def test_substitutes_env_placeholders_in_account( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test account string fields resolve ${ENV_VAR} placeholders.""" + monkeypatch.setenv("MT5_PASSWORD", "secret") + account = AccountSpec( + symbols=["EURUSD"], + login="${MT5_LOGIN}", + password="${MT5_PASSWORD}", + ) + monkeypatch.setenv("MT5_LOGIN", "999") + + resolved = resolve_account_spec(account) + + assert resolved.login == "999" + assert resolved.password == "secret" # noqa: S105 + assert resolved.symbols == ["EURUSD"] + + def test_explicit_overrides_take_precedence(self) -> None: + """Test explicit override values win over account fields.""" + account = AccountSpec(symbols=["EURUSD"], login=111, server="Acct") + + resolved = resolve_account_spec( + account, + login=222, + server="Override", + timeout=5000, + ) + + assert resolved.login == 222 + assert resolved.server == "Override" + assert resolved.timeout == 5000 + + def test_resolves_string_login_override( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test string login overrides expand ${ENV_VAR} placeholders.""" + monkeypatch.setenv("MT5_LOGIN", "777") + account = AccountSpec(symbols=["EURUSD"], login=111) + + resolved = resolve_account_spec(account, login="${MT5_LOGIN}") + + assert resolved.login == "777" + + def test_preserves_integer_login_without_coercion(self) -> None: + """Test integer logins remain integers after resolution.""" + account = AccountSpec(symbols=["EURUSD"], login=111) + + resolved = resolve_account_spec(account) + + assert resolved.login == 111 + assert isinstance(resolved.login, int) + + def test_raises_on_missing_env_variable( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test missing environment variables raise ValueError.""" + monkeypatch.delenv("MT5_NOPE", raising=False) + account = AccountSpec(symbols=["EURUSD"], server="${MT5_NOPE}") + + with pytest.raises(ValueError, match="'MT5_NOPE' is not set"): + resolve_account_spec(account) + + def test_resolve_account_specs_applies_to_all( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test resolve_account_specs resolves every account in order.""" + monkeypatch.setenv("MT5_SERVER", "Shared") + accounts = [ + AccountSpec(symbols=["EURUSD"], server="${MT5_SERVER}"), + AccountSpec(symbols=["GBPUSD"], server="Fixed"), + ] + + resolved = resolve_account_specs(accounts, timeout=1000) + + assert [a.server for a in resolved] == ["Shared", "Fixed"] + assert all(a.timeout == 1000 for a in resolved) + + +class TestThrottledHistoryUpdater: + """Tests for the throttled incremental history updater.""" + + def test_updates_every_call_when_interval_non_positive( + self, + mocker: MockerFixture, + ) -> None: + """Test interval_seconds <= 0 updates on every call.""" + update = mocker.patch("mt5cli.sdk.update_history") + client = MagicMock() + updater = ThrottledHistoryUpdater(output="history.db", interval_seconds=0) + + assert updater.update(client, ["EURUSD"]) is True + assert updater.update(client, ["EURUSD"]) is True + assert update.call_count == 2 + + def test_throttles_within_interval(self, mocker: MockerFixture) -> None: + """Test updates are skipped until the interval elapses.""" + update = mocker.patch("mt5cli.sdk.update_history") + monotonic = mocker.patch("mt5cli.sdk.time.monotonic") + # Calls: set(t=100), check(t=105), check(t=200), set(t=200). + monotonic.side_effect = [100.0, 105.0, 200.0, 200.0] + client = MagicMock() + updater = ThrottledHistoryUpdater(output="history.db", interval_seconds=60) + + assert updater.update(client, ["EURUSD"]) is True # first update at t=100 + assert updater.update(client, ["EURUSD"]) is False # t=105, throttled + assert updater.update(client, ["EURUSD"]) is True # t=200, elapsed + assert update.call_count == 2 + + def test_update_passes_expected_arguments( + self, + mocker: MockerFixture, + ) -> None: + """Test update_history is called with the configured arguments.""" + update = mocker.patch("mt5cli.sdk.update_history") + client = MagicMock() + updater = ThrottledHistoryUpdater( + output="history.db", + datasets={Dataset.rates}, + timeframes=["M1", "H1"], + flags="INFO", + lookback_hours=12.0, + with_views=True, + include_account_events=False, + ) + + updater.update(client, ["EURUSD", "GBPUSD"]) + + update.assert_called_once_with( + client=client, + output="history.db", + symbols=["EURUSD", "GBPUSD"], + datasets={Dataset.rates}, + timeframes=["M1", "H1"], + flags="INFO", + lookback_hours=12.0, + with_views=True, + include_account_events=False, + ) + + def test_propagates_errors_by_default(self, mocker: MockerFixture) -> None: + """Test MT5/SQLite errors propagate and do not advance the throttle.""" + mocker.patch( + "mt5cli.sdk.update_history", + side_effect=Mt5RuntimeError("boom"), + ) + updater = ThrottledHistoryUpdater(output="history.db") + + with pytest.raises(Mt5RuntimeError, match="boom"): + updater.update(MagicMock(), ["EURUSD"]) + + assert updater.last_update_monotonic is None + + @pytest.mark.parametrize( + "error", + [ + Mt5RuntimeError("boom"), + Mt5TradingError("trade failed"), + sqlite3.OperationalError("locked"), + ], + ) + def test_suppresses_errors_when_requested( + self, + mocker: MockerFixture, + error: Exception, + ) -> None: + """Test suppress_errors swallows recoverable errors and returns False.""" + mocker.patch( + "mt5cli.sdk.update_history", + side_effect=error, + ) + updater = ThrottledHistoryUpdater( + output="history.db", + suppress_errors=True, + ) + + assert updater.update(MagicMock(), ["EURUSD"]) is False + assert updater.last_update_monotonic is None diff --git a/uv.lock b/uv.lock index ee9db9f..b5683a0 100644 --- a/uv.lock +++ b/uv.lock @@ -487,7 +487,7 @@ wheels = [ [[package]] name = "mt5cli" -version = "0.5.2" +version = "0.5.3" source = { editable = "." } dependencies = [ { name = "click" },