optimize

pymt5adapter/context.py

@@ -27,7 +27,7 @@ def connected(*,
               logger: Callable = None,
               raise_on_errors: bool = False,
               debug_logging: bool = False,
-              # force_namedtuple: bool = False,
+              convert_namedtuples_to_dict:bool=False,
               **kwargs
               ) -> None:
     """Context manager for managing the connection with a MT5 terminal using the python ``with`` statement.
@@ -59,8 +59,9 @@ def connected(*,
     args = helpers.reduce_args(args)
     mt5_keys = ('path', 'portable', 'server', 'login', 'password', 'timeout')
     mt5_kwargs = {k: v for k, v in args.items() if k in mt5_keys}
-    _state.global_debugging = bool(debug_logging)
-    _state.raise_on_errors = bool(raise_on_errors)
+    _state.global_debugging = debug_logging
+    _state.raise_on_errors = raise_on_errors
+    # _state.convert_namedtuples_to_dict = convert_namedtuples_to_dict
     # _state.force_namedtuple = bool(force_namedtuple)
     _state.log = logger or print
     log = _state.log

pymt5adapter/core.py

@@ -10,26 +10,29 @@
 from .state import global_state as _state
 from .types import *
 
+# _OMIT_FUNCS_FROM_CONVERSION = []
 
 class MT5Error(Exception):
     pass
 
 
 def _context_manager_modified(f):
     @functools.wraps(f)
-    def wrapper(*args, **kwargs):
+    def context_manager_modified_wrapper(*args, **kwargs):
         result = f(*args, **kwargs)
         if _state.global_debugging:
             call_sig = f"{f.__name__}({_h.args_to_str(args, kwargs)})"
             _state.log(f"[{call_sig}][{last_error()}][{str(result)[:80]}]")
         # make sure we log before we raise
-        if _state.raise_on_errors:
+        if _state.raise_on_errors and not result: # no need to check last error if we got a result
             error_code, description = last_error()
             if error_code != _const.RES_S_OK:
                 raise MT5Error(error_code, description)
+        # if _state.convert_namedtuples_to_dict and result:
+        #     result = _h.as_dict_all(result)
         return result
 
-    return wrapper
+    return context_manager_modified_wrapper
 
 
 @_context_manager_modified
@@ -167,10 +170,9 @@ def symbols_get(*,
     symbols = _mt5.symbols_get(group=group) if group else _mt5.symbols_get()
     if symbols is None and _state.raise_on_errors:
         build = version()
-        if build:
-            if build[1] < _const.MIN_TERMINAL_BUILD :
-                raise MT5Error(_const.RES_X_TERMINAL_VERSION_OUTDATED,
-                               "The terminal build needs to be updated to support this feature.")
+        if build and build[1] < _const.MIN_TERMINAL_BUILD :
+            raise MT5Error(_const.RES_X_TERMINAL_VERSION_OUTDATED,
+                           "The terminal build needs to be updated to support this feature.")
         else:
             error_code, des = last_error()
             if error_code == _const.RES_S_OK:
@@ -215,9 +217,14 @@ def symbol_info_tick(symbol) -> Tick:
     :param symbol:
     :return:
     """
-    symbol = _h.any_symbol(symbol)
-    return _mt5.symbol_info_tick(symbol)
+    try:
+        return _mt5.symbol_info_tick(symbol)
+    except Exception:
+        symbol = _h.any_symbol(symbol)
+        return _mt5.symbol_info_tick(symbol)
 
+#direct access to API function without any added overhead
+symbol_info_tick_fast = _mt5.symbol_info_tick
 
 @_context_manager_modified
 def symbol_select(symbol, enable: bool = True) -> bool:

pymt5adapter/state.py

@@ -6,12 +6,13 @@ class _GlobalState:
 
     def __init__(self):
         self.__dict__ = self.__shared_state
-        if 'global_debugging' not in self.__shared_state:
+        if 'max_bars' not in self.__shared_state:
             self.set_defaults()
 
     def set_defaults(self):
-        self.raise_on_errors = False
+        # self.convert_namedtuples_to_dict = False
         self.global_debugging = False
+        self.raise_on_errors = False
         self.max_bars = 100_000
         self.log = print
 

tests/test_calendar.py

@@ -15,7 +15,7 @@ def test_calendar_events():
     t2 = perf_counter()
     events = calendar_events(time_from=f, time_to=t, importance=Importance.HIGH)
     t2 = perf_counter() - t2
-    assert t2 * 1000 < t1 # verify cache speed
+    assert t2 * 1000 < t1  # verify cache speed
     assert type(events) is list
     assert len(events) > 0
     for e in events:
@@ -46,8 +46,9 @@ def test_camel_to_snake():
         'NItems'      : 'n_items'
     }
     for k, v in controls.items():
-        assert cal._camel_to_snake(k) == v
-
+        res =  cal._camel_to_snake(k)
+    for k, v in controls.items():
+        assert (res :=  cal._camel_to_snake(k)) == v
 
 def test_normalize_time():
     control_minutes = 15