Repository: liampauling/flumine
Pull request #565 (2022-02-24 13:49:51)

initial wire frame for sports_data processing
#564 


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 80ac61e..143c673 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,14 @@
 Release History
 ---------------
 
+1.22.0 (2022-02-28)
++++++++++++++++++++
+
+**Improvements**
+
+- #564 sports data functionality added
+- blotter lookups updated to lists only (*breaking change)
+
 1.21.6 (2022-02-18)
 +++++++++++++++++++
 
diff --git a/docs/sportsdata.md b/docs/sportsdata.md
new file mode 100644
index 0000000..6456bdf
--- /dev/null
+++ b/docs/sportsdata.md
@@ -0,0 +1,89 @@
+# Sports Data
+
+Flumine is able to connect to the sports-data-stream provided by Betfair for live data on cricket and races.
+
+!!! tip
+    Your appKey must be authorised to access the sports-data stream, contact bdp@betfair.com
+
+## Cricket Subscription
+
+A cricket subscription can be added via the `sports_data_filter` on a strategy
+
+```python 
+strategy = ExampleStrategy(
+    market_filter=streaming_market_filter(
+        event_type_ids=["4"], market_types=["MATCH_ODDS"]
+    ),
+    sports_data_filter=["cricketSubscription"],
+)
+```
+
+## Race Subscription
+
+A race subscription can be added via the `sports_data_filter` on a strategy
+
+```python 
+strategy = ExampleStrategy(
+    market_filter=streaming_market_filter(
+        event_type_ids=["7"], market_types=["WIN"]
+    ),
+    sports_data_filter=["raceSubscription"],
+)
+```
+
+## Strategy
+
+Any sports data stream updates will be available in the strategy via the `process_sports_data` function
+
+```python
+class ExampleStrategy(BaseStrategy):
+    def process_sports_data(
+        self, market: Market, sports_data: Union[Race, CricketMatch]
+    ) -> None:
+        # called on each update from sports-data-stream
+        print(market, sports_data)
+```
+
+
+## Data Recorder
+
+The example `marketrecorder.py` can be modified to record race and cricket data by updating the `process_raw_data` with the matching `op` and data keys.
+
+- marketSubscription `mcm` and `mc`
+- orderSubscription `ocm` and `oc`
+- cricketSubscription `ccm` and `cc`
+- raceSubscription `rcm` and `rc`
+
+And using the correct stream class:
+
+#### Cricket Recorder
+```python
+from flumine.streams.datastream import CricketDataStream
+
+strategy= MarketRecorder(
+    market_filter=None,
+    stream_class=CricketDataStream,
+    context={
+        "local_dir": "/tmp",
+        "force_update": False,
+        "remove_file": True,
+        "remove_gz_file": False,
+    },
+)
+```
+
+#### Race Recorder
+```python
+from flumine.streams.datastream import RaceDataStream
+
+strategy= MarketRecorder(
+    market_filter=None,
+    stream_class=RaceDataStream,
+    context={
+        "local_dir": "/tmp",
+        "force_update": False,
+        "remove_file": True,
+        "remove_gz_file": False,
+    },
+)
+```
diff --git a/examples/example-sportsdata.py b/examples/example-sportsdata.py
new file mode 100644
index 0000000..d9a482a
--- /dev/null
+++ b/examples/example-sportsdata.py
@@ -0,0 +1,44 @@
+import time
+import logging
+import betfairlightweight
+from betfairlightweight.filters import streaming_market_filter
+from pythonjsonlogger import jsonlogger
+
+from flumine import Flumine, clients, BaseStrategy
+from flumine.order.trade import Trade
+from flumine.order.ordertype import LimitOrder
+from flumine.order.order import OrderStatus
+
+logger = logging.getLogger()
+
+custom_format = "%(asctime) %(levelname) %(message)"
+log_handler = logging.StreamHandler()
+formatter = jsonlogger.JsonFormatter(custom_format)
+formatter.converter = time.gmtime
+log_handler.setFormatter(formatter)
+logger.addHandler(log_handler)
+logger.setLevel(logging.INFO)
+
+
+class ExampleStrategy(BaseStrategy):
+    def process_sports_data(self, market, sports_data) -> None:
+        # called on each update from sports-data-stream
+        print(market, sports_data)
+
+
+trading = betfairlightweight.APIClient("username")
+client = clients.BetfairClient(trading)
+
+framework = Flumine(client)
+
+strategy = ExampleStrategy(
+    market_filter=streaming_market_filter(
+        event_type_ids=["4"], market_types=["MATCH_ODDS"]
+    ),
+    sports_data_filter=[
+        "cricketSubscription"
+    ],  # "cricketSubscription" and/or "raceSubscription"
+)
+framework.add_strategy(strategy)
+
+framework.run()
diff --git a/flumine/__version__.py b/flumine/__version__.py
index b745569..f7eea2a 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.21.6"
+__version__ = "1.22.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 20be37d..bb0046c 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -160,6 +160,23 @@ class BaseFlumine:
                         strategy.process_market_book, market, market_book
                     )
 
+    def _process_sports_data(self, event: events.SportsDataEvent) -> None:
+        for sports_data in event.event:
+            # get marketId
+            market_id = sports_data.market_id
+            # get market
+            market = self.markets.markets.get(market_id)
+            if market is None:
+                logger.error(
+                    "Market not present for sports data", extra={"market_id": market_id}
+                )
+                continue
+            for strategy in self.strategies:
+                if sports_data.streaming_unique_id in strategy.stream_ids:
+                    utils.call_strategy_error_handling(
+                        strategy.process_sports_data, market, sports_data
+                    )
+
     def process_order_package(self, order_package) -> None:
         """Execute through client."""
         order_package.client.execution.handler(order_package)
diff --git a/flumine/events/events.py b/flumine/events/events.py
index 69a37c6..e638628 100644
--- a/flumine/events/events.py
+++ b/flumine/events/events.py
@@ -8,6 +8,7 @@ class EventType(Enum):
     # betfair objects
     MARKET_CATALOGUE = "MarketCatalogue"
     MARKET_BOOK = "MarketBook"
+    SPORTS_DATA = "SportsData"
     RAW_DATA = "Raw streaming data"
     CURRENT_ORDERS = "CurrentOrders"
     CLEARED_MARKETS = "ClearedMarkets"
@@ -60,6 +61,11 @@ class MarketBookEvent(BaseEvent):
     QUEUE_TYPE = QueueType.HANDLER
 
 
+class SportsDataEvent(BaseEvent):
+    EVENT_TYPE = EventType.SPORTS_DATA
+    QUEUE_TYPE = QueueType.HANDLER
+
+
 class RawDataEvent(BaseEvent):
     EVENT_TYPE = EventType.RAW_DATA
     QUEUE_TYPE = QueueType.HANDLER
diff --git a/flumine/flumine.py b/flumine/flumine.py
index 17aff7c..6a885f0 100644
--- a/flumine/flumine.py
+++ b/flumine/flumine.py
@@ -24,6 +24,9 @@ class Flumine(BaseFlumine):
                 elif event.EVENT_TYPE == EventType.MARKET_BOOK:
                     self._process_market_books(event)
 
+                elif event.EVENT_TYPE == EventType.SPORTS_DATA:
+                    self._process_sports_data(event)
+
                 elif event.EVENT_TYPE == EventType.RAW_DATA:
                     self._process_raw_data(event)
 
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index eb78816..0524228 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -1,5 +1,4 @@
 import logging
-import warnings
 from typing import Iterable, Optional, List
 from collections import defaultdict
 
@@ -63,14 +62,7 @@ class Blotter:
         """Returns all orders related to a strategy."""
         orders = self._strategy_orders[strategy]
         if order_status:
-            if isinstance(order_status, OrderStatus):
-                warnings.warn(
-                    "`order_status` to accept List[OrderStatus] only from v1.22.0",
-                    DeprecationWarning,
-                )
-                orders = [o for o in orders if o.status == order_status]
-            else:
-                orders = [o for o in orders if o.status in order_status]
+            orders = [o for o in orders if o.status in order_status]
         if matched_only:
             orders = [o for o in orders if o.size_matched > 0]
         return orders
@@ -86,14 +78,7 @@ class Blotter:
         """Returns all orders related to a strategy selection."""
         orders = self._strategy_selection_orders[(strategy, selection_id, handicap)]
         if order_status:
-            if isinstance(order_status, OrderStatus):
-                warnings.warn(
-                    "`order_status` to accept List[OrderStatus] only from v1.22.0",
-                    DeprecationWarning,
-                )
-                orders = [o for o in orders if o.status == order_status]
-            else:
-                orders = [o for o in orders if o.status in order_status]
+            orders = [o for o in orders if o.status in order_status]
         if matched_only:
             orders = [o for o in orders if o.size_matched > 0]
         return orders
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index db3ca70..a705632 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -1,7 +1,7 @@
 import logging
-from typing import Type, Iterator, Union
+from typing import Type, Iterator, Union, List
 from betfairlightweight import filters
-from betfairlightweight.resources import MarketBook
+from betfairlightweight.resources import MarketBook, Race, CricketMatch
 
 from .runnercontext import RunnerContext
 from ..clients import BaseClient
@@ -40,6 +40,9 @@ class BaseStrategy:
         self,
         market_filter: Union[dict, list],
         market_data_filter: dict = None,
+        sports_data_filter: List[
+            str
+        ] = None,  # 'raceSubscription', 'cricketSubscription'
         streaming_timeout: float = None,
         conflate_ms: int = None,
         stream_class: Type[BaseStream] = MarketStream,
@@ -55,6 +58,7 @@ class BaseStrategy:
         """
         :param market_filter: Streaming market filter dict or list of market filters
         :param market_data_filter: Streaming market data filter
+        :param sports_data_filter: Streaming sports data filter (e.g. ["raceSubscription"])
         :param streaming_timeout: Streaming timeout in seconds, will call snap() on cache
         :param conflate_ms: Streaming conflation
         :param stream_class: Can be Market or Data (raw)
@@ -69,6 +73,7 @@ class BaseStrategy:
         """
         self.market_filter = market_filter
         self.market_data_filter = market_data_filter or DEFAULT_MARKET_DATA_FILTER
+        self.sports_data_filter = sports_data_filter or []
         self.streaming_timeout = streaming_timeout
         self.conflate_ms = conflate_ms
         self.stream_class = stream_class
@@ -112,6 +117,12 @@ class BaseStrategy:
         # process marketBook; place/cancel/replace orders
         return
 
+    def process_sports_data(
+        self, market: Market, sports_data: Union[Race, CricketMatch]
+    ) -> None:
+        # process sports data
+        return
+
     def process_raw_data(self, clk: str, publish_time: int, datum: dict) -> None:
         return
 
diff --git a/flumine/streams/basestream.py b/flumine/streams/basestream.py
index 625372b..e317457 100644
--- a/flumine/streams/basestream.py
+++ b/flumine/streams/basestream.py
@@ -22,6 +22,7 @@ class BaseStream(threading.Thread):
         conflate_ms: int = None,
         market_filter: dict = None,
         market_data_filter: dict = None,
+        sports_data_filter: str = None,
         custom: bool = False,
         client=None,
         output_queue: bool = True,
@@ -35,6 +36,7 @@ class BaseStream(threading.Thread):
         self.stream_id = stream_id
         self.market_filter = market_filter
         self.market_data_filter = market_data_filter
+        self.sports_data_filter = sports_data_filter
         self.streaming_timeout = streaming_timeout
         self.conflate_ms = conflate_ms
         self._client = client
diff --git a/flumine/streams/datastream.py b/flumine/streams/datastream.py
index f1cc0e1..39abb3e 100644
--- a/flumine/streams/datastream.py
+++ b/flumine/streams/datastream.py
@@ -26,6 +26,8 @@ class FlumineListener(StreamListener):
             return FlumineOrderStream(self, unique_id)
         elif operation == "raceSubscription":
             return FlumineRaceStream(self, unique_id)
+        elif operation == "cricketSubscription":
+            return FlumineCricketStream(self, unique_id)
 
 
 class FlumineStream(BFBaseStream):
@@ -111,6 +113,26 @@ class FlumineRaceStream(FlumineStream):
         return False
 
 
+class FlumineCricketStream(FlumineStream):
+
+    _lookup = "cc"
+
+    def _process(self, data: list, publish_time: int) -> bool:
+        for update in data:
+            market_id = update["marketId"]
+            if market_id not in self._caches:
+                # adds empty object to cache to track live market count
+                self._caches[market_id] = object()
+                logger.info(
+                    "[CricketStream: %s] %s added, %s markets in cache"
+                    % (self.unique_id, market_id, len(self._caches))
+                )
+            self._updates_processed += 1
+
+        self.on_process([self.unique_id, self._clk, publish_time, data])
+        return False
+
+
 class DataStream(BaseStream):
 
     LISTENER = FlumineListener
@@ -193,3 +215,59 @@ class OrderDataStream(DataStream):
             )
             raise
         logger.info("Stopped OrderDataStream {0}".format(self.stream_id))
+
+
+class RaceDataStream(DataStream):
+    @retry(wait=RETRY_WAIT)
+    def run(self) -> None:
+        logger.info(
+            "Starting RaceDataStream {0}".format(self.stream_id),
+            extra={
+                "stream_id": self.stream_id,
+            },
+        )
+        self._stream = self.betting_client.streaming.create_stream(
+            unique_id=self.stream_id, listener=self._listener, host="sports_data"
+        )
+        try:
+            self.stream_id = self._stream.subscribe_to_races()
+            self._stream.start()
+        except BetfairError:
+            logger.error(
+                "RaceDataStream {0} run error".format(self.stream_id), exc_info=True
+            )
+            raise
+        except Exception:
+            logger.critical(
+                "RaceDataStream {0} run error".format(self.stream_id), exc_info=True
+            )
+            raise
+        logger.info("Stopped RaceDataStream {0}".format(self.stream_id))
+
+
+class CricketDataStream(DataStream):
+    @retry(wait=RETRY_WAIT)
+    def run(self) -> None:
+        logger.info(
+            "Starting CricketDataStream {0}".format(self.stream_id),
+            extra={
+                "stream_id": self.stream_id,
+            },
+        )
+        self._stream = self.betting_client.streaming.create_stream(
+            unique_id=self.stream_id, listener=self._listener, host="sports_data"
+        )
+        try:
+            self.stream_id = self._stream.subscribe_to_cricket_matches()
+            self._stream.start()
+        except BetfairError:
+            logger.error(
+                "CricketDataStream {0} run error".format(self.stream_id), exc_info=True
+            )
+            raise
+        except Exception:
+            logger.critical(
+                "CricketDataStream {0} run error".format(self.stream_id), exc_info=True
+            )
+            raise
+        logger.info("Stopped CricketDataStream {0}".format(self.stream_id))
diff --git a/flumine/streams/sportsdatastream.py b/flumine/streams/sportsdatastream.py
new file mode 100644
index 0000000..cfd0163
--- /dev/null
+++ b/flumine/streams/sportsdatastream.py
@@ -0,0 +1,72 @@
+import time
+import queue
+import logging
+from betfairlightweight import BetfairError
+from tenacity import retry
+
+from .basestream import BaseStream
+from ..events.events import SportsDataEvent
+
+logger = logging.getLogger(__name__)
+
+RETRY_WAIT = BaseStream.RETRY_WAIT
+
+
+class SportsDataStream(BaseStream):
+    @retry(wait=RETRY_WAIT)
+    def run(self) -> None:
+        time.sleep(2)  # 2s delay to wait for market streams to start
+        logger.info(
+            "Starting SportsDataStream {0}".format(self.stream_id),
+            extra={
+                "stream_id": self.stream_id,
+                "sports_data_filter": self.sports_data_filter,
+                "streaming_timeout": self.streaming_timeout,
+            },
+        )
+        if not self._output_thread.is_alive():
+            logger.info(
+                "Starting output_thread (SportsDataStream {0})".format(self.stream_id)
+            )
+            self._output_thread.start()
+
+        self._stream = self.betting_client.streaming.create_stream(
+            unique_id=self.stream_id, host="sports_data", listener=self._listener
+        )
+        try:
+            if self.sports_data_filter == "raceSubscription":
+                self.stream_id = self._stream.subscribe_to_races()
+            elif self.sports_data_filter == "cricketSubscription":
+                self.stream_id = self._stream.subscribe_to_cricket_matches()
+            else:
+                raise ValueError("Unknown sports data filter")
+            self._stream.start()
+        except BetfairError:
+            logger.error(
+                "SportsDataStream {0} run error".format(self.stream_id), exc_info=True
+            )
+            raise
+        except Exception:
+            logger.critical(
+                "SportsDataStream {0} run error".format(self.stream_id), exc_info=True
+            )
+            raise
+        logger.info("Stopped SportsDataStream {0}".format(self.stream_id))
+
+    def handle_output(self) -> None:
+        """Handles output from stream."""
+        while self.is_alive():
+            try:
+                sports_data = self._output_queue.get(
+                    block=True, timeout=self.streaming_timeout
+                )
+            except queue.Empty:
+                sports_data = self._listener.snap(
+                    market_ids=self.flumine.markets.open_market_ids
+                )
+            if sports_data:
+                self.flumine.handler_queue.put(SportsDataEvent(sports_data))
+
+        logger.info(
+            "Stopped output_thread (SportsDataStream {0})".format(self.stream_id)
+        )
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index cf1a345..127054c 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -4,6 +4,7 @@ from typing import Union, Iterator
 
 from ..strategy.strategy import BaseStrategy
 from .marketstream import MarketStream
+from .sportsdatastream import SportsDataStream
 from .datastream import DataStream
 from .historicalstream import HistoricalStream
 from .orderstream import OrderStream
@@ -79,6 +80,7 @@ class Streams:
     """ market data """
 
     def add_stream(self, strategy: BaseStrategy) -> None:
+        # markets
         if isinstance(strategy.market_filter, dict) or strategy.market_filter is None:
             market_filters = [strategy.market_filter]
         else:
@@ -116,6 +118,36 @@ class Streams:
                 )
                 self._streams.append(stream)
                 strategy.streams.append(stream)
+        # sports data
+        for subscription in strategy.sports_data_filter:
+            for stream in self:  # check if sports data stream already exists
+                if (
+                    isinstance(stream, SportsDataStream)
+                    and stream.sports_data_filter == subscription
+                    and stream.streaming_timeout == strategy.streaming_timeout
+                ):
+                    logger.info(
+                        "Using {0} ({1}) for strategy {2}".format(
+                            strategy.stream_class, stream.stream_id, strategy
+                        )
+                    )
+                    strategy.streams.append(stream)
+                    break
+            else:  # nope? lets create a new one
+                stream_id = self._increment_stream_id()
+                logger.info(
+                    "Creating new {0} ({1}) for strategy {2}".format(
+                        strategy.stream_class, stream_id, strategy
+                    )
+                )
+                stream = SportsDataStream(
+                    flumine=self.flumine,
+                    stream_id=stream_id,
+                    sports_data_filter=subscription,
+                    streaming_timeout=strategy.streaming_timeout,
+                )
+                self._streams.append(stream)
+                strategy.streams.append(stream)
 
     def add_historical_stream(
         self,
diff --git a/mkdocs.yml b/mkdocs.yml
index a794f1f..f0afc18 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -27,6 +27,7 @@ nav:
       - Markets: 'markets.md'
       - Controls: 'controls.md'
       - Workers: 'workers.md'
+      - Sports Data: 'sportsdata.md'
       - Advanced Usage: 'advanced.md'
       - Performance: 'performance.md'
       - Architecture: 'architecture.md'


Test patch:
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 0d3ea05..14f5c6c 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -43,7 +43,7 @@ class BaseFlumineTest(unittest.TestCase):
     @mock.patch("flumine.baseflumine.events")
     @mock.patch("flumine.baseflumine.BaseFlumine.log_control")
     def test_add_strategy(self, mock_log_control, mock_events):
-        mock_strategy = mock.Mock(market_filter={})
+        mock_strategy = mock.Mock(market_filter={}, sports_data_filter=[])
         mock_client = mock.Mock()
         self.base_flumine.add_strategy(mock_strategy, mock_client)
         self.assertEqual(len(self.base_flumine.strategies), 1)
@@ -97,6 +97,29 @@ class BaseFlumineTest(unittest.TestCase):
         mock_event.event = [mock_market_book]
         self.base_flumine._process_market_books(mock_event)
 
+    @mock.patch("flumine.baseflumine.utils.call_strategy_error_handling")
+    def test__process_sports_data(self, mock_call_strategy_error_handling):
+        mock_market = mock.Mock()
+        self.base_flumine.markets._markets = {"1.1": mock_market}
+        mock_strategy_one = mock.Mock(stream_ids=[123])
+        mock_strategy_two = mock.Mock(stream_ids=[])
+        self.base_flumine.strategies = [
+            mock_strategy_one,
+            mock_strategy_two,
+        ]
+        mock_sports_data = mock.Mock(streaming_unique_id=123, market_id="1.1")
+        mock_event = mock.Mock(event=[mock_sports_data])
+        self.base_flumine._process_sports_data(mock_event)
+        mock_call_strategy_error_handling.assert_has_calls(
+            [
+                mock.call(
+                    mock_strategy_one.process_sports_data,
+                    mock_market,
+                    mock_sports_data,
+                )
+            ]
+        )
+
     def test_process_order_package(self):
         mock_order_package = mock.Mock()
         self.base_flumine.process_order_package(mock_order_package)
diff --git a/tests/test_flumine.py b/tests/test_flumine.py
index 8656147..5db6fd4 100644
--- a/tests/test_flumine.py
+++ b/tests/test_flumine.py
@@ -19,10 +19,12 @@ class FlumineTest(unittest.TestCase):
     @mock.patch("flumine.flumine.Flumine._process_end_flumine")
     @mock.patch("flumine.flumine.Flumine._process_market_catalogues")
     @mock.patch("flumine.flumine.Flumine._process_raw_data")
+    @mock.patch("flumine.flumine.Flumine._process_sports_data")
     @mock.patch("flumine.flumine.Flumine._process_market_books")
     def test_run(
         self,
         mock__process_market_books,
+        mock__process_sports_data,
         mock__process_raw_data,
         mock__process_market_catalogues,
         mock__process_end_flumine,
@@ -36,6 +38,7 @@ class FlumineTest(unittest.TestCase):
         mock_events = [
             events.MarketCatalogueEvent(None),
             events.MarketBookEvent(None),
+            events.SportsDataEvent(None),
             events.RawDataEvent(None),
             events.CurrentOrdersEvent(None),
             events.ClearedMarketsEvent(None),
@@ -49,14 +52,15 @@ class FlumineTest(unittest.TestCase):
         self.flumine.run()
 
         mock__process_market_books.assert_called_with(mock_events[1])
-        mock__process_raw_data.assert_called_with(mock_events[2])
+        mock__process_sports_data.assert_called_with(mock_events[2])
+        mock__process_raw_data.assert_called_with(mock_events[3])
         mock__process_market_catalogues.assert_called_with(mock_events[0])
         mock__process_end_flumine.assert_called_with()
-        mock__process_current_orders.assert_called_with(mock_events[3])
-        mock__process_close_market.assert_called_with(mock_events[6])
-        mock__process_cleared_markets.assert_called_with(mock_events[4])
-        mock__process_cleared_orders.assert_called_with(mock_events[5])
-        mock__process_custom_event.assert_called_with(mock_events[7])
+        mock__process_current_orders.assert_called_with(mock_events[4])
+        mock__process_close_market.assert_called_with(mock_events[7])
+        mock__process_cleared_markets.assert_called_with(mock_events[5])
+        mock__process_cleared_orders.assert_called_with(mock_events[6])
+        mock__process_custom_event.assert_called_with(mock_events[8])
         mock__add_default_workers.assert_called()
 
     @mock.patch("flumine.worker.BackgroundWorker")
diff --git a/tests/test_strategy.py b/tests/test_strategy.py
index 4c5b7eb..7f45435 100644
--- a/tests/test_strategy.py
+++ b/tests/test_strategy.py
@@ -50,6 +50,7 @@ class BaseStrategyTest(unittest.TestCase):
         self.strategy = strategy.BaseStrategy(
             market_filter=self.mock_market_filter,
             market_data_filter=self.mock_market_data_filter,
+            sports_data_filter=["cricketSubscription"],
             streaming_timeout=self.streaming_timeout,
             conflate_ms=self.conflate_ms,
             stream_class=strategy.MarketStream,
@@ -66,6 +67,7 @@ class BaseStrategyTest(unittest.TestCase):
     def test_init(self):
         self.assertEqual(self.strategy.market_filter, self.mock_market_filter)
         self.assertEqual(self.strategy.market_data_filter, self.mock_market_data_filter)
+        self.assertEqual(self.strategy.sports_data_filter, ["cricketSubscription"])
         self.assertEqual(self.strategy.streaming_timeout, self.streaming_timeout)
         self.assertEqual(self.strategy.conflate_ms, self.conflate_ms)
         self.assertEqual(self.strategy.stream_class, strategy.MarketStream)
@@ -149,6 +151,9 @@ class BaseStrategyTest(unittest.TestCase):
     def test_process_market_book(self):
         self.strategy.process_market_book(None, None)
 
+    def test_process_sports_data(self):
+        self.strategy.process_sports_data(None, None)
+
     def test_process_raw_data(self):
         self.strategy.process_raw_data(None, None, None)
 
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 418d387..0aad2df 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -177,7 +177,7 @@ class StreamsTest(unittest.TestCase):
 
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_new(self, mock_increment):
-        mock_strategy = mock.Mock(market_filter={})
+        mock_strategy = mock.Mock(market_filter={}, sports_data_filter=[])
         mock_stream_class = mock.Mock()
         mock_strategy.stream_class = mock_stream_class
 
@@ -195,7 +195,7 @@ class StreamsTest(unittest.TestCase):
 
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_none(self, mock_increment):
-        mock_strategy = mock.Mock(market_filter=None)
+        mock_strategy = mock.Mock(market_filter=None, sports_data_filter=[])
         mock_stream_class = mock.Mock()
         mock_strategy.stream_class = mock_stream_class
 
@@ -214,7 +214,9 @@ class StreamsTest(unittest.TestCase):
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_multi(self, mock_increment):
         mock_strategy = mock.Mock(
-            market_filter=[{1: 2}, {3: 4}], stream_class=streams.MarketStream
+            market_filter=[{1: 2}, {3: 4}],
+            stream_class=streams.MarketStream,
+            sports_data_filter=[],
         )
         self.streams.add_stream(mock_strategy)
         self.assertEqual(len(self.streams), 2)
@@ -222,18 +224,65 @@ class StreamsTest(unittest.TestCase):
 
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
     def test_add_stream_old(self, mock_increment):
-        mock_strategy = mock.Mock(market_filter={})
-        mock_strategy.market_data_filter = 2
-        mock_strategy.streaming_timeout = 3
-        mock_strategy.conflate_ms = 4
-        mock_strategy.stream_class = streams.MarketStream
+        mock_strategy = mock.Mock(
+            market_data_filter=2,
+            stream_class=streams.MarketStream,
+            streaming_timeout=3,
+            conflate_ms=4,
+            market_filter={},
+            sports_data_filter=[],
+        )
+        stream = mock.Mock(
+            spec=streams.MarketStream,
+            market_filter={},
+            market_data_filter=2,
+            streaming_timeout=3,
+            conflate_ms=4,
+            stream_id=1001,
+        )
+        self.streams._streams = [stream]
+
+        self.streams.add_stream(mock_strategy)
+        self.assertEqual(len(self.streams), 1)
+        mock_increment.assert_not_called()
+
+    @mock.patch(
+        "flumine.streams.streams.SportsDataStream", autospec=streams.SportsDataStream
+    )
+    @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
+    def test_add_stream_sports_data(self, mock_increment, sports_data_stream_class):
+        mock_strategy = mock.Mock(
+            market_filter=[],
+            sports_data_filter=["raceSubscription"],
+        )
+        self.streams.add_stream(mock_strategy)
+        self.assertEqual(len(self.streams), 1)
+        mock_increment.assert_called_with()
+        sports_data_stream_class.assert_has_calls(
+            [
+                call(
+                    flumine=self.mock_flumine,
+                    stream_id=mock_increment(),
+                    sports_data_filter="raceSubscription",
+                    streaming_timeout=mock_strategy.streaming_timeout,
+                )
+            ]
+        )
 
-        stream = mock.Mock(spec=streams.MarketStream)
-        stream.market_filter = {}
-        stream.market_data_filter = 2
-        stream.streaming_timeout = 3
-        stream.conflate_ms = 4
-        stream.stream_id = 1001
+    @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
+    def test_add_stream_sports_data_old(self, mock_increment):
+        mock_strategy = mock.Mock(
+            market_filter=[],
+            sports_data_filter=["cricketSubscription"],
+            streaming_timeout=3,
+        )
+        stream = mock.Mock(
+            stream_id=1,
+            spec=streams.SportsDataStream,
+            market_filter={},
+            sports_data_filter="cricketSubscription",
+            streaming_timeout=3,
+        )
         self.streams._streams = [stream]
 
         self.streams.add_stream(mock_strategy)
@@ -428,6 +477,7 @@ class TestBaseStream(unittest.TestCase):
             100,
             {"test": "me"},
             {"please": "now"},
+            "yes",
             client=self.mock_client,
             output_queue=False,
             event_processing=True,
@@ -441,6 +491,7 @@ class TestBaseStream(unittest.TestCase):
         self.assertEqual(self.stream.stream_id, 123)
         self.assertEqual(self.stream.market_filter, {"test": "me"})
         self.assertEqual(self.stream.market_data_filter, {"please": "now"})
+        self.assertEqual(self.stream.sports_data_filter, "yes")
         self.assertEqual(self.stream.streaming_timeout, 0.01)
         self.assertEqual(self.stream.conflate_ms, 100)
         self.assertFalse(self.stream.custom)
@@ -531,11 +582,16 @@ class TestDataStream(unittest.TestCase):
     # def test_handle_output(self):
     #     pass
 
+    @mock.patch("flumine.streams.datastream.FlumineCricketStream")
     @mock.patch("flumine.streams.datastream.FlumineRaceStream")
     @mock.patch("flumine.streams.datastream.FlumineOrderStream")
     @mock.patch("flumine.streams.datastream.FlumineMarketStream")
     def test_flumine_listener(
-        self, mock_market_stream, mock_order_stream, mock_race_stream
+        self,
+        mock_market_stream,
+        mock_order_stream,
+        mock_race_stream,
+        mock_cricket_stream,
     ):
         listener = datastream.FlumineListener()
         self.assertEqual(
@@ -547,6 +603,9 @@ class TestDataStream(unittest.TestCase):
         self.assertEqual(
             listener._add_stream(123, "raceSubscription"), mock_race_stream()
         )
+        self.assertEqual(
+            listener._add_stream(123, "cricketSubscription"), mock_cricket_stream()
+        )
 
     def test_flumine_stream(self):
         mock_listener = mock.Mock()
@@ -620,6 +679,25 @@ class TestDataStream(unittest.TestCase):
             [mock_listener.stream_unique_id, "AAA", 123, race_updates]
         )
 
+    @mock.patch("flumine.streams.datastream.FlumineCricketStream.on_process")
+    def test_flumine_cricket_stream(self, mock_on_process):
+        mock_listener = mock.Mock(stream_unique_id=0)
+        stream = datastream.FlumineCricketStream(mock_listener, 0)
+        stream._clk = "AAA"
+        race_updates = [
+            {"marketId": "1.123"},
+            {"marketId": "1.456"},
+            {"marketId": "1.123"},
+        ]
+        stream._process(race_updates, 123)
+
+        self.assertEqual(stream._lookup, "cc")
+        self.assertEqual(len(stream._caches), 2)
+        self.assertEqual(stream._updates_processed, 3)
+        mock_on_process.assert_called_with(
+            [mock_listener.stream_unique_id, "AAA", 123, race_updates]
+        )
+
 
 class TestHistoricalStream(unittest.TestCase):
     def setUp(self) -> None:
@@ -939,3 +1017,26 @@ class TestSimulatedOrderStream(unittest.TestCase):
         mock_market.blotter = [order_one, order_two, order_three]
         self.stream.flumine.markets = [mock_market, mock.Mock(closed=True)]
         self.assertEqual(self.stream._get_current_orders(), [order_one])
+
+
+class TestSportsDataStream(unittest.TestCase):
+    def setUp(self) -> None:
+        self.mock_flumine = mock.Mock()
+        self.stream = streams.SportsDataStream(
+            self.mock_flumine, 123, sports_data_filter="test", streaming_timeout=10
+        )
+
+    def test_init(self):
+        self.assertEqual(self.stream.flumine, self.mock_flumine)
+        self.assertEqual(self.stream.stream_id, 123)
+        self.assertEqual(self.stream.sports_data_filter, "test")
+        self.assertEqual(self.stream.streaming_timeout, 10)
+        self.assertIsNone(self.stream._stream)
+        self.assertEqual(orderstream.START_DELAY, 2)
+        self.assertEqual(orderstream.SNAP_DELTA, 5)
+
+    # def test_run(self):
+    #     pass
+    #
+    # def test_handle_output(self):
+    #     pass