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 }} 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" diff --git a/mqpy/trade.py b/mqpy/trade.py index 5484fa6..a635178 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 @@ -383,27 +359,24 @@ 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) - - # 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 self.trading_time(): + 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/tests/conftest.py b/tests/conftest.py index f1d8f21..ee59242 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,11 +2,35 @@ from __future__ import annotations +import ctypes import logging -from typing import Generator +import time +from typing import TYPE_CHECKING, Generator import pytest +if TYPE_CHECKING: + from mqpy.trade import Trade + +VK_CONTROL = 0x11 +VK_E = 0x45 + +logger = logging.getLogger(__name__) + + +def send_ctrl_e() -> None: + """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 +56,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 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..e9a251f 100644 --- a/tests/test_trade.py +++ b/tests/test_trade.py @@ -4,7 +4,9 @@ import contextlib import logging +import os import time +from pathlib import Path from typing import Generator import MetaTrader5 as Mt5 @@ -24,14 +26,49 @@ 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]: +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 not Mt5.initialize(): - pytest.skip("MetaTrader5 could not be initialized") + if is_headless(): + logger.info("Running in headless mode - skipping MT5 initialization") + yield + 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", + ] + + for path in common_paths: + 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.") + yield try: @@ -84,12 +121,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 +353,155 @@ 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: +def test_close_position(trade: Trade) -> None: # noqa: C901 - Test needs to handle many edge cases """Test closing a position with real trades.""" - try: - trade.open_buy_position("Test Position to Close") - - time.sleep(2) + # 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}") - positions = Mt5.positions_get(symbol=trade.symbol) - assert positions is not None + logger.info(f"Symbol {trade.symbol} trade mode: {symbol_info.trade_mode}") - has_position = False + # 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 - trade.close_position("Test Closing Position") + # 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 - time.sleep(2) + assert has_position, "Failed to find opened position" - positions = Mt5.positions_get(symbol=trade.symbol) + 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 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: + logger.exception("Error during position test") + raise + + +@pytest.fixture(autouse=True) +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")