dienakdz
87f2845483
- Cleaned up whitespace and formatting in various files including http.py, language.py, logger.py, safe_exec.py, and SQL migration scripts. - Consolidated import statements and removed unnecessary blank lines. - Updated logging configuration for better clarity. - Enhanced the safe execution code with improved error handling and logging. - Removed commented-out code and unnecessary variables in backfill_zero_trades.py and other scripts. - Added a pyproject.toml for Ruff and Vulture configuration. - Introduced requirements-dev.txt for development dependencies. - Removed commented-out stock entries in init.sql for cleaner migration scripts.
796 lines
25 KiB
Python
796 lines
25 KiB
Python
"""
|
|
MetaTrader 5 Trading Client
|
|
|
|
Uses official MetaTrader5 Python library to connect to MT5 terminal for trading.
|
|
Note: Requires Windows platform and MT5 terminal installed.
|
|
"""
|
|
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from app.services.mt5_trading.symbols import normalize_symbol
|
|
from app.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Lazy import MetaTrader5 to allow other features to work without it installed
|
|
mt5 = None
|
|
|
|
|
|
def _ensure_mt5():
|
|
"""Ensure MetaTrader5 is imported."""
|
|
global mt5
|
|
if mt5 is None:
|
|
try:
|
|
import MetaTrader5 as _mt5
|
|
|
|
mt5 = _mt5
|
|
except ImportError:
|
|
raise ImportError(
|
|
"MetaTrader5 is not installed. Run: pip install MetaTrader5\n"
|
|
"Note: This library only works on Windows with MT5 terminal installed."
|
|
)
|
|
return mt5
|
|
|
|
|
|
@dataclass
|
|
class MT5Config:
|
|
"""MT5 connection configuration."""
|
|
|
|
login: int = 0 # MT5 account number
|
|
password: str = "" # MT5 password
|
|
server: str = "" # Broker server name (e.g., "ICMarkets-Demo")
|
|
terminal_path: str = "" # Optional: path to terminal64.exe
|
|
timeout: int = 60000 # Connection timeout in milliseconds
|
|
magic_number: int = 123456 # EA magic number for identifying orders
|
|
|
|
|
|
@dataclass
|
|
class OrderResult:
|
|
"""Order execution result."""
|
|
|
|
success: bool
|
|
order_id: int = 0
|
|
deal_id: int = 0
|
|
filled: float = 0.0
|
|
price: float = 0.0
|
|
status: str = ""
|
|
message: str = ""
|
|
raw: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class MT5Client:
|
|
"""
|
|
MetaTrader 5 Trading Client
|
|
|
|
Usage:
|
|
config = MT5Config(
|
|
login=12345678,
|
|
password="your_password",
|
|
server="ICMarkets-Demo"
|
|
)
|
|
client = MT5Client(config)
|
|
|
|
if client.connect():
|
|
# Place order
|
|
result = client.place_market_order("EURUSD", "buy", 0.1)
|
|
|
|
# Get positions
|
|
positions = client.get_positions()
|
|
|
|
client.disconnect()
|
|
"""
|
|
|
|
def __init__(self, config: Optional[MT5Config] = None):
|
|
self.config = config or MT5Config()
|
|
self._connected = False
|
|
self._lock = threading.Lock()
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
"""Check if connected to MT5 terminal."""
|
|
if not self._connected:
|
|
return False
|
|
try:
|
|
_ensure_mt5()
|
|
info = mt5.terminal_info()
|
|
return info is not None and info.connected
|
|
except Exception:
|
|
return False
|
|
|
|
def connect(self) -> bool:
|
|
"""
|
|
Connect to MT5 terminal.
|
|
|
|
Returns:
|
|
True if connected successfully
|
|
"""
|
|
with self._lock:
|
|
if self.connected:
|
|
return True
|
|
|
|
try:
|
|
_ensure_mt5()
|
|
|
|
# Initialize MT5 connection
|
|
init_params = {}
|
|
|
|
if self.config.terminal_path:
|
|
init_params["path"] = self.config.terminal_path
|
|
|
|
if self.config.login and self.config.password and self.config.server:
|
|
init_params["login"] = self.config.login
|
|
init_params["password"] = self.config.password
|
|
init_params["server"] = self.config.server
|
|
init_params["timeout"] = self.config.timeout
|
|
|
|
logger.info(f"Connecting to MT5: server={self.config.server}, login={self.config.login}")
|
|
|
|
if init_params:
|
|
initialized = mt5.initialize(**init_params)
|
|
else:
|
|
# Connect to already running terminal
|
|
initialized = mt5.initialize()
|
|
|
|
if not initialized:
|
|
error = mt5.last_error()
|
|
logger.error(f"MT5 initialization failed: {error}")
|
|
return False
|
|
|
|
self._connected = True
|
|
|
|
# Log account info
|
|
account_info = mt5.account_info()
|
|
if account_info:
|
|
logger.info(
|
|
f"MT5 connected: account={account_info.login}, "
|
|
f"server={account_info.server}, balance={account_info.balance}"
|
|
)
|
|
else:
|
|
logger.warning("MT5 connected but account info not available")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"MT5 connection failed: {e}")
|
|
self._connected = False
|
|
return False
|
|
|
|
def disconnect(self):
|
|
"""Disconnect from MT5 terminal."""
|
|
with self._lock:
|
|
if self._connected:
|
|
try:
|
|
_ensure_mt5()
|
|
mt5.shutdown()
|
|
except Exception as e:
|
|
logger.warning(f"MT5 disconnect exception: {e}")
|
|
finally:
|
|
self._connected = False
|
|
logger.info("MT5 disconnected")
|
|
|
|
def _ensure_connected(self):
|
|
"""Ensure connection is established."""
|
|
if not self.connected:
|
|
if not self.connect():
|
|
raise ConnectionError("Cannot connect to MT5 terminal")
|
|
|
|
# ==================== Order Methods ====================
|
|
|
|
def place_market_order(
|
|
self,
|
|
symbol: str,
|
|
side: str,
|
|
volume: float,
|
|
deviation: int = 20,
|
|
comment: str = "QuantDinger",
|
|
) -> OrderResult:
|
|
"""
|
|
Place a market order.
|
|
|
|
Args:
|
|
symbol: Trading symbol (e.g., "EURUSD")
|
|
side: Direction ("buy" or "sell")
|
|
volume: Lot size (e.g., 0.1 = 1 mini lot)
|
|
deviation: Maximum price deviation in points
|
|
comment: Order comment
|
|
|
|
Returns:
|
|
OrderResult
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
# Normalize symbol
|
|
symbol = normalize_symbol(symbol)
|
|
|
|
# Get symbol info
|
|
symbol_info = mt5.symbol_info(symbol)
|
|
if symbol_info is None:
|
|
return OrderResult(success=False, message=f"Symbol not found: {symbol}")
|
|
|
|
# Validate volume against symbol constraints
|
|
volume_float = float(volume)
|
|
if volume_float < symbol_info.volume_min:
|
|
return OrderResult(
|
|
success=False, message=f"Volume {volume_float} is less than minimum {symbol_info.volume_min}"
|
|
)
|
|
if volume_float > symbol_info.volume_max:
|
|
return OrderResult(
|
|
success=False, message=f"Volume {volume_float} exceeds maximum {symbol_info.volume_max}"
|
|
)
|
|
# Round volume to lot step
|
|
volume_step = symbol_info.volume_step
|
|
if volume_step > 0:
|
|
volume_float = round(volume_float / volume_step) * volume_step
|
|
|
|
if not symbol_info.visible:
|
|
# Enable symbol in Market Watch
|
|
if not mt5.symbol_select(symbol, True):
|
|
return OrderResult(success=False, message=f"Failed to select symbol: {symbol}")
|
|
|
|
# Get current price
|
|
tick = mt5.symbol_info_tick(symbol)
|
|
if tick is None:
|
|
return OrderResult(success=False, message=f"Failed to get tick for: {symbol}")
|
|
|
|
# Determine order type and price
|
|
if side.lower() == "buy":
|
|
order_type = mt5.ORDER_TYPE_BUY
|
|
price = tick.ask
|
|
else:
|
|
order_type = mt5.ORDER_TYPE_SELL
|
|
price = tick.bid
|
|
|
|
# Determine filling mode based on symbol properties
|
|
# Different brokers support different filling modes
|
|
filling_mode = mt5.ORDER_FILLING_IOC # Default
|
|
if symbol_info.filling_mode & mt5.ORDER_FILLING_IOC:
|
|
filling_mode = mt5.ORDER_FILLING_IOC
|
|
elif symbol_info.filling_mode & mt5.ORDER_FILLING_FOK:
|
|
filling_mode = mt5.ORDER_FILLING_FOK
|
|
elif symbol_info.filling_mode & mt5.ORDER_FILLING_RETURN:
|
|
filling_mode = mt5.ORDER_FILLING_RETURN
|
|
|
|
# Prepare order request
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_DEAL,
|
|
"symbol": symbol,
|
|
"volume": volume_float, # Use validated and rounded volume
|
|
"type": order_type,
|
|
"price": price,
|
|
"deviation": deviation,
|
|
"magic": self.config.magic_number,
|
|
"comment": comment,
|
|
"type_time": mt5.ORDER_TIME_GTC,
|
|
"type_filling": filling_mode,
|
|
}
|
|
|
|
# Send order
|
|
result = mt5.order_send(request)
|
|
|
|
if result is None:
|
|
error = mt5.last_error()
|
|
return OrderResult(success=False, message=f"Order send failed: {error}")
|
|
|
|
if result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
return OrderResult(
|
|
success=False,
|
|
order_id=result.order if hasattr(result, "order") else 0,
|
|
status=str(result.retcode),
|
|
message=f"Order rejected: {result.comment}",
|
|
raw=result._asdict() if hasattr(result, "_asdict") else {},
|
|
)
|
|
|
|
return OrderResult(
|
|
success=True,
|
|
order_id=result.order,
|
|
deal_id=result.deal,
|
|
filled=result.volume,
|
|
price=result.price,
|
|
status="filled",
|
|
message="Order executed",
|
|
raw=result._asdict() if hasattr(result, "_asdict") else {},
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Market order failed: {e}")
|
|
return OrderResult(success=False, message=str(e))
|
|
|
|
def place_limit_order(
|
|
self,
|
|
symbol: str,
|
|
side: str,
|
|
volume: float,
|
|
price: float,
|
|
comment: str = "QuantDinger",
|
|
) -> OrderResult:
|
|
"""
|
|
Place a pending limit order.
|
|
|
|
Args:
|
|
symbol: Trading symbol
|
|
side: Direction ("buy" or "sell")
|
|
volume: Lot size
|
|
price: Limit price
|
|
comment: Order comment
|
|
|
|
Returns:
|
|
OrderResult
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
symbol = normalize_symbol(symbol)
|
|
|
|
symbol_info = mt5.symbol_info(symbol)
|
|
if symbol_info is None:
|
|
return OrderResult(success=False, message=f"Symbol not found: {symbol}")
|
|
|
|
if not symbol_info.visible:
|
|
mt5.symbol_select(symbol, True)
|
|
|
|
tick = mt5.symbol_info_tick(symbol)
|
|
if tick is None:
|
|
return OrderResult(success=False, message=f"Failed to get tick for: {symbol}")
|
|
|
|
# Determine order type based on side and price relative to market
|
|
if side.lower() == "buy":
|
|
if price < tick.ask:
|
|
order_type = mt5.ORDER_TYPE_BUY_LIMIT
|
|
else:
|
|
order_type = mt5.ORDER_TYPE_BUY_STOP
|
|
else:
|
|
if price > tick.bid:
|
|
order_type = mt5.ORDER_TYPE_SELL_LIMIT
|
|
else:
|
|
order_type = mt5.ORDER_TYPE_SELL_STOP
|
|
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_PENDING,
|
|
"symbol": symbol,
|
|
"volume": float(volume),
|
|
"type": order_type,
|
|
"price": price,
|
|
"magic": self.config.magic_number,
|
|
"comment": comment,
|
|
"type_time": mt5.ORDER_TIME_GTC,
|
|
}
|
|
|
|
result = mt5.order_send(request)
|
|
|
|
if result is None:
|
|
error = mt5.last_error()
|
|
return OrderResult(success=False, message=f"Order send failed: {error}")
|
|
|
|
if result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
return OrderResult(
|
|
success=False,
|
|
status=str(result.retcode),
|
|
message=f"Order rejected: {result.comment}",
|
|
)
|
|
|
|
return OrderResult(
|
|
success=True,
|
|
order_id=result.order,
|
|
price=price,
|
|
status="pending",
|
|
message="Pending order placed",
|
|
raw=result._asdict() if hasattr(result, "_asdict") else {},
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Limit order failed: {e}")
|
|
return OrderResult(success=False, message=str(e))
|
|
|
|
def close_position(
|
|
self,
|
|
ticket: int,
|
|
volume: Optional[float] = None,
|
|
deviation: int = 20,
|
|
comment: str = "QuantDinger close",
|
|
) -> OrderResult:
|
|
"""
|
|
Close an open position.
|
|
|
|
Args:
|
|
ticket: Position ticket number
|
|
volume: Volume to close (None = close all)
|
|
deviation: Maximum price deviation
|
|
comment: Order comment
|
|
|
|
Returns:
|
|
OrderResult
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
# Get position info
|
|
position = mt5.positions_get(ticket=ticket)
|
|
if not position:
|
|
return OrderResult(success=False, message=f"Position not found: {ticket}")
|
|
|
|
pos = position[0]
|
|
symbol = pos.symbol
|
|
|
|
# Get symbol info for filling mode
|
|
symbol_info = mt5.symbol_info(symbol)
|
|
if symbol_info is None:
|
|
return OrderResult(success=False, message=f"Symbol not found: {symbol}")
|
|
|
|
# Get tick
|
|
tick = mt5.symbol_info_tick(symbol)
|
|
if tick is None:
|
|
return OrderResult(success=False, message=f"Failed to get tick for: {symbol}")
|
|
|
|
# Determine close direction and price
|
|
if pos.type == mt5.POSITION_TYPE_BUY:
|
|
order_type = mt5.ORDER_TYPE_SELL
|
|
price = tick.bid
|
|
else:
|
|
order_type = mt5.ORDER_TYPE_BUY
|
|
price = tick.ask
|
|
|
|
close_volume = volume if volume else pos.volume
|
|
|
|
# Determine filling mode based on symbol properties
|
|
filling_mode = mt5.ORDER_FILLING_IOC # Default
|
|
if symbol_info.filling_mode & mt5.ORDER_FILLING_IOC:
|
|
filling_mode = mt5.ORDER_FILLING_IOC
|
|
elif symbol_info.filling_mode & mt5.ORDER_FILLING_FOK:
|
|
filling_mode = mt5.ORDER_FILLING_FOK
|
|
elif symbol_info.filling_mode & mt5.ORDER_FILLING_RETURN:
|
|
filling_mode = mt5.ORDER_FILLING_RETURN
|
|
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_DEAL,
|
|
"symbol": symbol,
|
|
"volume": float(close_volume),
|
|
"type": order_type,
|
|
"position": ticket,
|
|
"price": price,
|
|
"deviation": deviation,
|
|
"magic": self.config.magic_number,
|
|
"comment": comment,
|
|
"type_time": mt5.ORDER_TIME_GTC,
|
|
"type_filling": filling_mode,
|
|
}
|
|
|
|
result = mt5.order_send(request)
|
|
|
|
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
return OrderResult(
|
|
success=False, message=f"Close failed: {result.comment if result else 'Unknown error'}"
|
|
)
|
|
|
|
return OrderResult(
|
|
success=True,
|
|
order_id=result.order,
|
|
deal_id=result.deal,
|
|
filled=result.volume,
|
|
price=result.price,
|
|
status="closed",
|
|
message="Position closed",
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Close position failed: {e}")
|
|
return OrderResult(success=False, message=str(e))
|
|
|
|
def cancel_order(self, ticket: int) -> bool:
|
|
"""
|
|
Cancel a pending order.
|
|
|
|
Args:
|
|
ticket: Order ticket number
|
|
|
|
Returns:
|
|
True if cancelled successfully
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_REMOVE,
|
|
"order": ticket,
|
|
}
|
|
|
|
result = mt5.order_send(request)
|
|
|
|
if result is None or result.retcode != mt5.TRADE_RETCODE_DONE:
|
|
logger.warning(f"Cancel order failed: {result.comment if result else 'Unknown'}")
|
|
return False
|
|
|
|
logger.info(f"Order {ticket} cancelled")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Cancel order failed: {e}")
|
|
return False
|
|
|
|
# ==================== Query Methods ====================
|
|
|
|
def get_account_info(self) -> Dict[str, Any]:
|
|
"""
|
|
Get account information.
|
|
|
|
Returns:
|
|
Account info dictionary
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
info = mt5.account_info()
|
|
if info is None:
|
|
return {"success": False, "error": "Failed to get account info"}
|
|
|
|
return {
|
|
"success": True,
|
|
"login": info.login,
|
|
"server": info.server,
|
|
"name": info.name,
|
|
"currency": info.currency,
|
|
"balance": info.balance,
|
|
"equity": info.equity,
|
|
"margin": info.margin,
|
|
"margin_free": info.margin_free,
|
|
"margin_level": info.margin_level,
|
|
"profit": info.profit,
|
|
"leverage": info.leverage,
|
|
"trade_allowed": info.trade_allowed,
|
|
"trade_expert": info.trade_expert,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Get account info failed: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def get_positions(self, symbol: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get open positions.
|
|
|
|
Args:
|
|
symbol: Filter by symbol (optional)
|
|
|
|
Returns:
|
|
List of positions
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
if symbol:
|
|
positions = mt5.positions_get(symbol=normalize_symbol(symbol))
|
|
else:
|
|
positions = mt5.positions_get()
|
|
|
|
if positions is None:
|
|
return []
|
|
|
|
result = []
|
|
for pos in positions:
|
|
result.append(
|
|
{
|
|
"ticket": pos.ticket,
|
|
"symbol": pos.symbol,
|
|
"type": "buy" if pos.type == mt5.POSITION_TYPE_BUY else "sell",
|
|
"volume": pos.volume,
|
|
"price_open": pos.price_open,
|
|
"price_current": pos.price_current,
|
|
"sl": pos.sl,
|
|
"tp": pos.tp,
|
|
"profit": pos.profit,
|
|
"swap": pos.swap,
|
|
"magic": pos.magic,
|
|
"comment": pos.comment,
|
|
"time": datetime.fromtimestamp(pos.time).isoformat(),
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Get positions failed: {e}")
|
|
return []
|
|
|
|
def get_orders(self, symbol: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get pending orders.
|
|
|
|
Args:
|
|
symbol: Filter by symbol (optional)
|
|
|
|
Returns:
|
|
List of orders
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
if symbol:
|
|
orders = mt5.orders_get(symbol=normalize_symbol(symbol))
|
|
else:
|
|
orders = mt5.orders_get()
|
|
|
|
if orders is None:
|
|
return []
|
|
|
|
result = []
|
|
for order in orders:
|
|
order_type_map = {
|
|
mt5.ORDER_TYPE_BUY_LIMIT: "buy_limit",
|
|
mt5.ORDER_TYPE_SELL_LIMIT: "sell_limit",
|
|
mt5.ORDER_TYPE_BUY_STOP: "buy_stop",
|
|
mt5.ORDER_TYPE_SELL_STOP: "sell_stop",
|
|
}
|
|
|
|
result.append(
|
|
{
|
|
"ticket": order.ticket,
|
|
"symbol": order.symbol,
|
|
"type": order_type_map.get(order.type, str(order.type)),
|
|
"volume_initial": order.volume_initial,
|
|
"volume_current": order.volume_current,
|
|
"price_open": order.price_open,
|
|
"price_current": order.price_current,
|
|
"sl": order.sl,
|
|
"tp": order.tp,
|
|
"magic": order.magic,
|
|
"comment": order.comment,
|
|
"time_setup": datetime.fromtimestamp(order.time_setup).isoformat(),
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Get orders failed: {e}")
|
|
return []
|
|
|
|
def get_quote(self, symbol: str) -> Dict[str, Any]:
|
|
"""
|
|
Get real-time quote.
|
|
|
|
Args:
|
|
symbol: Symbol code
|
|
|
|
Returns:
|
|
Quote data
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
symbol = normalize_symbol(symbol)
|
|
|
|
# Select symbol
|
|
symbol_info = mt5.symbol_info(symbol)
|
|
if symbol_info is None:
|
|
return {"success": False, "error": f"Symbol not found: {symbol}"}
|
|
|
|
if not symbol_info.visible:
|
|
mt5.symbol_select(symbol, True)
|
|
|
|
tick = mt5.symbol_info_tick(symbol)
|
|
if tick is None:
|
|
return {"success": False, "error": f"Failed to get tick: {symbol}"}
|
|
|
|
return {
|
|
"success": True,
|
|
"symbol": symbol,
|
|
"bid": tick.bid,
|
|
"ask": tick.ask,
|
|
"last": tick.last,
|
|
"volume": tick.volume,
|
|
"time": datetime.fromtimestamp(tick.time).isoformat(),
|
|
"spread": round((tick.ask - tick.bid) / symbol_info.point, 1),
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Get quote failed: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def get_symbols(self, group: str = "*") -> List[Dict[str, Any]]:
|
|
"""
|
|
Get available symbols.
|
|
|
|
Args:
|
|
group: Filter by group pattern (e.g., "*USD*", "Forex*")
|
|
|
|
Returns:
|
|
List of symbol info
|
|
"""
|
|
try:
|
|
self._ensure_connected()
|
|
_ensure_mt5()
|
|
|
|
symbols = mt5.symbols_get(group=group)
|
|
if symbols is None:
|
|
return []
|
|
|
|
result = []
|
|
for s in symbols:
|
|
result.append(
|
|
{
|
|
"name": s.name,
|
|
"description": s.description,
|
|
"path": s.path,
|
|
"currency_base": s.currency_base,
|
|
"currency_profit": s.currency_profit,
|
|
"digits": s.digits,
|
|
"point": s.point,
|
|
"trade_mode": s.trade_mode,
|
|
"volume_min": s.volume_min,
|
|
"volume_max": s.volume_max,
|
|
"volume_step": s.volume_step,
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Get symbols failed: {e}")
|
|
return []
|
|
|
|
def get_connection_status(self) -> Dict[str, Any]:
|
|
"""Get connection status."""
|
|
try:
|
|
_ensure_mt5()
|
|
terminal_info = mt5.terminal_info() if self._connected else None
|
|
account_info = mt5.account_info() if self._connected else None
|
|
|
|
return {
|
|
"connected": self.connected,
|
|
"login": self.config.login,
|
|
"server": self.config.server,
|
|
"account_login": account_info.login if account_info else None,
|
|
"account_server": account_info.server if account_info else None,
|
|
"terminal_connected": terminal_info.connected if terminal_info else False,
|
|
"trade_allowed": terminal_info.trade_allowed if terminal_info else False,
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"connected": False,
|
|
"error": str(e),
|
|
}
|
|
|
|
|
|
# Global singleton (optional)
|
|
_global_client: Optional[MT5Client] = None
|
|
_global_lock = threading.Lock()
|
|
|
|
|
|
def get_mt5_client(config: Optional[MT5Config] = None) -> MT5Client:
|
|
"""
|
|
Get global MT5 client singleton.
|
|
|
|
Args:
|
|
config: Configuration (only effective on first call)
|
|
|
|
Returns:
|
|
MT5Client instance
|
|
"""
|
|
global _global_client
|
|
|
|
with _global_lock:
|
|
if _global_client is None:
|
|
_global_client = MT5Client(config)
|
|
return _global_client
|
|
|
|
|
|
def reset_mt5_client():
|
|
"""Reset global client (disconnect and clear instance)."""
|
|
global _global_client
|
|
|
|
with _global_lock:
|
|
if _global_client is not None:
|
|
_global_client.disconnect()
|
|
_global_client = None
|