Repository: liampauling/flumine
Pull request #218 (2020-06-15 14:52:19)

custom historical listener/stream added to ignore processing on certa…
…in market states


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index c7e3762..2ecbc7a 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,24 @@
 Release History
 ---------------
 
+1.8.0 (2020-06-29)
++++++++++++++++++++
+
+**Improvements**
+
+- Custom historical listener/stream added
+- Large order count (per market) optimisations
+- #203 client transaction count
+- #224 multi market processing
+
+**Bug Fixes**
+
+- #221 RuntimeError: market/order looping
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.5.0
+
 1.7.0 (2020-06-15)
 +++++++++++++++++++
 
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 0af11c9..226b345 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.7.0"
+__version__ = "1.8.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 6a03d42..1e573da 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -128,11 +128,12 @@ class BaseFlumine:
                         strategy.process_market_book, market, market_book
                     )
 
-            self._process_market_orders(market)
+            self._process_market_orders()
 
-    def _process_market_orders(self, market: Market) -> None:
-        for order_package in market.blotter.process_orders(self.client):
-            self.handler_queue.put(order_package)
+    def _process_market_orders(self) -> None:
+        for market in self.markets:
+            for order_package in market.blotter.process_orders(self.client):
+                self.handler_queue.put(order_package)
 
     def _process_order_package(self, order_package) -> None:
         """Validate trading controls and
@@ -182,7 +183,7 @@ class BaseFlumine:
                 for strategy in self.strategies:
                     strategy_orders = market.blotter.strategy_orders(strategy)
                     strategy.process_orders(market, strategy_orders)
-            self._process_market_orders(market)
+        self._process_market_orders()
 
     def _process_custom_event(self, event: events.CustomEvent) -> None:
         try:
@@ -193,8 +194,7 @@ class BaseFlumine:
                     e, event.callback
                 )
             )
-        for market in self.markets:
-            self._process_market_orders(market)
+        self._process_market_orders()
 
     def _process_close_market(self, event: events.CloseMarketEvent) -> None:
         market_id = event.event.market_id
diff --git a/flumine/clients/baseclient.py b/flumine/clients/baseclient.py
index 00acd31..713c176 100644
--- a/flumine/clients/baseclient.py
+++ b/flumine/clients/baseclient.py
@@ -1,4 +1,5 @@
 from typing import Optional
+from betfairlightweight.metadata import transaction_limit as betfair_transaction_limit
 
 from ..utils import create_short_uuid
 from .clients import ExchangeType
@@ -17,7 +18,7 @@ class BaseClient:
     def __init__(
         self,
         betting_client=None,
-        transaction_limit: int = 1000,
+        transaction_limit: Optional[int] = betfair_transaction_limit,
         capital_base: int = DEFAULT_CAPITAL_BASE,
         commission_base: float = DEFAULT_COMMISSION_BASE,
         interactive_login: bool = False,
diff --git a/flumine/controls/clientcontrols.py b/flumine/controls/clientcontrols.py
index 24779a8..bf1d31b 100644
--- a/flumine/controls/clientcontrols.py
+++ b/flumine/controls/clientcontrols.py
@@ -1,5 +1,6 @@
 import datetime
 import logging
+from typing import Optional
 from betfairlightweight.metadata import transaction_limit
 
 from ..order.orderpackage import OrderPackageType
@@ -31,9 +32,8 @@ class MaxOrderCount(BaseControl):
         self._next_hour = None
         self._set_next_hour()
         self.transaction_count = 0
-        self.transaction_limit = self.client.transaction_limit
 
-    def _validate(self, order_package):
+    def _validate(self, order_package) -> None:
         self._check_hour()
         self.total += 1
         if order_package.package_type == OrderPackageType.PLACE:
@@ -54,7 +54,7 @@ class MaxOrderCount(BaseControl):
         elif order_package.package_type == OrderPackageType.REPLACE:
             self.replace_requests += len(order_package)
 
-    def _check_hour(self):
+    def _check_hour(self) -> None:
         if datetime.datetime.utcnow() > self._next_hour:
             logger.info(
                 "Execution new hour",
@@ -74,7 +74,7 @@ class MaxOrderCount(BaseControl):
                 )
             self.transaction_count = 0
 
-    def _check_transaction_count(self, transaction_count):
+    def _check_transaction_count(self, transaction_count: int) -> None:
         self.transaction_count += transaction_count
         if self.transaction_count > self.transaction_limit:
             logger.error(
@@ -86,16 +86,18 @@ class MaxOrderCount(BaseControl):
                 },
             )
 
-    def _set_next_hour(self):
+    def _set_next_hour(self) -> None:
         now = datetime.datetime.utcnow()
         self._next_hour = (now + datetime.timedelta(hours=1)).replace(
             minute=0, second=0, microsecond=0
         )
 
     @property
-    def safe(self):
+    def safe(self) -> bool:
         self._check_hour()
-        if self.transaction_count < self.transaction_limit:
+        if self.transaction_limit is None:
+            return True
+        elif self.transaction_count < self.transaction_limit:
             return True
         else:
             logger.error(
@@ -106,3 +108,8 @@ class MaxOrderCount(BaseControl):
                     "client": self.client,
                 },
             )
+            return False
+
+    @property
+    def transaction_limit(self) -> Optional[int]:
+        return self.client.transaction_limit
diff --git a/flumine/markets/blotter.py b/flumine/markets/blotter.py
index 04f6814..1a81421 100644
--- a/flumine/markets/blotter.py
+++ b/flumine/markets/blotter.py
@@ -15,6 +15,7 @@ class Blotter:
     def __init__(self, market):
         self.market = market
         self._orders = {}  # {Order.id: Order}
+        self._live_orders = []  # cached list of live orders
         # pending orders
         self.pending_place = []
         self.pending_cancel = []
@@ -70,11 +71,12 @@ class Blotter:
         return packages
 
     @property
-    def live_orders(self) -> bool:
-        for order in self._orders.values():
-            if order.complete is False or order.trade.complete is False:
-                return True
-        return False
+    def live_orders(self):
+        return iter(self._live_orders)
+
+    @property
+    def has_live_orders(self):
+        return bool(self._live_orders)
 
     def process_closed_market(self, market_book):
         for order in self:
@@ -99,7 +101,11 @@ class Blotter:
         """
         mb, ml = [], []  # (price, size)
         for order in self:
-            if order.trade.strategy == strategy and order.lookup == lookup:
+            if (
+                order.trade.strategy == strategy
+                and order.size_matched
+                and order.lookup == lookup
+            ):
                 if order.side == "BACK":
                     mb.append((order.average_price_matched, order.size_matched))
                 else:
@@ -112,6 +118,9 @@ class Blotter:
 
     """ getters / setters """
 
+    def complete_order(self, order) -> None:
+        self._live_orders.remove(order)
+
     def has_order(self, customer_order_ref: str) -> bool:
         return customer_order_ref in self._orders
 
@@ -119,12 +128,13 @@ class Blotter:
 
     def __setitem__(self, customer_order_ref: str, order) -> None:
         self._orders[customer_order_ref] = order
+        self._live_orders.append(order)
 
     def __getitem__(self, customer_order_ref: str):
         return self._orders[customer_order_ref]
 
     def __iter__(self) -> Iterable[BaseOrder]:
-        return iter(self._orders.values())
+        return iter(list(self._orders.values()))
 
     def __len__(self) -> int:
         return len(self._orders)
diff --git a/flumine/markets/market.py b/flumine/markets/market.py
index 90af463..e72743b 100644
--- a/flumine/markets/market.py
+++ b/flumine/markets/market.py
@@ -1,6 +1,7 @@
 import datetime
 import logging
 from typing import Optional
+from collections import defaultdict
 from betfairlightweight.resources.bettingresources import MarketBook, MarketCatalogue
 
 from .blotter import Blotter
@@ -61,6 +62,14 @@ class Market:
         order.replace(new_price)
         self.blotter.pending_replace.append(order)
 
+    @property
+    def event(self) -> dict:
+        event = defaultdict(list)
+        for market in self.flumine.markets:
+            if market.event_id == self.event_id:
+                event[market.market_type].append(market)
+        return event
+
     @property
     def event_type_id(self) -> str:
         if self.market_book:
@@ -71,6 +80,11 @@ class Market:
         if self.market_book:
             return self.market_book.market_definition.event_id
 
+    @property
+    def market_type(self) -> str:
+        if self.market_book:
+            return self.market_book.market_definition.market_type
+
     @property
     def seconds_to_start(self):
         return (self.market_start_datetime - datetime.datetime.utcnow()).total_seconds()
diff --git a/flumine/markets/markets.py b/flumine/markets/markets.py
index 714bcbc..2b9e24b 100644
--- a/flumine/markets/markets.py
+++ b/flumine/markets/markets.py
@@ -49,12 +49,12 @@ class Markets:
     @property
     def live_orders(self) -> bool:
         for market in self:
-            if market.closed is False and market.blotter.live_orders is True:
+            if market.closed is False and market.blotter.has_live_orders:
                 return True
         return False
 
     def __iter__(self) -> Iterator[Market]:
-        return iter(self.markets.values())
+        return iter(list(self.markets.values()))
 
     def __len__(self) -> int:
         return len(self.markets)
diff --git a/flumine/markets/middleware.py b/flumine/markets/middleware.py
index de95269..1330b1c 100644
--- a/flumine/markets/middleware.py
+++ b/flumine/markets/middleware.py
@@ -35,7 +35,7 @@ class SimulatedMiddleware(Middleware):
 
     @staticmethod
     def _process_simulated_orders(market, market_analytics: dict) -> None:
-        for order in market.blotter:
+        for order in market.blotter.live_orders:
             if order.simulated and order.status == OrderStatus.EXECUTABLE:
                 runner_analytics = market_analytics.get(
                     (order.selection_id, order.handicap)
diff --git a/flumine/order/order.py b/flumine/order/order.py
index 7b69613..9e4952d 100644
--- a/flumine/order/order.py
+++ b/flumine/order/order.py
@@ -55,6 +55,7 @@ class BaseOrder:
         self.update_data = {}  # stores cancel/update/replace data
         self.responses = Responses()  # raw api responses
         self.simulated = Simulated(self)  # used in simulated execution
+        self._simulated = bool(self.simulated)  # cache in current class (2x quicker)
         self.publish_time = None  # marketBook.publish_time
 
         self.date_time_created = datetime.datetime.utcnow()
@@ -131,7 +132,7 @@ class BaseOrder:
 
     @property
     def current_order(self) -> Union[CurrentOrder, Simulated]:
-        if self.simulated:
+        if self._simulated:
             return self.simulated
         elif self.responses.current_order:
             return self.responses.current_order
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index 28444de..d38166d 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -1,6 +1,6 @@
 from typing import Type, Iterator
 from betfairlightweight import filters
-from betfairlightweight.resources import MarketBook, RaceCard
+from betfairlightweight.resources import MarketBook
 
 from ..streams.marketstream import BaseStream, MarketStream
 from ..markets.market import Market
diff --git a/flumine/streams/basestream.py b/flumine/streams/basestream.py
index 71e7119..33f863d 100644
--- a/flumine/streams/basestream.py
+++ b/flumine/streams/basestream.py
@@ -21,6 +21,8 @@ class BaseStream(threading.Thread):
         market_filter: dict = None,
         market_data_filter: dict = None,
         client=None,
+        output_queue: bool = True,
+        **listener_kwargs,
     ):
         threading.Thread.__init__(self, daemon=True, name=self.__class__.__name__)
         self.flumine = flumine
@@ -31,9 +33,11 @@ class BaseStream(threading.Thread):
         self.conflate_ms = conflate_ms
         self._client = client
         self._stream = None
-        self._output_queue = queue.Queue()
+        self._output_queue = queue.Queue() if output_queue else None
         self._listener = self.LISTENER(
-            output_queue=self._output_queue, max_latency=self.MAX_LATENCY
+            output_queue=self._output_queue,
+            max_latency=self.MAX_LATENCY,
+            **listener_kwargs,
         )
         self._output_thread = threading.Thread(
             name="{0}_output_thread".format(self._name),
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
index 367f30f..667aa0d 100644
--- a/flumine/streams/historicalstream.py
+++ b/flumine/streams/historicalstream.py
@@ -1,13 +1,96 @@
 import logging
-from betfairlightweight.streaming import HistoricalGeneratorStream
+import datetime
+from betfairlightweight.streaming import StreamListener, HistoricalGeneratorStream
+from betfairlightweight.streaming.stream import BaseStream as BFLWBaseStream
+from betfairlightweight.streaming.cache import MarketBookCache
+from betfairlightweight.resources.baseresource import BaseResource
 
 from .basestream import BaseStream
 
 logger = logging.getLogger(__name__)
 
 
+class Stream(BFLWBaseStream):
+    """
+    Custom bflw stream to speed up processing
+    by limiting to inplay/not inplay or limited
+    seconds to start.
+    """
+
+    _lookup = "mc"
+
+    def _process(self, data: list, publish_time: int) -> None:
+        for market_book in data:
+            market_id = market_book["id"]
+            market_book_cache = self._caches.get(market_id)
+            if (
+                market_book.get("img") or market_book_cache is None
+            ):  # historic data does not contain img
+                if "marketDefinition" not in market_book:
+                    logger.error(
+                        "[MarketStream: %s] Unable to add %s to cache due to marketDefinition "
+                        "not being present (make sure EX_MARKET_DEF is requested)"
+                        % (self.unique_id, market_id)
+                    )
+                    continue
+                market_book_cache = MarketBookCache(
+                    publish_time=publish_time, **market_book
+                )
+                self._caches[market_id] = market_book_cache
+                logger.info(
+                    "[MarketStream: %s] %s added, %s markets in cache"
+                    % (self.unique_id, market_id, len(self._caches))
+                )
+            else:
+                market_book_cache.update_cache(market_book, publish_time)
+                self._updates_processed += 1
+
+    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 has closed send regardless
+            if cache.market_definition["status"] != "CLOSED":
+                if self._listener.inplay:
+                    if not cache.market_definition["inPlay"]:
+                        continue
+                elif self._listener.seconds_to_start:
+                    _now = datetime.datetime.utcfromtimestamp(cache.publish_time / 1e3)
+                    _market_time = BaseResource.strip_datetime(
+                        cache.market_definition["marketTime"]
+                    )
+                    seconds_to_start = (_market_time - _now).total_seconds()
+                    if seconds_to_start > self._listener.seconds_to_start:
+                        continue
+                if self._listener.inplay is False:
+                    if cache.market_definition["inPlay"]:
+                        continue
+            market_books.append(
+                cache.create_resource(self.unique_id, None, self._lightweight)
+            )
+        return market_books
+
+
+class HistoricListener(StreamListener):
+    """
+    Custom listener to restrict processing by
+    inplay or seconds_to_start.
+    """
+
+    def __init__(self, inplay: bool = None, seconds_to_start: float = None, **kwargs):
+        super(HistoricListener, self).__init__(**kwargs)
+        self.inplay = inplay
+        self.seconds_to_start = seconds_to_start
+
+    def _add_stream(self, unique_id, stream_type):
+        if stream_type == "marketSubscription":
+            return Stream(self)
+
+
 class HistoricalStream(BaseStream):
 
+    LISTENER = HistoricListener
     MAX_LATENCY = None
 
     def run(self) -> None:
@@ -17,9 +100,8 @@ class HistoricalStream(BaseStream):
         pass
 
     def create_generator(self):
-        self._listener.output_queue = None
         self._listener.register_stream(0, "marketSubscription")
         stream = HistoricalGeneratorStream(
-            directory=self.market_filter, listener=self._listener
+            file_path=self.market_filter, listener=self._listener
         )
         return stream.get_generator()
diff --git a/flumine/streams/marketstream.py b/flumine/streams/marketstream.py
index 46dbe85..258e8b9 100644
--- a/flumine/streams/marketstream.py
+++ b/flumine/streams/marketstream.py
@@ -50,6 +50,7 @@ class MarketStream(BaseStream):
                 market_books = self._listener.snap(
                     market_ids=self.flumine.markets.open_market_ids
                 )
-            self.flumine.handler_queue.put(MarketBookEvent(market_books))
+            if market_books:
+                self.flumine.handler_queue.put(MarketBookEvent(market_books))
 
         logger.info("Stopped output_thread (MarketStream {0})".format(self.stream_id))
diff --git a/flumine/streams/orderstream.py b/flumine/streams/orderstream.py
index 5e65888..a29f812 100644
--- a/flumine/streams/orderstream.py
+++ b/flumine/streams/orderstream.py
@@ -55,7 +55,7 @@ class OrderStream(BaseStream):
                     )
                 else:
                     continue
-
-            self.flumine.handler_queue.put(CurrentOrdersEvent(order_books))
+            if order_books:
+                self.flumine.handler_queue.put(CurrentOrdersEvent(order_books))
 
         logger.info("Stopped output_thread (OrderStream {0})".format(self.stream_id))
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 98214d4..c538c63 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -20,11 +20,14 @@ class Streams:
     def __call__(self, strategy: BaseStrategy) -> None:
         if self.flumine.BACKTEST:
             markets = strategy.market_filter.get("markets")
+            listener_kwargs = strategy.market_filter.get("listener_kwargs", {})
             if markets is None:
                 logging.warning("No markets found for strategy {0}".format(strategy))
             else:
                 for market in markets:
-                    stream = self.add_historical_stream(strategy, market)
+                    stream = self.add_historical_stream(
+                        strategy, market, **listener_kwargs
+                    )
                     strategy.streams.append(stream)
         else:
             stream = self.add_stream(strategy)
@@ -71,7 +74,9 @@ class Streams:
             self._streams.append(stream)
             return stream
 
-    def add_historical_stream(self, strategy: BaseStrategy, market) -> HistoricalStream:
+    def add_historical_stream(
+        self, strategy: BaseStrategy, market, **listener_kwargs
+    ) -> HistoricalStream:
         for stream in self:
             if stream.market_filter == market:
                 return stream
@@ -89,6 +94,8 @@ class Streams:
                 market_data_filter=strategy.market_data_filter,
                 streaming_timeout=strategy.streaming_timeout,
                 conflate_ms=strategy.conflate_ms,
+                output_queue=False,
+                **listener_kwargs,
             )
             self._streams.append(stream)
             return stream
diff --git a/flumine/utils.py b/flumine/utils.py
index 99dbb33..a8b56c7 100644
--- a/flumine/utils.py
+++ b/flumine/utils.py
@@ -118,10 +118,12 @@ def price_ticks_away(price: float, n_ticks: int) -> float:
 
 
 # todo LRU cache?
-def calculate_exposure(mb: list, ml: list) -> int:
+def calculate_exposure(mb: list, ml: list) -> float:
     """Calculates exposure based on list
     of (price, size)
     """
+    if not mb and not ml:
+        return 0.0
     back_exp = sum(-i[1] for i in mb)
     back_profit = sum((i[0] - 1) * i[1] for i in mb)
     lay_exp = sum((i[0] - 1) * -i[1] for i in ml)
diff --git a/flumine/worker.py b/flumine/worker.py
index 2531062..150bbe7 100644
--- a/flumine/worker.py
+++ b/flumine/worker.py
@@ -65,6 +65,7 @@ class BackgroundWorker(threading.Thread):
                         "function": self.function,
                         "context": self.context,
                     },
+                    exc_info=True,
                 )
             time.sleep(self.interval)
 
diff --git a/requirements.txt b/requirements.txt
index 57c25cb..46e3c2d 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
-betfairlightweight==2.4.0
+betfairlightweight==2.5.0
 tenacity==5.0.3
 python-json-logger==0.1.11
 requests


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index 2c7fdd3..032696d 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -7,7 +7,6 @@ from .. import config, utils
 from ..clients import ExchangeType
 from ..exceptions import RunError
 from .utils import PendingPackages
-from ..markets.market import Market
 from ..order.orderpackage import OrderPackageType
 from ..order import process
 from ..markets.middleware import SimulatedMiddleware
@@ -94,17 +93,20 @@ class FlumineBacktest(BaseFlumine):
 
             # process current orders
             blotter = market.blotter
-            for order in blotter:
+            for order in blotter.live_orders:
                 process.process_current_order(order)
+                if order.trade.status.value == "Complete":
+                    blotter.complete_order(order)
             for strategy in self.strategies:
                 strategy_orders = blotter.strategy_orders(strategy)
                 strategy.process_orders(market, strategy_orders)
 
-            self._process_market_orders(market)
+            self._process_market_orders()
 
-    def _process_market_orders(self, market: Market) -> None:
-        for order_package in market.blotter.process_orders(self.client):
-            self._pending_packages.append(order_package)
+    def _process_market_orders(self) -> None:
+        for market in self.markets:
+            for order_package in market.blotter.process_orders(self.client):
+                self._pending_packages.append(order_package)
 
     def _process_order_package(self, order_package) -> None:
         """Validate trading controls and
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 84b1c7c..b06be67 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -81,7 +81,8 @@ class BaseFlumineTest(unittest.TestCase):
     def test__process_market_orders(self):
         mock_market = mock.Mock()
         mock_market.blotter.process_orders.return_value = [1, 2, 3]
-        self.base_flumine._process_market_orders(mock_market)
+        self.base_flumine.markets = [mock_market]
+        self.base_flumine._process_market_orders()
         mock_market.blotter.process_orders.assert_called_with(self.mock_client)
 
     def test__process_order_package(self):
@@ -168,7 +169,7 @@ class BaseFlumineTest(unittest.TestCase):
         mock_event = mock.Mock()
         self.base_flumine._process_custom_event(mock_event)
         mock_event.callback.assert_called_with(self.base_flumine, mock_event)
-        mock__process_market_orders.assert_called_with(mock_market)
+        mock__process_market_orders.assert_called_with()
 
     @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 fed480f..11e64c2 100644
--- a/tests/test_blotter.py
+++ b/tests/test_blotter.py
@@ -16,6 +16,7 @@ class BlotterTest(unittest.TestCase):
         self.assertEqual(self.blotter.pending_cancel, [])
         self.assertEqual(self.blotter.pending_update, [])
         self.assertEqual(self.blotter.pending_replace, [])
+        self.assertEqual(self.blotter._live_orders, [])
 
     def test_strategy_orders(self):
         mock_order = mock.Mock()
@@ -69,17 +70,15 @@ class BlotterTest(unittest.TestCase):
         self.assertEqual(mock_orders, [])
 
     def test_live_orders(self):
-        self.assertFalse(self.blotter.live_orders)
+        self.assertEqual(list(self.blotter.live_orders), [])
         mock_order = mock.Mock(complete=False)
-        self.blotter._orders = {"12345": mock_order}
-        self.assertTrue(self.blotter.live_orders)
+        self.blotter._live_orders = [mock_order]
+        self.assertEqual(list(self.blotter.live_orders), [mock_order])
 
-    def test_live_orders_trade(self):
-        self.assertFalse(self.blotter.live_orders)
-        mock_order = mock.Mock()
-        mock_order.trade.complete = False
-        self.blotter._orders = {"12345": mock_order}
-        self.assertTrue(self.blotter.live_orders)
+    def test_has_live_orders(self):
+        self.assertFalse(self.blotter.has_live_orders)
+        self.blotter._live_orders = [mock.Mock()]
+        self.assertTrue(self.blotter.has_live_orders)
 
     def test_process_closed_market(self):
         mock_market_book = mock.Mock()
@@ -105,17 +104,32 @@ class BlotterTest(unittest.TestCase):
             size_matched=2.0,
         )
         self.blotter._orders = {"12345": mock_order}
+        self.assertEqual(
+            self.blotter.selection_exposure(mock_strategy, mock_order.lookup), -2,
+        )
 
+    def test_selection_exposure_no_match(self):
+        mock_strategy = mock.Mock()
+        mock_trade = mock.Mock(strategy=mock_strategy)
+        mock_order = mock.Mock(
+            trade=mock_trade,
+            lookup=(self.blotter.market_id, 123, 0),
+            side="BACK",
+            average_price_matched=5.6,
+            size_matched=0.0,
+        )
+        self.blotter._orders = {"12345": mock_order}
         self.assertEqual(
-            self.blotter.selection_exposure(
-                mock_strategy, (self.blotter.market_id, 123, 0)
-            ),
-            -2,
+            self.blotter.selection_exposure(mock_strategy, mock_order.lookup), 0,
         )
 
     def test_market_id(self):
         self.assertEqual(self.blotter.market_id, self.mock_market.market_id)
 
+    def test_complete_order(self):
+        self.blotter._live_orders = ["test"]
+        self.blotter.complete_order("test")
+
     def test__contains(self):
         self.blotter._orders = {"123": "test"}
         self.assertIn("123", self.blotter)
@@ -124,6 +138,7 @@ class BlotterTest(unittest.TestCase):
     def test__setitem(self):
         self.blotter["123"] = "test"
         self.assertEqual(self.blotter._orders, {"123": "test"})
+        self.assertEqual(self.blotter._live_orders, ["test"])
 
     def test__getitem(self):
         self.blotter._orders = {"12345": "test", "54321": "test2"}
diff --git a/tests/test_clientcontrols.py b/tests/test_clientcontrols.py
index 2ac21b6..c70e545 100644
--- a/tests/test_clientcontrols.py
+++ b/tests/test_clientcontrols.py
@@ -146,6 +146,15 @@ class TestMaxOrderCount(unittest.TestCase):
         self.assertTrue(self.trading_control.safe)
         mock_check_hour.assert_called_with()
 
-        self.trading_control.transaction_limit = -10
-        self.assertIsNone(self.trading_control.safe)
+        self.trading_control.client.transaction_limit = -10
+        self.assertFalse(self.trading_control.safe)
         mock_check_hour.assert_called_with()
+
+        self.trading_control.client.transaction_limit = None
+        self.assertTrue(self.trading_control.safe)
+        mock_check_hour.assert_called_with()
+
+    def test_transaction_limit(self):
+        self.assertEqual(
+            self.trading_control.transaction_limit, self.mock_client.transaction_limit
+        )
diff --git a/tests/test_fluminebacktest.py b/tests/test_fluminebacktest.py
index f0edefa..e5b24af 100644
--- a/tests/test_fluminebacktest.py
+++ b/tests/test_fluminebacktest.py
@@ -66,7 +66,9 @@ class FlumineBacktestTest(unittest.TestCase):
         mock_order_package = mock.Mock()
         mock_market = mock.Mock()
         mock_market.blotter.process_orders.return_value = [mock_order_package]
-        self.flumine._process_market_orders(mock_market)
+        self.flumine.markets = [mock_market]
+        self.flumine._process_market_orders()
+        mock_market.blotter.process_orders.assert_called_with(self.flumine.client)
         self.assertEqual(self.flumine._pending_packages, [mock_order_package])
 
     @mock.patch("flumine.backtest.backtest.BaseFlumine._process_order_package")
diff --git a/tests/test_markets.py b/tests/test_markets.py
index c12d411..a17022f 100644
--- a/tests/test_markets.py
+++ b/tests/test_markets.py
@@ -80,9 +80,11 @@ class MarketsTest(unittest.TestCase):
         self.assertFalse(self.markets.live_orders)
         mock_market = mock.Mock()
         mock_market.closed = False
-        mock_market.blotter.live_orders = True
+        mock_market.blotter.has_live_orders = True
         self.markets._markets = {"1.234": mock_market}
         self.assertTrue(self.markets.live_orders)
+        mock_market.blotter.has_live_orders = False
+        self.assertFalse(self.markets.live_orders)
 
     def test_iter(self):
         self.assertEqual(len([i for i in self.markets]), 0)
@@ -179,6 +181,21 @@ class MarketTest(unittest.TestCase):
         mock_order.replace.assert_called_with(1.01)
         self.assertEqual(mock_blotter.pending_replace, [mock_order])
 
+    def test_event(self):
+        mock_market_book = mock.Mock()
+        mock_market_book.market_definition = mock.Mock(event_id=12)
+        self.market.market_book = mock_market_book
+
+        self.market.flumine.markets = []
+        self.assertEqual(self.market.event, {})
+
+        m_one = mock.Mock(market_type=1, event_id=12)
+        m_two = mock.Mock(market_type=2, event_id=12)
+        m_three = mock.Mock(market_type=3, event_id=123)
+        m_four = mock.Mock(market_type=1, event_id=12)
+        self.market.flumine.markets = [m_one, m_two, m_three, m_four]
+        self.assertEqual(self.market.event, {1: [m_one, m_four], 2: [m_two]})
+
     def test_event_type_id(self):
         mock_market_book = mock.Mock()
         self.market.market_book = mock_market_book
@@ -193,6 +210,13 @@ class MarketTest(unittest.TestCase):
             self.market.event_id, mock_market_book.market_definition.event_id
         )
 
+    def test_market_type(self):
+        mock_market_book = mock.Mock()
+        self.market.market_book = mock_market_book
+        self.assertEqual(
+            self.market.market_type, mock_market_book.market_definition.market_type
+        )
+
     def test_seconds_to_start(self):
         mock_market_catalogue = mock.Mock()
         mock_market_catalogue.market_start_time = datetime.datetime.utcfromtimestamp(1)
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index fdd417a..5c7c52d 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -50,8 +50,12 @@ class SimulatedMiddlewareTest(unittest.TestCase):
         mock_order_two.status = OrderStatus.PENDING
         mock_order_three = mock.Mock()
         mock_order_three.status = OrderStatus.EXECUTABLE
-        mock_order_three.simulated = False
-        mock_market.blotter = [mock_order, mock_order_two, mock_order_three]
+        mock_order_three._simulated = False
+        mock_market.blotter.live_orders = [
+            mock_order,
+            mock_order_two,
+            mock_order_three,
+        ]
         mock_market_analytics = {(mock_order.selection_id, mock_order.handicap): "test"}
         mock_market.market_book = mock_market_book
         self.middleware._process_simulated_orders(mock_market, mock_market_analytics)
diff --git a/tests/test_order.py b/tests/test_order.py
index ac2edd7..0b3e181 100644
--- a/tests/test_order.py
+++ b/tests/test_order.py
@@ -33,6 +33,8 @@ class BaseOrderTest(unittest.TestCase):
         self.assertIsNone(self.order.publish_time)
         self.assertIsNotNone(self.order.date_time_created)
         self.assertIsNone(self.order.date_time_execution_complete)
+        self.assertFalse(self.order.simulated)
+        self.assertFalse(self.order._simulated)
 
     @mock.patch("flumine.order.order.BaseOrder.info")
     def test__update_status(self, mock_info):
@@ -142,9 +144,12 @@ class BaseOrderTest(unittest.TestCase):
         mock_responses.current_order = 1
         self.assertEqual(self.order.current_order, 1)
 
-    def test_current_order_simulated(self):
-        self.order.simulated = True
-        self.assertTrue(self.order.current_order)
+    @mock.patch("flumine.backtest.simulated.config")
+    def test_current_order_simulated(self, mock_config):
+        mock_config.simulated = True
+        order = BaseOrder(mock.Mock(), "", mock.Mock())
+        self.assertTrue(order.simulated)
+        self.assertTrue(order._simulated)
 
     def test_complete(self):
         self.order.status = None
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 6bd467c..04f48b7 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -4,6 +4,7 @@ from unittest import mock
 from flumine.streams import streams, datastream
 from flumine.streams.basestream import BaseStream
 from flumine.exceptions import ListenerError
+from flumine.streams import historicalstream
 
 
 class StreamsTest(unittest.TestCase):
@@ -42,11 +43,14 @@ class StreamsTest(unittest.TestCase):
         self.mock_flumine.BACKTEST = True
         mock_strategy = mock.Mock()
         mock_strategy.streams = []
-        mock_strategy.market_filter = {"markets": ["dubs of the mad skint and british"]}
+        mock_strategy.market_filter = {
+            "markets": ["dubs of the mad skint and british"],
+            "listener_kwargs": {"canary_yellow": True},
+        }
         self.streams(mock_strategy)
 
         mock_add_historical_stream.assert_called_with(
-            mock_strategy, "dubs of the mad skint and british"
+            mock_strategy, "dubs of the mad skint and british", canary_yellow=True
         )
         self.assertEqual(len(mock_strategy.streams), 1)
 
@@ -123,7 +127,7 @@ class StreamsTest(unittest.TestCase):
         mock_strategy.conflate_ms = 4
         mock_strategy.stream_class = streams.MarketStream
 
-        self.streams.add_historical_stream(mock_strategy, "GANG")
+        self.streams.add_historical_stream(mock_strategy, "GANG", inplay=True)
         self.assertEqual(len(self.streams), 1)
         mock_increment.assert_called_with()
         mock_historical_stream_class.assert_called_with(
@@ -133,6 +137,8 @@ class StreamsTest(unittest.TestCase):
             market_data_filter=mock_strategy.market_data_filter,
             streaming_timeout=mock_strategy.streaming_timeout,
             conflate_ms=mock_strategy.conflate_ms,
+            output_queue=False,
+            inplay=True,
         )
 
     def test_add_historical_stream_old(self):
@@ -148,7 +154,9 @@ class StreamsTest(unittest.TestCase):
 
     @mock.patch("flumine.streams.streams.OrderStream")
     @mock.patch("flumine.streams.streams.Streams._increment_stream_id")
-    def test_add_historical_stream(self, mock_increment, mock_order_stream_class):
+    def test_add_historical_stream_conflate(
+        self, mock_increment, mock_order_stream_class
+    ):
         conflate_ms = 500
         streaming_timeout = 0.5
         mock_client = mock.Mock()
@@ -205,6 +213,7 @@ class TestBaseStream(unittest.TestCase):
             {"test": "me"},
             {"please": "now"},
             client=self.mock_client,
+            output_queue=False,
         )
 
     def test_init(self):
@@ -217,6 +226,7 @@ class TestBaseStream(unittest.TestCase):
         self.assertIsNone(self.stream._stream)
         self.assertEqual(self.stream._client, self.mock_client)
         self.assertEqual(self.stream.MAX_LATENCY, 0.5)
+        self.assertIsNone(self.stream._output_queue)
 
     def test_run(self):
         with self.assertRaises(NotImplementedError):
@@ -371,7 +381,14 @@ class TestHistoricalStream(unittest.TestCase):
     def setUp(self) -> None:
         self.mock_flumine = mock.Mock()
         self.stream = streams.HistoricalStream(
-            self.mock_flumine, 123, 0.01, 100, {"test": "me"}, {"please": "now"}
+            self.mock_flumine,
+            123,
+            0.01,
+            100,
+            {"test": "me"},
+            {"please": "now"},
+            inplay=True,
+            seconds_to_start=123,
         )
 
     def test_init(self):
@@ -383,6 +400,8 @@ class TestHistoricalStream(unittest.TestCase):
         self.assertEqual(self.stream.conflate_ms, 100)
         self.assertIsNone(self.stream._stream)
         self.assertIsNone(self.stream.MAX_LATENCY)
+        self.assertTrue(self.stream._listener.inplay)
+        self.assertEqual(self.stream._listener.seconds_to_start, 123)
 
     def test_run(self):
         self.stream.run()
@@ -397,6 +416,98 @@ class TestHistoricalStream(unittest.TestCase):
         )
 
 
+class TestStream(unittest.TestCase):
+    def setUp(self) -> None:
+        self.listener = mock.Mock()
+        self.stream = historicalstream.Stream(self.listener)
+
+    def test_init(self):
+        self.assertEqual(self.stream._listener, self.listener)
+        self.assertEqual(self.stream._lookup, "mc")
+
+    def test__process(self):
+        self.stream._process(
+            [{"id": "1.23", "img": {1: 2}, "marketDefinition": {"runners": []}}], 12345
+        )
+        self.assertEqual(len(self.stream._caches), 1)
+
+    def test_snap_inplay(self):
+        # inPlay
+        self.stream = historicalstream.Stream(
+            mock.Mock(inplay=True, seconds_to_start=None)
+        )
+        self.stream._caches = {
+            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": False})
+        }
+        self.assertEqual(len(self.stream.snap()), 0)
+        self.stream._caches = {
+            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": True})
+        }
+        self.assertEqual(len(self.stream.snap()), 1)
+        self.stream._caches = {
+            "1.123": mock.Mock(market_definition={"status": "CLOSED", "inPlay": False})
+        }
+        self.assertEqual(len(self.stream.snap()), 1)
+
+        self.stream = historicalstream.Stream(
+            mock.Mock(inplay=False, seconds_to_start=None)
+        )
+        self.stream._caches = {
+            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": False})
+        }
+        self.assertEqual(len(self.stream.snap()), 1)
+        self.stream._caches = {
+            "1.123": mock.Mock(market_definition={"status": "OPEN", "inPlay": True})
+        }
+        self.assertEqual(len(self.stream.snap()), 0)
+
+    def test_snap_seconds_to_start(self):
+        # secondsToStart
+        self.stream = historicalstream.Stream(
+            mock.Mock(inplay=None, seconds_to_start=600)
+        )
+        self.stream._caches = {
+            "1.123": mock.Mock(
+                publish_time=123,
+                market_definition={
+                    "status": "OPEN",
+                    "inPlay": False,
+                    "marketTime": 456,
+                },
+            )
+        }
+        self.assertEqual(len(self.stream.snap()), 1)
+        self.stream._caches = {
+            "1.123": mock.Mock(
+                publish_time=123,
+                market_definition={
+                    "status": "OPEN",
+                    "inPlay": False,
+                    "marketTime": 1234567,
+                },
+            )
+        }
+        self.assertEqual(len(self.stream.snap()), 0)
+
+
+class TestHistoricListener(unittest.TestCase):
+    def setUp(self) -> None:
+        self.mock_flumine = mock.Mock()
+        self.listener = historicalstream.HistoricListener(
+            inplay=True, seconds_to_start=123
+        )
+
+    def test_init(self):
+        self.assertTrue(self.listener.inplay)
+        self.assertEqual(self.listener.seconds_to_start, 123)
+
+    @mock.patch("flumine.streams.historicalstream.Stream")
+    def test__add_stream(self, mock_stream):
+        self.assertEqual(
+            self.listener._add_stream(123, "marketSubscription"), mock_stream()
+        )
+
+
 class TestOrderStream(unittest.TestCase):
     def setUp(self) -> None:
         self.mock_flumine = mock.Mock()