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

feat: centralize tick price validation and add resilient position margin helpers (#49, #50)

Browse Source 
Add _valid_tick_price() internal helper that returns a positive finite float
from a tick dict or None for any invalid value (missing, None, NaN, infinite,
zero, negative, or unsupported type). Refactor five existing bid/ask validation
sites in trading.py to use it, removing duplicated isinstance/isfinite checks.

Add calculate_positions_margin_by_symbol() which computes margin per unique
symbol independently using the existing strict calculate_positions_margin(),
with first-seen deduplication and configurable error suppression
(Mt5TradingError, Mt5RuntimeError, AttributeError) via suppress_errors=.

Add calculate_positions_margin_safe() as a thin sum wrapper with
suppress_errors=True, returning 0.0 on empty or fully-failed inputs.

Both new helpers are exported from mt5cli, added to STABLE_SDK_EXPORTS, and
documented in docs/api/public-contract.md. Existing strict behavior of
calculate_positions_margin() is unchanged.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
agent
2026-06-23 07:30:02 +00:00
parent 8e53212a24
commit d292fbb9d9
5 changed files with 292 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/api/public-contract.md
+14 -12
View File
@@ -99,18 +99,20 @@ diagrams.
These helpers implement broker-facing calculations only. They do not encode
strategy entries, exits, Kelly sizing, or signal logic.
| Symbol | Role |
| -------------------------------------------------------------------------------------------------- | --------------------------------------------- |
| `get_account_snapshot`, `get_symbol_snapshot`, `get_tick_snapshot`, `get_positions_frame` | Normalized account/symbol/tick/position views |
| `detect_position_side` | Net long / short / flat from open positions |
| `calculate_spread_ratio` | Relative bid-ask spread |
| `calculate_margin_and_volume`, `calculate_volume_by_margin`, `calculate_new_position_margin_ratio` | Margin budget and volume sizing |
| `normalize_order_volume`, `estimate_order_margin`, `calculate_positions_margin` | Broker volume normalization and margin totals |
| `determine_order_limits` | SL/TP price levels from ratios |
| `ensure_symbol_selected` | Select/verify Market Watch visibility |
| `place_market_order`, `close_open_positions`, `update_sltp_for_open_positions` | Order execution helpers (`dry_run` supported) |
| `MarginVolume`, `OrderLimits`, `OrderExecutionResult` | Typed return contracts for order helpers |
| `OrderSide`, `OrderFillingMode`, `OrderTimeMode`, `PositionSide`, `ExecutionStatus` | Typed enums for order helpers |
| Symbol | Role |
| -------------------------------------------------------------------------------------------------- | ----------------------------------------------------------- |
| `get_account_snapshot`, `get_symbol_snapshot`, `get_tick_snapshot`, `get_positions_frame` | Normalized account/symbol/tick/position views |
| `detect_position_side` | Net long / short / flat from open positions |
| `calculate_spread_ratio` | Relative bid-ask spread |
| `calculate_margin_and_volume`, `calculate_volume_by_margin`, `calculate_new_position_margin_ratio` | Margin budget and volume sizing |
| `normalize_order_volume`, `estimate_order_margin`, `calculate_positions_margin` | Broker volume normalization and margin totals |
| `calculate_positions_margin_by_symbol` | Per-symbol margin map (resilient, first-seen order) |
| `calculate_positions_margin_safe` | Summed total margin across symbols (failed symbols skipped) |
| `determine_order_limits` | SL/TP price levels from ratios |
| `ensure_symbol_selected` | Select/verify Market Watch visibility |
| `place_market_order`, `close_open_positions`, `update_sltp_for_open_positions` | Order execution helpers (`dry_run` supported) |
| `MarginVolume`, `OrderLimits`, `OrderExecutionResult` | Typed return contracts for order helpers |
| `OrderSide`, `OrderFillingMode`, `OrderTimeMode`, `PositionSide`, `ExecutionStatus` | Typed enums for order helpers |
`MT5Client.order_send()` and CLI `order-send --yes` are live execution paths.
mt5cli/__init__.py
+4
View File
@@ -119,6 +119,8 @@ from .trading import (
calculate_margin_and_volume,
calculate_new_position_margin_ratio,
calculate_positions_margin,
calculate_positions_margin_by_symbol,
calculate_positions_margin_safe,
calculate_spread_ratio,
calculate_volume_by_margin,
close_open_positions,
@@ -188,6 +190,8 @@ __all__ = [
"calculate_margin_and_volume",
"calculate_new_position_margin_ratio",
"calculate_positions_margin",
"calculate_positions_margin_by_symbol",
"calculate_positions_margin_safe",
"calculate_spread_ratio",
"calculate_volume_by_margin",
"call_with_normalized_errors",
mt5cli/contract.py
+2
View File
@@ -31,6 +31,8 @@ STABLE_SDK_EXPORTS: frozenset[str] = frozenset({
"calculate_margin_and_volume",
"calculate_new_position_margin_ratio",
"calculate_positions_margin",
"calculate_positions_margin_by_symbol",
"calculate_positions_margin_safe",
"calculate_spread_ratio",
"calculate_volume_by_margin",
"call_with_normalized_errors",
mt5cli/trading.py
+115 -16
View File
@@ -2,13 +2,14 @@
from __future__ import annotations
import logging
from contextlib import contextmanager
from math import floor, isfinite
from numbers import Integral, Real
from typing import TYPE_CHECKING, Literal, TypedDict, cast
import pandas as pd
from pdmt5 import Mt5Config, Mt5TradingClient, Mt5TradingError
from pdmt5 import Mt5Config, Mt5RuntimeError, Mt5TradingClient, Mt5TradingError
from .history import drop_forming_rate_bar
from .sdk import build_config
@@ -16,7 +17,9 @@ from .utils import coerce_login as _coerce_login
from .utils import parse_timeframe
if TYPE_CHECKING:
from collections.abc import Iterator, Sequence
from collections.abc import Iterator, Mapping, Sequence
_logger = logging.getLogger(__name__)
PositionSide = Literal["long", "short"]
OrderSide = Literal["BUY", "SELL"]
@@ -128,6 +131,8 @@ __all__ = [
"calculate_margin_and_volume",
"calculate_new_position_margin_ratio",
"calculate_positions_margin",
"calculate_positions_margin_by_symbol",
"calculate_positions_margin_safe",
"calculate_spread_ratio",
"calculate_volume_by_margin",
"close_open_positions",
@@ -423,6 +428,30 @@ def _optional_price(value: object) -> float | None:
return price
def _valid_tick_price(tick: Mapping[str, object], key: str) -> float | None:
"""Return a positive finite float from tick[key], or None if invalid.
Accepts int, float, or numeric string values. Returns None when the key is
missing, the value is None, non-numeric, NaN, infinite, zero, or negative.
Booleans are treated as non-numeric and return None.
"""
value = tick.get(key)
if value is None or isinstance(value, bool):
return None
if isinstance(value, int | float):
price = float(value)
elif isinstance(value, str):
try:
price = float(value)
except ValueError:
return None
else:
return None
if not isfinite(price) or price <= 0:
return None
return price
def _success_retcodes(mt5: object) -> frozenset[int]:
values = {
value
@@ -461,9 +490,10 @@ def _calculate_min_volume_if_affordable(
msg = f"Invalid volume constraints for {symbol!r}."
raise Mt5TradingError(msg)
side = _normalize_order_side(order_side)
tick = get_tick_snapshot(client, symbol)
price = tick["ask"] if side == "BUY" else tick["bid"]
if not isinstance(price, int | float) or price <= 0:
price = _valid_tick_price(
get_tick_snapshot(client, symbol), "ask" if side == "BUY" else "bid"
)
if price is None:
msg = f"Tick price is unavailable for {symbol!r}."
raise Mt5TradingError(msg)
order_type = (
@@ -621,14 +651,14 @@ def estimate_order_margin(
raise Mt5TradingError(msg)
side = _normalize_order_side(order_side)
tick = get_tick_snapshot(client, symbol)
price = tick["ask"] if side == "BUY" else tick["bid"]
if not isinstance(price, int | float) or price <= 0 or not isfinite(price):
price = _valid_tick_price(tick, "ask" if side == "BUY" else "bid")
if price is None:
msg = f"Tick price is unavailable for {symbol!r}."
raise Mt5TradingError(msg)
order_type = (
client.mt5.ORDER_TYPE_BUY if side == "BUY" else client.mt5.ORDER_TYPE_SELL
)
raw_margin = client.order_calc_margin(order_type, symbol, volume, float(price))
raw_margin = client.order_calc_margin(order_type, symbol, volume, price)
try:
margin = float(raw_margin)
except (TypeError, ValueError) as exc:
@@ -682,6 +712,72 @@ def calculate_positions_margin(
return total
def calculate_positions_margin_by_symbol(
client: Mt5TradingClient,
*,
symbols: Sequence[str],
suppress_errors: bool = True,
) -> dict[str, float]:
"""Return per-symbol estimated margin for open positions.
Computes margin for each unique input symbol independently using the strict
:func:`calculate_positions_margin` helper. Duplicates are deduplicated in
first-seen order.
Args:
client: Connected ``Mt5TradingClient`` instance.
symbols: Symbols to compute margin for.
suppress_errors: When ``True``, log and skip symbols that raise
``Mt5TradingError``, ``Mt5RuntimeError``, or ``AttributeError``.
When ``False``, re-raise the first failure.
Returns:
Mapping of symbol to margin total in first-seen unique-symbol order.
Returns an empty dict when ``symbols`` is empty or all symbols fail
with ``suppress_errors=True``.
Raises:
Mt5TradingError: When a symbol raises ``Mt5TradingError`` and
``suppress_errors=False``.
Mt5RuntimeError: When a symbol raises ``Mt5RuntimeError`` and
``suppress_errors=False``.
AttributeError: When a symbol raises ``AttributeError`` and
``suppress_errors=False``.
"""
result: dict[str, float] = {}
for symbol in dict.fromkeys(symbols):
try:
result[symbol] = calculate_positions_margin(client, symbols=[symbol])
except (Mt5TradingError, Mt5RuntimeError, AttributeError) as exc:
if not suppress_errors:
raise
_logger.warning("Skipping margin for %r: %s", symbol, exc)
return result
def calculate_positions_margin_safe(
client: Mt5TradingClient,
*,
symbols: Sequence[str],
) -> float:
"""Return the total estimated margin for open positions across symbols.
Internally calls :func:`calculate_positions_margin_by_symbol` with
``suppress_errors=True``. Failed symbols are silently skipped.
Args:
client: Connected ``Mt5TradingClient`` instance.
symbols: Symbols to include.
Returns:
Sum of per-symbol margins; ``0.0`` when no symbols or all fail.
"""
return sum(
calculate_positions_margin_by_symbol(client, symbols=symbols).values(),
0.0,
)
def calculate_spread_ratio(client: Mt5TradingClient, symbol: str) -> float:
"""Return ``(ask - bid) / ((ask + bid) / 2)`` for the latest tick.
@@ -720,9 +816,10 @@ def calculate_new_position_margin_ratio(
margin = float(account.get("margin") or 0.0)
if new_position_side is not None and new_position_volume > 0:
side = _normalize_order_side(new_position_side)
tick = get_tick_snapshot(client, symbol)
price = tick["ask"] if side == "BUY" else tick["bid"]
if not isinstance(price, int | float) or price <= 0:
price = _valid_tick_price(
get_tick_snapshot(client, symbol), "ask" if side == "BUY" else "bid"
)
if price is None:
msg = f"Tick price is unavailable for {symbol!r}."
raise Mt5TradingError(msg)
order_type = (
@@ -827,8 +924,10 @@ def calculate_volume_by_margin(
msg = f"Invalid volume constraints for {symbol!r}."
raise Mt5TradingError(msg)
side = _normalize_order_side(order_side)
price = get_tick_snapshot(client, symbol)["ask" if side == "BUY" else "bid"]
if not isinstance(price, int | float) or price <= 0:
price = _valid_tick_price(
get_tick_snapshot(client, symbol), "ask" if side == "BUY" else "bid"
)
if price is None:
msg = f"Tick price is unavailable for {symbol!r}."
raise Mt5TradingError(msg)
order_type = (
@@ -984,8 +1083,8 @@ def place_market_order(
if not dry_run:
ensure_symbol_selected(client, symbol)
tick = get_tick_snapshot(client, symbol)
price = tick["ask"] if side == "BUY" else tick["bid"]
if not isinstance(price, int | float) or price <= 0:
price = _valid_tick_price(tick, "ask" if side == "BUY" else "bid")
if price is None:
msg = f"Tick price is unavailable for {symbol!r}."
raise Mt5TradingError(msg)
request = {
@@ -995,7 +1094,7 @@ def place_market_order(
"type": (
client.mt5.ORDER_TYPE_BUY if side == "BUY" else client.mt5.ORDER_TYPE_SELL
),
"price": float(price),
"price": price,
"type_filling": _resolve_mt5_constant(
client.mt5,
"ORDER_FILLING",
tests/test_trading.py
+157
View File
@@ -18,9 +18,12 @@ from mt5cli.trading import (
MarginVolume,
OrderExecutionResult,
OrderLimits,
_valid_tick_price, # type: ignore[reportPrivateUsage]
calculate_margin_and_volume,
calculate_new_position_margin_ratio,
calculate_positions_margin,
calculate_positions_margin_by_symbol,
calculate_positions_margin_safe,
calculate_spread_ratio,
calculate_volume_by_margin,
close_open_positions,
@@ -2979,3 +2982,157 @@ class TestFetchLatestClosedRatesIndexed:
assert "open" in result.columns
assert "close" in result.columns
assert isinstance(result.index, pd.DatetimeIndex)
class TestValidTickPrice:
"""Tests for the _valid_tick_price internal helper (#49)."""
def test_valid_positive_float(self) -> None:
"""Returns a positive float value unchanged."""
_assert_close(_valid_tick_price({"bid": 1.1000}, "bid"), 1.1000)
def test_valid_positive_int(self) -> None:
"""Accepts an integer value and returns it as float."""
result = _valid_tick_price({"bid": 2}, "bid")
_assert_close(result, 2.0)
assert isinstance(result, float)
def test_valid_numeric_string(self) -> None:
"""Accepts a numeric string and returns the parsed float."""
_assert_close(_valid_tick_price({"bid": "1.5"}, "bid"), 1.5)
def test_missing_key(self) -> None:
"""Returns None when the key is absent from the tick dict."""
assert _valid_tick_price({}, "bid") is None
def test_none_value(self) -> None:
"""Returns None when the stored value is None."""
assert _valid_tick_price({"bid": None}, "bid") is None
def test_invalid_string(self) -> None:
"""Returns None for a non-numeric string."""
assert _valid_tick_price({"bid": "not_a_number"}, "bid") is None
def test_nan(self) -> None:
"""Returns None for a NaN float."""
assert _valid_tick_price({"bid": float("nan")}, "bid") is None
def test_positive_infinity(self) -> None:
"""Returns None for positive infinity."""
assert _valid_tick_price({"bid": float("inf")}, "bid") is None
def test_negative_infinity(self) -> None:
"""Returns None for negative infinity."""
assert _valid_tick_price({"bid": float("-inf")}, "bid") is None
def test_zero(self) -> None:
"""Returns None for zero (not a valid price)."""
assert _valid_tick_price({"bid": 0.0}, "bid") is None
def test_negative_value(self) -> None:
"""Returns None for a negative price."""
assert _valid_tick_price({"bid": -1.0}, "bid") is None
def test_unsupported_type(self) -> None:
"""Returns None for unsupported value types such as list."""
assert _valid_tick_price({"bid": [1.0]}, "bid") is None
class TestCalculatePositionsMarginBySymbol:
"""Tests for calculate_positions_margin_by_symbol (#50)."""
def test_all_symbols_succeed(self, mocker: MockerFixture) -> None:
"""Returns one entry per symbol when all margin calls succeed."""
client = _mock_trade_client()
mocker.patch(
"mt5cli.trading.calculate_positions_margin",
side_effect=[12.5, 30.0],
)
result = calculate_positions_margin_by_symbol(
client, symbols=["EURUSD", "GBPUSD"]
)
assert result == {"EURUSD": 12.5, "GBPUSD": 30.0}
def test_one_symbol_fails_suppress_errors_true(self, mocker: MockerFixture) -> None:
"""Skips the failing symbol and returns the successful one."""
client = _mock_trade_client()
mocker.patch(
"mt5cli.trading.calculate_positions_margin",
side_effect=[Mt5TradingError("tick unavailable"), 30.0],
)
result = calculate_positions_margin_by_symbol(
client, symbols=["EURUSD", "GBPUSD"], suppress_errors=True
)
assert result == {"GBPUSD": 30.0}
def test_one_symbol_fails_suppress_errors_false(
self, mocker: MockerFixture
) -> None:
"""Re-raises the first failure when suppress_errors=False."""
client = _mock_trade_client()
mocker.patch(
"mt5cli.trading.calculate_positions_margin",
side_effect=[Mt5TradingError("tick unavailable"), 30.0],
)
with pytest.raises(Mt5TradingError, match="tick unavailable"):
calculate_positions_margin_by_symbol(
client, symbols=["EURUSD", "GBPUSD"], suppress_errors=False
)
def test_all_symbols_fail_suppress_errors_true(self, mocker: MockerFixture) -> None:
"""Returns an empty dict when all symbols fail with suppress_errors=True."""
client = _mock_trade_client()
mocker.patch(
"mt5cli.trading.calculate_positions_margin",
side_effect=[Mt5TradingError("err1"), Mt5TradingError("err2")],
)
result = calculate_positions_margin_by_symbol(
client, symbols=["EURUSD", "GBPUSD"], suppress_errors=True
)
assert result == {}
def test_empty_symbol_list(self) -> None:
"""Returns an empty dict for an empty input list."""
client = _mock_trade_client()
result = calculate_positions_margin_by_symbol(client, symbols=[])
assert result == {}
def test_duplicate_symbols_preserve_first_seen_order(
self, mocker: MockerFixture
) -> None:
"""Processes each unique symbol once in first-seen order."""
client = _mock_trade_client()
mock_calc = mocker.patch(
"mt5cli.trading.calculate_positions_margin",
return_value=12.5,
)
result = calculate_positions_margin_by_symbol(
client, symbols=["EURUSD", "GBPUSD", "EURUSD"]
)
assert list(result.keys()) == ["EURUSD", "GBPUSD"]
assert mock_calc.call_count == 2
class TestCalculatePositionsMarginSafe:
"""Tests for calculate_positions_margin_safe (#50)."""
def test_returns_summed_total(self, mocker: MockerFixture) -> None:
"""Returns the sum of all per-symbol margins."""
client = _mock_trade_client()
mocker.patch(
"mt5cli.trading.calculate_positions_margin",
side_effect=[12.5, 30.0],
)
total = calculate_positions_margin_safe(client, symbols=["EURUSD", "GBPUSD"])
_assert_close(total, 42.5)