gavindiaz/mt5cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add trading session helpers and extend ThrottledHistoryUpdater (#25)

Browse Source 
* Add trading session helpers and extend ThrottledHistoryUpdater

Introduce mt5cli.trading with mt5_trading_session() for Mt5TradingClient
lifecycle management and reusable operational helpers for position-side
detection, margin/volume sizing, and protective order price derivation.

Extend ThrottledHistoryUpdater to validate inputs before updates and to
optionally suppress ValueError, OSError, and missing-method errors without
advancing the throttle timestamp.

Export the new helpers from mt5cli.__init__, add unit tests with mocked
clients, and document migration guidance for downstream projects such as
mteor.

Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>

* Narrow ThrottledHistoryUpdater suppress_errors handling (#27)

* Narrow ThrottledHistoryUpdater suppress_errors for MT5 capability only

Remove broad AttributeError/TypeError handling from recoverable errors.
Add _is_mt5_client_capability_error() to detect missing history API methods
or non-callable client attributes by message and attribute name.

Generic AttributeError/TypeError values always propagate even when
suppress_errors=True. Update docs and tests accordingly.

Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>

* Detect non-callable history client methods in suppress_errors

Address review feedback: when a history API attribute exists but is not
callable, Python raises a generic TypeError. Inspect the traceback for
mt5cli.history client call sites so these capability mismatches are still
suppressed without matching all TypeError values.

Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>

* Address PR review feedback on trading helpers

- Resolve history module path once at import time
- Only treat non-callable TypeErrors as capability errors at the raise site
- Validate SL/TP ratios in determine_order_limits
- Add tests for margin_free edge cases, body-raise shutdown, and internal TypeError propagation
- Clarify ThrottledHistoryUpdater suppress_errors docs
- Split README migration example into trading vs read-only history sessions

Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>

* Tighten protective ratio validation and clamp negative margin_free

Add _require_protective_ratio enforcing 0 <= ratio < 1 for SL/TP limits so
a ratio of 1.0 cannot produce zero protective prices. Clamp negative
margin_free to 0.0 in calculate_margin_and_volume before sizing.

Add boundary and negative-margin tests; document constraints in trading API
docs.

Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Daichi Narushima <dceoy@users.noreply.github.com>
This commit is contained in:
Daichi Narushima
2026-06-11 19:32:52 +09:00
committed by GitHub
parent 18df96872b
commit b5e82e71c7
10 changed files with 951 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+59 -1
View File
@@ -152,7 +152,8 @@ 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` and let the caller decide logging.
- **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.
@@ -164,6 +165,63 @@ eurusd_m1 = rates["EURUSD", "M1"] # closed bars only
- 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/index.md
+7 -2
View File
@@ -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
+14 -3
View File
@@ -98,6 +98,17 @@ finally:
client.shutdown()
```
By default `Mt5TradingError`, `Mt5RuntimeError`, and `sqlite3.Error` propagate so
the caller controls logging; pass `suppress_errors=True` to swallow them and
return `False` without advancing the throttle.
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
+70
View File
@@ -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
+1
View File
@@ -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
+10
View File
@@ -61,6 +61,12 @@ 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,
@@ -89,6 +95,7 @@ __all__ = [
"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",
@@ -101,6 +108,8 @@ __all__ = [
"copy_ticks_from",
"copy_ticks_range",
"detect_format",
"detect_position_side",
"determine_order_limits",
"drop_forming_rate_bar",
"export_dataframe",
"export_dataframe_to_sqlite",
@@ -117,6 +126,7 @@ __all__ = [
"mt5_session",
"mt5_summary",
"mt5_summary_as_df",
"mt5_trading_session",
"mt5_version",
"orders",
"parse_datetime",
mt5cli/sdk.py
+85 -9
View File
@@ -44,6 +44,62 @@ 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",
@@ -1014,10 +1070,14 @@ class ThrottledHistoryUpdater:
include_account_events: Include account-level cash events.
interval_seconds: Minimum seconds between successful updates. Values
``<= 0`` update on every call.
suppress_errors: When True, ``Mt5TradingError``, ``Mt5RuntimeError``,
and ``sqlite3.Error`` raised during an update are swallowed and
:meth:`update` returns False without advancing the throttle. When
False (default), such errors propagate so callers control logging.
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
@@ -1057,16 +1117,27 @@ class ThrottledHistoryUpdater:
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:
Mt5TradingError: If the update fails and ``suppress_errors`` is False.
Mt5RuntimeError: If the update fails and ``suppress_errors`` is False.
sqlite3.Error: If the SQLite write fails and ``suppress_errors`` is
False.
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,
@@ -1078,11 +1149,16 @@ class ThrottledHistoryUpdater:
with_views=self.with_views,
include_account_events=self.include_account_events,
)
except (Mt5TradingError, Mt5RuntimeError, sqlite3.Error):
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
mt5cli/trading.py
+210
View File
@@ -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()
tests/test_sdk.py
+139 -2
View File
@@ -19,7 +19,7 @@ if TYPE_CHECKING:
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,
@@ -61,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):
@@ -1913,6 +1913,16 @@ class TestThrottledHistoryUpdater:
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(
@@ -1932,3 +1942,130 @@ class TestThrottledHistoryUpdater:
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
+356
View File
@@ -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()