Repository: liampauling/flumine
Pull request #517 (2021-10-26 07:23:26)

Extend selection exposures to whole market
Builds on the existing selection exposure calculations (which does the heavy lifting) to derive the worst case outcome (exposure) for the whole market.

The underlying approach is to sum all potential losses and then adjust to reflect the worst case for the given number of winners in the market.

Limitations: this will not work for multi-line markets (e,g, each way, or most asian handicaps) or markets with a variable number of winners such as correct score for which Betfair treats each line as a separate market.


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index a982757..728993f 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,28 +3,6 @@
 Release History
 ---------------
 
-1.20.9 (2021-11-11)
-+++++++++++++++++++
-
-**Improvements**
-
-- #522 add clk to output in market recorder
-
-1.20.8 (2021-11-01)
-+++++++++++++++++++
-
-**Improvements**
-
-- Handle list of market filters in strategy
-
-**Bug Fixes**
-
-- #519 reset real datetime added to allow s3 download (RequestTimeTooSkewed)
-
-**Libraries**
-
-- black upgraded to 21.10b0
-
 1.20.7 (2021-10-25)
 +++++++++++++++++++
 
diff --git a/docs/markets.md b/docs/markets.md
index a33cd06..3f18ae4 100644
--- a/docs/markets.md
+++ b/docs/markets.md
@@ -59,6 +59,7 @@ The blotter is a simple and fast class to hold all orders for a particular marke
 - `strategy_orders(strategy)` Returns all orders related to a strategy
 - `strategy_selection_orders(strategy, selection_id, handicap)` Returns all orders related to a strategy selection
 - `selection_exposure(strategy, lookup)` Returns strategy/selection exposure
+- `market_exposure(strategy, num_runners, num_winners)` Returns strategy/market exposure (default num_winners=1)
 
 ### Properties
 
diff --git a/docs/strategies.md b/docs/strategies.md
index 044cb65..1ab0ed1 100644
--- a/docs/strategies.md
+++ b/docs/strategies.md
@@ -6,7 +6,7 @@ The base strategy class should be used for all strategies and contains the follo
 
 ### Parameters
 
-- `market_filter` Streaming market filter or list of filters required
+- `market_filter` Streaming market filter required
 - `market_data_filter` Streaming market data filter required
 - `streaming_timeout` Streaming timeout, will call snap() on cache every x seconds
 - `conflate_ms` Streaming conflate
diff --git a/examples/strategies/marketrecorder.py b/examples/strategies/marketrecorder.py
index 80df491..0c08fdf 100644
--- a/examples/strategies/marketrecorder.py
+++ b/examples/strategies/marketrecorder.py
@@ -61,12 +61,12 @@ class MarketRecorder(BaseStrategy):
             daemon=True,
         ).start()
 
-    def process_raw_data(self, clk: str, publish_time: int, data: dict):
+    def process_raw_data(self, publish_time, data):
         market_id = data.get(self.MARKET_ID_LOOKUP)
         file_directory = os.path.join(self.local_dir, self.recorder_id, market_id)
         with open(file_directory, "a") as f:
             f.write(
-                json.dumps({"op": "mcm", "clk": clk, "pt": publish_time, "mc": [data]})
+                json.dumps({"op": "mcm", "clk": None, "pt": publish_time, "mc": [data]})
                 + "\n"
             )
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index cb87f3b..441f4ed 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.20.9"
+__version__ = "1.20.7"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index fc6d3aa..e4227b8 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -181,7 +181,7 @@ class BaseFlumine:
         self.markets.remove_market(market.market_id)
 
     def _process_raw_data(self, event: events.RawDataEvent) -> None:
-        stream_id, clk, publish_time, data = event.event
+        stream_id, publish_time, data = event.event
         for datum in data:
             if "id" in datum:
                 market_id = datum["id"]
@@ -200,7 +200,7 @@ class BaseFlumine:
 
             for strategy in self.strategies:
                 if stream_id in strategy.stream_ids:
-                    strategy.process_raw_data(clk, publish_time, datum)
+                    strategy.process_raw_data(publish_time, datum)
 
     def _process_market_catalogues(self, event: events.MarketCatalogueEvent) -> None:
         for market_catalogue in event.event:
diff --git a/flumine/controls/loggingcontrols.py b/flumine/controls/loggingcontrols.py
index 2c84b9e..d08b223 100644
--- a/flumine/controls/loggingcontrols.py
+++ b/flumine/controls/loggingcontrols.py
@@ -34,7 +34,7 @@ class LoggingControl(Thread):
                     self.process_event(event)
                 except Exception as e:
                     logger.critical(
-                        "{0} exception raised in {1}".format(e, self.NAME),
+                        "{0} exception raised in {0}".format(e, self.NAME),
                         exc_info=True,
                         extra={"event": event},
                     )
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index a7fc2e3..b9cd3ef 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -103,6 +103,25 @@ class Blotter:
 
     """ position """
 
+    def market_exposure(self, strategy, num_runners, num_winners: int = 1) -> float:
+        """Returns worst-case exposure for market, which is the maximum potential loss (negative),
+        arising from the worst race outcome, or the minimum potential profit (positive).
+        """
+        orders = self.strategy_orders(strategy)
+        runners = set([order.lookup for order in orders])
+        worst_possible_profits = [
+            self.get_exposures(strategy, lookup) for lookup in runners
+        ]
+        worst_possible_profits_on_loses = [
+            wpp["worst_possible_profit_on_lose"] for wpp in worst_possible_profits
+        ]
+        differences = [
+            wpp["worst_possible_profit_on_win"] - wpp["worst_possible_profit_on_lose"]
+            for wpp in worst_possible_profits
+        ] + (num_runners - len(runners)) * [0]
+        worst_differences = sorted(differences)[:num_winners]
+        return sum(worst_possible_profits_on_loses) + sum(worst_differences)
+
     def selection_exposure(self, strategy, lookup: tuple) -> float:
         """Returns strategy/selection exposure, which is the worse-case loss arising
         from the selection either winning or losing. Can be positive or zero.
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index 3a88aa4..4925bd4 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Type, Iterator, Union
+from typing import Type, Iterator
 from betfairlightweight import filters
 from betfairlightweight.resources import MarketBook
 
@@ -38,7 +38,7 @@ class BaseStrategy:
 
     def __init__(
         self,
-        market_filter: Union[dict, list],
+        market_filter: dict,
         market_data_filter: dict = None,
         streaming_timeout: float = None,
         conflate_ms: int = None,
@@ -53,7 +53,7 @@ class BaseStrategy:
         multi_order_trades: bool = False,
     ):
         """
-        :param market_filter: Streaming market filter dict or list of market filters
+        :param market_filter: Streaming market filter
         :param market_data_filter: Streaming market data filter
         :param streaming_timeout: Streaming timeout in seconds, will call snap() on cache
         :param conflate_ms: Streaming conflation
@@ -112,7 +112,7 @@ class BaseStrategy:
         # process marketBook; place/cancel/replace orders
         return
 
-    def process_raw_data(self, clk: str, publish_time: int, datum: dict) -> None:
+    def process_raw_data(self, publish_time: int, datum: dict) -> None:
         return
 
     def process_orders(self, market: Market, orders: list) -> None:
diff --git a/flumine/streams/datastream.py b/flumine/streams/datastream.py
index 9fbf60f..789b1a7 100644
--- a/flumine/streams/datastream.py
+++ b/flumine/streams/datastream.py
@@ -11,6 +11,7 @@ logger = logging.getLogger(__name__)
 
 RETRY_WAIT = BaseStream.RETRY_WAIT
 
+
 """
 Custom listener that doesn't do any processing,
 helps reduce CPU.
@@ -66,7 +67,7 @@ class FlumineMarketStream(FlumineStream):
                 )
             self._updates_processed += 1
 
-        self.on_process([self.unique_id, self._clk, publish_time, data])
+        self.on_process([self.unique_id, publish_time, data])
         return False
 
 
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 33b912c..26c89d7 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -76,7 +76,8 @@ class Streams:
             elif events:
                 raise NotImplementedError()
         else:
-            self.add_stream(strategy)
+            stream = self.add_stream(strategy)
+            strategy.streams.append(stream)
 
     def add_client(self, client: BaseClient) -> None:
         if client.order_stream:
@@ -87,44 +88,38 @@ class Streams:
 
     """ market data """
 
-    def add_stream(self, strategy: BaseStrategy) -> None:
-        if isinstance(strategy.market_filter, dict):
-            market_filters = [strategy.market_filter]
-        else:
-            market_filters = strategy.market_filter
-        for market_filter in market_filters:
-            for stream in self:  # check if market stream already exists
-                if (
-                    isinstance(stream, strategy.stream_class)
-                    and stream.market_filter == market_filter
-                    and stream.market_data_filter == strategy.market_data_filter
-                    and stream.streaming_timeout == strategy.streaming_timeout
-                    and stream.conflate_ms == strategy.conflate_ms
-                ):
-                    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()
+    def add_stream(self, strategy: BaseStrategy) -> Union[MarketStream, DataStream]:
+        for stream in self:  # check if market stream already exists
+            if (
+                isinstance(stream, strategy.stream_class)
+                and stream.market_filter == strategy.market_filter
+                and stream.market_data_filter == strategy.market_data_filter
+                and stream.streaming_timeout == strategy.streaming_timeout
+                and stream.conflate_ms == strategy.conflate_ms
+            ):
                 logger.info(
-                    "Creating new {0} ({1}) for strategy {2}".format(
-                        strategy.stream_class, stream_id, strategy
+                    "Using {0} ({1}) for strategy {2}".format(
+                        strategy.stream_class, stream.stream_id, strategy
                     )
                 )
-                stream = strategy.stream_class(
-                    flumine=self.flumine,
-                    stream_id=stream_id,
-                    market_filter=market_filter,
-                    market_data_filter=strategy.market_data_filter,
-                    streaming_timeout=strategy.streaming_timeout,
-                    conflate_ms=strategy.conflate_ms,
+                return stream
+        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
                 )
-                self._streams.append(stream)
-                strategy.streams.append(stream)
+            )
+            stream = strategy.stream_class(
+                flumine=self.flumine,
+                stream_id=stream_id,
+                market_filter=strategy.market_filter,
+                market_data_filter=strategy.market_data_filter,
+                streaming_timeout=strategy.streaming_timeout,
+                conflate_ms=strategy.conflate_ms,
+            )
+            self._streams.append(stream)
+            return stream
 
     def add_historical_stream(
         self,


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index 8f34813..38441c2 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -55,7 +55,6 @@ class FlumineBacktest(BaseFlumine):
                                 "markets": [s.market_filter for s in streams],
                             },
                         )
-                        self.simulated_datetime.reset_real_datetime()
                         # create cycles
                         cycles = []  # [[epoch, [MarketBook], gen], ..]
                         for stream in streams:
@@ -91,7 +90,6 @@ class FlumineBacktest(BaseFlumine):
                                 ),
                                 extra={"market": stream.market_filter},
                             )
-                            self.simulated_datetime.reset_real_datetime()
                             stream_gen = stream.create_generator()
                             for event in stream_gen():
                                 self._process_market_books(
diff --git a/flumine/backtest/utils.py b/flumine/backtest/utils.py
index 21fef75..b3941c3 100644
--- a/flumine/backtest/utils.py
+++ b/flumine/backtest/utils.py
@@ -17,9 +17,6 @@ class SimulatedDateTime:
     def __call__(self, pt: datetime.datetime):
         config.current_time = pt
 
-    def reset_real_datetime(self):
-        config.current_time = self._real_datetime.utcnow()
-
     @contextmanager
     def real_time(self):
         datetime.datetime = self._real_datetime
diff --git a/requirements-test.txt b/requirements-test.txt
index ef646f0..f1710c0 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -1,5 +1,5 @@
 # Tests & Linting
-black==21.10b0
+black==21.9b0
 coverage
 
 # Documentation
diff --git a/tests/test_backtest_utils.py b/tests/test_backtest_utils.py
index 4a08ec5..ab04b35 100644
--- a/tests/test_backtest_utils.py
+++ b/tests/test_backtest_utils.py
@@ -19,14 +19,6 @@ class SimulatedDateTimeTest(unittest.TestCase):
     def test_init(self):
         self.assertIsNone(self.s._real_datetime)
 
-    @mock.patch("flumine.backtest.utils.config")
-    def test_reset_real_datetime(self, mock_config):
-        mock_real_datetime = mock.Mock()
-        self.s._real_datetime = mock_real_datetime
-        self.s.reset_real_datetime()
-        mock_real_datetime.utcnow.assert_called()
-        self.assertEqual(mock_config.current_time, mock_real_datetime.utcnow())
-
     @mock.patch("flumine.backtest.utils.config")
     def test_call(self, mock_config):
         mock_dt = mock.Mock()
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 24aff21..653565a 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()
         mock_client = mock.Mock()
         self.base_flumine.add_strategy(mock_strategy, mock_client)
         self.assertEqual(len(self.base_flumine.strategies), 1)
@@ -132,7 +132,7 @@ class BaseFlumineTest(unittest.TestCase):
     @mock.patch("flumine.baseflumine.BaseFlumine._add_market")
     def test__process_raw_data(self, mock__add_market):
         mock_event = mock.Mock()
-        mock_event.event = (12, "AAA", 12345, [{"id": "1.23"}])
+        mock_event.event = (12, 12345, [{"id": "1.23"}])
         self.base_flumine._process_raw_data(mock_event)
         mock__add_market.assert_called_with("1.23", None)
 
@@ -144,19 +144,18 @@ class BaseFlumineTest(unittest.TestCase):
         mock_event = mock.Mock()
         mock_event.event = (
             12,
-            "AAA",
             12345,
             [{"id": "1.23", "marketDefinition": {"status": "CLOSED"}}],
         )
         self.base_flumine._process_raw_data(mock_event)
         mock__add_market.assert_called_with("1.23", None)
         mock_queue.put.assert_called_with(mock_events.CloseMarketEvent())
-        self.assertEqual(mock_event.event[3][0]["_stream_id"], mock_event.event[0])
+        self.assertEqual(mock_event.event[2][0]["_stream_id"], mock_event.event[0])
 
     @mock.patch("flumine.baseflumine.BaseFlumine._add_market")
     def test__process_raw_data_no_id(self, mock__add_market):
         mock_event = mock.Mock()
-        mock_event.event = (12, "AAA", 12345, [{"mid": "1.23"}])
+        mock_event.event = (12, 12345, [{"mid": "1.23"}])
         self.base_flumine._process_raw_data(mock_event)
         mock__add_market.assert_not_called()
 
diff --git a/tests/test_blotter.py b/tests/test_blotter.py
index 9f7e569..fc7ed4c 100644
--- a/tests/test_blotter.py
+++ b/tests/test_blotter.py
@@ -442,6 +442,75 @@ class BlotterTest(unittest.TestCase):
             },
         )
 
+    def test_market_position(self):
+        mock_strategy = mock.Mock()
+        mock_trade = mock.Mock(strategy=mock_strategy)
+        order_data = [
+            # (order_id, selection_id, side, average_price_matched, size_matched, size_remaining, order_type, complete)
+            (1001, 123, "BACK", 5.6, 2.0, 0.0, LimitOrder(price=5.6, size=2.0), True),
+            (1002, 123, "LAY", 5.2, 3.0, 0.0, LimitOrder(price=5.2, size=3.0), True),
+            (1003, 234, "LAY", 4.8, 4.0, 1.0, LimitOrder(price=4.8, size=5.0), True),
+            (1004, 456, "BACK", None, 0.0, 0.0, LimitOrder(price=5.6, size=2.0), False),
+            (1005, 678, "BACK", None, 0, 0, MarketOnCloseOrder(liability=6), False),
+            (
+                1006,
+                678,
+                "LAY",
+                None,
+                0,
+                0,
+                LimitOnCloseOrder(price=1.01, liability=10),
+                False,
+            ),
+        ]
+        for order in order_data:
+            self.blotter[order[0]] = mock.Mock(
+                trade=mock_trade,
+                lookup=(self.blotter.market_id, order[1], 0),
+                side=order[2],
+                average_price_matched=order[3],
+                size_matched=order[4],
+                size_remaining=order[5],
+                order_type=order[6],
+                complete=order[7],
+            )
+        self.assertEqual(  # single winner
+            self.blotter.market_exposure(mock_strategy, 6), -20.2
+        )
+        self.assertEqual(  # muliple winners
+            self.blotter.market_exposure(mock_strategy, 6, 2), -24.6
+        )
+        self.assertEqual(  # num winmers > num runners traded
+            self.blotter.market_exposure(mock_strategy, 20, 7), -28.6
+        )
+
+    def test_greened_market_position(self):
+        mock_strategy = mock.Mock()
+        mock_trade = mock.Mock(strategy=mock_strategy)
+        orders = [
+            # (order_id, selection_id, side, average_price_matched, size_matched, size_remaining, order_type, complete)
+            (1001, 123, "BACK", 5.6, 2.0, 0.0, LimitOrder(price=5.6, size=2.0), True),
+            (1002, 123, "LAY", 5.2, 2.1, 0.0, LimitOrder(price=5.2, size=2.1), True),
+            (1003, 234, "BACK", 4.8, 4.0, 0.0, LimitOrder(price=4.8, size=4.0), True),
+            (1004, 234, "LAY", 4, 4.2, 0.0, LimitOrder(price=4, size=4.2), True),
+            (1005, 345, "BACK", 10, 2.0, 0.0, LimitOrder(price=10, size=2.0), True),
+            (1006, 345, "LAY", 8, 2.2, 0.0, LimitOrder(price=8, size=2.2), True),
+        ]
+        for order in orders:
+            self.blotter[order[0]] = mock.Mock(
+                trade=mock_trade,
+                lookup=(self.blotter.market_id, order[1], 0),
+                side=order[2],
+                average_price_matched=order[3],
+                size_matched=order[4],
+                size_remaining=order[5],
+                order_type=order[6],
+                complete=order[7],
+            )
+        self.assertEqual(  # single winner
+            self.blotter.market_exposure(mock_strategy, 6), 0.5
+        )
+
     def test_complete_order(self):
         self.blotter._live_orders = ["test"]
         self.blotter.complete_order("test")
diff --git a/tests/test_strategy.py b/tests/test_strategy.py
index e1e7e95..6db5311 100644
--- a/tests/test_strategy.py
+++ b/tests/test_strategy.py
@@ -143,7 +143,7 @@ class BaseStrategyTest(unittest.TestCase):
         self.strategy.process_market_book(None, None)
 
     def test_process_raw_data(self):
-        self.strategy.process_raw_data(None, None, None)
+        self.strategy.process_raw_data(None, None)
 
     def test_process_orders(self):
         self.strategy.process_orders(None, None)
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 3f8d249..af5e90f 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -25,13 +25,17 @@ class StreamsTest(unittest.TestCase):
     def test_call(self, mock_add_stream):
         mock_strategy = mock.Mock(streams=[], raw_data=False)
         self.streams(mock_strategy)
+
         mock_add_stream.assert_called_with(mock_strategy)
+        self.assertEqual(len(mock_strategy.streams), 1)
 
     @mock.patch("flumine.streams.streams.Streams.add_stream")
     def test_call_data_stream(self, mock_add_stream):
         mock_strategy = mock.Mock(streams=[], stream_class=datastream.DataStream)
         self.streams(mock_strategy)
+
         mock_add_stream.assert_called_with(mock_strategy)
+        self.assertEqual(len(mock_strategy.streams), 1)
 
     @mock.patch("flumine.streams.streams.get_file_md")
     @mock.patch("flumine.streams.streams.Streams.add_historical_stream")
@@ -177,7 +181,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()
         mock_stream_class = mock.Mock()
         mock_strategy.stream_class = mock_stream_class
 
@@ -193,33 +197,26 @@ class StreamsTest(unittest.TestCase):
             conflate_ms=mock_strategy.conflate_ms,
         )
 
-    @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
-        )
-        self.streams.add_stream(mock_strategy)
-        self.assertEqual(len(self.streams), 2)
-        mock_increment.assert_called_with()
-
     @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 = mock.Mock()
+        mock_strategy.market_filter = 1
         mock_strategy.market_data_filter = 2
         mock_strategy.streaming_timeout = 3
         mock_strategy.conflate_ms = 4
         mock_strategy.stream_class = streams.MarketStream
 
         stream = mock.Mock(spec=streams.MarketStream)
-        stream.market_filter = {}
+        stream.market_filter = 1
         stream.market_data_filter = 2
         stream.streaming_timeout = 3
         stream.conflate_ms = 4
         stream.stream_id = 1001
         self.streams._streams = [stream]
 
-        self.streams.add_stream(mock_strategy)
+        new_stream = self.streams.add_stream(mock_strategy)
         self.assertEqual(len(self.streams), 1)
+        self.assertEqual(stream, new_stream)
         mock_increment.assert_not_called()
 
     @mock.patch("flumine.streams.streams.get_file_md")
@@ -537,14 +534,13 @@ class TestDataStream(unittest.TestCase):
     def test_flumine_market_stream(self, mock_on_process):
         mock_listener = mock.Mock(stream_unique_id=0)
         stream = datastream.FlumineMarketStream(mock_listener, 0)
-        stream._clk = "AAA"
         market_books = [{"id": "1.123"}, {"id": "1.456"}, {"id": "1.123"}]
         stream._process(market_books, 123)
 
         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, market_books]
+            [mock_listener.stream_unique_id, 123, market_books]
         )
 
     @mock.patch("flumine.streams.datastream.FlumineMarketStream.on_process")
@@ -552,7 +548,6 @@ class TestDataStream(unittest.TestCase):
         mock_listener = mock.Mock(stream_unique_id=0)
         stream = datastream.FlumineMarketStream(mock_listener, 0)
         stream._caches = {"1.123": object}
-        stream._clk = "AAA"
         market_books = [{"id": "1.123", "marketDefinition": {"status": "CLOSED"}}]
         stream._process(market_books, 123)
 
@@ -560,7 +555,7 @@ class TestDataStream(unittest.TestCase):
         self.assertEqual(len(stream._caches), 0)
         self.assertEqual(stream._updates_processed, 1)
         mock_on_process.assert_called_with(
-            [mock_listener.stream_unique_id, "AAA", 123, market_books]
+            [mock_listener.stream_unique_id, 123, market_books]
         )
 
     @mock.patch("flumine.streams.datastream.FlumineOrderStream.on_process")