Repository: liampauling/flumine (python)

Issue:
clk is not output in market recorder
The `clk` field is hardcoded to `None`, ignoring whatever was received in the raw stream:

https://github.com/liampauling/flumine/blob/8ae3879a2cb82ae0ebbb648beee93608551fe9f4/examples/strategies/marketrecorder.py#L64-L71

In the interests of

1. Replicating the raw stream as closely as possible
2. Not discarding any data

it would be good if the received `clk` value could be output

Pull request description:
#522 add clk to output in market recorder
close #522

Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index ff573cf..31f709c 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,19 @@
 Release History
 ---------------
 
+1.20.9 (2021-11-11)
++++++++++++++++++++
+
+**Improvements**
+
+- #522 add clk to output in market recorder (Breaking Change)
+- #517 Extend selection exposures to whole market (@petercoles)
+- Example strategies updated to remove whitespace on dump (1mb saved per raw file)
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.15.1
+
 1.20.8 (2021-11-01)
 +++++++++++++++++++
 
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/examples/controls/jupyterloggingcontrol.py b/examples/controls/jupyterloggingcontrol.py
index c7cd02b..337b5c1 100644
--- a/examples/controls/jupyterloggingcontrol.py
+++ b/examples/controls/jupyterloggingcontrol.py
@@ -69,7 +69,7 @@ class JupyterLoggingControl(LoggingControl):
     def _create_json(self, data):
         # save json to file
         with open(self.file_name, "w") as f:
-            json.dump(data, f)
+            json.dump(data, f, separators=(",", ":"))
 
     @staticmethod
     def launch(self, argv=None, **kwargs):
diff --git a/examples/strategies/marketrecorder.py b/examples/strategies/marketrecorder.py
index 0c08fdf..1264271 100644
--- a/examples/strategies/marketrecorder.py
+++ b/examples/strategies/marketrecorder.py
@@ -61,12 +61,15 @@ class MarketRecorder(BaseStrategy):
             daemon=True,
         ).start()
 
-    def process_raw_data(self, publish_time, data):
+    def process_raw_data(self, clk: str, publish_time: int, data: dict):
         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": None, "pt": publish_time, "mc": [data]})
+                json.dumps(
+                    {"op": "mcm", "clk": clk, "pt": publish_time, "mc": [data]},
+                    separators=(",", ":"),
+                )
                 + "\n"
             )
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 243ac63..cb87f3b 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.8"
+__version__ = "1.20.9"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index e4227b8..fc6d3aa 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, publish_time, data = event.event
+        stream_id, clk, 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(publish_time, datum)
+                    strategy.process_raw_data(clk, publish_time, datum)
 
     def _process_market_catalogues(self, event: events.MarketCatalogueEvent) -> None:
         for market_catalogue in 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 f487495..3a88aa4 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -112,7 +112,7 @@ class BaseStrategy:
         # process marketBook; place/cancel/replace orders
         return
 
-    def process_raw_data(self, publish_time: int, datum: dict) -> None:
+    def process_raw_data(self, clk: str, 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 789b1a7..9fbf60f 100644
--- a/flumine/streams/datastream.py
+++ b/flumine/streams/datastream.py
@@ -11,7 +11,6 @@ logger = logging.getLogger(__name__)
 
 RETRY_WAIT = BaseStream.RETRY_WAIT
 
-
 """
 Custom listener that doesn't do any processing,
 helps reduce CPU.
@@ -67,7 +66,7 @@ class FlumineMarketStream(FlumineStream):
                 )
             self._updates_processed += 1
 
-        self.on_process([self.unique_id, publish_time, data])
+        self.on_process([self.unique_id, self._clk, publish_time, data])
         return False
 
 
diff --git a/requirements.txt b/requirements.txt
index da631f1..e4f2649 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.15.0
+betfairlightweight==2.15.1
 tenacity>=7.0.0,<=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 5e05723..24aff21 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -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, 12345, [{"id": "1.23"}])
+        mock_event.event = (12, "AAA", 12345, [{"id": "1.23"}])
         self.base_flumine._process_raw_data(mock_event)
         mock__add_market.assert_called_with("1.23", None)
 
@@ -144,18 +144,19 @@ 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[2][0]["_stream_id"], mock_event.event[0])
+        self.assertEqual(mock_event.event[3][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, 12345, [{"mid": "1.23"}])
+        mock_event.event = (12, "AAA", 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 6db5311..e1e7e95 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)
+        self.strategy.process_raw_data(None, 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 c42ec1d..3f8d249 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -537,13 +537,14 @@ 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, 123, market_books]
+            [mock_listener.stream_unique_id, "AAA", 123, market_books]
         )
 
     @mock.patch("flumine.streams.datastream.FlumineMarketStream.on_process")
@@ -551,6 +552,7 @@ 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)
 
@@ -558,7 +560,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, 123, market_books]
+            [mock_listener.stream_unique_id, "AAA", 123, market_books]
         )
 
     @mock.patch("flumine.streams.datastream.FlumineOrderStream.on_process")