test calendar

Author: nicker-bocker

README.md

@@ -126,5 +126,32 @@ def out_deal(deal: mt5.TradeDeal):
 out_deals = mt5.history_deals_get(function=out_deal)
 ```
 
+### Calendar Events
+Get economic calendar events from mql5.com using the `calendar_events` function. The function is memoized so
+so repeated calls pull event data from cache instead of repeated calls to the server. 
+
+Example:
+```python
+from pymt5adapter.calendar import calendar_events
+from pymt5adapter.calendar import Currency
+from pymt5adapter.calendar import Importance
+
+symbol = mt5.symbols_get(function=forex_symbol)[0]
+one_week = timedelta(weeks=1)
+now = datetime.now()
+default_one_week_ahead_all_events = calendar_events()
+filtered_by_callback = calendar_events(function=lambda e: 'fed' in e['event_name'].lower())
+filtered_by_flags = calendar_events(importance=Importance.MEDIUM | Importance.HIGH,
+                                    currencies=Currency.USD | Currency.JPY)
+filtered_by_strings = calendar_events(importance='medium high',
+                                      currencies='usdjpy')
+filtered_by_iterables = calendar_events(importance=('medium', ' high'),
+                                        currencies=('usd', 'jpy'))
+filtered_by_specific_times = calendar_events(time_from=now, time_to=now + one_week)
+filtered_by_timedeltas = calendar_events(time_from=(-one_week), time_to=one_week)
+filtered_by_timedelta_lookback = calendar_events(-one_week)
+calendar_events_in_russian = calendar_events(language='ru')
+```
+
 [intellisence_screen]: https://github.com/nicholishen/pymt5adapter/raw/master/images/intellisense_screen.jpg "intellisence example"
 [docs_screen]: https://github.com/nicholishen/pymt5adapter/raw/master/images/docs_screen.jpg "quick docs example"

pymt5adapter/calendar.py

@@ -45,6 +45,47 @@ class Currency(enum.IntFlag):
     INR = enum.auto()
 
 
+def calendar_events(time_to: Union[datetime, timedelta] = None,
+                    *,
+                    time_from: Union[datetime, timedelta] = None,
+                    importance: Union[Iterable[str], str, int] = None,
+                    currencies: Union[Iterable[str], str, int] = None,
+                    function: Callable = None,
+                    round_minutes: int = 15,
+                    cache_clear: bool = False,
+                    language: str = None,
+                    **kwargs,
+                    ) -> List[dict]:
+    """Get economic events from mql5.com/calendar. A call with empty args will results in all events for the next week.
+    Since the function is memoized, the time is rounded to ``round_minutes`` and cached. This avoids repeat requests
+    to the mql5.com server. In order to refresh the results on subsequest function calls you'll need to set the
+    ``cache_clear`` param to True.
+
+    :param time_to: Can be a timedelta or datetime object. If timedelta then the to time is calculated as
+        datetime.now() + time_to. If the the time_from param is specified the the time_to will be calculated as
+        time_from + time_to. Can also do a history look-back by passing in a negative timedelta to this param.
+    :param time_from: Can be a timedelta or datetime object. If a timedelta object is passed then the time_from is
+        calculated as time_from + now(), thus it needs to be a negative delta.
+    :param importance: Can be int flags from the Importance enum class, and iterable of strings or a (space and/or
+        comma separated string)
+    :param currencies: Can be int flags from the Importance enum class, and iterable of strings or a (space and/or
+        comma separated string) Pairs are automatically separated to the respective currency codes.
+    :param function: A callback function that receives an event dict and returns bool for filtering.
+    :param round_minutes: Round the number of minutes to this factor. Rounding aides the memoization of parameters.
+    :param cache_clear: Clear the memo cache to refresh events from the server.
+    :param language: The language code for the calendar. Default is 'en'
+    :param kwargs:
+    :return:
+    """
+    cal_args = _construct_args(**locals())
+    if cache_clear:
+        _get_calendar_events.cache_clear()
+    events = _get_calendar_events(language=language, **cal_args)
+    if function:
+        events = list(filter(function, events))
+    return events
+
+
 @functools.lru_cache
 def _get_calendar_events(datetime_from: datetime,
                          datetime_to: datetime,
@@ -72,11 +113,11 @@ def _get_calendar_events(datetime_from: datetime,
             e['Url'] = _BASE_URL + e['Url']
             e['ReleaseDate'] = time
             e['request'] = data
-            filtered_events.append(e)
-    filtered_events = [
-        {_camel_to_snake(k): v for k, v in x.items() if k not in _OMIT_RESULT_KEYS}
-        for x in filtered_events
-    ]
+            formatted_event = {}
+            for k, v in e.items():
+                if k not in _OMIT_RESULT_KEYS:
+                    formatted_event[_camel_to_snake(k)] = v
+            filtered_events.append(formatted_event)
     return filtered_events
 
 
@@ -106,10 +147,10 @@ def _camel_to_snake(w):
     return w
 
 
-_camel_to_snake.pattern = re.compile(r'[A-Z][a-z]+')
+_camel_to_snake.pattern = re.compile(r'[A-Z][a-z]*')
 
 
-@functools.lru_cache
+@functools.lru_cache(maxsize=128, typed=True)
 def _make_flag(enum_cls: Union[Type[Importance], Type[Currency]],
                flags: Union[Iterable[str], int, str] = None
                ) -> int:
@@ -126,40 +167,12 @@ def _make_flag(enum_cls: Union[Type[Importance], Type[Currency]],
     return int(flag)
 
 
-def calendar_events(time_to: Union[datetime, timedelta] = None,
-                    *,
-                    time_from: Union[datetime, timedelta] = None,
-                    importance: Union[Iterable[str], str, int] = None,
-                    currencies: Union[Iterable[str], str, int] = None,
-                    function: Callable = None,
-                    round_minutes: int = 15,
-                    cache_clear: bool = False,
-                    language: str = None,
-                    **kwargs,
-                    ) -> List[dict]:
-    """Get economic events from mql5.com/calendar. A call with empty args will results in all events for the next week.
-    Since the function is memoized, the time is rounded to ``round_minutes`` and cached. This avoids repeat requests
-    to the mql5.com server. In order to refresh the results on subsequest function calls you'll need to set the
-    ``cache_clear`` param to True.
-
-    :param time_to: Can be a timedelta or datetime object. If timedelta then the to time is calculated as
-        datetime.now() + time_to. If the the time_from param is specified the the time_to will be calculated as
-        time_from + time_to. Can also do a history look-back by passing in a negative timedelta to this param.
-    :param time_from: Can be a timedelta or datetime object. If a timedelta object is passed then the time_from is
-        calculated as time_from + now(), thus it needs to be a negative delta.
-    :param importance: Can be int flags from the Importance enum class, and iterable of strings or a (space and/or
-        comma separated string)
-    :param currencies: Can be int flags from the Importance enum class, and iterable of strings or a (space and/or
-        comma separated string) Pairs are automatically separated to the respective currency codes.
-    :param function: A callback function that receives an event dict and returns bool for filtering.
-    :param round_minutes: Round the number of minutes to this factor. Rounding aides the memoization of parameters.
-    :param cache_clear: Clear the memo cache to refresh events from the server.
-    :param language: The language code for the calendar. Default is 'en'
-    :param kwargs:
-    :return:
-    """
-    if cache_clear:
-        _get_calendar_events.cache_clear()
+def _construct_args(**kw):
+    time_to = kw.get('time_to')
+    time_from = kw.get('time_from')
+    round_minutes = kw.get('round_minutes')
+    importance = kw.get('importance')
+    currencies = kw.get('currencies')
     now = datetime.now()
     if time_to is None and time_from is None:
         time_to = now + timedelta(weeks=1)
@@ -174,11 +187,5 @@ def calendar_events(time_to: Union[datetime, timedelta] = None,
     _f = lambda x: tuple(x) if isinstance(x, Iterable) and not isinstance(x, str) else x
     importance, currencies = _f(importance), _f(currencies)
     i_flag, c_flag = _make_flag(Importance, importance), _make_flag(Currency, currencies)
-    events = _get_calendar_events(datetime_from=time_from,
-                                  datetime_to=time_to,
-                                  importance=i_flag,
-                                  currencies=c_flag,
-                                  language=language)
-    if function:
-        events = list(filter(function, events))
-    return events
+    res_args = dict(datetime_to=time_to, datetime_from=time_from, importance=i_flag, currencies=c_flag)
+    return res_args

tests/test_calendar.py

@@ -0,0 +1,94 @@
+from .context import pymt5adapter as mt5
+from pymt5adapter.calendar import calendar_events, Importance, Currency
+import pymt5adapter.calendar as cal
+from datetime import datetime, timedelta
+import pytest
+
+
+def test_calendar_events():
+    from time import perf_counter
+    f = datetime.now()
+    t = f + timedelta(days=30)
+    t1 = perf_counter()
+    events = calendar_events(time_from=f, time_to=t, importance=Importance.HIGH)
+    t1 = perf_counter() - t1
+    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 type(events) is list
+    assert len(events) > 0
+    for e in events:
+        etime = e['release_date']
+        assert f <= etime <= t
+        assert e['importance'].upper() not in ['MEDIUM', 'LOW', 'HOLIDAY']
+
+
+def test_construct_args():
+    test_args = dict(time_to=timedelta(days=-30), round_minutes=15)
+    res_args = cal._construct_args(**test_args)
+    assert isinstance(res_args['datetime_to'], datetime)
+    assert isinstance(res_args['datetime_from'], datetime)
+    assert isinstance(res_args['importance'], int)
+    assert isinstance(res_args['currencies'], int)
+    t = test_args['time_to'] = datetime(2020, 1, 1)
+    test_args['time_from'] = t + timedelta(weeks=1)
+    res_args = cal._construct_args(**test_args)
+    assert res_args['datetime_from'] < res_args['datetime_to']
+
+
+def test_camel_to_snake():
+    controls = {
+        'Url'         : 'url',
+        'EventName'   : 'event_name',
+        'CamelCaseKey': 'camel_case_key',
+        'ok_id'       : 'ok_id',
+        'NItems'      : 'n_items'
+    }
+    for k, v in controls.items():
+        assert cal._camel_to_snake(k) == v
+
+
+def test_normalize_time():
+    control_minutes = 15
+    control_time = datetime(2020, 1, 1, 1, control_minutes)
+    delta = timedelta(minutes=7)
+    test_time = control_time - delta
+    result = cal._time_ceil(test_time, minutes=control_minutes)
+    assert result == control_time
+    test_time = control_time + delta
+    result = cal._time_floor(test_time, minutes=control_minutes)
+    assert result == control_time
+
+
+def test_split_pairs():
+    control = set('EUR USD AUD'.split())
+
+    def clean(res):
+        return set(map(str.upper, res))
+
+    result = clean(cal._split_pairs(['eurusd', 'audusd', 'eur', 'aud', 'usd']))
+    assert result == control
+
+
+def test_make_flags():
+    control_currency = Currency.USD | Currency.EUR | Currency.AUD
+    control_importance = Importance.HOLIDAY | Importance.LOW
+    test_currency = cal._make_flag(Currency, control_currency)
+    assert test_currency == control_currency
+    test_currency = cal._make_flag(Currency, 'eur aud usd')
+    assert test_currency == control_currency
+    test_currency = cal._make_flag(Currency, 'eur, aud, usd')
+    assert test_currency == control_currency
+    test_currency = cal._make_flag(Currency, ('eurusd', 'audusd'))
+    assert test_currency == control_currency
+    test_currency = cal._make_flag(Currency, ('eur', 'usd', 'aud'))
+    assert test_currency == control_currency
+    test_currency = cal._make_flag(Currency, 'eurusd, audusd, usdjpy')
+    assert test_currency != control_currency
+    test_importance = cal._make_flag(Importance, 'low, holiday')
+    assert test_importance == control_importance
+
+
+if __name__ == '__main__':
+    pass