Context manager:
The main pattern for MetaTrader5 scripts in python is:
mt5.initialize()
#user code
mt5.shutdown()
The challenge is if the user code throws an exception then shutdown() won't be reached.
mt5.initialize() err = 1 / 0 # ZeroDivisionError # shutdown not called
pymt5adapter overcomes this challenge by implementing this login in a context manager, thus ensuring that no matter what, shutdown is always called.
import pymt5adapter as mt5 with mt5.connected(debug_logging=True): x = 1 / 0 # MT5 connection has been initialized. # [shutdown()][(1, 'Success')] # MT5 connection has been shutdown. # Traceback (most recent call last): # File "C:/Users/user/AppData/Roaming/JetBrains/PyCharm2020.1/scratches/mt5.py", line 37, in <module> # x = 1 / 0 # ZeroDivisionError: division by zero
The context manager can also modify the behavior of the API. With one simple flag, the entire API can raise exceptions instead of the default behavior of failing silently.
import pymt5adapter as mt5 with mt5.connected(raise_on_errors=True) as conn: try: invalid = mt5.history_deals_get('sdfasdf', 'sdfadd') except mt5.MT5Error as e: print('Errors are raised globally for each function') print('Error code =', e.error_code) print('Error description =', e.description) conn.raise_on_errors = False invalid = mt5.history_deals_get('sdfasdf', 'sdfadd') print(invalid) print("Default behavior toggled at runtime. No Exceptions Raised.") # Errors are raised globally for each function # Error code = ERROR_CODE.INVALID_PARAMS # Error description = Invalid arguments('sdfasdf', 'sdfadd'){} # () # Default behavior toggled at runtime. No Exceptions Raised.
See builtin docs for more info on params available for the context manager.
Example script:
import pymt5adapter as mt5 import pandas as pd def main(): s = pd.Series(mt5.account_info()._asdict(), name='AccountInfo') print(s) if __name__ == '__main__': with mt5.connected(): main()
socket stream server for MT5
import argparse import asyncio import json import pymt5adapter as mta API_FUNCS = mta.get_function_dispatch() async def handle_api(reader, writer): data = await reader.read(1000) message = data.decode() try: req = json.loads(message) func = API_FUNCS[req['function']] args = req.get('args', []) kwargs = req.get('kwargs', {}) response = func(*args, **kwargs) res = dict(error=mta.mt5_last_error(), response=response) except Exception as e: res = dict(error=[-1, f'{type(e).__name__} {e.args[0]}'], response=None) res = json.dumps(res) writer.write(res.encode()) await writer.drain() writer.close() async def main(): server = await asyncio.start_server(handle_api, '127.0.0.1', 8888) addr = server.sockets[0].getsockname() print('Serving on {}:{}'.format(*addr)) async with server: await server.serve_forever() if __name__ == '__main__': parser = argparse.ArgumentParser(description='Socket server for MT5 API') parser.add_argument('--path', type=str, help='Absolute path to the terminal64.exe') parser.add_argument('--login', type=int, help='Account number') parser.add_argument('--password', type=str, help='Account password') parser.add_argument('--server', type=str, help='Name of the trade server eg. "MetaQuotes-Demo"') parser.add_argument('--portable', action='store_true', default=None, help='Will launch the terminal in portable mode if this flag is set') ns = parser.parse_args() kw = vars(ns) val_states = [v is not None for v in kw.values()] if any(val_states) and not all(val_states): print(ns) raise Exception('Missing commandline arguments.') mt5_connected = mta.connected( ensure_trade_enabled=True, native_python_objects=True, raise_on_errors=True, **kw ) with mt5_connected: asyncio.run(main())
Example python client. (can be anything that uses sockets and JSON)
import asyncio import json async def socket_client(message): reader, writer = await asyncio.open_connection('127.0.0.1', 8888) print(f'Send: {message!r}') writer.write(message.encode()) data = await reader.read() data = data.decode() writer.close() return json.loads(data) if __name__ == '__main__': req = json.dumps({ 'function': 'copy_rates_from_pos', 'kwargs' : { 'symbol' : 'EURUSD', 'timeframe': 1, 'start_pos': 1, 'count' : 1000 }, }) res = asyncio.run(socket_client(req)) print(str(res)[:100]) errno, strerr = res['error'] if errno == 1: rates = res['response'] print(len(rates))
Thanks for this code. I was able to get pending orders implemented using it:
https://gist.github.com/metaperl/dc728891166963661ad6f66e80db2de5
I do have few questions/comments:
- I added a .pips() method because I think in pips not points - perhaps such a method could be added to the Symbol class?
- How should one test for successful execution of a method? Should I simply see if the retcode is 10009?
Again, thanks a lot. I was struggling with the vanilla MT5 interface. But it kicks serious butt over the old ways of doing things in MT4!
![Create pending orders in Python MT5 via pymt5adapter Create pending orders in Python MT5 via pymt5adapter](https://c.mql5.com/36/37/create-pending-orders-in-python.png)
- gist.github.com
- I added a .pips() method because I think in pips not points - perhaps such a method could be added to the Symbol
Thanks for the kind words. I like your idea about adding a tick calculator to the Symbol class. I added an explicit function titled "tick_calc" to the Symbol class because you will always want to use Symbol.trade_tick_size for calculations, however, if you prefer you can sublclass and alias tick_calc to pips quite easily.
Declaration:
def tick_calc(self, price: float, num_ticks: int): """Calculate a new price by number of ticks from the price param. The result is normalized to the tick-size of the instrument. :param price: The price to add or subtract ticks from. :param num_ticks: number of ticks. If subtracting ticks then this should be a negative number. :return: A new price adjusted by the number of ticks and normalized to tick-size. """ return self.normalize_price(price + num_ticks * self.trade_tick_size)
Alias:
class MySymbol(Symbol): pips = Symbol.tick_calc
Usage:
symbol = MySymbol('EURUSD') # alt constructor order = Order.as_buy_limit( volume=symbol.volume_min, expiration=mta.ORDER_TIME.DAY, symbol=symbol.name, ) # call the object directly "callable object" order(magic=12345, comment="python") # set properties directly order.price = symbol.pips(symbol.bid, -10) order.sl = symbol.pips(order.price, -100) order.tp = symbol.pips(order.price, +100) res = order.send()
Make sure you update pymt5adapter
pip install -U pymt5adapter
- How should one test for successful execution of a method? Should I simply see if the retcode is 10009?
I would recommend that you use logging and the pymt5adapter.TRADE_RETCODE enum class. The new version of pymt5adapter includes a major upgrade with regards to logging. The API logging is set by the "logger" param in the connected context manager. The logger param excepts a logging.Logger instance (you can define your own) or you can get the recommended Logger instance using the pymt5adapter.get_logger function, or you can pass in a path to the loggin file and the program will return a logging.Logger with loglevel set to logging.INFO.
Example:
import logging from pathlib import Path import pymt5adapter as mta from pymt5adapter.order import Order from pymt5adapter.symbol import Symbol class MySymbol(Symbol): pips = Symbol.tick_calc def main(conn): print(conn.ping()) print(conn.terminal_info) symbol = MySymbol('EURUSD') # alt constructor order = Order.as_buy_limit( volume=symbol.volume_min, expiration=mta.ORDER_TIME.DAY, symbol=symbol.name, ) # call the object directly "callable object" order(magic=12345, comment="python") # set properties directly order.price = symbol.pips(symbol.bid, -10) order.sl = symbol.pips(order.price, -100) order.tp = symbol.pips(order.price, +100) res = order.send() retcode = mta.TRADE_RETCODE(res.retcode) if retcode is mta.TRADE_RETCODE.DONE: print('done') else: print('trade error', mta.trade_retcode_description(retcode)) try: print(1 / 0) except ZeroDivisionError as e: conn.logger.error(mta.LogJson(e, { 'type' : 'exception', 'exception': { 'type' : type(e).__name__, 'message': str(e) } })) raise if __name__ == '__main__': print(mta.__version__) print(mta.__author__) desktop = Path.home() / 'Desktop' logger = mta.get_logger(loglevel=logging.DEBUG, path_to_logfile=desktop / 'example_mt5.log', time_utc=True) mt5_connected = mta.connected( path=desktop / 'terminal1/terminal64.exe', portable=True, server='MetaQuotes-Demo', login=18237468, password='kjhfdslk', timeout=5000, logger=logger, ensure_trade_enabled=True, raise_on_errors=True, ) with mt5_connected as conn: main(conn)
Which will result in a logfile of:
2020-06-21 21:03:41,857 INFO Terminal Initialize Success {"type":"terminal_connection_state","state":true} 2020-06-21 21:03:41,858 INFO Init TerminalInfo {"type":"init_terminal_info","terminal_info":{"community_account":false,"community_connection":false,"connected":true,"dlls_allowed":false,"trade_allowed":true,"tradeapi_disabled":false,"email_enabled":false,"ftp_enabled":false,"notifications_enabled":false,"mqid":false,"build":2497,"maxbars":100000000,"codepage":0,"ping_last":147583,"community_balance":0.0,"retransmission":0.47393364928909953,"company":"MetaQuotes Software Corp.","name":"MetaTrader 5","language":"English","path":"C:\\Users\\nicho\\Desktop\\Terminal1","data_path":"C:\\Users\\nicho\\Desktop\\Terminal1","commondata_path":"C:\\Users\\nicho\\AppData\\Roaming\\MetaQuotes\\Terminal\\Common"}} 2020-06-21 21:03:41,870 INFO Init AccountInfo {"type":"init_account_info","account_info":{"login":7658765,"trade_mode":0,"leverage":100,"limit_orders":200,"margin_so_mode":0,"trade_allowed":true,"trade_expert":true,"margin_mode":2,"currency_digits":2,"fifo_close":false,"balance":1000000.0,"credit":0.0,"profit":0.0,"equity":1000000.0,"margin":0.0,"margin_free":1000000.0,"margin_level":0.0,"margin_so_call":50.0,"margin_so_so":30.0,"margin_initial":0.0,"margin_maintenance":0.0,"assets":0.0,"liabilities":0.0,"commission_blocked":0.0,"name":"nicholishen","server":"MetaQuotes-Demo","currency":"USD","company":"MetaQuotes Software Corp."}} 2020-06-21 21:03:41,876 DEBUG Function Debugging: symbol_info {"type":"function_debugging","latency_ms":0.322,"last_error":[1,"Success"],"call_signature":{"function":"symbol_info","args":["EURUSD"],"kwargs":{}}} 2020-06-21 21:03:41,876 DEBUG Function Debugging: symbol_info_tick {"type":"function_debugging","latency_ms":0.174,"last_error":[1,"Success"],"call_signature":{"function":"symbol_info_tick","args":["EURUSD"],"kwargs":{}}} 2020-06-21 21:03:42,033 DEBUG Function Debugging: order_send {"type":"function_debugging","latency_ms":155.543,"last_error":[1,"Success"],"call_signature":{"function":"order_send","args":[{"action":5,"magic":12345,"volume":0.01,"price":1.11791,"sl":1.11691,"tp":1.11891,"type":2,"expiration":1,"comment":"python"}],"kwargs":{}}} 2020-06-21 21:03:42,034 INFO Order Request: BUY_LIMIT {"type":"order_request","request":{"action":5,"magic":12345,"order":0,"symbol":"EURUSD","volume":0.01,"price":1.11791,"stoplimit":0.0,"sl":1.11691,"tp":1.11891,"deviation":0,"type":2,"type_filling":0,"type_time":0,"expiration":1,"comment":"python","position":0,"position_by":0}} 2020-06-21 21:03:42,035 INFO Order Response: DONE {"type":"order_response","latency_ms":155.543,"response":{"retcode":10009,"deal":0,"order":634208352,"volume":0.01,"price":0.0,"bid":0.0,"ask":0.0,"comment":"Request executed","request_id":7,"retcode_external":0}} 2020-06-21 21:03:42,035 ERROR division by zero {"type":"exception","exception":{"type":"ZeroDivisionError","message":"division by zero"}} 2020-06-21 21:03:42,036 CRITICAL UNCAUGHT EXCEPTION: division by zero {"type":"exception","last_error":[1,"Success"],"exception":{"type":"ZeroDivisionError","message":"division by zero"}} 2020-06-21 21:03:42,036 INFO Terminal Shutdown {"type":"terminal_connection_state","state":false}
Which can be easily parsed using python:
import logging from pathlib import Path import dateutil.parser as dp import pandas as pd try: import ujson as json except ImportError: import json class LogLine: def __init__(self, log_line): time, loglevel, short_message, json_dump, *crap = log_line.split('\t') self.time = dp.parse(time) self.loglevel = self.loglevel_mapped(loglevel) self.short_message = short_message self.json_dump = json.loads(json_dump) def loglevel_mapped(self, loglevel): return getattr(logging, loglevel, None) def iter_json(path_to_logfile): p = Path(path_to_logfile) with p.open() as f: for line in f: try: yield LogLine(line) except Exception as e: print(e) if __name__ == '__main__': logfile = Path.home() / 'Desktop/testinglog.log' function_calls = [] for line in iter_json(logfile): d = line.json_dump if d['type'] == 'function_debugging': function_calls.append({'latency_ms': d.get('latency_ms'), **d['call_signature']}) df = pd.DataFrame(function_calls).groupby('function').mean() df = df[df['latency_ms'] >= 0.5] print(df)
Result dataframe
latency_ms function copy_rates 0.503000 history_deals_get 7.749205 order_send 51.148026 symbols_get 20.833088 terminal_info 9.934125
pymt5adapter release 0.4.3
pip install -U pymt5adapter
- fixed bug with Python 3.6 compatibility
- fixed bug in Order.as_sell constructor
- enhanced logging capabilities
Thanks for your python wrapper. It will help me transition from using MQL to Python.
nice going, I also developed a new drag and drop able to use python metatrader4 and metatrader5.
easy to set up, lot's of features!
Hi, I want to know if we can call built-in MQL5 functions via python or with this pymt5adapter?
For example: I want to call this ChartScreenShot MQL5 function from python to take screenshot of the chart whenever a new position is initiated and read the saved image in the local folder of from mql5.com cloud image repository?
![Documentation on MQL5: Chart Operations / ChartScreenShot Documentation on MQL5: Chart Operations / ChartScreenShot](https://c.mql5.com/36/54/documentation-on-mql5-chart-operations.png)
- www.mql5.com
No, you can only access the functions provided by the standard MetaTrader 5 Python Integration API, not the MQL5 functionality. For that you will have to write it in MQL5.
![MQL5 - Language of trade strategies built-in the MetaTrader 5 client terminal](https://c.mql5.com/i/registerlandings/logo-2.png)
- Free trading apps
- Over 8,000 signals for copying
- Economic news for exploring financial markets
You agree to website policy and terms of use
I wanted to spin off a separate thread for this package so topics and discussions won't clutter up the original Python thread.
Summary: pymt5adapter is a drop-in replacement (wrapper) for the `MetaTrader5` python package by MetaQuotes. The API functions simply pass through values from the `MetaTrader5` functions, but adds functionality as well as a more "pythonic" interface.
github: https://github.com/nicholishen/pymt5adapter
pypi: https://pypi.org/project/pymt5adapter/
Install: