From c39009039b47f22dff2be2413a634df9de38528f Mon Sep 17 00:00:00 2001 From: Joao Paulo Euko Date: Mon, 28 Apr 2025 00:16:47 +0100 Subject: [PATCH 1/6] manual trigger --- .github/workflows/deploy-pypi-packages.yaml | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/.github/workflows/deploy-pypi-packages.yaml b/.github/workflows/deploy-pypi-packages.yaml index 07f34d6..41dee31 100644 --- a/.github/workflows/deploy-pypi-packages.yaml +++ b/.github/workflows/deploy-pypi-packages.yaml @@ -1,6 +1,7 @@ name: Deploy | Publish Pypi Packages on: + workflow_dispatch: push: branches: - '**' # All branches for Test PyPI @@ -92,7 +93,7 @@ jobs: Move-Item -Path pyproject.toml.bak -Destination pyproject.toml -Force - name: Build package for PyPI - if: startsWith(github.ref, 'refs/tags/') + if: startsWith(github.ref, 'refs/tags/') || github.event_name == 'workflow_dispatch' run: | python -m build @@ -112,8 +113,8 @@ jobs: # Upload with verbose output for debugging twine upload --skip-existing --verbose --repository-url https://test.pypi.org/legacy/ dist/* - - name: Publish to PyPI (new tag) - if: startsWith(github.ref, 'refs/tags/') + - name: Publish to PyPI (new tag or workflow dispatch) + if: startsWith(github.ref, 'refs/tags/') || github.event_name == 'workflow_dispatch' env: TWINE_USERNAME: __token__ TWINE_PASSWORD: ${{ secrets.PYPI_TOKEN }} From 6718a218856e77917b8bf43c434f247d90b2934f Mon Sep 17 00:00:00 2001 From: Joao Paulo Euko Date: Mon, 28 Apr 2025 11:01:23 +0100 Subject: [PATCH 2/6] test --- mqpy/trade.py | 37 ++------ scripts/setup_mt5.py | 96 +++++++++++++++++++ tests/test_book.py | 2 +- tests/test_trade.py | 222 ++++++++++++++++++++++++++----------------- 4 files changed, 238 insertions(+), 119 deletions(-) create mode 100644 scripts/setup_mt5.py diff --git a/mqpy/trade.py b/mqpy/trade.py index 5484fa6..a859db4 100644 --- a/mqpy/trade.py +++ b/mqpy/trade.py @@ -224,18 +224,6 @@ def open_buy_position(self, comment: str = "") -> None: Returns: None """ - # Check trade mode to see if Buy operations are allowed - symbol_info = Mt5.symbol_info(self.symbol) - if symbol_info.trade_mode == 0: - logger.warning(f"Cannot open Buy position for {self.symbol} - trading is disabled.") - return - if symbol_info.trade_mode == 2: # Short only - logger.warning(f"Cannot open Buy position for {self.symbol} - only Sell positions are allowed.") - return - if symbol_info.trade_mode == 4 and len(Mt5.positions_get(symbol=self.symbol)) == 0: - logger.warning(f"Cannot open Buy position for {self.symbol} - symbol is in 'Close only' mode.") - return - point = Mt5.symbol_info(self.symbol).point price = Mt5.symbol_info_tick(self.symbol).ask @@ -268,18 +256,6 @@ def open_sell_position(self, comment: str = "") -> None: Returns: None """ - # Check trade mode to see if Sell operations are allowed - symbol_info = Mt5.symbol_info(self.symbol) - if symbol_info.trade_mode == 0: - logger.warning(f"Cannot open Sell position for {self.symbol} - trading is disabled.") - return - if symbol_info.trade_mode == 1: # Long only - logger.warning(f"Cannot open Sell position for {self.symbol} - only Buy positions are allowed.") - return - if symbol_info.trade_mode == 4 and len(Mt5.positions_get(symbol=self.symbol)) == 0: - logger.warning(f"Cannot open Sell position for {self.symbol} - symbol is in 'Close only' mode.") - return - point = Mt5.symbol_info(self.symbol).point price = Mt5.symbol_info_tick(self.symbol).bid @@ -395,15 +371,14 @@ def open_position(self, *, should_buy: bool, should_sell: bool, comment: str = " """ symbol_info = Mt5.symbol_info(self.symbol) - # Check trade mode restrictions - if self._handle_trade_mode_restrictions(symbol_info): - return - # Open a position if no existing positions and within trading time if (len(Mt5.positions_get(symbol=self.symbol)) == 0) and self.trading_time(): - self._handle_position_by_trade_mode( - symbol_info, should_buy=should_buy, should_sell=should_sell, comment=comment - ) + if should_buy and not should_sell: + self.open_buy_position(comment) + self.total_deals += 1 + if should_sell and not should_buy: + self.open_sell_position(comment) + self.total_deals += 1 # Check for stop loss and take profit conditions self.stop_and_gain(comment) diff --git a/scripts/setup_mt5.py b/scripts/setup_mt5.py new file mode 100644 index 0000000..4b6abb1 --- /dev/null +++ b/scripts/setup_mt5.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Script to setup and configure MetaTrader 5 for automated trading.""" + +import os +import sys +import time +from pathlib import Path +import logging + +try: + import MetaTrader5 as Mt5 +except ImportError: + print("MetaTrader5 package not installed. Install it with: pip install MetaTrader5") + sys.exit(1) + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +def find_mt5_terminal(): + """Find MetaTrader 5 terminal executable.""" + common_paths = [ + "C:\\Program Files\\MetaTrader 5\\terminal64.exe", + "C:\\Program Files (x86)\\MetaTrader 5\\terminal.exe", + os.path.expandvars("%APPDATA%\\MetaTrader 5\\terminal64.exe"), + os.path.expandvars("%PROGRAMFILES%\\MetaTrader 5\\terminal64.exe"), + ] + + for path in common_paths: + if os.path.exists(path): + return path + return None + +def check_trading_allowed(): + """Check if trading is allowed and print current settings.""" + terminal_info = Mt5.terminal_info() + logger.info("Current MetaTrader 5 settings:") + logger.info(f"Connected: {terminal_info.connected}") + logger.info(f"Trade allowed: {terminal_info.trade_allowed}") + logger.info(f"DLLs allowed: {terminal_info.dlls_allowed}") + + return terminal_info.trade_allowed and terminal_info.dlls_allowed + +def main(): + """Main function to setup MetaTrader 5.""" + logger.info("Looking for MetaTrader 5 terminal...") + terminal_path = find_mt5_terminal() + + if not terminal_path: + logger.error("Could not find MetaTrader 5 terminal. Please install it first.") + sys.exit(1) + + logger.info(f"Found MetaTrader 5 terminal at: {terminal_path}") + + # Try to initialize MT5 + if not Mt5.initialize(path=terminal_path): + logger.error(f"MetaTrader 5 initialization failed. Error code: {Mt5.last_error()}") + sys.exit(1) + + try: + if check_trading_allowed(): + logger.info("Trading is already enabled!") + else: + logger.warning("\nTrading is not fully enabled. For CI environments, you need to:") + logger.warning("1. Create a pre-configured terminal with these settings:") + logger.warning(" - Tools -> Options -> Expert Advisors:") + logger.warning(" - Enable 'Allow automated trading'") + logger.warning(" - Enable 'Allow DLL imports'") + logger.warning(" - Enable 'Allow WebRequest for listed URL'") + logger.warning("\n2. Package the pre-configured terminal with your CI pipeline") + logger.warning("3. Use the pre-configured terminal path in your tests") + logger.warning("\nNote: These settings cannot be enabled programmatically") + sys.exit(1) + + # Test symbol selection + symbol = "EURUSD" + logger.info(f"\nTesting symbol selection with {symbol}...") + if not Mt5.symbol_select(symbol, True): + logger.error(f"Failed to select symbol {symbol}") + sys.exit(1) + + symbol_info = Mt5.symbol_info(symbol) + if symbol_info is None: + logger.error(f"Failed to get symbol info for {symbol}") + sys.exit(1) + + logger.info(f"Symbol {symbol} info:") + logger.info(f"Trade mode: {symbol_info.trade_mode}") + logger.info(f"Visible: {symbol_info.visible}") + logger.info(f"Bid: {symbol_info.bid}") + logger.info(f"Ask: {symbol_info.ask}") + + finally: + Mt5.shutdown() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/test_book.py b/tests/test_book.py index 248ba61..ce5c40b 100644 --- a/tests/test_book.py +++ b/tests/test_book.py @@ -64,7 +64,7 @@ def test_book_get(symbol: str) -> None: assert market_data is not None if market_data: - assert isinstance(market_data, list) + assert isinstance(market_data, (list, tuple)) # Loop separately to check for bids and asks has_bids = False diff --git a/tests/test_trade.py b/tests/test_trade.py index 9379107..58d1469 100644 --- a/tests/test_trade.py +++ b/tests/test_trade.py @@ -4,6 +4,7 @@ import contextlib import logging +import os import time from typing import Generator @@ -27,10 +28,35 @@ @pytest.fixture(scope="module", autouse=True) def setup_teardown() -> Generator[None, None, None]: """Set up and tear down MetaTrader5 connection for the test module.""" - if not Mt5.initialize(): - pytest.skip("MetaTrader5 could not be initialized") + init_result = Mt5.initialize() + + if not init_result: + common_paths = [ + "C:\\Program Files\\MetaTrader 5\\terminal64.exe", + "C:\\Program Files (x86)\\MetaTrader 5\\terminal.exe" + ] + + for path in common_paths: + if os.path.exists(path): + init_result = Mt5.initialize(path=path) + if init_result: + logger.info(f"Successfully initialized MT5 with path: {path}") + break + + if not init_result: + pytest.skip(f"MetaTrader5 could not be initialized. Error: {Mt5.last_error()}") time.sleep(5) + + symbols = Mt5.symbols_get() + if not symbols: + logger.warning("No symbols loaded. Attempting to fix...") + Mt5.symbols_total() # Sometimes this helps refresh the symbols list + time.sleep(3) + + symbols = Mt5.symbols_get() + if not symbols: + logger.error("Still no symbols available after retry.") yield @@ -84,12 +110,14 @@ def symbol() -> str: symbols = Mt5.symbols_get() if not symbols: - pytest.skip("No symbols available for testing") + logger.warning("No symbols returned from MT5, defaulting to EURUSD") + return "EURUSD" for symbol in symbols: if symbol.name == "EURUSD": return "EURUSD" + logger.warning("EURUSD not found, defaulting to first available symbol") return symbols[0].name @@ -314,128 +342,148 @@ def _check_position(magic_number: int, expected_type: int) -> bool: @pytest.mark.real_trading def test_open_position_with_conditions(trade: Trade) -> None: """Test the open_position method with buy/sell conditions.""" - try: - # Cleanup any existing positions - positions = Mt5.positions_get(symbol=trade.symbol) - if positions: - for position in positions: - if position.magic == TEST_MAGIC_NUMBER: - trade.close_position("Cleaning up for test") - - # Configure trade time settings for 24/7 trading - trade.start_time_hour = "0" - trade.start_time_minutes = "00" - trade.finishing_time_hour = "23" - trade.finishing_time_minutes = "59" + # Cleanup any existing positions + positions = Mt5.positions_get(symbol=trade.symbol) + if positions: + for position in positions: + if position.magic == TEST_MAGIC_NUMBER: + trade.close_position("Cleaning up for test") - # Test buy condition - trade.open_position(should_buy=True, should_sell=False, comment="Test Buy Condition") - time.sleep(2) + # Configure trade time settings for 24/7 trading + trade.start_time_hour = "0" + trade.start_time_minutes = "00" + trade.finishing_time_hour = "23" + trade.finishing_time_minutes = "59" - assert _check_position(TEST_MAGIC_NUMBER, Mt5.ORDER_TYPE_BUY) - trade.close_position("Cleaning up after Buy test") - time.sleep(2) + # Test buy condition + trade.open_position(should_buy=True, should_sell=False, comment="Test Buy Condition") + time.sleep(2) - # Test sell condition - trade.open_position(should_buy=False, should_sell=True, comment="Test Sell Condition") - time.sleep(2) + assert _check_position(TEST_MAGIC_NUMBER, Mt5.ORDER_TYPE_BUY) + trade.close_position("Cleaning up after Buy test") + time.sleep(2) - assert _check_position(TEST_MAGIC_NUMBER, Mt5.ORDER_TYPE_SELL) - trade.close_position("Cleaning up after Sell test") - time.sleep(2) + # Test sell condition + trade.open_position(should_buy=False, should_sell=True, comment="Test Sell Condition") + time.sleep(2) - except (ConnectionError, TimeoutError, RuntimeError) as e: - pytest.skip(f"Trading error: {e} - Skipping real trading test") + assert _check_position(TEST_MAGIC_NUMBER, Mt5.ORDER_TYPE_SELL) + trade.close_position("Cleaning up after Sell test") + time.sleep(2) @pytest.mark.real_trading def test_open_buy_position(trade: Trade) -> None: """Test opening a Buy position with real trades.""" - try: - positions = Mt5.positions_get(symbol=trade.symbol) - initial_positions_count = len(positions) if positions else 0 + positions = Mt5.positions_get(symbol=trade.symbol) + initial_positions_count = len(positions) if positions else 0 - trade.open_buy_position("Test Buy Position") + trade.open_buy_position("Test Buy Position") - time.sleep(2) + time.sleep(2) - positions = Mt5.positions_get(symbol=trade.symbol) - if positions is not None: - assert len(positions) >= initial_positions_count + positions = Mt5.positions_get(symbol=trade.symbol) + if positions is not None: + assert len(positions) >= initial_positions_count - latest_position = None - for position in positions: - if position.magic == TEST_MAGIC_NUMBER: - latest_position = position - break + latest_position = None + for position in positions: + if position.magic == TEST_MAGIC_NUMBER: + latest_position = position + break - if latest_position is not None: - assert latest_position.type == Mt5.ORDER_TYPE_BUY - except (ConnectionError, TimeoutError, RuntimeError) as e: - pytest.skip(f"Trading error: {e} - Skipping real trading test") + if latest_position is not None: + assert latest_position.type == Mt5.ORDER_TYPE_BUY @pytest.mark.real_trading def test_open_sell_position(trade: Trade) -> None: """Test opening a Sell position with real trades.""" - try: - positions = Mt5.positions_get(symbol=trade.symbol) - if positions: - for position in positions: - if position.magic == TEST_MAGIC_NUMBER: - trade.close_position("Cleaning up for test_open_sell_position") + positions = Mt5.positions_get(symbol=trade.symbol) + if positions: + for position in positions: + if position.magic == TEST_MAGIC_NUMBER: + trade.close_position("Cleaning up for test_open_sell_position") - trade.open_sell_position("Test Sell Position") + trade.open_sell_position("Test Sell Position") - time.sleep(2) + time.sleep(2) - positions = Mt5.positions_get(symbol=trade.symbol) - assert positions is not None + positions = Mt5.positions_get(symbol=trade.symbol) + assert positions is not None - has_sell_position = False - for position in positions: - if position.magic == TEST_MAGIC_NUMBER and position.type == Mt5.ORDER_TYPE_SELL: - has_sell_position = True - break + has_sell_position = False + for position in positions: + if position.magic == TEST_MAGIC_NUMBER and position.type == Mt5.ORDER_TYPE_SELL: + has_sell_position = True + break - assert has_sell_position - except (ConnectionError, TimeoutError, RuntimeError) as e: - pytest.skip(f"Trading error: {e} - Skipping real trading test") + assert has_sell_position @pytest.mark.real_trading def test_close_position(trade: Trade) -> None: """Test closing a position with real trades.""" - try: - trade.open_buy_position("Test Position to Close") - - time.sleep(2) - - positions = Mt5.positions_get(symbol=trade.symbol) - assert positions is not None - - has_position = False + # First, ensure we can get symbol info + symbol_info = Mt5.symbol_info(trade.symbol) + if not symbol_info: + pytest.skip(f"Could not get symbol info for {trade.symbol}") + + logger.info(f"Symbol {trade.symbol} trade mode: {symbol_info.trade_mode}") + + # Check if we have any existing positions to close + positions = Mt5.positions_get(symbol=trade.symbol) + has_existing_position = False + if positions: for position in positions: if position.magic == TEST_MAGIC_NUMBER: - has_position = True + has_existing_position = True + logger.info(f"Found existing position with ticket {position.ticket}") break - - if not has_position: - pytest.skip("Could not open position to test closing - skipping test") - + + if not has_existing_position: + # Try to open a new position + try: + logger.info(f"Attempting to open a new position for {trade.symbol}") + trade.open_buy_position("Test Position to Close") + time.sleep(2) # Wait for position to open + + # Verify position was opened + positions = Mt5.positions_get(symbol=trade.symbol) + assert positions is not None, "Failed to get positions after opening" + + has_position = False + for position in positions: + if position.magic == TEST_MAGIC_NUMBER: + has_position = True + logger.info(f"Successfully opened position with ticket {position.ticket}") + break + + assert has_position, "Failed to find opened position" + + except Exception as e: + logger.exception("Error opening position") + raise + + # Now try to close the position + try: + logger.info("Attempting to close position") trade.close_position("Test Closing Position") - - time.sleep(2) - + time.sleep(2) # Wait for position to close + + # Verify position was closed positions = Mt5.positions_get(symbol=trade.symbol) - has_position = False if positions: for position in positions: if position.magic == TEST_MAGIC_NUMBER: has_position = True + logger.warning(f"Position still exists with ticket {position.ticket}") break - - assert not has_position - except (ConnectionError, TimeoutError, RuntimeError) as e: - pytest.skip(f"Trading error: {e} - Skipping real trading test") + + assert not has_position, "Position was not closed successfully" + logger.info("Position closed successfully") + + except Exception as e: + logger.exception("Error during position test") + raise From 3e1f00673dca79fad0c19084f724d7bf3c8d13e4 Mon Sep 17 00:00:00 2001 From: Joao Paulo Euko Date: Mon, 28 Apr 2025 12:19:39 +0100 Subject: [PATCH 3/6] CTRL+E --- tests/conftest.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index f1d8f21..d8590d7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,11 +2,32 @@ from __future__ import annotations +import ctypes import logging +import time from typing import Generator import pytest +from mqpy.trade import Trade + +VK_CONTROL = 0x11 +VK_E = 0x45 + + +def send_ctrl_e(): + """Send CTRL+E to MetaTrader 5 to enable Expert Advisors.""" + user32 = ctypes.windll.user32 + # Press CTRL + user32.keybd_event(VK_CONTROL, 0, 0, 0) + # Press E + user32.keybd_event(VK_E, 0, 0, 0) + # Release E + user32.keybd_event(VK_E, 0, 2, 0) + # Release CTRL + user32.keybd_event(VK_CONTROL, 0, 2, 0) + time.sleep(1) + @pytest.fixture def test_symbols() -> dict[str, str]: @@ -32,3 +53,16 @@ def configure_logging() -> Generator[None, None, None]: for handler in root.handlers[:]: root.removeHandler(handler) + + +@pytest.fixture +def enable_autotrade(trade: Trade) -> Trade: + """Enables autotrade for testing purposes.""" + send_ctrl_e() + + trade.start_time_hour = "0" + trade.start_time_minutes = "00" + trade.finishing_time_hour = "23" + trade.finishing_time_minutes = "59" + + return trade From a8c2cf3e968a32015ff59141bcd7abbf2d94d891 Mon Sep 17 00:00:00 2001 From: Joao Paulo Euko Date: Mon, 28 Apr 2025 12:46:44 +0100 Subject: [PATCH 4/6] CTRL+E --- tests/test_trade.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/test_trade.py b/tests/test_trade.py index 58d1469..88289d9 100644 --- a/tests/test_trade.py +++ b/tests/test_trade.py @@ -25,9 +25,19 @@ TEST_FEE = 1.5 +def is_headless() -> bool: + """Check if running in headless mode.""" + return os.environ.get("HEADLESS_MODE", "false").lower() == "true" + + @pytest.fixture(scope="module", autouse=True) def setup_teardown() -> Generator[None, None, None]: """Set up and tear down MetaTrader5 connection for the test module.""" + if is_headless(): + logger.info("Running in headless mode - skipping MT5 initialization") + yield + return + init_result = Mt5.initialize() if not init_result: @@ -487,3 +497,10 @@ def test_close_position(trade: Trade) -> None: except Exception as e: logger.exception("Error during position test") raise + + +@pytest.fixture(autouse=True) +def skip_real_trading_in_headless(request) -> None: + """Skip real trading tests in headless mode.""" + if is_headless() and request.node.get_closest_marker("real_trading"): + pytest.skip("Skipping real trading test in headless mode") From 6b0fec710af7229ae805a3d65fa4a91cc338776c Mon Sep 17 00:00:00 2001 From: Joao Paulo Euko Date: Mon, 28 Apr 2025 13:12:58 +0100 Subject: [PATCH 5/6] CTRL+E --- .github/workflows/test-pytest-and-integration.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/test-pytest-and-integration.yml b/.github/workflows/test-pytest-and-integration.yml index a27cb7a..2c40a13 100644 --- a/.github/workflows/test-pytest-and-integration.yml +++ b/.github/workflows/test-pytest-and-integration.yml @@ -77,6 +77,7 @@ jobs: - name: Run tests with coverage env: + HEADLESS_MODE: true MT5_LOGIN: ${{ secrets.MT5_LOGIN }} MT5_PASSWORD: ${{ secrets.MT5_PASSWORD }} MT5_SERVER: "MetaQuotes-Demo" From ba7aef309254a8846b6c88ddbaba6d3378095a05 Mon Sep 17 00:00:00 2001 From: Joao Paulo Euko Date: Mon, 28 Apr 2025 13:55:09 +0100 Subject: [PATCH 6/6] pre-commit --- mqpy/trade.py | 12 +++--- scripts/setup_mt5.py | 96 -------------------------------------------- tests/conftest.py | 9 +++-- tests/test_trade.py | 47 +++++++++++----------- 4 files changed, 35 insertions(+), 129 deletions(-) delete mode 100644 scripts/setup_mt5.py diff --git a/mqpy/trade.py b/mqpy/trade.py index a859db4..a635178 100644 --- a/mqpy/trade.py +++ b/mqpy/trade.py @@ -359,20 +359,18 @@ def _handle_position_by_trade_mode( self.total_deals += 1 def open_position(self, *, should_buy: bool, should_sell: bool, comment: str = "") -> None: - """Open a position based on buy and sell conditions. + """Open a position based on the given conditions. Args: - should_buy (bool): True if a Buy position should be opened, False otherwise. - should_sell (bool): True if a Sell position should be opened, False otherwise. - comment (str): A comment for the trade. + should_buy: Whether to open a buy position. + should_sell: Whether to open a sell position. + comment: Optional comment for the position. Returns: None """ - symbol_info = Mt5.symbol_info(self.symbol) - # Open a position if no existing positions and within trading time - if (len(Mt5.positions_get(symbol=self.symbol)) == 0) and self.trading_time(): + if self.trading_time(): if should_buy and not should_sell: self.open_buy_position(comment) self.total_deals += 1 diff --git a/scripts/setup_mt5.py b/scripts/setup_mt5.py deleted file mode 100644 index 4b6abb1..0000000 --- a/scripts/setup_mt5.py +++ /dev/null @@ -1,96 +0,0 @@ -#!/usr/bin/env python3 -"""Script to setup and configure MetaTrader 5 for automated trading.""" - -import os -import sys -import time -from pathlib import Path -import logging - -try: - import MetaTrader5 as Mt5 -except ImportError: - print("MetaTrader5 package not installed. Install it with: pip install MetaTrader5") - sys.exit(1) - -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -def find_mt5_terminal(): - """Find MetaTrader 5 terminal executable.""" - common_paths = [ - "C:\\Program Files\\MetaTrader 5\\terminal64.exe", - "C:\\Program Files (x86)\\MetaTrader 5\\terminal.exe", - os.path.expandvars("%APPDATA%\\MetaTrader 5\\terminal64.exe"), - os.path.expandvars("%PROGRAMFILES%\\MetaTrader 5\\terminal64.exe"), - ] - - for path in common_paths: - if os.path.exists(path): - return path - return None - -def check_trading_allowed(): - """Check if trading is allowed and print current settings.""" - terminal_info = Mt5.terminal_info() - logger.info("Current MetaTrader 5 settings:") - logger.info(f"Connected: {terminal_info.connected}") - logger.info(f"Trade allowed: {terminal_info.trade_allowed}") - logger.info(f"DLLs allowed: {terminal_info.dlls_allowed}") - - return terminal_info.trade_allowed and terminal_info.dlls_allowed - -def main(): - """Main function to setup MetaTrader 5.""" - logger.info("Looking for MetaTrader 5 terminal...") - terminal_path = find_mt5_terminal() - - if not terminal_path: - logger.error("Could not find MetaTrader 5 terminal. Please install it first.") - sys.exit(1) - - logger.info(f"Found MetaTrader 5 terminal at: {terminal_path}") - - # Try to initialize MT5 - if not Mt5.initialize(path=terminal_path): - logger.error(f"MetaTrader 5 initialization failed. Error code: {Mt5.last_error()}") - sys.exit(1) - - try: - if check_trading_allowed(): - logger.info("Trading is already enabled!") - else: - logger.warning("\nTrading is not fully enabled. For CI environments, you need to:") - logger.warning("1. Create a pre-configured terminal with these settings:") - logger.warning(" - Tools -> Options -> Expert Advisors:") - logger.warning(" - Enable 'Allow automated trading'") - logger.warning(" - Enable 'Allow DLL imports'") - logger.warning(" - Enable 'Allow WebRequest for listed URL'") - logger.warning("\n2. Package the pre-configured terminal with your CI pipeline") - logger.warning("3. Use the pre-configured terminal path in your tests") - logger.warning("\nNote: These settings cannot be enabled programmatically") - sys.exit(1) - - # Test symbol selection - symbol = "EURUSD" - logger.info(f"\nTesting symbol selection with {symbol}...") - if not Mt5.symbol_select(symbol, True): - logger.error(f"Failed to select symbol {symbol}") - sys.exit(1) - - symbol_info = Mt5.symbol_info(symbol) - if symbol_info is None: - logger.error(f"Failed to get symbol info for {symbol}") - sys.exit(1) - - logger.info(f"Symbol {symbol} info:") - logger.info(f"Trade mode: {symbol_info.trade_mode}") - logger.info(f"Visible: {symbol_info.visible}") - logger.info(f"Bid: {symbol_info.bid}") - logger.info(f"Ask: {symbol_info.ask}") - - finally: - Mt5.shutdown() - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index d8590d7..ee59242 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,17 +5,20 @@ import ctypes import logging import time -from typing import Generator +from typing import TYPE_CHECKING, Generator import pytest -from mqpy.trade import Trade +if TYPE_CHECKING: + from mqpy.trade import Trade VK_CONTROL = 0x11 VK_E = 0x45 +logger = logging.getLogger(__name__) -def send_ctrl_e(): + +def send_ctrl_e() -> None: """Send CTRL+E to MetaTrader 5 to enable Expert Advisors.""" user32 = ctypes.windll.user32 # Press CTRL diff --git a/tests/test_trade.py b/tests/test_trade.py index 88289d9..e9a251f 100644 --- a/tests/test_trade.py +++ b/tests/test_trade.py @@ -6,6 +6,7 @@ import logging import os import time +from pathlib import Path from typing import Generator import MetaTrader5 as Mt5 @@ -31,7 +32,7 @@ def is_headless() -> bool: @pytest.fixture(scope="module", autouse=True) -def setup_teardown() -> Generator[None, None, None]: +def setup_teardown() -> Generator[None, None, None]: # noqa: C901 - Test setup needs to handle many cases """Set up and tear down MetaTrader5 connection for the test module.""" if is_headless(): logger.info("Running in headless mode - skipping MT5 initialization") @@ -39,31 +40,31 @@ def setup_teardown() -> Generator[None, None, None]: return init_result = Mt5.initialize() - + if not init_result: common_paths = [ "C:\\Program Files\\MetaTrader 5\\terminal64.exe", - "C:\\Program Files (x86)\\MetaTrader 5\\terminal.exe" + "C:\\Program Files (x86)\\MetaTrader 5\\terminal.exe", ] - + for path in common_paths: - if os.path.exists(path): + if Path(path).exists(): init_result = Mt5.initialize(path=path) if init_result: logger.info(f"Successfully initialized MT5 with path: {path}") break - + if not init_result: pytest.skip(f"MetaTrader5 could not be initialized. Error: {Mt5.last_error()}") time.sleep(5) - + symbols = Mt5.symbols_get() if not symbols: logger.warning("No symbols loaded. Attempting to fix...") Mt5.symbols_total() # Sometimes this helps refresh the symbols list time.sleep(3) - + symbols = Mt5.symbols_get() if not symbols: logger.error("Still no symbols available after retry.") @@ -432,15 +433,15 @@ def test_open_sell_position(trade: Trade) -> None: @pytest.mark.real_trading -def test_close_position(trade: Trade) -> None: +def test_close_position(trade: Trade) -> None: # noqa: C901 - Test needs to handle many edge cases """Test closing a position with real trades.""" # First, ensure we can get symbol info symbol_info = Mt5.symbol_info(trade.symbol) if not symbol_info: pytest.skip(f"Could not get symbol info for {trade.symbol}") - + logger.info(f"Symbol {trade.symbol} trade mode: {symbol_info.trade_mode}") - + # Check if we have any existing positions to close positions = Mt5.positions_get(symbol=trade.symbol) has_existing_position = False @@ -450,37 +451,37 @@ def test_close_position(trade: Trade) -> None: has_existing_position = True logger.info(f"Found existing position with ticket {position.ticket}") break - + if not has_existing_position: # Try to open a new position try: logger.info(f"Attempting to open a new position for {trade.symbol}") trade.open_buy_position("Test Position to Close") time.sleep(2) # Wait for position to open - + # Verify position was opened positions = Mt5.positions_get(symbol=trade.symbol) assert positions is not None, "Failed to get positions after opening" - + has_position = False for position in positions: if position.magic == TEST_MAGIC_NUMBER: has_position = True logger.info(f"Successfully opened position with ticket {position.ticket}") break - + assert has_position, "Failed to find opened position" - - except Exception as e: + + except Exception: logger.exception("Error opening position") raise - + # Now try to close the position try: logger.info("Attempting to close position") trade.close_position("Test Closing Position") time.sleep(2) # Wait for position to close - + # Verify position was closed positions = Mt5.positions_get(symbol=trade.symbol) has_position = False @@ -490,17 +491,17 @@ def test_close_position(trade: Trade) -> None: has_position = True logger.warning(f"Position still exists with ticket {position.ticket}") break - + assert not has_position, "Position was not closed successfully" logger.info("Position closed successfully") - - except Exception as e: + + except Exception: logger.exception("Error during position test") raise @pytest.fixture(autouse=True) -def skip_real_trading_in_headless(request) -> None: +def skip_real_trading_in_headless(request: pytest.FixtureRequest) -> None: """Skip real trading tests in headless mode.""" if is_headless() and request.node.get_closest_marker("real_trading"): pytest.skip("Skipping real trading test in headless mode")