Repository: liampauling/flumine
Pull request #502 (2021-09-17 07:27:43)

Release/1.20.0


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 3071f22..bf3b6ab 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,21 @@
 Release History
 ---------------
 
+1.20.0 (2021-09-17)
++++++++++++++++++++
+
+**Improvements**
+
+- Update to use `listener_kwargs` in `_process` rather than `snap`
+
+**Bug Fixes**
+
+- #499 missing market call on closure
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.14.0
+
 1.19.17 (2021-09-14)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 8da91c8..54f464a 100644
--- a/flumine/__version__.py
+++ b/flumine/__version__.py
@@ -1,6 +1,6 @@
 __title__ = "flumine"
 __description__ = "Betfair trading framework"
 __url__ = "https://github.com/liampauling/flumine"
-__version__ = "1.19.17"
+__version__ = "1.20.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index ab744ee..f8e3f5f 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -272,6 +272,8 @@ class BaseFlumine:
                 extra={"market_id": market_id, **self.info},
             )
             return
+        # process market
+        market(market_book)
         if market.closed is False:
             market.close_market()
         if recorder is False:
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index 1f49818..a7fc2e3 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -125,7 +125,11 @@ class Blotter:
         for order in self.strategy_selection_orders(strategy, *lookup[1:]):
             if order == exclusion:
                 continue
-            if order.status == OrderStatus.VIOLATION:
+            if order.status in [
+                OrderStatus.PENDING,
+                OrderStatus.VIOLATION,
+                OrderStatus.EXPIRED,
+            ]:
                 continue
             if order.order_type.ORDER_TYPE == OrderTypes.LIMIT:
                 _size_matched = order.size_matched  # cache
@@ -134,12 +138,13 @@ class Blotter:
                         mb.append((order.average_price_matched, _size_matched))
                     else:
                         ml.append((order.average_price_matched, _size_matched))
-                _size_remaining = order.size_remaining  # cache
-                if order.order_type.price and _size_remaining:
-                    if order.side == "BACK":
-                        ub.append((order.order_type.price, _size_remaining))
-                    else:
-                        ul.append((order.order_type.price, _size_remaining))
+                if not order.complete:
+                    _size_remaining = order.size_remaining  # cache
+                    if order.order_type.price and _size_remaining:
+                        if order.side == "BACK":
+                            ub.append((order.order_type.price, _size_remaining))
+                        else:
+                            ul.append((order.order_type.price, _size_remaining))
             elif order.order_type.ORDER_TYPE in (
                 OrderTypes.LIMIT_ON_CLOSE,
                 OrderTypes.MARKET_ON_CLOSE,
diff --git a/flumine/markets/middleware.py b/flumine/markets/middleware.py
index 8a27f66..3420544 100644
--- a/flumine/markets/middleware.py
+++ b/flumine/markets/middleware.py
@@ -183,7 +183,7 @@ class SimulatedMiddleware(Middleware):
     @staticmethod
     def _process_simulated_orders(market, market_analytics: dict) -> None:
         for order in market.blotter.live_orders:
-            if order.simulated and order.status == OrderStatus.EXECUTABLE:
+            if order.status == OrderStatus.EXECUTABLE and order.simulated:
                 runner_analytics = market_analytics[
                     (order.selection_id, order.handicap)
                 ]
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
index ad0a32b..44cef31 100644
--- a/flumine/streams/historicalstream.py
+++ b/flumine/streams/historicalstream.py
@@ -9,6 +9,7 @@ from betfairlightweight.compat import json
 
 from .basestream import BaseStream
 from ..exceptions import ListenerError
+from ..utils import create_time
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +24,7 @@ class FlumineMarketStream(MarketStream):
     """
 
     def _process(self, data: list, publish_time: int) -> bool:
+        active = False
         for market_book in data:
             if "id" not in market_book:
                 continue
@@ -53,43 +55,52 @@ class FlumineMarketStream(MarketStream):
                     % (self, self.unique_id, market_id, len(self._caches))
                 )
 
-            market_book_cache.update_cache(market_book, publish_time)
-            self._updates_processed += 1
-        return False
-
-    def snap(self, market_ids: list = None) -> list:
-        market_books = []
-        for cache in list(self._caches.values()):
-            if market_ids and cache.market_id not in market_ids:
-                continue
-            # if market is not open (closed/suspended) send regardless
-            if cache._definition_status == "OPEN":
+            # listener_kwargs filtering
+            active = True
+            if "marketDefinition" in market_book:
+                _definition_status = market_book["marketDefinition"].get("status")
+                _definition_in_play = market_book["marketDefinition"].get("inPlay")
+                _definition_market_time = market_book["marketDefinition"].get(
+                    "marketTime"
+                )
+            else:
+                _definition_status = market_book_cache._definition_status
+                _definition_in_play = market_book_cache._definition_in_play
+                _definition_market_time = market_book_cache.market_definition[
+                    "marketTime"
+                ]
+
+            # if market is not open (closed/suspended) process regardless
+            if _definition_status == "OPEN":
                 if self._listener.inplay:
-                    if not cache._definition_in_play:
-                        continue
+                    if not _definition_in_play:
+                        active = False
                 elif self._listener.seconds_to_start:
-                    _now = datetime.datetime.utcfromtimestamp(cache.publish_time / 1e3)
-                    _market_time = BaseResource.strip_datetime(
-                        cache.market_definition["marketTime"]
-                    )
+                    _now = datetime.datetime.utcfromtimestamp(publish_time / 1e3)
+                    _market_time = BaseResource.strip_datetime(_definition_market_time)
                     seconds_to_start = (_market_time - _now).total_seconds()
                     if seconds_to_start > self._listener.seconds_to_start:
-                        continue
+                        active = False
                 if self._listener.inplay is False:
-                    if cache._definition_in_play:
-                        continue
-            market_books.append(cache.create_resource(self.unique_id, snap=True))
-        return market_books
+                    if _definition_in_play:
+                        active = False
+            # check if refresh required
+            if active and not market_book_cache.active:
+                market_book_cache.refresh_cache()
+
+            market_book_cache.update_cache(market_book, publish_time, active=active)
+            self._updates_processed += 1
+        return active
 
 
 class FlumineRaceStream(RaceStream):
     """
     `_process` updated to not call `on_process`
     which reduces some function calls.
-    # todo snap optimisation?
     """
 
     def _process(self, race_updates: list, publish_time: int) -> bool:
+        active = False
         for update in race_updates:
             market_id = update["mid"]
             race_cache = self._caches.get(market_id)
@@ -105,7 +116,19 @@ class FlumineRaceStream(RaceStream):
                 )
             race_cache.update_cache(update, publish_time)
             self._updates_processed += 1
-        return False
+
+            # filter after start time
+            if not hasattr(race_cache, "start_time"):
+                race_cache.start_time = create_time(
+                    race_cache.publish_time, race_cache.race_id
+                )
+            diff = (
+                race_cache.start_time
+                - datetime.datetime.utcfromtimestamp(race_cache.publish_time / 1e3)
+            ).total_seconds()
+            if diff <= 0:
+                active = True
+        return active
 
 
 class HistoricListener(StreamListener):
@@ -138,7 +161,7 @@ class HistoricListener(StreamListener):
 
         # skip on_change / on_update as we know it is always an update
         publish_time = data["pt"]
-        self.stream._process(data[self.stream._lookup], publish_time)
+        return self.stream._process(data[self.stream._lookup], publish_time)
 
 
 class FlumineHistoricalGeneratorStream(HistoricalGeneratorStream):
@@ -150,10 +173,8 @@ class FlumineHistoricalGeneratorStream(HistoricalGeneratorStream):
         stream_snap = self.listener.stream.snap
         with open(self.file_path, "r") as f:
             for update in f:
-                listener_on_data(update)
-                data = stream_snap()
-                if data:  # can return empty list
-                    yield data
+                if listener_on_data(update):
+                    yield stream_snap()
 
 
 class HistoricalStream(BaseStream):
diff --git a/flumine/utils.py b/flumine/utils.py
index 46b2900..69c98e9 100644
--- a/flumine/utils.py
+++ b/flumine/utils.py
@@ -2,6 +2,7 @@ import uuid
 import json
 import logging
 import hashlib
+import datetime
 from typing import Optional, Tuple, Callable
 from decimal import Decimal, ROUND_HALF_UP
 from betfairlightweight.resources.bettingresources import MarketBook, RunnerBook
@@ -284,3 +285,10 @@ def get_event_ids(markets: list, event_type_id: str) -> list:
         if not market.closed and market.event_type_id == event_type_id:
             event_ids.append(market.event_id)
     return list(set(event_ids))
+
+
+def create_time(publish_time: int, id_: str) -> datetime.datetime:
+    pt_datetime = datetime.datetime.utcfromtimestamp(publish_time / 1e3)
+    event_id, start_time = id_.split(".")
+    hour, minute = int(start_time[:2]), int(start_time[2:])
+    return pt_datetime.replace(hour=hour, minute=minute, second=0, microsecond=0)
diff --git a/requirements.txt b/requirements.txt
index a9f662c..89484a9 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.13.2
+betfairlightweight==2.14.0
 tenacity==8.0.1
 python-json-logger==2.0.2
 requests


Test patch:
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index a803088..f77b701 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -234,6 +234,7 @@ class BaseFlumineTest(unittest.TestCase):
             mock_market, mock_market_book
         )
         mock_log_control.assert_called_with(mock_event)
+        mock_market.assert_called_with(mock_market_book)
 
     @mock.patch("flumine.baseflumine.BaseFlumine.info")
     @mock.patch("flumine.baseflumine.BaseFlumine.log_control")
diff --git a/tests/test_blotter.py b/tests/test_blotter.py
index 1a4dc15..9f7e569 100644
--- a/tests/test_blotter.py
+++ b/tests/test_blotter.py
@@ -306,6 +306,8 @@ class BlotterTest(unittest.TestCase):
             size_matched=2.0,
             size_remaining=2.0,
             order_type=LimitOrder(price=6, size=4.0),
+            status=OrderStatus.EXECUTABLE,
+            complete=False,
         )
         self.blotter["12345"] = mock_order
         # On the win side, we have 2.0 * (5.6-1.0) = 9.2
@@ -333,6 +335,8 @@ class BlotterTest(unittest.TestCase):
             size_matched=2.0,
             size_remaining=2.0,
             order_type=LimitOrder(price=6, size=4.0),
+            status=OrderStatus.EXECUTABLE,
+            complete=False,
         )
         self.blotter["12345"] = mock_order
         # On the win side, we have -2.0 * (5.6-1.0) -2.0 * (6.0-1.0) = -19.2
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 77e3795..c33a7a7 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -1,4 +1,5 @@
 import unittest
+import datetime
 from unittest import mock
 from unittest.mock import call
 
@@ -645,9 +646,10 @@ class TestFlumineMarketStream(unittest.TestCase):
 
     @mock.patch("flumine.streams.historicalstream.MarketBookCache")
     def test__process(self, mock_cache):
-        self.assertFalse(
+        update = [{"id": "1.23", "img": {1: 2}, "marketDefinition": {"runners": []}}]
+        self.assertTrue(
             self.stream._process(
-                [{"id": "1.23", "img": {1: 2}, "marketDefinition": {"runners": []}}],
+                update,
                 12345,
             )
         )
@@ -660,67 +662,121 @@ class TestFlumineMarketStream(unittest.TestCase):
             self.stream._calculate_market_tv,
             self.stream._cumulative_runner_tv,
         )
-        mock_cache().update_cache.assert_called_with(
-            {"id": "1.23", "img": {1: 2}, "marketDefinition": {"runners": []}}, 12345
-        )
-
-    def test_snap_inplay(self):
-        # inPlay
-        self.stream = historicalstream.FlumineMarketStream(
-            mock.Mock(inplay=True, seconds_to_start=None), 0
-        )
-        self.stream._caches = {
-            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=False),
-        }
-        self.assertEqual(len(self.stream.snap()), 0)
-        self.stream._caches = {
-            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=True),
-        }
-        self.assertEqual(len(self.stream.snap()), 1)
-        self.stream._caches = {
-            "1.123": mock.Mock(_definition_status="CLOSED", _definition_in_play=False),
-        }
-        self.assertEqual(len(self.stream.snap()), 1)
-
-        self.stream = historicalstream.FlumineMarketStream(
-            mock.Mock(inplay=False, seconds_to_start=None), 0
-        )
-        self.stream._caches = {
-            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=False),
-        }
-        self.assertEqual(len(self.stream.snap()), 1)
-        self.stream._caches = {
-            "1.123": mock.Mock(_definition_status="OPEN", _definition_in_play=True),
-        }
-        self.assertEqual(len(self.stream.snap()), 0)
-
-    def test_snap_seconds_to_start(self):
-        # secondsToStart
-        self.stream = historicalstream.FlumineMarketStream(
-            mock.Mock(inplay=None, seconds_to_start=600), 0
-        )
-        self.stream._caches = {
-            "1.123": mock.Mock(
-                publish_time=123,
-                market_definition={
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=True)
+
+    @mock.patch("flumine.streams.historicalstream.MarketBookCache")
+    def test__process_inplay(self, mock_cache):
+        self.stream._listener.inplay = True
+        update = [
+            {
+                "id": "1.23",
+                "img": {1: 2},
+                "marketDefinition": {"status": "OPEN", "inPlay": False, "runners": []},
+            }
+        ]
+        self.assertFalse(
+            self.stream._process(
+                update,
+                12345,
+            )
+        )
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=False)
+
+        self.stream._listener.inplay = True
+        update = [
+            {
+                "id": "1.23",
+                "img": {1: 2},
+                "marketDefinition": {
+                    "status": "SUSPENDED",
+                    "inPlay": False,
+                    "runners": [],
+                },
+            }
+        ]
+        self.assertTrue(
+            self.stream._process(
+                update,
+                12345,
+            )
+        )
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=True)
+
+        update = [
+            {
+                "id": "1.23",
+                "img": {1: 2},
+                "marketDefinition": {"status": "OPEN", "inPlay": True, "runners": []},
+            }
+        ]
+        self.assertTrue(
+            self.stream._process(
+                update,
+                12345,
+            )
+        )
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=True)
+
+    @mock.patch("flumine.streams.historicalstream.MarketBookCache")
+    def test__process_seconds_to_start(self, mock_cache):
+        self.stream._listener.inplay = None
+        self.stream._listener.seconds_to_start = 600
+        update = [
+            {
+                "id": "1.23",
+                "img": {1: 2},
+                "marketDefinition": {
+                    "status": "OPEN",
                     "marketTime": 456,
+                    "runners": [],
+                },
+            }
+        ]
+        self.assertTrue(
+            self.stream._process(
+                update,
+                12345,
+            )
+        )
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=True)
+
+        update = [
+            {
+                "id": "1.23",
+                "img": {1: 2},
+                "marketDefinition": {
+                    "status": "OPEN",
+                    "marketTime": 1234567,
+                    "runners": [],
                 },
-                _definition_status="OPEN",
-                _definition_in_play=False,
+            }
+        ]
+        self.assertFalse(
+            self.stream._process(
+                update,
+                12345,
             )
-        }
-        self.assertEqual(len(self.stream.snap()), 1)
-        self.stream._caches = {
-            "1.123": mock.Mock(
-                publish_time=123,
-                market_definition={
+        )
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=False)
+
+        update = [
+            {
+                "id": "1.23",
+                "img": {1: 2},
+                "marketDefinition": {
+                    "status": "SUSPENDED",
                     "marketTime": 1234567,
+                    "runners": [],
                 },
-                _definition_status="OPEN",
-                _definition_in_play=False,
+            }
+        ]
+        self.assertTrue(
+            self.stream._process(
+                update,
+                12345,
             )
-        }
-        self.assertEqual(len(self.stream.snap()), 0)
+        )
+        mock_cache().update_cache.assert_called_with(update[0], 12345, active=True)
 
 
 class TestFlumineRaceStream(unittest.TestCase):
@@ -728,22 +784,35 @@ class TestFlumineRaceStream(unittest.TestCase):
         self.listener = mock.Mock()
         self.stream = historicalstream.FlumineRaceStream(self.listener, 0)
 
-    @mock.patch("flumine.streams.historicalstream.RaceCache")
-    def test__process(self, mock_cache):
-        self.assertFalse(
+    @mock.patch(
+        "flumine.streams.historicalstream.create_time",
+        return_value=datetime.datetime(1970, 1, 1, 13, 10),
+    )
+    def test__process(self, mock_create_time):
+        self.assertTrue(
             self.stream._process(
-                [{"mid": "1.23", "id": 1, "img": {1: 2}}],
-                12345,
+                [{"mid": "1.23", "id": "13.10", "img": {1: 2}}],
+                11111111111111,
             )
         )
         self.assertEqual(len(self.stream._caches), 1)
         self.assertEqual(self.stream._updates_processed, 1)
-        mock_cache.assert_called_with(
-            "1.23", 12345, 1, self.stream._listener.lightweight
-        )
-        mock_cache().update_cache.assert_called_with(
-            {"mid": "1.23", "id": 1, "img": {1: 2}}, 12345
+        mock_create_time.assert_called_with(11111111111111, "13.10")
+
+    @mock.patch(
+        "flumine.streams.historicalstream.create_time",
+        return_value=datetime.datetime(1970, 1, 1, 13, 10),
+    )
+    def test__process_false(self, mock_create_time):
+        self.assertFalse(
+            self.stream._process(
+                [{"mid": "1.23", "id": "13.10", "img": {1: 2}}],
+                1234,
+            )
         )
+        self.assertEqual(len(self.stream._caches), 1)
+        self.assertEqual(self.stream._updates_processed, 1)
+        mock_create_time.assert_called_with(1234, "13.10")
 
 
 class TestHistoricListener(unittest.TestCase):
diff --git a/tests/test_tradingcontrols.py b/tests/test_tradingcontrols.py
index e0919d8..d714bea 100644
--- a/tests/test_tradingcontrols.py
+++ b/tests/test_tradingcontrols.py
@@ -10,6 +10,7 @@ from flumine.controls.tradingcontrols import (
     MarketValidation,
 )
 from flumine.markets.blotter import Blotter
+from flumine.order.order import OrderStatus
 
 
 class TestOrderValidation(unittest.TestCase):
@@ -408,25 +409,34 @@ class TestStrategyExposure(unittest.TestCase):
         strategy.max_order_exposure = 10
         strategy.max_selection_exposure = 10
 
-        order1 = mock.Mock(market_id="market_id", lookup=(1, 2, 3))
+        order1 = mock.Mock(
+            market_id="market_id",
+            lookup=(1, 2, 3),
+            side="BACK",
+            size_remaining=9.0,
+            average_price_matched=0.0,
+            size_matched=0,
+            status=OrderStatus.EXECUTABLE,
+            complete=False,
+        )
         order1.trade.strategy = strategy
         order1.order_type.ORDER_TYPE = OrderTypes.LIMIT
-        order1.side = "BACK"
         order1.order_type.price = 2.0
         order1.order_type.size = 9.0
-        order1.size_remaining = 9.0
-        order1.average_price_matched = 0.0
-        order1.size_matched = 0
 
-        order2 = mock.Mock(lookup=(1, 2, 3))
+        order2 = mock.Mock(
+            lookup=(1, 2, 3),
+            side="BACK",
+            size_remaining=5.0,
+            average_price_matched=0.0,
+            size_matched=0,
+            status=OrderStatus.EXECUTABLE,
+            complete=False,
+        )
         order2.trade.strategy = strategy
         order2.order_type.ORDER_TYPE = OrderTypes.LIMIT
-        order2.side = "BACK"
         order2.order_type.price = 3.0
         order2.order_type.size = 9.0
-        order2.size_remaining = 5.0
-        order2.average_price_matched = 0.0
-        order2.size_matched = 0
 
         self.market.blotter["order2"] = order2
 
@@ -571,17 +581,22 @@ class TestStrategyExposure(unittest.TestCase):
         strategy.max_order_exposure = 10
         strategy.max_selection_exposure = 10
 
-        order1 = mock.Mock(market_id="market_id", lookup=(1, 2, 3))
+        order1 = mock.Mock(
+            market_id="market_id",
+            lookup=(1, 2, 3),
+            selection_id=1234,
+            side="BACK",
+            average_price_matched=0.0,
+            size_matched=0,
+            handicap=0,
+            status=OrderStatus.EXECUTABLE,
+            complete=False,
+        )
         order1.trade.strategy = strategy
         order1.order_type.ORDER_TYPE = OrderTypes.LIMIT
-        order1.side = "BACK"
         order1.order_type.price = 2.0
         order1.order_type.size = 9.0
         order1.size_remaining = 9.0
-        order1.average_price_matched = 0.0
-        order1.size_matched = 0
-        order1.selection_id = 1234
-        order1.handicap = 0
 
         self.market.blotter._strategy_selection_orders = {(strategy, 2, 3): [order1]}
 
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 959068f..9e1fdeb 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,5 +1,6 @@
 import logging
 import unittest
+import datetime
 from unittest import mock
 
 from flumine import utils, FlumineException
@@ -316,3 +317,8 @@ class UtilsTest(unittest.TestCase):
             mock.Mock(event_id=4, event_type_id="7", closed=False),
         ]
         self.assertEqual(utils.get_event_ids(mock_markets, "1"), [1, 2])
+
+    def test_create_time(self):
+        self.assertEqual(
+            utils.create_time(123, "12345.1310"), datetime.datetime(1970, 1, 1, 13, 10)
+        )