Bump version from 0.6.0 to 0.6.1.

Co-authored-by: Cursor <cursoragent@cursor.com>

README.md

@@ -133,8 +133,29 @@ update_history_with_config(
 - **`update_history`**: incremental append based on existing SQLite `MAX(time)` per symbol (and timeframe for rates); account-level deals use a separate cursor when `include_account_events=True`.
 - **`rates` table**: normalized storage with `symbol` and `timeframe` columns.
 - **Rate compatibility views**: mt5cli manages all `rate_*` views. Naming is `rate_<symbol>__<timeframe>` when a symbol has one timeframe, otherwise `rate_<symbol>__<granularity>_<timeframe>` (for example `rate_EURUSD__M1_1`). Stale `rate_*` views are dropped and recreated when rates change for offline tools such as mteor optimize.
-- **Rate view resolution**: use `mt5cli.history.resolve_rate_view_name()` / `resolve_rate_view_names()` to map symbols and granularities to existing SQLite compatibility views without creating databases.
+- **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. 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.
+- **Latest closed bars**: use `collect_latest_closed_rates_for_accounts()` when downstream logic must exclude the still-forming current bar. It fetches `count + 1` bars at `start_pos=0`, drops the last row with `drop_forming_rate_bar()`, and validates each series is non-empty. `collect_latest_closed_rates_by_granularity()` returns the same data keyed by `(symbol, granularity_name)` such as `("EURUSD", "M1")`.
+
+```python
+from mt5cli import AccountSpec, collect_latest_closed_rates_by_granularity
+
+rates = collect_latest_closed_rates_by_granularity(
+    [AccountSpec(symbols=["EURUSD", "GBPUSD"], login=12345)],
+    ["M1", "H1"],
+    count=500,
+    retry_count=3,
+)
+eurusd_m1 = rates["EURUSD", "M1"]  # closed bars only
+```
+
+- **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`, `ValueError`, `OSError`, and MT5 client capability errors for history API methods without advancing the throttle (other `AttributeError` / `TypeError` values always propagate).
+- **Trading session helpers**: use `mt5_trading_session()` for a trading-capable `pdmt5.Mt5TradingClient` that initializes/logs in via `Mt5Config.path` and always shuts down safely. Pair with `detect_position_side()`, `calculate_margin_and_volume()`, and `determine_order_limits()` for generic position and sizing utilities. The read-only `mt5_session()` / `Mt5CliClient` SDK is unchanged.
+- **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.
 
@@ -144,6 +165,63 @@ update_history_with_config(
 - Windows OS (MetaTrader 5 requirement)
 - MetaTrader 5 platform installed
 
+### Migration note for mteor
+
+Replace local MT5 lifecycle and trading helper code with mt5cli imports:
+
+```python
+# Before (local mteor helpers)
+# with local_mt5_trading_session(config) as client:
+#     side = local_detect_position_side(client, symbol)
+#     sizing = local_calculate_margin_and_volume(client, symbol, unit_ratio, preserved_ratio)
+#     limits = local_determine_order_limits(client, symbol, side, sl_ratio, tp_ratio)
+
+# After (mt5cli shared layer)
+from pdmt5 import Mt5Config
+from mt5cli import (
+    calculate_margin_and_volume,
+    detect_position_side,
+    determine_order_limits,
+    mt5_trading_session,
+)
+
+with mt5_trading_session(
+    Mt5Config(path=terminal_path, login=login), retry_count=2
+) as client:
+    side = detect_position_side(client, symbol)
+    sizing = calculate_margin_and_volume(
+        client, symbol, unit_margin_ratio=0.5, preserved_margin_ratio=0.2
+    )
+    if side is not None:
+        limits = determine_order_limits(
+            client,
+            symbol,
+            side,
+            stop_loss_limit_ratio=0.01,
+            take_profit_limit_ratio=0.02,
+        )
+```
+
+Throttled history updates use a separate read-only session:
+
+```python
+from pdmt5 import Mt5Config, Mt5DataClient
+
+from mt5cli import ThrottledHistoryUpdater
+
+updater = ThrottledHistoryUpdater(
+    output="history.db", interval_seconds=60, suppress_errors=True
+)
+client = Mt5DataClient(config=Mt5Config(login=login))
+client.initialize_and_login_mt5()
+try:
+    updater.update(client, ["EURUSD"])
+finally:
+    client.shutdown()
+```
+
+Read-only collectors can keep using `mt5_session()` and `Mt5CliClient` without changes.
+
 ## Development
 
 ```bash

docs/api/history.md

@@ -133,8 +133,8 @@ The `update_history` SDK path uses the same base tables and optional
 ### Rate view resolution
 
 Downstream tools can resolve mt5cli-managed compatibility view names from an
-existing SQLite history database without creating files or guessing legacy
-naming schemes:
+existing SQLite history database without creating files or guessing naming
+schemes:
 
 ```python
 from pathlib import Path
@@ -183,3 +183,47 @@ rates = load_rate_data(Path("history.db"), view, count=1000)
 The loader accepts close-based OHLC rate data or tick-like bid/ask data. It
 validates that `time` exists, parses timestamps with pandas, and returns a
 DataFrame indexed by ascending `DatetimeIndex` named `time`.
+
+### Multi-series rate loading
+
+For loading many rate series at once, build neutral `RateTarget` pairs and load
+them from SQLite in one call. View names are resolved via the same
+compatibility-view rules, or you can pass `explicit_tables` to bypass resolution:
+
+```python
+from pathlib import Path
+
+from mt5cli import build_rate_targets, load_rate_series_from_sqlite
+
+targets = build_rate_targets(["EURUSD", "GBPUSD"], ["M1", "H1"])
+series = load_rate_series_from_sqlite(Path("history.db"), targets, count=1000)
+frame = series["EURUSD", 1]  # keyed by (symbol, integer timeframe)
+```
+
+- `build_rate_targets()` returns `RateTarget(symbol, timeframe)` pairs in
+  row-major order, normalizing timeframe names such as `"M1"` to their integer
+  values; set `allow_missing_symbol=True` to address series solely by
+  `explicit_tables` (targets carry `symbol=None`).
+- `resolve_rate_tables()` maps targets to table or view names and validates that
+  any `explicit_tables` count matches the target count. Pass
+  `require_existing=True` to raise `ValueError` instead of returning a
+  best-guess name when the database or managed view is missing. When
+  `explicit_tables` is provided, names are returned as-is and
+  `require_existing` is ignored.
+- `load_rate_series_from_sqlite()` returns a mapping keyed by
+  `(symbol, integer timeframe)`. Unless `explicit_tables` is supplied, it
+  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)
+  ```

docs/api/index.md

@@ -18,6 +18,10 @@ Utility module providing constants, enums, Click parameter types, and helper fun
 
 Programmatic SDK for read-only MetaTrader 5 data collection. Returns pandas DataFrames and provides `collect_history` for SQLite bulk collection.
 
+### [Trading](trading.md)
+
+Trading-capable session management and operational helpers built on `pdmt5.Mt5TradingClient`. Complements the read-only SDK without changing existing `Mt5CliClient` behavior.
+
 ### [History Collection (SQLite)](history.md)
 
 SQLite storage helpers for the `collect-history` command schema, incremental updates, deduplication, indexes, and optional views.
@@ -28,8 +32,9 @@ The package follows a simple architecture built on top of pdmt5:
 
 1. **CLI Layer** (`cli.py`): Typer application with subcommands that delegate to the SDK and export results.
 2. **SDK Layer** (`sdk.py`): Read-only data access functions, `Mt5CliClient`, and `collect_history` orchestration.
-3. **Utils Layer** (`utils.py`): Constants, enums, custom Click parameter types, parsing helpers, and format detection/export utilities.
-4. **Data Layer** (via `pdmt5`): Uses `Mt5DataClient` and `Mt5Config` from the pdmt5 package for all MetaTrader 5 data access.
+3. **Trading Layer** (`trading.py`): Trading-capable sessions and operational helpers on `Mt5TradingClient`.
+4. **Utils Layer** (`utils.py`): Constants, enums, custom Click parameter types, parsing helpers, and format detection/export utilities.
+5. **Data Layer** (via `pdmt5`): Uses `Mt5DataClient`, `Mt5TradingClient`, and `Mt5Config` from the pdmt5 package for MetaTrader 5 access.
 
 ## Usage Guidelines
 

docs/api/sdk.md

@@ -1,3 +1,114 @@
 # 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
+)
+```
+
+### Latest closed rate bars
+
+MetaTrader 5 `start_pos=0` includes the still-forming current bar as the last
+row. `collect_latest_closed_rates_for_accounts()` fetches `count + 1` bars,
+drops that row with `drop_forming_rate_bar()`, and validates each series is
+non-empty. Use `collect_latest_closed_rates_by_granularity()` when callers
+prefer keys such as `("EURUSD", "M1")` instead of integer timeframes.
+
+```python
+from mt5cli import AccountSpec, collect_latest_closed_rates_by_granularity
+
+rates = collect_latest_closed_rates_by_granularity(
+    [AccountSpec(symbols=["EURUSD"], login=12345)],
+    ["M1", "H1"],
+    count=500,
+    retry_count=3,
+)
+closed_m1 = rates["EURUSD", "M1"]
+```
+
+### 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 recoverable errors (`Mt5TradingError`, `Mt5RuntimeError`,
+`sqlite3.Error`, `ValueError`, `OSError`, and MT5 client capability
+`AttributeError` / `TypeError` for history API methods) propagate so the caller
+controls logging; pass `suppress_errors=True` to swallow them and return
+`False` without advancing the throttle. Other `AttributeError` / `TypeError`
+values always propagate. Input validation (`_resolve_update_history_request`)
+runs before any MT5 or SQLite calls, but when `suppress_errors=True` the
+resulting `ValueError` is suppressed along with other recoverable errors.
+
+## Trading-capable sessions
+
+For order placement and trading calculations, use the dedicated
+[Trading module](trading.md). The read-only `Mt5CliClient` and `mt5_session()`
+helpers in this module are unchanged.

docs/api/trading.md

@@ -0,0 +1,70 @@
+# Trading Module
+
+::: mt5cli.trading
+
+## Trading-capable MT5 sessions
+
+`mt5_trading_session()` complements the read-only `mt5_session()` helper in
+`sdk.py`. It yields a connected `pdmt5.Mt5TradingClient`, uses
+`Mt5Config.path` to launch the terminal when configured, and always calls
+`shutdown()` on exit.
+
+```python
+from pdmt5 import Mt5Config
+
+from mt5cli import mt5_trading_session
+
+with mt5_trading_session(
+    Mt5Config(path=r"C:\Program Files\MetaTrader 5\terminal64.exe", login=12345),
+    retry_count=2,
+) as client:
+    positions = client.positions_get_as_df(symbol="EURUSD")
+```
+
+The read-only `Mt5CliClient` / `mt5_session()` API is unchanged.
+
+## Operational trading helpers
+
+These helpers are strategy-agnostic and do not depend on signal detection,
+betting logic, or scheduling code in downstream applications.
+
+```python
+from mt5cli import (
+    calculate_margin_and_volume,
+    detect_position_side,
+    determine_order_limits,
+)
+
+side = detect_position_side(client, "EURUSD")
+sizing = calculate_margin_and_volume(
+    client,
+    "EURUSD",
+    unit_margin_ratio=0.5,
+    preserved_margin_ratio=0.2,
+)
+limits = determine_order_limits(
+    client,
+    "EURUSD",
+    side="long",
+    stop_loss_limit_ratio=0.01,
+    take_profit_limit_ratio=0.02,
+)
+```
+
+Protective ratios must satisfy `0 <= ratio < 1`; `0` omits that level.
+`calculate_margin_and_volume()` clamps negative `margin_free` to `0.0`
+before sizing.
+
+## Migration from mteor-local helpers
+
+| mteor-local concern                                      | mt5cli replacement                              |
+| -------------------------------------------------------- | ----------------------------------------------- |
+| Manual terminal spawn/kill around trading code           | `mt5_trading_session()`                         |
+| Local position-side detection                            | `detect_position_side()`                        |
+| Local margin/volume sizing                               | `calculate_margin_and_volume()`                 |
+| Local SL/TP price derivation                             | `determine_order_limits()`                      |
+| Throttled SQLite history loop with ad-hoc error handling | `ThrottledHistoryUpdater(suppress_errors=True)` |
+
+Keep read-only data collection on `mt5_session()` / `Mt5CliClient`; use
+`mt5_trading_session()` only where order placement or trading calculations are
+required.

mkdocs.yml

@@ -58,6 +58,7 @@ nav:
       - Overview: api/index.md
       - CLI: api/cli.md
       - SDK: api/sdk.md
+      - Trading: api/trading.md
       - History Collection (SQLite): api/history.md
       - Utils: api/utils.md
 

mt5cli/__init__.py

@@ -2,13 +2,34 @@
 
 from importlib.metadata import version
 
-from .history import load_rate_data, load_rate_data_from_connection
+from .history import (
+    RateTarget,
+    build_rate_targets,
+    build_rate_view_name,
+    drop_forming_rate_bar,
+    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,
+    resolve_history_timeframes,
+    resolve_rate_tables,
+    resolve_rate_view_name,
+    resolve_rate_view_names,
+)
 from .sdk import (
+    AccountSpec,
     Mt5CliClient,
+    ThrottledHistoryUpdater,
     account_info,
     build_config,
     collect_history,
+    collect_latest_closed_rates_by_granularity,
+    collect_latest_closed_rates_for_accounts,
     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,
@@ -20,12 +41,16 @@ from .sdk import (
     latest_rates,
     market_book,
     minimum_margins,
+    mt5_session,
     mt5_summary,
     mt5_summary_as_df,
     orders,
     positions,
     recent_history_deals,
     recent_ticks,
+    resolve_account_spec,
+    resolve_account_specs,
+    substitute_env_placeholders,
     symbol_info,
     symbol_info_tick,
     symbols,
@@ -36,30 +61,56 @@ from .sdk import (
 from .sdk import (
     version as mt5_version,
 )
+from .trading import (
+    calculate_margin_and_volume,
+    detect_position_side,
+    determine_order_limits,
+    mt5_trading_session,
+)
 from .utils import (
+    TICK_FLAG_MAP,
+    TIMEFRAME_MAP,
     Dataset,
     IfExists,
     detect_format,
     export_dataframe,
     export_dataframe_to_sqlite,
+    parse_datetime,
+    parse_tick_flags,
+    parse_timeframe,
 )
 
 __version__ = version(__package__) if __package__ else None
 
 __all__ = [
+    "TICK_FLAG_MAP",
+    "TIMEFRAME_MAP",
+    "AccountSpec",
     "Dataset",
     "IfExists",
     "Mt5CliClient",
+    "RateTarget",
+    "ThrottledHistoryUpdater",
     "account_info",
     "build_config",
+    "build_rate_targets",
+    "build_rate_view_name",
+    "calculate_margin_and_volume",
     "collect_history",
+    "collect_latest_closed_rates_by_granularity",
+    "collect_latest_closed_rates_for_accounts",
     "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",
     "copy_ticks_from",
     "copy_ticks_range",
     "detect_format",
+    "detect_position_side",
+    "determine_order_limits",
+    "drop_forming_rate_bar",
     "export_dataframe",
     "export_dataframe_to_sqlite",
     "history_deals",
@@ -68,15 +119,31 @@ __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",
+    "mt5_session",
     "mt5_summary",
     "mt5_summary_as_df",
+    "mt5_trading_session",
     "mt5_version",
     "orders",
+    "parse_datetime",
+    "parse_tick_flags",
+    "parse_timeframe",
     "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",

mt5cli/history.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import sqlite3
+from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal, cast
@@ -20,7 +21,7 @@ from .utils import (
 )
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Sequence
+    from collections.abc import Callable, Mapping, Sequence
 
     from pdmt5 import Mt5DataClient
 
@@ -105,6 +106,23 @@ def resolve_granularity_name(timeframe: int) -> str:
     return str(timeframe)
 
 
+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 build_rate_view_name(
     *,
     symbol: str,
@@ -135,14 +153,17 @@ def _require_non_empty_identifier(identifier: str, kind: str) -> str:
 
 
 def _open_history_connection(
-    conn_or_path: SqliteConnOrPath,
+    conn_or_path: SqliteConnOrPath | None,
 ) -> tuple[sqlite3.Connection | None, bool]:
     """Open a read-only SQLite connection when given a path.
 
     Returns:
-        A connection and whether the caller should close it. When the path does
-        not exist, returns ``(None, False)`` without creating a database file.
+        A connection and whether the caller should close it. When ``conn_or_path``
+        is None or the path does not exist, returns ``(None, False)`` without
+        creating a database file.
     """
+    if conn_or_path is None:
+        return None, False
     if isinstance(conn_or_path, sqlite3.Connection):
         return conn_or_path, False
     path = Path(conn_or_path)
@@ -376,7 +397,7 @@ def _resolve_rate_view_name_from_context(
 
 
 def resolve_rate_view_name(
-    conn_or_path: SqliteConnOrPath,
+    conn_or_path: SqliteConnOrPath | None,
     symbol: str,
     granularity: str,
     *,
@@ -385,7 +406,9 @@ def resolve_rate_view_name(
     """Resolve the mt5cli-managed rate compatibility view name.
 
     Args:
-        conn_or_path: SQLite database path or open connection.
+        conn_or_path: SQLite database path or open connection. When None or a
+            non-existing path and ``require_existing`` is False, the deterministic
+            default view name is returned without creating a database file.
         symbol: Symbol stored in the normalized ``rates`` table.
         granularity: Timeframe name (for example ``M1``) or integer string.
         require_existing: When True, require the database and a managed view to exist.
@@ -429,7 +452,7 @@ def resolve_rate_view_name(
 
 
 def resolve_rate_view_names(
-    conn_or_path: SqliteConnOrPath,
+    conn_or_path: SqliteConnOrPath | None,
     symbols: Sequence[str],
     granularities: Sequence[str],
     *,
@@ -438,7 +461,9 @@ def resolve_rate_view_names(
     """Resolve rate compatibility view names for symbol and granularity pairs.
 
     Args:
-        conn_or_path: SQLite database path or open connection.
+        conn_or_path: SQLite database path or open connection. When None or a
+            non-existing path and ``require_existing`` is False, deterministic
+            default view names are returned without creating a database file.
         symbols: Symbols stored in the normalized ``rates`` table.
         granularities: Timeframe names (for example ``M1``) or integer strings.
         require_existing: When True, require the database and managed views to exist.
@@ -482,6 +507,271 @@ def resolve_rate_view_names(
             conn.close()
 
 
+@dataclass(frozen=True)
+class RateTarget:
+    """A single rate series identified by symbol and timeframe.
+
+    Attributes:
+        symbol: MT5 symbol name, or None when the rate series is addressed only
+            by an explicit table (for example a custom SQLite view).
+        timeframe: MT5 timeframe as an integer or name (for example ``M1``).
+    """
+
+    symbol: str | None
+    timeframe: int | str
+
+    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))
+
+    @property
+    def timeframe_int(self) -> int:
+        """Return the timeframe as its integer MT5 value."""
+        return cast("int", self.timeframe)
+
+
+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 resolve_rate_tables(
+    conn_or_path: SqliteConnOrPath | None,
+    targets: Sequence[RateTarget],
+    explicit_tables: Sequence[str] | None = None,
+    *,
+    require_existing: bool = False,
+) -> list[str]:
+    """Resolve SQLite table or view names for rate targets.
+
+    Args:
+        conn_or_path: SQLite database path or open connection. May be None when
+            ``explicit_tables`` is provided, or when ``require_existing`` is
+            False and deterministic default view names are sufficient.
+        targets: Rate targets to resolve.
+        explicit_tables: Optional explicit table or view names. When provided,
+            they are used as-is and must match the number of targets.
+        require_existing: When True, require the database and managed views to
+            exist for each symbol target. Ignored when ``explicit_tables`` is
+            provided.
+
+    Returns:
+        Table or view names aligned with ``targets``.
+
+    Raises:
+        ValueError: If ``targets`` is empty, ``explicit_tables`` length does not
+            match the target count, a target without a symbol is resolved
+            without an explicit table, or ``require_existing`` is True and the
+            database or a managed view is missing.
+    """
+    target_list = list(targets)
+    if not target_list:
+        msg = "At least one rate target is required."
+        raise ValueError(msg)
+    if explicit_tables is not None:
+        tables = list(explicit_tables)
+        if len(tables) != len(target_list):
+            msg = (
+                f"Expected {len(target_list)} explicit table(s) "
+                f"to match the targets, got {len(tables)}."
+            )
+            raise ValueError(msg)
+        return tables
+    if any(target.symbol is None for target in target_list):
+        msg = (
+            "Cannot resolve a rate table for a target without a symbol; "
+            "provide explicit_tables."
+        )
+        raise ValueError(msg)
+    conn, should_close = _open_history_connection(conn_or_path)
+    try:
+        if conn is None:
+            if require_existing:
+                path = (
+                    conn_or_path
+                    if isinstance(conn_or_path, (Path, str))
+                    else "database"
+                )
+                msg = f"SQLite database not found: {path}"
+                raise ValueError(msg)
+            timeframe_counts = None
+            existing_views: set[str] = set()
+        else:
+            timeframe_counts = _load_rates_timeframe_counts(conn)
+            existing_views = _load_existing_rate_views(conn)
+        resolved: list[str] = []
+        for target in target_list:
+            symbol = cast("str", target.symbol)
+            timeframe = target.timeframe_int
+            resolved.append(
+                _resolve_rate_view_name_from_context(
+                    symbol=symbol,
+                    timeframe=timeframe,
+                    granularity_name=resolve_granularity_name(timeframe),
+                    timeframe_counts=timeframe_counts,
+                    existing_views=existing_views,
+                    require_existing=require_existing,
+                ),
+            )
+        return resolved
+    finally:
+        if should_close and conn is not None:
+            conn.close()
+
+
+def load_rate_series_from_sqlite(
+    conn_or_path: SqliteConnOrPath,
+    targets: Sequence[RateTarget],
+    count: int,
+    explicit_tables: Sequence[str] | None = None,
+) -> dict[tuple[str | None, int], pd.DataFrame]:
+    """Load multiple rate series from a SQLite database.
+
+    Args:
+        conn_or_path: SQLite database path or open connection.
+        targets: Rate targets to load. Each ``(symbol, timeframe_int)`` pair
+            must be unique.
+        count: Number of most recent rows to load per series.
+        explicit_tables: Optional explicit table or view names matching targets.
+            When omitted, managed ``rate_*`` compatibility views must already
+            exist in the database.
+
+    Returns:
+        Mapping keyed by ``(symbol, timeframe_int)`` to each rate DataFrame.
+
+    Raises:
+        ValueError: If ``count`` is not positive, targets are empty, duplicate
+            ``(symbol, timeframe_int)`` pairs are present, or table resolution
+            fails.
+    """
+    if count <= 0:
+        msg = "count must be positive."
+        raise ValueError(msg)
+    target_list = list(targets)
+    if not target_list:
+        msg = "At least one rate target is required."
+        raise ValueError(msg)
+    if explicit_tables is None and any(target.symbol is None for target in target_list):
+        msg = (
+            "Cannot resolve a rate table for a target without a symbol; "
+            "provide explicit_tables."
+        )
+        raise ValueError(msg)
+    seen_keys: set[tuple[str | None, int]] = set()
+    for target in target_list:
+        key = (target.symbol, target.timeframe_int)
+        if key in seen_keys:
+            symbol_repr = repr(target.symbol)
+            msg = f"Duplicate rate target: ({symbol_repr}, {target.timeframe_int})"
+            raise ValueError(msg)
+        seen_keys.add(key)
+    tables = (
+        resolve_rate_tables(None, target_list, explicit_tables)
+        if explicit_tables is not None
+        else None
+    )
+    conn, should_close = _open_existing_sqlite_database(conn_or_path)
+    try:
+        resolved_tables = tables or resolve_rate_tables(
+            conn,
+            target_list,
+            require_existing=True,
+        )
+        return {
+            (target.symbol, target.timeframe_int): load_rate_data_from_connection(
+                conn,
+                table,
+                count=count,
+            )
+            for target, table in zip(target_list, resolved_tables, strict=True)
+        }
+    finally:
+        if should_close:
+            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)
@@ -765,7 +1055,20 @@ def drop_duplicates_in_table(
     )
 
 
-DedupScope = tuple[str, tuple[object, ...]]
+@dataclass(frozen=True)
+class DedupScope:
+    """Scoped deduplication predicate and the columns it references.
+
+    Attributes:
+        where: SQL predicate appended to the duplicate-removal query.
+        params: Parameters bound to the scope predicate.
+        required_columns: Columns that must be present in the written table for
+            the scope to run.
+    """
+
+    where: str
+    params: tuple[object, ...]
+    required_columns: frozenset[str]
 
 
 def _record_dedup_scope(
@@ -773,17 +1076,25 @@ def _record_dedup_scope(
     dataset: Dataset,
     scope_where: str,
     scope_params: tuple[object, ...],
+    required_columns: frozenset[str],
 ) -> None:
-    dedup_scopes.setdefault(dataset, []).append((scope_where, scope_params))
+    dedup_scopes.setdefault(dataset, []).append(
+        DedupScope(scope_where, scope_params, required_columns),
+    )
 
 
 def deduplicate_history_tables(
     conn: sqlite3.Connection,
     written_columns: dict[Dataset, set[str]],
     written_tables: set[Dataset],
-    dedup_scopes: dict[Dataset, list[DedupScope]] | None = None,
+    dedup_scopes: Mapping[Dataset, Sequence[DedupScope]] | None = None,
 ) -> None:
-    """Deduplicate appended history tables by stable identifiers."""
+    """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())
@@ -802,16 +1113,19 @@ def deduplicate_history_tables(
                 table,
             )
             continue
-        scopes = dedup_scopes.get(dataset, []) if dedup_scopes else []
+        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_where, scope_params in scopes:
+            for scope in scopes:
                 drop_duplicates_in_table(
                     cursor,
                     table,
                     list(keys),
                     keep="last",
-                    scope_where=scope_where,
-                    scope_params=scope_params,
+                    scope_where=scope.where,
+                    scope_params=scope.params,
                 )
             continue
         drop_duplicates_in_table(cursor, table, list(keys), keep="last")
@@ -1178,6 +1492,7 @@ def _write_incremental_rates(
                     Dataset.rates,
                     "symbol = ? AND timeframe = ? AND time >= ?",
                     (symbol, timeframe, start_date),
+                    frozenset({"symbol", "timeframe", "time"}),
                 )
 
 
@@ -1216,6 +1531,7 @@ def _write_incremental_ticks(
                 Dataset.ticks,
                 "symbol = ? AND time >= ?",
                 (symbol, start_date),
+                frozenset({"symbol", "time"}),
             )
 
 
@@ -1254,6 +1570,7 @@ def _write_incremental_history_orders(
                 Dataset.history_orders,
                 "symbol = ? AND time >= ?",
                 (symbol, start_date),
+                frozenset({"symbol", "time"}),
             )
 
 
@@ -1307,6 +1624,7 @@ def _write_incremental_history_deals(
                         Dataset.history_deals,
                         "symbol = ? AND time >= ?",
                         (symbol, start_by_symbol[symbol, None]),
+                        frozenset({"symbol", "time"}),
                     )
             if "type" in columns:
                 _record_dedup_scope(
@@ -1314,6 +1632,7 @@ def _write_incremental_history_deals(
                     Dataset.history_deals,
                     f"type NOT IN {_TRADE_DEAL_TYPES_SQL} AND time >= ?",
                     (account_event_start,),
+                    frozenset({"type", "time"}),
                 )
             if "type" not in columns and "symbol" in columns:
                 _record_dedup_scope(
@@ -1321,6 +1640,7 @@ def _write_incremental_history_deals(
                     Dataset.history_deals,
                     "(symbol IS NULL OR symbol = '') AND time >= ?",
                     (account_event_start,),
+                    frozenset({"symbol", "time"}),
                 )
         return
     start_by_symbol = load_incremental_start_datetimes(
@@ -1348,6 +1668,7 @@ def _write_incremental_history_deals(
                 Dataset.history_deals,
                 "symbol = ? AND time >= ?",
                 (symbol, start_date),
+                frozenset({"symbol", "time"}),
             )
 
 

mt5cli/sdk.py

@@ -4,20 +4,25 @@ from __future__ import annotations
 
 import json
 import logging
+import os
+import re
 import sqlite3
+import time
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 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,
     create_history_indexes,
     create_positions_reconstructed_view,
+    drop_forming_rate_bar,
+    resolve_granularity_name,
     resolve_history_datasets,
     resolve_history_tick_flags,
     resolve_history_timeframes,
@@ -39,12 +44,74 @@ T = TypeVar("T")
 
 logger = logging.getLogger(__name__)
 
+_RECOVERABLE_HISTORY_UPDATE_ERRORS: tuple[type[BaseException], ...] = (
+    Mt5TradingError,
+    Mt5RuntimeError,
+    sqlite3.Error,
+    ValueError,
+    OSError,
+)
+
+_MT5_CLIENT_CAPABILITY_METHODS: frozenset[str] = frozenset({
+    "copy_rates_range_as_df",
+    "copy_ticks_range_as_df",
+    "history_deals_get_as_df",
+    "history_orders_get_as_df",
+})
+_MT5_HISTORY_MODULE = Path(__file__).with_name("history.py").resolve()
+_MT5_HISTORY_CLIENT_CALL_FUNCTIONS: frozenset[str] = frozenset({
+    "write_rates_dataset",
+    "write_ticks_dataset",
+    "write_history_dataset",
+    "_write_incremental_history_deals",
+})
+_NON_CALLABLE_TYPE_ERROR = re.compile(r"^'[^']+' object is not callable$")
+
+
+def _is_non_callable_history_client_type_error(exc: TypeError) -> bool:
+    """Return whether a TypeError came from calling a history client API attribute."""
+    if not _NON_CALLABLE_TYPE_ERROR.match(str(exc)):
+        return False
+    tb = exc.__traceback__
+    if tb is None:
+        return False
+    while tb.tb_next is not None:
+        tb = tb.tb_next
+    frame = tb.tb_frame
+    return (
+        frame.f_code.co_name in _MT5_HISTORY_CLIENT_CALL_FUNCTIONS
+        and Path(frame.f_code.co_filename).resolve() == _MT5_HISTORY_MODULE
+    )
+
+
+def _is_mt5_client_capability_error(exc: BaseException) -> bool:
+    """Return whether an error indicates an incompatible MT5 client API surface."""
+    if isinstance(exc, AttributeError):
+        msg = str(exc)
+        if msg.startswith("MT5 client is missing required method:"):
+            return True
+        name = getattr(exc, "name", None)
+        return isinstance(name, str) and name in _MT5_CLIENT_CAPABILITY_METHODS
+    if isinstance(exc, TypeError):
+        msg = str(exc)
+        if msg.startswith("MT5 client attribute is not callable:"):
+            return True
+        return _is_non_callable_history_client_type_error(exc)
+    return False
+
+
 __all__ = [
+    "AccountSpec",
     "Mt5CliClient",
+    "ThrottledHistoryUpdater",
     "account_info",
     "build_config",
     "collect_history",
+    "collect_latest_closed_rates_by_granularity",
+    "collect_latest_closed_rates_for_accounts",
     "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",
@@ -56,12 +123,16 @@ __all__ = [
     "latest_rates",
     "market_book",
     "minimum_margins",
+    "mt5_session",
     "mt5_summary",
     "mt5_summary_as_df",
     "orders",
     "positions",
     "recent_history_deals",
     "recent_ticks",
+    "resolve_account_spec",
+    "resolve_account_specs",
+    "substitute_env_placeholders",
     "symbol_info",
     "symbol_info_tick",
     "symbols",
@@ -118,6 +189,12 @@ def _require_positive(value: float, name: str) -> None:
         raise ValueError(msg)
 
 
+def _require_non_negative(value: int, name: str) -> None:
+    if value < 0:
+        msg = f"{name} must be non-negative."
+        raise ValueError(msg)
+
+
 def _call_required_client_method(client: Mt5DataClient, name: str) -> object:
     try:
         method = getattr(client, name)
@@ -277,6 +354,26 @@ def _run_with_client(
         return fetch_fn(client)
 
 
+@contextmanager
+def mt5_session(config: Mt5Config | None = None) -> Iterator[Mt5CliClient]:
+    """Open an MT5 terminal session and yield a connected client.
+
+    Launches the MetaTrader 5 terminal using ``Mt5Config.path`` (when set),
+    logs in, yields a connected :class:`Mt5CliClient`, and always shuts the
+    terminal down on exit.
+
+    Args:
+        config: MT5 connection configuration. Defaults to an empty config that
+            attaches to a running terminal.
+
+    Yields:
+        Connected ``Mt5CliClient`` bound to the session.
+    """
+    mt5_config = config or build_config()
+    with _connected_client(mt5_config) as client:
+        yield Mt5CliClient.from_connected_client(client)
+
+
 class Mt5CliClient:
     """Programmatic client for read-only MetaTrader 5 data access."""
 
@@ -937,6 +1034,135 @@ 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, recoverable errors (``Mt5TradingError``,
+                ``Mt5RuntimeError``, ``sqlite3.Error``, ``ValueError``,
+                ``OSError``, and MT5 client capability ``AttributeError`` /
+                ``TypeError`` for history API methods) raised during an update
+                are swallowed and :meth:`update` returns False without advancing
+                the throttle. Other ``AttributeError`` / ``TypeError`` values
+                always propagate. When False (default), recoverable 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.
+            When ``suppress_errors`` is False, recoverable update failures
+            propagate to the caller.
+
+        Raises:
+            AttributeError: MT5 client capability mismatch when
+                ``suppress_errors`` is False, or any other attribute error.
+            TypeError: MT5 client capability mismatch when ``suppress_errors``
+                is False, or any other type error.
+        """
+        if not self.should_update():
+            return False
+        try:
+            _resolve_update_history_request(
+                output=self.output,
+                symbols=symbols,
+                datasets=self.datasets,
+                timeframes=self.timeframes,
+                flags=self.flags,
+                lookback_hours=self.lookback_hours,
+                date_to=None,
+            )
+            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 _RECOVERABLE_HISTORY_UPDATE_ERRORS:
+            if self.suppress_errors:
+                logger.warning("Suppressed history update error", exc_info=True)
+                return False
+            raise
+        except (AttributeError, TypeError) as exc:
+            if self.suppress_errors and _is_mt5_client_capability_error(exc):
+                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],
@@ -1075,6 +1301,433 @@ def collect_latest_rates(
     )
 
 
+@dataclass(frozen=True)
+class AccountSpec:
+    """Connection parameters and symbols for one MT5 account group.
+
+    Attributes:
+        symbols: Symbols to load latest rates for under this account.
+        login: Trading account login. String values are coerced to int when
+            non-empty.
+        password: Trading account password.
+        server: Trading server name.
+        path: Path to the MetaTrader5 terminal EXE file.
+        timeout: Connection timeout in milliseconds.
+    """
+
+    symbols: Sequence[str]
+    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<name>[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.
+
+    Returns:
+        Integer login, or None when unset or an empty string.
+    """
+    if login is None or isinstance(login, int):
+        return login
+    text = login.strip()
+    if not text:
+        return None
+    return int(text)
+
+
+def _build_account_config(
+    account: AccountSpec,
+    base_config: Mt5Config | None,
+) -> Mt5Config:
+    """Build an ``Mt5Config`` for an account, falling back to ``base_config``.
+
+    Returns:
+        Merged MT5 configuration for the account.
+    """
+    login = _coerce_login(account.login)
+    if login is None and base_config is not None:
+        login = base_config.login
+    return build_config(
+        path=account.path or (base_config.path if base_config else None),
+        login=login,
+        password=account.password or (base_config.password if base_config else None),
+        server=account.server or (base_config.server if base_config else None),
+        timeout=account.timeout
+        if account.timeout is not None
+        else (base_config.timeout if base_config else None),
+    )
+
+
+def collect_latest_rates_for_accounts(
+    accounts: Sequence[AccountSpec],
+    timeframes: Sequence[int | str],
+    count: int,
+    *,
+    start_pos: int = 0,
+    base_config: Mt5Config | None = None,
+) -> dict[tuple[str, int], pd.DataFrame]:
+    """Collect latest rates across multiple MT5 account groups.
+
+    Each account is connected in turn, its symbols are read for every
+    timeframe, and the resulting frames are merged into a single mapping.
+
+    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.
+
+    Returns:
+        Mapping keyed by ``(symbol, timeframe_int)``. When accounts share a
+        symbol/timeframe pair, the last account processed wins.
+
+    Raises:
+        ValueError: If ``accounts``, ``timeframes``, or any account's symbols are
+            empty, or ``count`` is not positive.
+    """
+    account_list = list(accounts)
+    if not account_list:
+        msg = "At least one account is required."
+        raise ValueError(msg)
+    if not timeframes:
+        msg = "At least one timeframe is required."
+        raise ValueError(msg)
+    if any(not account.symbols for account in account_list):
+        msg = "Each account requires at least one symbol."
+        raise ValueError(msg)
+    _require_positive(count, "count")
+    result: dict[tuple[str, int], pd.DataFrame] = {}
+    for account in account_list:
+        config = _build_account_config(account, base_config)
+        with Mt5CliClient(config=config) as client:
+            result.update(
+                client.collect_latest_rates(
+                    account.symbols,
+                    timeframes,
+                    count=count,
+                    start_pos=start_pos,
+                ),
+            )
+    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 collect_latest_closed_rates_for_accounts(
+    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 closed rate bars across multiple MT5 account groups.
+
+    When ``start_pos`` is ``0`` (the default), MetaTrader 5 includes the
+    still-forming current bar as the last row. This helper fetches
+    ``count + 1`` bars, drops that bar with :func:`drop_forming_rate_bar`, and
+    validates that each resulting frame is non-empty. When ``start_pos`` is
+    greater than zero the forming bar is not in range, so only ``count`` bars
+    are fetched and no row is dropped.
+
+    Wraps :func:`collect_latest_rates_for_accounts_with_retries` for transient
+    MT5 error handling.
+
+    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 closed bars to return per symbol/timeframe.
+        start_pos: Initial bar position offset passed to the underlying collector.
+        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 between retry attempts.
+
+    Returns:
+        Mapping keyed by ``(symbol, timeframe_int)``.
+
+    Raises:
+        ValueError: If inputs are invalid, or any series is empty (after
+            dropping the still-forming bar when ``start_pos`` is ``0``).
+    """
+    _require_positive(count, "count")
+    _require_non_negative(start_pos, "start_pos")
+    fetch_count = count + 1 if start_pos == 0 else count
+    loaded = collect_latest_rates_for_accounts_with_retries(
+        accounts,
+        timeframes,
+        fetch_count,
+        start_pos=start_pos,
+        base_config=base_config,
+        retry_count=retry_count,
+        backoff_base=backoff_base,
+    )
+    result: dict[tuple[str, int], pd.DataFrame] = {}
+    for key, df_rate in loaded.items():
+        closed = drop_forming_rate_bar(df_rate) if start_pos == 0 else df_rate
+        if closed.empty:
+            symbol, timeframe = key
+            msg = f"Rate data is empty for {symbol!r} at timeframe {timeframe}."
+            raise ValueError(msg)
+        result[key] = closed
+    return result
+
+
+def collect_latest_closed_rates_by_granularity(
+    accounts: Sequence[AccountSpec],
+    granularities: 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, str], pd.DataFrame]:
+    """Collect latest closed rate bars keyed by symbol and granularity name.
+
+    Thin wrapper around :func:`collect_latest_closed_rates_for_accounts` that
+    rekeys the result by granularity name (for example ``M1``) instead of the
+    integer timeframe.
+
+    Args:
+        accounts: Account groups to read. Each must define at least one symbol.
+        granularities: MT5 timeframes as integers or names (for example ``M1``).
+        count: Number of closed bars to return per symbol/timeframe.
+        start_pos: Initial bar position offset passed to the underlying collector.
+        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 between retry attempts.
+
+    Returns:
+        Mapping keyed by ``(symbol, granularity_name)``. Propagates
+        ``ValueError`` from :func:`collect_latest_closed_rates_for_accounts`.
+    """
+    loaded = collect_latest_closed_rates_for_accounts(
+        accounts,
+        granularities,
+        count,
+        start_pos=start_pos,
+        base_config=base_config,
+        retry_count=retry_count,
+        backoff_base=backoff_base,
+    )
+    return {
+        (symbol, resolve_granularity_name(timeframe)): frame
+        for (symbol, timeframe), frame in loaded.items()
+    }
+
+
 def copy_rates_range(
     symbol: str,
     timeframe: int | str,

mt5cli/trading.py

@@ -0,0 +1,210 @@
+"""Trading-capable MetaTrader 5 session helpers and operational utilities."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Literal
+
+from pdmt5 import Mt5Config, Mt5TradingClient
+
+from .sdk import build_config
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    import pandas as pd
+
+PositionSide = Literal["long", "short"]
+OrderSide = Literal["long", "short"]
+
+__all__ = [
+    "OrderSide",
+    "PositionSide",
+    "calculate_margin_and_volume",
+    "detect_position_side",
+    "determine_order_limits",
+    "mt5_trading_session",
+]
+
+
+def _require_unit_ratio(value: float, name: str) -> None:
+    if not 0.0 <= value <= 1.0:
+        msg = f"{name} must be between 0 and 1 inclusive."
+        raise ValueError(msg)
+
+
+def _require_protective_ratio(value: float, name: str) -> None:
+    if not 0.0 <= value < 1.0:
+        msg = f"{name} must be at least 0 and less than 1."
+        raise ValueError(msg)
+
+
+def _sum_position_volume(positions: pd.DataFrame, position_type: object) -> float:
+    matched = positions.loc[positions["type"] == position_type, "volume"]
+    if matched.empty:
+        return 0.0
+    return float(matched.to_numpy(dtype=float).sum())
+
+
+def _normalize_order_side(side: str) -> OrderSide:
+    normalized = side.lower()
+    if normalized in {"long", "buy"}:
+        return "long"
+    if normalized in {"short", "sell"}:
+        return "short"
+    msg = (
+        f"Unsupported order side: {side!r}. Expected 'long', 'short', 'buy', or 'sell'."
+    )
+    raise ValueError(msg)
+
+
+def detect_position_side(
+    client: Mt5TradingClient,
+    symbol: str,
+) -> PositionSide | None:
+    """Detect the net open position side for a symbol.
+
+    Args:
+        client: Connected ``Mt5TradingClient`` instance.
+        symbol: Symbol to inspect.
+
+    Returns:
+        ``"long"`` when net buy volume exceeds sell volume, ``"short"`` when
+        net sell volume exceeds buy volume, or ``None`` when no positions exist
+        or buy/sell volumes are exactly balanced.
+    """
+    positions = client.positions_get_as_df(symbol=symbol)
+    if positions.empty:
+        return None
+
+    buy_type = client.mt5.POSITION_TYPE_BUY
+    sell_type = client.mt5.POSITION_TYPE_SELL
+    buy_volume = _sum_position_volume(positions, buy_type)
+    sell_volume = _sum_position_volume(positions, sell_type)
+    net_volume = buy_volume - sell_volume
+    if net_volume > 0:
+        return "long"
+    if net_volume < 0:
+        return "short"
+    return None
+
+
+def calculate_margin_and_volume(
+    client: Mt5TradingClient,
+    symbol: str,
+    unit_margin_ratio: float,
+    preserved_margin_ratio: float,
+) -> dict[str, float]:
+    """Calculate tradable margin and volumes from account free margin.
+
+    Applies ``preserved_margin_ratio`` to keep a reserve off ``margin_free``,
+    then allocates ``unit_margin_ratio`` of the remainder as the margin budget
+    for volume sizing on both buy and sell sides.
+
+    Args:
+        client: Connected ``Mt5TradingClient`` instance.
+        symbol: Symbol used for minimum-lot margin and volume calculations.
+        unit_margin_ratio: Fraction of post-reserve margin to allocate per unit.
+        preserved_margin_ratio: Fraction of ``margin_free`` to preserve.
+
+    Returns:
+        Dictionary with ``margin_free``, ``available_margin``, ``trade_margin``,
+        ``buy_volume``, and ``sell_volume``. Negative ``margin_free`` values are
+        clamped to ``0.0`` before sizing.
+    """
+    _require_unit_ratio(unit_margin_ratio, "unit_margin_ratio")
+    _require_unit_ratio(preserved_margin_ratio, "preserved_margin_ratio")
+
+    account = client.account_info_as_dict()
+    margin_free = max(0.0, float(account.get("margin_free") or 0.0))
+    available_margin = margin_free * (1.0 - preserved_margin_ratio)
+    trade_margin = available_margin * unit_margin_ratio
+    buy_volume = client.calculate_volume_by_margin(symbol, trade_margin, "BUY")
+    sell_volume = client.calculate_volume_by_margin(symbol, trade_margin, "SELL")
+    return {
+        "margin_free": margin_free,
+        "available_margin": available_margin,
+        "trade_margin": trade_margin,
+        "buy_volume": buy_volume,
+        "sell_volume": sell_volume,
+    }
+
+
+def determine_order_limits(
+    client: Mt5TradingClient,
+    symbol: str,
+    side: OrderSide | str,
+    stop_loss_limit_ratio: float,
+    take_profit_limit_ratio: float,
+) -> dict[str, float | None]:
+    """Derive entry and protective order prices from current market quotes.
+
+    Args:
+        client: Connected ``Mt5TradingClient`` instance.
+        symbol: Symbol used for the quote lookup.
+        side: Position side as ``"long"``/``"short"`` (``"buy"``/``"sell"``
+            aliases are accepted).
+        stop_loss_limit_ratio: Relative distance from entry for stop loss in
+            ``[0, 1)``. A value of ``0`` omits the stop loss.
+        take_profit_limit_ratio: Relative distance from entry for take profit in
+            ``[0, 1)``. A value of ``0`` omits the take profit.
+
+    Returns:
+        Dictionary with ``entry``, ``stop_loss``, and ``take_profit`` keys.
+        Omitted protective levels are returned as ``None``.
+    """
+    _require_protective_ratio(stop_loss_limit_ratio, "stop_loss_limit_ratio")
+    _require_protective_ratio(take_profit_limit_ratio, "take_profit_limit_ratio")
+    normalized_side = _normalize_order_side(side)
+    tick = client.symbol_info_tick_as_dict(symbol=symbol)
+    entry = float(tick["ask"] if normalized_side == "long" else tick["bid"])
+
+    stop_loss: float | None = None
+    if stop_loss_limit_ratio > 0:
+        if normalized_side == "long":
+            stop_loss = entry * (1.0 - stop_loss_limit_ratio)
+        else:
+            stop_loss = entry * (1.0 + stop_loss_limit_ratio)
+
+    take_profit: float | None = None
+    if take_profit_limit_ratio > 0:
+        if normalized_side == "long":
+            take_profit = entry * (1.0 + take_profit_limit_ratio)
+        else:
+            take_profit = entry * (1.0 - take_profit_limit_ratio)
+
+    return {
+        "entry": entry,
+        "stop_loss": stop_loss,
+        "take_profit": take_profit,
+    }
+
+
+@contextmanager
+def mt5_trading_session(
+    config: Mt5Config | None = None,
+    retry_count: int = 0,
+) -> Iterator[Mt5TradingClient]:
+    """Open a trading-capable MT5 session and always shut down safely.
+
+    Launches the MetaTrader 5 terminal using ``Mt5Config.path`` when set,
+    initializes and logs in via ``initialize_and_login_mt5()``, yields a
+    connected :class:`~pdmt5.Mt5TradingClient`, and calls ``shutdown()`` on
+    exit even when an error is raised inside the context.
+
+    Args:
+        config: MT5 connection configuration. Defaults to an empty config that
+            attaches to a running terminal.
+        retry_count: Number of initialization retries passed to
+            ``Mt5TradingClient``.
+
+    Yields:
+        Connected ``Mt5TradingClient`` bound to the session.
+    """
+    mt5_config = config or build_config()
+    client = Mt5TradingClient(config=mt5_config, retry_count=retry_count)
+    try:
+        client.initialize_and_login_mt5()
+        yield client
+    finally:
+        client.shutdown()

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mt5cli"
-version = "0.5.0"
+version = "0.6.1"
 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"}]

tests/test_history.py

@@ -10,14 +10,19 @@ from unittest.mock import MagicMock
 
 import pandas as pd
 import pytest
+from pytest_mock import MockerFixture  # noqa: TC002
 
 if TYPE_CHECKING:
     from pathlib import Path
 
+from mt5cli import history
 from mt5cli.history import (
     DEFAULT_HISTORY_TIMEFRAMES,
+    DedupScope,
+    RateTarget,
     append_dataframe,
     augment_written_columns_from_sqlite,
+    build_rate_targets,
     build_rate_view_name,
     create_cash_events_view,
     create_history_indexes,
@@ -25,6 +30,7 @@ from mt5cli.history import (
     create_rate_compatibility_views,
     deduplicate_history_tables,
     drop_duplicates_in_table,
+    drop_forming_rate_bar,
     filter_incremental_history_deals_frame,
     filter_trade_history_frame,
     get_history_deals_account_event_start_datetime,
@@ -33,6 +39,8 @@ 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,
     record_written_columns,
@@ -40,6 +48,7 @@ from mt5cli.history import (
     resolve_history_datasets,
     resolve_history_tick_flags,
     resolve_history_timeframes,
+    resolve_rate_tables,
     resolve_rate_view_name,
     resolve_rate_view_names,
     write_collected_datasets,
@@ -60,6 +69,21 @@ class TestResolveRateViewName:
         assert resolve_rate_view_name(db_path, "EURUSD", "M1") == "rate_EURUSD__1"
         assert not db_path.exists()
 
+    def test_none_path_returns_default_name(self) -> None:
+        """Test a None connection or path returns the deterministic default."""
+        assert resolve_rate_view_name(None, "EURUSD", "M1") == "rate_EURUSD__1"
+        assert resolve_rate_view_names(None, ["EURUSD"], ["M1", "H1"]) == [
+            "rate_EURUSD__1",
+            "rate_EURUSD__16385",
+        ]
+
+    def test_none_path_with_require_existing_raises(self) -> None:
+        """Test a None path under strict mode raises a clear error."""
+        with pytest.raises(ValueError, match="SQLite database not found"):
+            resolve_rate_view_name(None, "EURUSD", "M1", require_existing=True)
+        with pytest.raises(ValueError, match="SQLite database not found"):
+            resolve_rate_view_names(None, ["EURUSD"], ["M1"], require_existing=True)
+
     def test_no_rates_table_falls_back_to_single_timeframe_name(
         self,
         tmp_path: Path,
@@ -511,6 +535,46 @@ class TestResolveHistorySettings:
         assert resolve_granularity_name(1) == "M1"
 
 
+class TestDropFormingRateBar:
+    """Tests for drop_forming_rate_bar."""
+
+    def test_drops_still_forming_last_bar(self) -> None:
+        """Test the still-forming last bar is removed."""
+        df_rate = pd.DataFrame(
+            {"time": [1, 2, 3], "close": [1.1, 1.2, 1.3]},
+            index=pd.Index(["a", "b", "c"], name="idx"),
+        )
+
+        result = drop_forming_rate_bar(df_rate)
+
+        pd.testing.assert_frame_equal(
+            result,
+            pd.DataFrame(
+                {"time": [1, 2], "close": [1.1, 1.2]},
+                index=pd.Index(["a", "b"], name="idx"),
+            ),
+        )
+        assert df_rate.shape == (3, 2)
+
+    def test_returns_empty_frame_when_input_empty(self) -> None:
+        """Test empty frames stay empty."""
+        df_rate = pd.DataFrame(columns=["time", "close"])
+
+        result = drop_forming_rate_bar(df_rate)
+
+        assert result.empty
+        assert list(result.columns) == ["time", "close"]
+
+    def test_returns_empty_frame_when_only_forming_bar_present(self) -> None:
+        """Test a single-bar frame becomes empty after dropping the forming bar."""
+        df_rate = pd.DataFrame({"time": [1], "close": [1.1]})
+
+        result = drop_forming_rate_bar(df_rate)
+
+        assert result.empty
+        assert list(result.columns) == ["time", "close"]
+
+
 class TestParseSqliteTimestamp:
     """Tests for parse_sqlite_timestamp."""
 
@@ -597,7 +661,7 @@ class TestIncrementalStart:
     ) -> None:
         """Test rates tables without timeframe fail fast during incremental resume."""
         fallback = datetime(2024, 1, 1, tzinfo=UTC)
-        with sqlite3.connect(tmp_path / "legacy-rates.db") as conn:
+        with sqlite3.connect(tmp_path / "rates-without-timeframe.db") as conn:
             conn.execute("CREATE TABLE rates(symbol TEXT, time TEXT, open REAL)")
             conn.execute(
                 "INSERT INTO rates(symbol, time, open) VALUES (?, ?, ?)",
@@ -866,9 +930,10 @@ class TestDeduplication:
                 {Dataset.rates},
                 {
                     Dataset.rates: [
-                        (
+                        DedupScope(
                             "symbol = ? AND timeframe = ? AND time >= ?",
                             ("EURUSD", 1, boundary),
+                            frozenset({"symbol", "timeframe", "time"}),
                         ),
                     ],
                 },
@@ -881,6 +946,89 @@ class TestDeduplication:
             ("2024-01-02T00:00:00+00:00", 9.9),
         ]
 
+    def test_unusable_scope_falls_back_to_table_dedup(self, tmp_path: Path) -> None:
+        """Test scopes with missing columns do not break stable-key dedup."""
+        boundary = datetime(2024, 1, 1, tzinfo=UTC)
+        with sqlite3.connect(tmp_path / "orders-without-time.db") as conn:
+            conn.execute(
+                "CREATE TABLE history_orders("
+                " ticket INTEGER, symbol TEXT, time_setup TEXT, type INTEGER)",
+            )
+            conn.executemany(
+                "INSERT INTO history_orders(ticket, symbol, time_setup, type)"
+                " VALUES (?, ?, ?, ?)",
+                [
+                    (1, "EURUSD", "2024-01-01T00:00:00+00:00", 0),
+                    (1, "EURUSD", "2024-01-01T00:00:01+00:00", 1),
+                ],
+            )
+            deduplicate_history_tables(
+                conn,
+                {Dataset.history_orders: {"ticket", "symbol", "time_setup", "type"}},
+                {Dataset.history_orders},
+                {
+                    Dataset.history_orders: [
+                        DedupScope(
+                            "symbol = ? AND time >= ?",
+                            ("EURUSD", boundary),
+                            frozenset({"symbol", "time"}),
+                        ),
+                    ],
+                },
+            )
+            rows = conn.execute(
+                "SELECT ticket, time_setup, type FROM history_orders",
+            ).fetchall()
+        assert rows == [(1, "2024-01-01T00:00:01+00:00", 1)]
+
+    def test_partially_unusable_scopes_only_run_usable_scopes(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test mixed scope filtering skips only scopes with missing columns."""
+        boundary = datetime(2024, 1, 2, tzinfo=UTC)
+        with sqlite3.connect(tmp_path / "partial-scope-filter.db") as conn:
+            conn.execute(
+                "CREATE TABLE rates("
+                " symbol TEXT, timeframe INTEGER, time TEXT, open REAL)",
+            )
+            conn.executemany(
+                "INSERT INTO rates(symbol, timeframe, time, open) VALUES (?, ?, ?, ?)",
+                [
+                    ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 2.0),
+                    ("EURUSD", 1, "2024-01-02T00:00:00+00:00", 9.9),
+                    ("USDJPY", 1, "2024-01-02T00:00:00+00:00", 100.0),
+                    ("USDJPY", 1, "2024-01-02T00:00:00+00:00", 101.0),
+                ],
+            )
+            deduplicate_history_tables(
+                conn,
+                {Dataset.rates: {"symbol", "timeframe", "time", "open"}},
+                {Dataset.rates},
+                {
+                    Dataset.rates: [
+                        DedupScope(
+                            "symbol = ? AND timeframe = ? AND time >= ?",
+                            ("EURUSD", 1, boundary),
+                            frozenset({"symbol", "timeframe", "time"}),
+                        ),
+                        DedupScope(
+                            "symbol = ? AND timeframe = ? AND broker = ?",
+                            ("USDJPY", 1, "demo"),
+                            frozenset({"symbol", "timeframe", "broker"}),
+                        ),
+                    ],
+                },
+            )
+            rows = conn.execute(
+                "SELECT symbol, open FROM rates ORDER BY symbol, open",
+            ).fetchall()
+        assert rows == [
+            ("EURUSD", 9.9),
+            ("USDJPY", 100.0),
+            ("USDJPY", 101.0),
+        ]
+
 
 class TestRateCompatibilityViews:
     """Tests for rate compatibility view creation."""
@@ -1341,6 +1489,54 @@ class TestIncrementalIntegration:
             "rate_EURUSD_M1__1",
         }
 
+    def test_incremental_orders_without_time_deduplicate_by_ticket(
+        self,
+        tmp_path: Path,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Test incremental history_orders without time deduplicate safely."""
+
+        def history_orders_get_as_df(**kwargs: object) -> pd.DataFrame:
+            if kwargs["symbol"] == "GBPUSD":
+                return pd.DataFrame()
+            return pd.DataFrame({
+                "ticket": [1, 1],
+                "symbol": ["EURUSD", "EURUSD"],
+                "time_setup": [
+                    "2024-01-01T00:00:00+00:00",
+                    "2024-01-01T00:00:01+00:00",
+                ],
+                "type": [0, 1],
+            })
+
+        client = MagicMock()
+        client.history_orders_get_as_df.side_effect = history_orders_get_as_df
+        start = datetime(2024, 1, 1, tzinfo=UTC)
+        end = datetime(2024, 1, 2, tzinfo=UTC)
+        with (
+            sqlite3.connect(tmp_path / "incremental-orders-without-time.db") as conn,
+            caplog.at_level(logging.WARNING, logger="mt5cli.history"),
+        ):
+            write_incremental_datasets(
+                conn,
+                client,
+                ["EURUSD", "GBPUSD"],
+                {Dataset.history_orders},
+                [],
+                0,
+                start,
+                end,
+                deduplicate=True,
+                create_rate_views=False,
+                with_views=False,
+                include_account_events=False,
+            )
+            rows = conn.execute(
+                "SELECT ticket, time_setup, type FROM history_orders",
+            ).fetchall()
+        assert rows == [(1, "2024-01-01T00:00:01+00:00", 1)]
+        assert "Skipping history_orders: dataset returned no columns" in caplog.text
+
     def test_write_collected_datasets_and_edge_branches(
         self,
         tmp_path: Path,
@@ -1716,7 +1912,7 @@ class TestIncrementalHistoryDeals:
         })
         start = datetime(2024, 1, 1, tzinfo=UTC)
         end = datetime(2024, 1, 3, tzinfo=UTC)
-        with sqlite3.connect(tmp_path / "legacy-deals.db") as conn:
+        with sqlite3.connect(tmp_path / "deals-without-type.db") as conn:
             conn.execute(
                 "CREATE TABLE history_deals( ticket INTEGER, symbol TEXT, time TEXT)",
             )
@@ -1915,3 +2111,355 @@ class TestWriteHelpers:
             )
             assert get_table_columns(conn, "rates") == {"time", "open"}
             create_history_indexes(conn, written_columns)
+
+
+class TestRateSourceHelpers:
+    """Tests for generic rate-source SDK helpers."""
+
+    def test_rate_target_timeframe_int(self) -> None:
+        """Test RateTarget resolves named and integer timeframes."""
+        target = RateTarget(symbol="EURUSD", timeframe="M1")
+        assert target.timeframe == 1
+        assert target.timeframe_int == 1
+        assert RateTarget(symbol="EURUSD", timeframe=16385).timeframe_int == 16385
+
+    def test_build_rate_targets_row_major(self) -> None:
+        """Test targets are built in row-major symbol/timeframe order."""
+        targets = build_rate_targets(["EURUSD", "GBPUSD"], ["M1", "H1"])
+        assert [(t.symbol, t.timeframe) for t in targets] == [
+            ("EURUSD", 1),
+            ("EURUSD", 16385),
+            ("GBPUSD", 1),
+            ("GBPUSD", 16385),
+        ]
+
+    def test_build_rate_targets_allows_missing_symbol(self) -> None:
+        """Test missing symbols produce None-symbol targets when allowed."""
+        targets = build_rate_targets([], ["M1", "H1"], allow_missing_symbol=True)
+        assert [(t.symbol, t.timeframe) for t in targets] == [
+            (None, 1),
+            (None, 16385),
+        ]
+
+    @pytest.mark.parametrize(
+        ("symbols", "timeframes", "match"),
+        [
+            (["EURUSD"], [], "At least one timeframe"),
+            ([], ["M1"], "At least one symbol"),
+        ],
+    )
+    def test_build_rate_targets_rejects_empty(
+        self,
+        symbols: list[str],
+        timeframes: list[str],
+        match: str,
+    ) -> None:
+        """Test target building input validation."""
+        with pytest.raises(ValueError, match=match):
+            build_rate_targets(symbols, timeframes)
+
+    def test_resolve_rate_tables_uses_explicit_tables(self) -> None:
+        """Test explicit tables bypass view resolution when counts match."""
+        targets = build_rate_targets([], ["M1", "H1"], allow_missing_symbol=True)
+        assert resolve_rate_tables(None, targets, ["t1", "t2"]) == ["t1", "t2"]
+
+    def test_resolve_rate_tables_rejects_mismatched_explicit_count(self) -> None:
+        """Test explicit table count must match the number of targets."""
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        with pytest.raises(ValueError, match="Expected 1 explicit table"):
+            resolve_rate_tables(None, targets, ["t1", "t2"])
+
+    def test_resolve_rate_tables_rejects_empty_targets(self) -> None:
+        """Test resolving requires at least one target."""
+        with pytest.raises(ValueError, match="At least one rate target"):
+            resolve_rate_tables(None, [])
+
+    def test_resolve_rate_tables_requires_symbol_without_explicit(self) -> None:
+        """Test None-symbol targets require explicit tables."""
+        targets = build_rate_targets([], ["M1"], allow_missing_symbol=True)
+        with pytest.raises(ValueError, match="without a symbol"):
+            resolve_rate_tables(None, targets)
+
+    def test_resolve_rate_tables_resolves_view_names(self) -> None:
+        """Test symbol targets resolve to default view names without a database."""
+        targets = build_rate_targets(["EURUSD"], ["M1", "H1"])
+        assert resolve_rate_tables(None, targets) == [
+            "rate_EURUSD__1",
+            "rate_EURUSD__16385",
+        ]
+
+    def test_resolve_rate_tables_none_path_with_require_existing_raises(self) -> None:
+        """Test strict mode rejects a missing database path."""
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        with pytest.raises(ValueError, match="SQLite database not found"):
+            resolve_rate_tables(None, targets, require_existing=True)
+
+    def test_resolve_rate_tables_missing_db_with_require_existing_raises(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test strict mode rejects a non-existing database path."""
+        db_path = tmp_path / "missing.db"
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        with pytest.raises(ValueError, match="SQLite database not found"):
+            resolve_rate_tables(db_path, targets, require_existing=True)
+
+    def test_resolve_rate_tables_missing_view_with_require_existing_raises(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test strict mode rejects databases without managed rate views."""
+        db_path = tmp_path / "no-views.db"
+        with sqlite3.connect(db_path) as conn:
+            conn.execute(
+                "CREATE TABLE rates("
+                " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)",
+            )
+            conn.execute(
+                "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)",
+                ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0),
+            )
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        with pytest.raises(ValueError, match="No rate compatibility view exists"):
+            resolve_rate_tables(db_path, targets, require_existing=True)
+
+    def test_resolve_rate_tables_with_require_existing_resolves_views(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test strict mode resolves existing managed rate views."""
+        db_path = tmp_path / "strict-views.db"
+        with sqlite3.connect(db_path) as conn:
+            conn.execute(
+                "CREATE TABLE rates("
+                " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)",
+            )
+            conn.execute(
+                "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)",
+                ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0),
+            )
+            create_rate_compatibility_views(conn)
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        assert resolve_rate_tables(db_path, targets, require_existing=True) == [
+            "rate_EURUSD__1",
+        ]
+
+    def test_resolve_rate_tables_batches_sqlite_metadata(
+        self,
+        tmp_path: Path,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test resolving multiple targets loads SQLite metadata once."""
+        db_path = tmp_path / "batch-rate-tables.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-01T01:00:00+00:00", 1.1),
+                    ("GBPUSD", 1, "2024-01-01T00:00:00+00:00", 1.2),
+                ],
+            )
+            create_rate_compatibility_views(conn)
+        counts_spy = mocker.spy(history, "_load_rates_timeframe_counts")
+        views_spy = mocker.spy(history, "_load_existing_rate_views")
+
+        targets = build_rate_targets(["EURUSD", "GBPUSD"], ["M1", "H1"])
+        assert resolve_rate_tables(db_path, targets) == [
+            "rate_EURUSD__M1_1",
+            "rate_EURUSD__H1_16385",
+            "rate_GBPUSD__1",
+            "rate_GBPUSD__16385",
+        ]
+        assert counts_spy.call_count == 1
+        assert views_spy.call_count == 1
+
+    def test_load_rate_series_from_sqlite(self, tmp_path: Path) -> None:
+        """Test loading multiple rate series keyed by symbol and timeframe."""
+        db_path = tmp_path / "series.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", 1, "2024-01-01T00:01:00+00:00", 1.1),
+                ],
+            )
+            create_rate_compatibility_views(conn)
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        result = load_rate_series_from_sqlite(db_path, targets, count=2)
+        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,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test loading from a path opens SQLite once for resolve and reads."""
+        db_path = tmp_path / "single-open-series.db"
+        with sqlite3.connect(db_path) as conn:
+            conn.execute(
+                "CREATE TABLE rates("
+                " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)",
+            )
+            conn.execute(
+                "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)",
+                ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0),
+            )
+            create_rate_compatibility_views(conn)
+        connect_spy = mocker.spy(history.sqlite3, "connect")
+
+        result = load_rate_series_from_sqlite(
+            db_path,
+            build_rate_targets(["EURUSD"], ["M1"]),
+            count=1,
+        )
+
+        assert set(result) == {("EURUSD", 1)}
+        assert connect_spy.call_count == 1
+
+    def test_load_rate_series_with_explicit_tables(self, tmp_path: Path) -> None:
+        """Test explicit tables and None-symbol targets load series."""
+        db_path = tmp_path / "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),
+            )
+        targets = build_rate_targets([], ["M1"], allow_missing_symbol=True)
+        result = load_rate_series_from_sqlite(
+            db_path,
+            targets,
+            count=1,
+            explicit_tables=["custom_view"],
+        )
+        assert set(result) == {(None, 1)}
+
+    def test_load_rate_series_rejects_non_positive_count(self) -> None:
+        """Test loading requires a positive count."""
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        with pytest.raises(ValueError, match="count must be positive"):
+            load_rate_series_from_sqlite("unused.db", targets, count=0)
+
+    def test_load_rate_series_rejects_empty_targets(self) -> None:
+        """Test loading requires at least one target before opening SQLite."""
+        with pytest.raises(ValueError, match="At least one rate target"):
+            load_rate_series_from_sqlite("unused.db", [], count=1)
+
+    def test_load_rate_series_requires_symbol_without_explicit_tables(self) -> None:
+        """Test None-symbol targets require explicit tables before opening SQLite."""
+        targets = build_rate_targets([], ["M1"], allow_missing_symbol=True)
+        with pytest.raises(ValueError, match="without a symbol"):
+            load_rate_series_from_sqlite("unused.db", targets, count=1)
+
+    def test_load_rate_series_requires_existing_managed_views(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test loading without explicit tables requires managed rate views."""
+        db_path = tmp_path / "no-managed-views.db"
+        with sqlite3.connect(db_path) as conn:
+            conn.execute(
+                "CREATE TABLE rates("
+                " symbol TEXT, timeframe INTEGER, time TEXT, close REAL)",
+            )
+            conn.execute(
+                "INSERT INTO rates(symbol, timeframe, time, close) VALUES (?, ?, ?, ?)",
+                ("EURUSD", 1, "2024-01-01T00:00:00+00:00", 1.0),
+            )
+        targets = build_rate_targets(["EURUSD"], ["M1"])
+        with pytest.raises(ValueError, match="No rate compatibility view exists"):
+            load_rate_series_from_sqlite(db_path, targets, count=1)
+
+    def test_load_rate_series_rejects_duplicate_targets(self) -> None:
+        """Test duplicate (symbol, timeframe) targets are rejected."""
+        targets = [
+            RateTarget("EURUSD", 1),
+            RateTarget("EURUSD", "M1"),
+        ]
+        with pytest.raises(ValueError, match=r"Duplicate rate target: \('EURUSD', 1\)"):
+            load_rate_series_from_sqlite("unused.db", targets, count=1)
+
+    def test_load_rate_series_rejects_duplicate_targets_with_explicit_tables(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test duplicate targets are rejected even with explicit tables."""
+        db_path = tmp_path / "duplicate-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),
+            )
+        targets = [
+            RateTarget("EURUSD", 1),
+            RateTarget("EURUSD", 1),
+        ]
+        with pytest.raises(ValueError, match=r"Duplicate rate target: \('EURUSD', 1\)"):
+            load_rate_series_from_sqlite(
+                db_path,
+                targets,
+                count=1,
+                explicit_tables=["custom_view", "custom_view"],
+            )

tests/test_sdk.py

@@ -10,21 +10,28 @@ 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:
     from pathlib import Path
 
-    from pdmt5 import Mt5DataClient
+    from pdmt5 import Mt5Config, Mt5DataClient
 
 from mt5cli import sdk
-from mt5cli.history import DEFAULT_HISTORY_TIMEFRAMES
+from mt5cli.history import DEFAULT_HISTORY_TIMEFRAMES, write_rates_dataset
 from mt5cli.sdk import (
+    AccountSpec,
     Mt5CliClient,
+    ThrottledHistoryUpdater,
     account_info,
     build_config,
     collect_history,
+    collect_latest_closed_rates_by_granularity,
+    collect_latest_closed_rates_for_accounts,
     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,
@@ -36,12 +43,16 @@ from mt5cli.sdk import (
     latest_rates,
     market_book,
     minimum_margins,
+    mt5_session,
     mt5_summary,
     mt5_summary_as_df,
     orders,
     positions,
     recent_history_deals,
     recent_ticks,
+    resolve_account_spec,
+    resolve_account_specs,
+    substitute_env_placeholders,
     symbol_info,
     symbol_info_tick,
     symbols,
@@ -50,7 +61,7 @@ from mt5cli.sdk import (
     update_history_with_config,
     version,
 )
-from mt5cli.utils import Dataset
+from mt5cli.utils import Dataset, IfExists
 
 
 class _TerminalInfo(NamedTuple):
@@ -1248,3 +1259,813 @@ class TestMinimumMargins:
         )
         client.order_calc_margin.assert_any_call(0, "EURUSD", 0.01, 1.1010)
         client.order_calc_margin.assert_any_call(1, "EURUSD", 0.01, 1.1000)
+
+
+class TestMt5Session:
+    """Tests for the mt5_session context manager."""
+
+    def test_yields_connected_client_and_shuts_down(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test mt5_session connects, yields a client wrapper, and shuts down."""
+        mock_client = MagicMock()
+        mt5_data_client = mocker.patch(
+            "mt5cli.sdk.Mt5DataClient",
+            return_value=mock_client,
+        )
+
+        with mt5_session(build_config(path="/opt/mt5/terminal64.exe")) as client:
+            mock_client.initialize_and_login_mt5.assert_called_once()
+            assert isinstance(client, Mt5CliClient)
+
+        config = mt5_data_client.call_args.kwargs["config"]
+        assert config.path == "/opt/mt5/terminal64.exe"
+        mock_client.shutdown.assert_called_once()
+
+    def test_default_config_attaches_to_running_terminal(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test mt5_session builds a default config when none is supplied."""
+        mock_client = MagicMock()
+        mt5_data_client = mocker.patch(
+            "mt5cli.sdk.Mt5DataClient",
+            return_value=mock_client,
+        )
+
+        with mt5_session():
+            pass
+
+        mt5_data_client.assert_called_once()
+        mock_client.shutdown.assert_called_once()
+
+
+class TestAccountSpec:
+    """Tests for account configuration helpers."""
+
+    def test_repr_omits_password(self) -> None:
+        """Test AccountSpec repr does not expose plaintext passwords."""
+        spec = AccountSpec(symbols=["EURUSD"], login=123, password="secret")
+
+        assert "secret" not in repr(spec)
+        assert "password" not in repr(spec)
+
+    @pytest.mark.parametrize(
+        ("login", "expected"),
+        [
+            (None, None),
+            (123, 123),
+            ("", None),
+            ("   ", None),
+            ("456", 456),
+        ],
+    )
+    def test_coerce_login(
+        self,
+        login: int | str | None,
+        expected: int | None,
+    ) -> None:
+        """Test login values are normalized for account configs."""
+        assert sdk._coerce_login(login) == expected  # type: ignore[reportPrivateUsage]
+
+    def test_coerce_login_rejects_non_numeric_string(self) -> None:
+        """Test non-numeric login strings raise ValueError."""
+        with pytest.raises(ValueError, match="invalid literal"):
+            sdk._coerce_login("abc")  # type: ignore[reportPrivateUsage]
+
+
+class TestCollectLatestRatesForAccounts:
+    """Tests for collect_latest_rates_for_accounts."""
+
+    def test_merges_results_across_accounts(
+        self,
+        mock_client: MagicMock,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test rates are collected and merged for each account group."""
+        mt5_data_client = mocker.patch(
+            "mt5cli.sdk.Mt5DataClient",
+            return_value=mock_client,
+        )
+        accounts = [
+            AccountSpec(symbols=["EURUSD"], login="123"),
+            AccountSpec(symbols=["GBPUSD"], login=456),
+        ]
+
+        result = collect_latest_rates_for_accounts(accounts, ["M1"], count=2)
+
+        assert set(result) == {("EURUSD", 1), ("GBPUSD", 1)}
+        assert mt5_data_client.call_count == 2
+        assert mock_client.initialize_and_login_mt5.call_count == 2
+        assert mock_client.shutdown.call_count == 2
+
+    def test_builds_config_from_account_and_base(
+        self,
+        mock_client: MagicMock,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test account fields override base_config, empty login falls back."""
+        configs: list[object] = []
+
+        def _record_config(*, config: object) -> MagicMock:
+            configs.append(config)
+            return mock_client
+
+        mocker.patch("mt5cli.sdk.Mt5DataClient", side_effect=_record_config)
+        base = build_config(login=999, server="Base-Server", timeout=5000)
+        accounts = [
+            AccountSpec(symbols=["EURUSD"], login="", server="Acct-Server"),
+        ]
+
+        collect_latest_rates_for_accounts(accounts, ["M1"], count=1, base_config=base)
+
+        assert len(configs) == 1
+        config = cast("Mt5Config", configs[0])
+        assert config.login == 999
+        assert config.server == "Acct-Server"
+        assert config.timeout == 5000
+
+    @pytest.mark.parametrize(
+        ("accounts", "timeframes", "count", "match"),
+        [
+            ([], ["M1"], 1, "At least one account"),
+            ([AccountSpec(symbols=["EURUSD"])], [], 1, "At least one timeframe"),
+            (
+                [AccountSpec(symbols=[])],
+                ["M1"],
+                1,
+                "Each account requires at least one symbol",
+            ),
+            (
+                [AccountSpec(symbols=["EURUSD"])],
+                ["M1"],
+                0,
+                "count must be positive",
+            ),
+        ],
+    )
+    def test_rejects_invalid_inputs(
+        self,
+        accounts: list[AccountSpec],
+        timeframes: list[str],
+        count: int,
+        match: str,
+    ) -> None:
+        """Test input validation for account-level rate collection."""
+        with pytest.raises(ValueError, match=match):
+            collect_latest_rates_for_accounts(accounts, timeframes, count)
+
+    def test_rejects_empty_symbols_before_connecting(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test all account symbols are validated before any MT5 connection."""
+        mt5_data_client = mocker.patch("mt5cli.sdk.Mt5DataClient")
+        accounts = [
+            AccountSpec(symbols=["EURUSD"], login=123),
+            AccountSpec(symbols=[], login=456),
+        ]
+
+        with pytest.raises(
+            ValueError, match="Each account requires at least one symbol"
+        ):
+            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 TestCollectLatestClosedRatesForAccounts:
+    """Tests for collect_latest_closed_rates_for_accounts."""
+
+    def test_fetches_count_plus_one_and_drops_forming_bar(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test closed-bar collection requests one extra bar at start_pos=0."""
+        df_rate = pd.DataFrame({"time": [1, 2, 3], "close": [1.1, 1.2, 1.3]})
+        wrapped = mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+            return_value={("EURUSD", 1): df_rate},
+        )
+        accounts = [AccountSpec(symbols=["EURUSD"])]
+
+        result = collect_latest_closed_rates_for_accounts(
+            accounts,
+            ["M1"],
+            count=2,
+            retry_count=1,
+            backoff_base=3,
+        )
+
+        wrapped.assert_called_once_with(
+            accounts,
+            ["M1"],
+            3,
+            start_pos=0,
+            base_config=None,
+            retry_count=1,
+            backoff_base=3,
+        )
+        pd.testing.assert_frame_equal(
+            result["EURUSD", 1],
+            pd.DataFrame({"time": [1, 2], "close": [1.1, 1.2]}),
+        )
+
+    def test_rejects_forming_bar_only_frames(self, mocker: MockerFixture) -> None:
+        """Test empty results after dropping the forming bar raise ValueError."""
+        mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+            return_value={("EURUSD", 1): pd.DataFrame({"time": [1], "close": [1.1]})},
+        )
+
+        with pytest.raises(ValueError, match="Rate data is empty"):
+            collect_latest_closed_rates_for_accounts(
+                [AccountSpec(symbols=["EURUSD"])],
+                ["M1"],
+                count=1,
+            )
+
+    def test_skips_extra_fetch_when_start_pos_nonzero(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test start_pos > 0 fetches count bars without dropping the last row."""
+        df_rate = pd.DataFrame({"time": [1, 2], "close": [1.1, 1.2]})
+        wrapped = mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+            return_value={("EURUSD", 1): df_rate},
+        )
+
+        result = collect_latest_closed_rates_for_accounts(
+            [AccountSpec(symbols=["EURUSD"])],
+            ["M1"],
+            count=2,
+            start_pos=1,
+        )
+
+        wrapped.assert_called_once_with(
+            [AccountSpec(symbols=["EURUSD"])],
+            ["M1"],
+            2,
+            start_pos=1,
+            base_config=None,
+            retry_count=0,
+            backoff_base=2.0,
+        )
+        pd.testing.assert_frame_equal(result["EURUSD", 1], df_rate)
+
+    def test_rejects_zero_count_before_fetching(self, mocker: MockerFixture) -> None:
+        """Test count=0 is rejected before any MT5 collection attempt."""
+        wrapped = mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+        )
+
+        with pytest.raises(ValueError, match="count must be positive"):
+            collect_latest_closed_rates_for_accounts(
+                [AccountSpec(symbols=["EURUSD"])],
+                ["M1"],
+                count=0,
+            )
+
+        wrapped.assert_not_called()
+
+    def test_rejects_negative_start_pos(self, mocker: MockerFixture) -> None:
+        """Test negative start_pos is rejected before any MT5 collection attempt."""
+        wrapped = mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+        )
+
+        with pytest.raises(ValueError, match="start_pos must be non-negative"):
+            collect_latest_closed_rates_for_accounts(
+                [AccountSpec(symbols=["EURUSD"])],
+                ["M1"],
+                count=1,
+                start_pos=-1,
+            )
+
+        wrapped.assert_not_called()
+
+    def test_rejects_empty_frames_with_start_pos_nonzero(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test empty upstream frames raise ValueError when start_pos > 0."""
+        mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+            return_value={("EURUSD", 1): pd.DataFrame(columns=["time", "close"])},
+        )
+
+        with pytest.raises(ValueError, match="Rate data is empty"):
+            collect_latest_closed_rates_for_accounts(
+                [AccountSpec(symbols=["EURUSD"])],
+                ["M1"],
+                count=1,
+                start_pos=1,
+            )
+
+    def test_processes_multiple_symbol_timeframe_pairs(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test each returned series is trimmed and validated independently."""
+        mocker.patch(
+            "mt5cli.sdk.collect_latest_rates_for_accounts_with_retries",
+            return_value={
+                ("EURUSD", 1): pd.DataFrame(
+                    {"time": [1, 2, 3], "close": [1.1, 1.2, 1.3]},
+                ),
+                ("GBPUSD", 16385): pd.DataFrame(
+                    {"time": [4, 5, 6], "close": [2.1, 2.2, 2.3]},
+                ),
+            },
+        )
+
+        result = collect_latest_closed_rates_for_accounts(
+            [AccountSpec(symbols=["EURUSD", "GBPUSD"])],
+            ["M1", "H1"],
+            count=2,
+        )
+
+        assert set(result) == {("EURUSD", 1), ("GBPUSD", 16385)}
+        pd.testing.assert_frame_equal(
+            result["EURUSD", 1],
+            pd.DataFrame({"time": [1, 2], "close": [1.1, 1.2]}),
+        )
+        pd.testing.assert_frame_equal(
+            result["GBPUSD", 16385],
+            pd.DataFrame({"time": [4, 5], "close": [2.1, 2.2]}),
+        )
+
+
+class TestCollectLatestClosedRatesByGranularity:
+    """Tests for collect_latest_closed_rates_by_granularity."""
+
+    def test_rekeys_by_granularity_name(self, mocker: MockerFixture) -> None:
+        """Test closed rates are keyed by symbol and granularity name."""
+        df_rate = pd.DataFrame({"time": [1, 2], "close": [1.1, 1.2]})
+        wrapped = mocker.patch(
+            "mt5cli.sdk.collect_latest_closed_rates_for_accounts",
+            return_value={("EURUSD", 1): df_rate},
+        )
+
+        result = collect_latest_closed_rates_by_granularity(
+            [AccountSpec(symbols=["EURUSD"])],
+            ["M1"],
+            count=2,
+        )
+
+        wrapped.assert_called_once_with(
+            [AccountSpec(symbols=["EURUSD"])],
+            ["M1"],
+            2,
+            start_pos=0,
+            base_config=None,
+            retry_count=0,
+            backoff_base=2.0,
+        )
+        assert ("EURUSD", "M1") in result
+        pd.testing.assert_frame_equal(result["EURUSD", "M1"], df_rate)
+
+
+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"),
+            ValueError("invalid symbols"),
+            OSError("disk full"),
+            AttributeError(
+                "'StubClient' object has no attribute 'copy_rates_range_as_df'",
+                name="copy_rates_range_as_df",
+            ),
+            AttributeError(
+                "MT5 client is missing required method: copy_ticks_range_as_df"
+            ),
+            TypeError("MT5 client attribute is not callable: history_orders_get_as_df"),
+        ],
+    )
+    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
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            AttributeError("'dict' object has no attribute 'typo'"),
+            TypeError("unsupported operand types"),
+        ],
+    )
+    def test_suppress_errors_does_not_hide_programming_errors(
+        self,
+        mocker: MockerFixture,
+        error: Exception,
+    ) -> None:
+        """Test generic AttributeError/TypeError still propagate when suppressed."""
+        mocker.patch(
+            "mt5cli.sdk.update_history",
+            side_effect=error,
+        )
+        updater = ThrottledHistoryUpdater(
+            output="history.db",
+            suppress_errors=True,
+        )
+
+        with pytest.raises(type(error)):
+            updater.update(MagicMock(), ["EURUSD"])
+
+        assert updater.last_update_monotonic is None
+
+    @pytest.mark.parametrize(
+        ("error", "expected"),
+        [
+            (AttributeError("MT5 client is missing required method: version"), True),
+            (
+                AttributeError(
+                    "'Stub' object has no attribute 'copy_rates_range_as_df'",
+                    name="copy_rates_range_as_df",
+                ),
+                True,
+            ),
+            (AttributeError("'dict' object has no attribute 'typo'"), False),
+            (TypeError("MT5 client attribute is not callable: version"), True),
+            (TypeError("unsupported operand types"), False),
+            (TypeError("'NoneType' object is not callable"), False),
+            (ValueError("invalid"), False),
+        ],
+    )
+    def test_is_mt5_client_capability_error(
+        self,
+        error: BaseException,
+        expected: bool,
+    ) -> None:
+        """Test MT5 client capability error detection."""
+        assert sdk._is_mt5_client_capability_error(error) is expected  # type: ignore[reportPrivateUsage]
+
+    def test_is_mt5_client_capability_error_for_non_callable_history_client(
+        self,
+    ) -> None:
+        """Test non-callable history client attributes are capability errors."""
+        client = MagicMock()
+        client.copy_rates_range_as_df = None
+        with (
+            sqlite3.connect(":memory:") as conn,
+            pytest.raises(TypeError, match="not callable") as exc_info,
+        ):
+            write_rates_dataset(
+                conn,
+                client,
+                ["EURUSD"],
+                1,
+                datetime.now(UTC),
+                datetime.now(UTC),
+                IfExists.APPEND,
+                {},
+            )
+
+        assert sdk._is_mt5_client_capability_error(exc_info.value) is True  # type: ignore[reportPrivateUsage]
+
+    def test_suppresses_non_callable_history_client_method(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """Test suppress_errors swallows non-callable history client API attributes."""
+        client = MagicMock()
+        client.copy_rates_range_as_df = None
+        updater = ThrottledHistoryUpdater(
+            output=tmp_path / "history.db",
+            datasets={Dataset.rates},
+            timeframes=["M1"],
+            suppress_errors=True,
+        )
+
+        assert updater.update(client, ["EURUSD"]) is False
+        assert updater.last_update_monotonic is None
+
+    def test_suppress_errors_does_not_hide_internal_client_type_error(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test TypeError raised inside a callable client method still propagates."""
+        mocker.patch(
+            "mt5cli.sdk.update_history",
+            side_effect=TypeError("'int' object is not callable"),
+        )
+        updater = ThrottledHistoryUpdater(
+            output="history.db",
+            suppress_errors=True,
+        )
+
+        with pytest.raises(TypeError, match="not callable"):
+            updater.update(MagicMock(), ["EURUSD"])
+
+        assert updater.last_update_monotonic is None
+
+    def test_suppresses_validation_errors_before_update(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test validation failures are suppressed without calling update_history."""
+        update = mocker.patch("mt5cli.sdk.update_history")
+        updater = ThrottledHistoryUpdater(
+            output="history.db",
+            suppress_errors=True,
+        )
+
+        assert updater.update(MagicMock(), []) is False
+        update.assert_not_called()
+        assert updater.last_update_monotonic is None

tests/test_trading.py

@@ -0,0 +1,356 @@
+"""Tests for trading session helpers and operational utilities."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pandas as pd
+import pytest
+from pdmt5 import Mt5RuntimeError
+from pytest_mock import MockerFixture  # noqa: TC002
+
+from mt5cli.sdk import build_config
+from mt5cli.trading import (
+    calculate_margin_and_volume,
+    detect_position_side,
+    determine_order_limits,
+    mt5_trading_session,
+)
+
+
+class TestDetectPositionSide:
+    """Tests for detect_position_side."""
+
+    def test_returns_none_when_no_positions(self) -> None:
+        """Test None is returned when no open positions exist."""
+        client = MagicMock()
+        client.positions_get_as_df.return_value = pd.DataFrame()
+
+        assert detect_position_side(client, "EURUSD") is None
+
+    def test_returns_long_for_net_buy_volume(self) -> None:
+        """Test long is returned when buy volume exceeds sell volume."""
+        client = MagicMock()
+        client.mt5.POSITION_TYPE_BUY = 0
+        client.mt5.POSITION_TYPE_SELL = 1
+        client.positions_get_as_df.return_value = pd.DataFrame(
+            {
+                "type": [0, 0, 1],
+                "volume": [0.2, 0.1, 0.05],
+            },
+        )
+
+        assert detect_position_side(client, "EURUSD") == "long"
+
+    def test_returns_short_for_net_sell_volume(self) -> None:
+        """Test short is returned when sell volume exceeds buy volume."""
+        client = MagicMock()
+        client.mt5.POSITION_TYPE_BUY = 0
+        client.mt5.POSITION_TYPE_SELL = 1
+        client.positions_get_as_df.return_value = pd.DataFrame(
+            {
+                "type": [1, 1],
+                "volume": [0.3, 0.1],
+            },
+        )
+
+        assert detect_position_side(client, "EURUSD") == "short"
+
+    def test_returns_none_for_balanced_hedged_positions(self) -> None:
+        """Test None is returned when buy and sell volumes net to zero."""
+        client = MagicMock()
+        client.mt5.POSITION_TYPE_BUY = 0
+        client.mt5.POSITION_TYPE_SELL = 1
+        client.positions_get_as_df.return_value = pd.DataFrame(
+            {
+                "type": [0, 1],
+                "volume": [0.2, 0.2],
+            },
+        )
+
+        assert detect_position_side(client, "EURUSD") is None
+
+
+class TestCalculateMarginAndVolume:
+    """Tests for calculate_margin_and_volume."""
+
+    def test_calculates_margin_budget_and_volumes(self) -> None:
+        """Test margin budget and buy/sell volumes are derived from ratios."""
+        client = MagicMock()
+        client.account_info_as_dict.return_value = {"margin_free": 1000.0}
+        client.calculate_volume_by_margin.side_effect = [0.3, 0.2]
+
+        result = calculate_margin_and_volume(
+            client,
+            "EURUSD",
+            unit_margin_ratio=0.5,
+            preserved_margin_ratio=0.2,
+        )
+
+        assert result == {
+            "margin_free": 1000.0,
+            "available_margin": 800.0,
+            "trade_margin": 400.0,
+            "buy_volume": 0.3,
+            "sell_volume": 0.2,
+        }
+        client.calculate_volume_by_margin.assert_any_call("EURUSD", 400.0, "BUY")
+        client.calculate_volume_by_margin.assert_any_call("EURUSD", 400.0, "SELL")
+
+    @pytest.mark.parametrize(
+        ("account_dict", "expected_margin_free"),
+        [
+            ({"margin_free": 0.0}, 0.0),
+            ({}, 0.0),
+            ({"margin_free": None}, 0.0),
+        ],
+    )
+    def test_zero_or_missing_margin_free(
+        self,
+        account_dict: dict[str, float | None],
+        expected_margin_free: float,
+    ) -> None:
+        """Test missing or zero margin_free yields zero trade margin."""
+        client = MagicMock()
+        client.account_info_as_dict.return_value = account_dict
+        client.calculate_volume_by_margin.return_value = 0.0
+
+        result = calculate_margin_and_volume(
+            client,
+            "EURUSD",
+            unit_margin_ratio=0.5,
+            preserved_margin_ratio=0.2,
+        )
+
+        assert result["margin_free"] == expected_margin_free
+        client.calculate_volume_by_margin.assert_any_call("EURUSD", 0.0, "BUY")
+        client.calculate_volume_by_margin.assert_any_call("EURUSD", 0.0, "SELL")
+
+    def test_clamps_negative_margin_free_to_zero(self) -> None:
+        """Test negative margin_free is clamped to zero before sizing."""
+        client = MagicMock()
+        client.account_info_as_dict.return_value = {"margin_free": -500.0}
+        client.calculate_volume_by_margin.return_value = 0.0
+
+        result = calculate_margin_and_volume(
+            client,
+            "EURUSD",
+            unit_margin_ratio=0.5,
+            preserved_margin_ratio=0.2,
+        )
+
+        expected_margin_free = 0.0
+        assert result["margin_free"] == expected_margin_free
+        client.calculate_volume_by_margin.assert_any_call("EURUSD", 0.0, "BUY")
+        client.calculate_volume_by_margin.assert_any_call("EURUSD", 0.0, "SELL")
+
+    @pytest.mark.parametrize(
+        ("unit_ratio", "preserved_ratio"),
+        [
+            (-0.1, 0.0),
+            (1.1, 0.0),
+            (0.5, -0.1),
+            (0.5, 1.1),
+        ],
+    )
+    def test_rejects_invalid_ratios(
+        self,
+        unit_ratio: float,
+        preserved_ratio: float,
+    ) -> None:
+        """Test invalid ratio values raise ValueError."""
+        with pytest.raises(ValueError, match="must be between 0 and 1"):
+            calculate_margin_and_volume(
+                MagicMock(),
+                "EURUSD",
+                unit_margin_ratio=unit_ratio,
+                preserved_margin_ratio=preserved_ratio,
+            )
+
+
+class TestDetermineOrderLimits:
+    """Tests for determine_order_limits."""
+
+    @pytest.mark.parametrize(
+        ("side", "expected_entry_key"),
+        [
+            ("long", "ask"),
+            ("short", "bid"),
+            ("buy", "ask"),
+            ("sell", "bid"),
+        ],
+    )
+    def test_uses_expected_quote_for_entry(
+        self,
+        side: str,
+        expected_entry_key: str,
+    ) -> None:
+        """Test entry price is taken from ask for long/buy and bid for short/sell."""
+        client = MagicMock()
+        client.symbol_info_tick_as_dict.return_value = {"ask": 1.1010, "bid": 1.1000}
+
+        result = determine_order_limits(
+            client,
+            "EURUSD",
+            side,
+            stop_loss_limit_ratio=0.0,
+            take_profit_limit_ratio=0.0,
+        )
+
+        assert (
+            result["entry"]
+            == client.symbol_info_tick_as_dict.return_value[expected_entry_key]
+        )
+        assert result["stop_loss"] is None
+        assert result["take_profit"] is None
+
+    def test_calculates_long_protective_levels(self) -> None:
+        """Test long stop loss and take profit are placed below/above entry."""
+        client = MagicMock()
+        client.symbol_info_tick_as_dict.return_value = {"ask": 100.0, "bid": 99.0}
+
+        result = determine_order_limits(
+            client,
+            "EURUSD",
+            "long",
+            stop_loss_limit_ratio=0.02,
+            take_profit_limit_ratio=0.03,
+        )
+
+        assert result == {
+            "entry": 100.0,
+            "stop_loss": 98.0,
+            "take_profit": 103.0,
+        }
+
+    def test_calculates_short_protective_levels(self) -> None:
+        """Test short stop loss and take profit are placed above/below entry."""
+        client = MagicMock()
+        client.symbol_info_tick_as_dict.return_value = {"ask": 100.0, "bid": 99.0}
+
+        result = determine_order_limits(
+            client,
+            "EURUSD",
+            "short",
+            stop_loss_limit_ratio=0.02,
+            take_profit_limit_ratio=0.03,
+        )
+
+        assert result == {
+            "entry": 99.0,
+            "stop_loss": 100.98,
+            "take_profit": 96.03,
+        }
+
+    def test_rejects_unknown_side(self) -> None:
+        """Test unsupported side values raise ValueError."""
+        with pytest.raises(ValueError, match="Unsupported order side"):
+            determine_order_limits(
+                MagicMock(),
+                "EURUSD",
+                "flat",
+                stop_loss_limit_ratio=0.01,
+                take_profit_limit_ratio=0.01,
+            )
+
+    @pytest.mark.parametrize(
+        ("stop_loss_ratio", "take_profit_ratio"),
+        [
+            (-0.05, 0.01),
+            (0.01, 2.0),
+        ],
+    )
+    def test_rejects_invalid_protective_ratios(
+        self,
+        stop_loss_ratio: float,
+        take_profit_ratio: float,
+    ) -> None:
+        """Test out-of-range protective ratios raise ValueError."""
+        with pytest.raises(ValueError, match="must be at least 0 and less than 1"):
+            determine_order_limits(
+                MagicMock(),
+                "EURUSD",
+                "long",
+                stop_loss_limit_ratio=stop_loss_ratio,
+                take_profit_limit_ratio=take_profit_ratio,
+            )
+
+    @pytest.mark.parametrize(
+        ("field", "ratio"),
+        [
+            ("stop_loss_limit_ratio", 1.0),
+            ("take_profit_limit_ratio", 1.0),
+        ],
+    )
+    def test_rejects_unit_boundary_protective_ratios(
+        self,
+        field: str,
+        ratio: float,
+    ) -> None:
+        """Test protective ratios of exactly 1.0 are rejected."""
+        kwargs = {
+            "stop_loss_limit_ratio": 0.01,
+            "take_profit_limit_ratio": 0.01,
+            field: ratio,
+        }
+        with pytest.raises(ValueError, match="must be at least 0 and less than 1"):
+            determine_order_limits(
+                MagicMock(),
+                "EURUSD",
+                "long",
+                **kwargs,
+            )
+
+
+class TestMt5TradingSession:
+    """Tests for the mt5_trading_session context manager."""
+
+    def test_yields_connected_client_and_shuts_down(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test mt5_trading_session connects, yields a client, and shuts down."""
+        mock_client = MagicMock()
+        trading_client = mocker.patch(
+            "mt5cli.trading.Mt5TradingClient",
+            return_value=mock_client,
+        )
+
+        with mt5_trading_session(
+            build_config(path="/opt/mt5/terminal64.exe"),
+            retry_count=2,
+        ) as client:
+            mock_client.initialize_and_login_mt5.assert_called_once()
+            assert client is mock_client
+
+        trading_client.assert_called_once()
+        assert trading_client.call_args.kwargs["retry_count"] == 2
+        assert (
+            trading_client.call_args.kwargs["config"].path == "/opt/mt5/terminal64.exe"
+        )
+        mock_client.shutdown.assert_called_once()
+
+    def test_shuts_down_when_initialize_raises(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """Test shutdown is called when initialization fails."""
+        mock_client = MagicMock()
+        mock_client.initialize_and_login_mt5.side_effect = Mt5RuntimeError("boom")
+        mocker.patch("mt5cli.trading.Mt5TradingClient", return_value=mock_client)
+
+        with pytest.raises(Mt5RuntimeError, match="boom"), mt5_trading_session():
+            pass
+
+        mock_client.shutdown.assert_called_once()
+
+    def test_shuts_down_when_body_raises(self, mocker: MockerFixture) -> None:
+        """Test shutdown is called when the context body raises."""
+        mock_client = MagicMock()
+        mocker.patch("mt5cli.trading.Mt5TradingClient", return_value=mock_client)
+
+        body_error = "body error"
+        with pytest.raises(RuntimeError, match=body_error), mt5_trading_session():
+            raise RuntimeError(body_error)
+
+        mock_client.shutdown.assert_called_once()

uv.lock

@@ -487,7 +487,7 @@ wheels = [
 
 [[package]]
 name = "mt5cli"
-version = "0.5.0"
+version = "0.6.1"
 source = { editable = "." }
 dependencies = [
     { name = "click" },
@@ -836,11 +836,11 @@ wheels = [
 
 [[package]]
 name = "pygments"
-version = "2.19.2"
+version = "2.20.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" },
+    { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
 ]
 
 [[package]]