Merge pull request #349 from liampauling/release/1.15.0

bflw upgraded to 2.11.0 and cleanup

HISTORY.rst

@@ -3,6 +3,22 @@
 Release History
 ---------------
 
+1.15.0 (2020-12-07)
++++++++++++++++++++
+
+**Improvements**
+
+- Updates for bflw 2.11.0
+- logging improved on orphan orders
+
+**Bug Fixes**
+
+- #347 incorrect adjustment factor (sub 1.01)
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.11.0
+
 1.14.13 (2020-12-05)
 +++++++++++++++++++
 

flumine/__version__.py

@@ -1,6 +1,6 @@
 __title__ = "flumine"
 __description__ = "Betfair trading framework"
 __url__ = "https://github.com/liampauling/flumine"
-__version__ = "1.14.13"
+__version__ = "1.15.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"

flumine/markets/middleware.py

@@ -102,8 +102,8 @@ def _process_runner_removal(
                     ):
                         # todo place market
                         for match in order.simulated.matched:
-                            match[1] = round(
-                                match[1] * (1 - (removal_adjustment_factor / 100)), 2
+                            match[1] = self._calculate_reduction_factor(
+                                match[1], removal_adjustment_factor
                             )
                         _, order.simulated.average_price_matched = wap(
                             order.simulated.matched
@@ -115,6 +115,11 @@ def _process_runner_removal(
                             extra=order.info,
                         )
 
+    @staticmethod
+    def _calculate_reduction_factor(price: float, adjustment_factor: float) -> float:
+        price_adjusted = round(price * (1 - (adjustment_factor / 100)), 2)
+        return max(price_adjusted, 1.01)  # min: 1.01
+
     @staticmethod
     def _process_simulated_orders(market, market_analytics: dict) -> None:
         for order in market.blotter.live_orders:

flumine/order/process.py

@@ -77,12 +77,30 @@ def create_order_from_current(markets: Markets, strategies: Strategies, current_
     # get market
     market = markets.markets.get(current_order.market_id)
     if market is None:
-        # todo log
+        logger.warning(
+            "Market not available to create order {0}".format(order_id),
+            extra={
+                "bet_id": current_order.bet_id,
+                "market_id": current_order.market_id,
+                "customer_strategy_ref": current_order.customer_strategy_ref,
+                "customer_order_ref": current_order.customer_order_ref,
+                "strategy_name_hash": strategy_name_hash,
+            },
+        )
         return
     # get strategy
     strategy = strategies.hashes.get(strategy_name_hash)
     if strategy is None:
-        # todo log
+        logger.warning(
+            "Strategy not available to create order {0}".format(order_id),
+            extra={
+                "bet_id": current_order.bet_id,
+                "market_id": current_order.market_id,
+                "customer_strategy_ref": current_order.customer_strategy_ref,
+                "customer_order_ref": current_order.customer_order_ref,
+                "strategy_name_hash": strategy_name_hash,
+            },
+        )
         return
     # add trade/order
     trade = Trade(

flumine/streams/historicalstream.py

@@ -1,8 +1,8 @@
 import logging
 import datetime
 from betfairlightweight.streaming import StreamListener, HistoricalGeneratorStream
-from betfairlightweight.streaming.stream import MarketStream
-from betfairlightweight.streaming.cache import MarketBookCache
+from betfairlightweight.streaming.stream import MarketStream, RaceStream
+from betfairlightweight.streaming.cache import MarketBookCache, RaceCache
 from betfairlightweight.resources.baseresource import BaseResource
 
 from .basestream import BaseStream
@@ -75,6 +75,29 @@ def snap(self, market_ids: list = None) -> list:
         return market_books
 
 
+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:
+        for update in race_updates:
+            market_id = update["mid"]
+            race_cache = self._caches.get(market_id)
+            if race_cache is None:
+                race_cache = RaceCache(publish_time=publish_time, **update)
+                self._caches[market_id] = race_cache
+                logger.info(
+                    "[%s: %s]: %s added, %s markets in cache"
+                    % (self, self.unique_id, market_id, len(self._caches))
+                )
+            race_cache.update_cache(update, publish_time)
+            self._updates_processed += 1
+        return False
+
+
 class HistoricListener(StreamListener):
     """
     Custom listener to restrict processing by
@@ -89,6 +112,8 @@ def __init__(self, inplay: bool = None, seconds_to_start: float = None, **kwargs
     def _add_stream(self, unique_id, stream_type):
         if stream_type == "marketSubscription":
             return FlumineMarketStream(self)
+        elif stream_type == "raceSubscription":
+            return FlumineRaceStream(self)
 
 
 class HistoricalStream(BaseStream):

requirements.txt

@@ -1,4 +1,4 @@
-betfairlightweight==2.10.2
+betfairlightweight==2.11.0
 tenacity==5.0.3
 python-json-logger==2.0.1
 requests

tests/test_middleware.py

@@ -135,6 +135,14 @@ def test__process_runner_removal_none(self):
         self.middleware._process_runner_removal(mock_market, 12345, 0, None)
         self.assertEqual(mock_order.simulated.matched, [[123, 8.6, 10]])
 
+    def test__calculate_reduction_factor(self):
+        self.assertEqual(self.middleware._calculate_reduction_factor(10, 10), 9)
+        self.assertEqual(self.middleware._calculate_reduction_factor(1000, 0), 1000)
+        self.assertEqual(self.middleware._calculate_reduction_factor(1000, 5), 950)
+        self.assertEqual(self.middleware._calculate_reduction_factor(3.2, 75.18), 1.01)
+        self.assertEqual(self.middleware._calculate_reduction_factor(10, 75.18), 2.48)
+        self.assertEqual(self.middleware._calculate_reduction_factor(1.01, 75.18), 1.01)
+
     def test__process_simulated_orders(self):
         mock_market_book = mock.Mock()
         mock_order = mock.Mock()

tests/test_streams.py

@@ -533,6 +533,27 @@ def test_snap_seconds_to_start(self):
         self.assertEqual(len(self.stream.snap()), 0)
 
 
+class TestFlumineRaceStream(unittest.TestCase):
+    def setUp(self) -> None:
+        self.listener = mock.Mock()
+        self.stream = historicalstream.FlumineRaceStream(self.listener)
+
+    @mock.patch("flumine.streams.historicalstream.RaceCache")
+    def test__process(self, mock_cache):
+        self.assertFalse(
+            self.stream._process(
+                [{"mid": "1.23", "img": {1: 2}}],
+                12345,
+            )
+        )
+        self.assertEqual(len(self.stream._caches), 1)
+        self.assertEqual(self.stream._updates_processed, 1)
+        mock_cache.assert_called_with(publish_time=12345, mid="1.23", img={1: 2})
+        mock_cache().update_cache.assert_called_with(
+            {"mid": "1.23", "img": {1: 2}}, 12345
+        )
+
+
 class TestHistoricListener(unittest.TestCase):
     def setUp(self) -> None:
         self.mock_flumine = mock.Mock()
@@ -545,11 +566,17 @@ def test_init(self):
         self.assertEqual(self.listener.seconds_to_start, 123)
 
     @mock.patch("flumine.streams.historicalstream.FlumineMarketStream")
-    def test__add_stream(self, mock_stream):
+    def test__add_stream_market(self, mock_stream):
         self.assertEqual(
             self.listener._add_stream(123, "marketSubscription"), mock_stream()
         )
 
+    @mock.patch("flumine.streams.historicalstream.FlumineRaceStream")
+    def test__add_stream_race(self, mock_stream):
+        self.assertEqual(
+            self.listener._add_stream(123, "raceSubscription"), mock_stream()
+        )
+
 
 class TestOrderStream(unittest.TestCase):
     def setUp(self) -> None: