Repository: liampauling/flumine
Pull request #159 (2020-05-13 08:00:27)

process_closed_market added to strategy and called on closure


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 76af82d..081eded 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,20 @@
 Release History
 ---------------
 
+1.5.0 (2020-05-21)
++++++++++++++++++++
+
+**Improvements**
+
+- Logging control added and integrated
+- PriceRecorder example added
+- Balance polling added
+- Cleared Orders/Market polling added
+- Trade.notes added
+- Middleware moved to flumine level
+- SimulatedMiddleware refactored to handle all logic
+- Context added to worker functionality
+
 1.4.0 (2020-05-13)
 +++++++++++++++++++
 
diff --git a/docs/advanced.md b/docs/advanced.md
index 745416f..a97b461 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -79,7 +79,18 @@ _In development_
 
 ## Logging Controls
 
-_In development_
+Custom logging is available using the `LoggingControl` class, the base class creates debug logs and can be used as follows:
+
+```python
+from flumine.controls.loggingcontrols import LoggingControl
+
+control = LoggingControl()
+
+framework.add_logging_control(control)
+```
+
+!!! tip
+    More than one control can be added, for example a csv logger and db logger.
 
 ## Background Workers
 
diff --git a/examples/strategies/lowestlayer.py b/examples/strategies/lowestlayer.py
index cf512f4..d008804 100644
--- a/examples/strategies/lowestlayer.py
+++ b/examples/strategies/lowestlayer.py
@@ -14,11 +14,11 @@ class LowestLayer(BaseStrategy):
     - Order killed after 2 seconds if not matched
     """
 
-    def check_market_book(self, live_market, market_book):
+    def check_market_book(self, market, market_book):
         if market_book.status not in ["CLOSED", "SUSPENDED"] and market_book.inplay:
             return True
 
-    def process_market_book(self, live_market, market_book):
+    def process_market_book(self, market, market_book):
         # get lowest price runner
         prices = [
             (r.selection_id, r.last_price_traded)
@@ -41,7 +41,7 @@ class LowestLayer(BaseStrategy):
                 order = trade.create_order(
                     side="LAY", order_type=LimitOrder(lay, self.context["stake"]),
                 )
-                self.place_order(live_market, order)
+                self.place_order(market, order)
 
     def process_orders(self, market, orders):
         # kill order if unmatched in market for greater than 2 seconds
diff --git a/examples/strategies/pricerecorder.py b/examples/strategies/pricerecorder.py
new file mode 100644
index 0000000..533d15e
--- /dev/null
+++ b/examples/strategies/pricerecorder.py
@@ -0,0 +1,58 @@
+import os
+import csv
+
+from flumine import BaseStrategy
+from flumine.utils import get_price
+
+HEADERS = [
+    "market_id",
+    "publish_time",
+    "status",
+    "inplay",
+    "selection_id",
+    "last_price_traded",
+    "back",
+    "lay",
+]
+
+
+class PriceRecorder(BaseStrategy):
+    """
+    Example strategy for recording prices
+    from historical or live data.
+    """
+
+    def __init__(self, *args, **kwargs):
+        BaseStrategy.__init__(self, *args, **kwargs)
+        self.local_dir = self.context.get("local_dir", "/tmp")
+        self.filename = self.context.get("filename", "output.txt")
+        self.file_directory = os.path.join(self.local_dir, self.filename)
+
+    def add(self) -> None:
+        # check local dir
+        if not os.path.isdir(self.local_dir):
+            raise OSError("File dir %s does not exist" % self.local_dir)
+        # write headers
+        with open(self.file_directory, "w") as f:
+            writer = csv.DictWriter(f, fieldnames=HEADERS)
+            writer.writeheader()
+
+    def check_market_book(self, live_market, market_book):
+        return True
+
+    def process_market_book(self, live_market, market_book):
+        with open(self.file_directory, "a") as f:
+            writer = csv.DictWriter(f, fieldnames=HEADERS)
+            for runner in market_book.runners:
+                writer.writerow(
+                    {
+                        "market_id": market_book.market_id,
+                        "publish_time": market_book.publish_time,
+                        "status": market_book.status,
+                        "inplay": market_book.inplay,
+                        "selection_id": runner.selection_id,
+                        "last_price_traded": runner.last_price_traded,
+                        "back": get_price(runner.ex.available_to_back, 0),
+                        "lay": get_price(runner.ex.available_to_lay, 0),
+                    }
+                )
diff --git a/examples/workers/__init__.py b/examples/workers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/flumine/__version__.py b/flumine/__version__.py
index 5838118..84895d8 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.4.0"
+__version__ = "1.5.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index e30ec4a..55800c2 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -10,11 +10,13 @@ from .worker import BackgroundWorker
 from .clients.baseclient import BaseClient
 from .markets.markets import Markets
 from .markets.market import Market
+from .markets.middleware import Middleware
 from .execution.betfairexecution import BetfairExecution
 from .execution.simulatedexecution import SimulatedExecution
 from .order.process import process_current_orders
 from .controls.clientcontrols import BaseControl, MaxOrderCount
 from .controls.tradingcontrols import OrderValidation, StrategyExposure
+from .controls.loggingcontrols import LoggingControl
 from . import config
 
 
@@ -36,9 +38,11 @@ class BaseFlumine:
 
         # queues
         self.handler_queue = queue.Queue()
+        self.cleared_market_queue = queue.Queue()
 
         # all markets
         self.markets = Markets()
+        self._market_middleware = []
 
         # all strategies
         self.strategies = Strategies()
@@ -69,9 +73,9 @@ class BaseFlumine:
         raise NotImplementedError
 
     def add_strategy(self, strategy: BaseStrategy) -> None:
-        # create stream if required
         self.streams(strategy)  # create required streams
         self.strategies(strategy)  # store in strategies
+        self.log_control(events.StrategyEvent(strategy))
 
     def add_worker(self, worker: BackgroundWorker) -> None:
         self._workers.append(worker)
@@ -84,6 +88,18 @@ class BaseFlumine:
         logger.info("Adding trading control {0}".format(trading_control.NAME))
         self._trading_controls.append(trading_control(self, **kwargs))
 
+    def add_market_middleware(self, middleware: Middleware) -> None:
+        logger.info("Adding market middleware {0}".format(middleware))
+        self._market_middleware.append(middleware)
+
+    def add_logging_control(self, logging_control: LoggingControl) -> None:
+        logger.info("Adding logging control {0}".format(logging_control.NAME))
+        self._logging_controls.append(logging_control)
+
+    def log_control(self, event: events.BaseEvent) -> None:
+        for logging_control in self._logging_controls:
+            logging_control.logging_queue.put(event)
+
     def _add_default_workers(self) -> None:
         return
 
@@ -95,13 +111,16 @@ class BaseFlumine:
                 continue
 
             market = self.markets.markets.get(market_id)
-
-            if not market:
+            if market is None:
                 market = self._add_live_market(market_id, market_book)
 
             # process market
             market(market_book)
 
+            # process middleware
+            for middleware in self._market_middleware:
+                middleware(market)  # todo err handling?
+
             for strategy in self.strategies:
                 if strategy.check_market(market, market_book):
                     strategy.process_market_book(market, market_book)
@@ -128,7 +147,7 @@ class BaseFlumine:
     def _add_live_market(
         self, market_id: str, market_book: resources.MarketBook
     ) -> Market:
-        market = Market(market_id, market_book)
+        market = Market(self, market_id, market_book)
         self.markets.add_market(market_id, market)
         logger.info("Adding: {0} to markets".format(market.market_id))
         return market
@@ -145,10 +164,10 @@ class BaseFlumine:
             market = self.markets.markets.get(market_catalogue.market_id)
             if market:
                 if market.market_catalogue is None:
+                    self.log_control(events.MarketEvent(market))
                     logger.info(
                         "Updated marketCatalogue for {0}".format(market.market_id)
                     )
-                    # todo logging control
                 market.market_catalogue = market_catalogue
 
     def _process_current_orders(self, event: events.CurrentOrdersEvent) -> None:
@@ -160,14 +179,47 @@ class BaseFlumine:
                     strategy.process_orders(market, strategy_orders)
             self._process_market_orders(market)
 
+    def _process_custom_event(self, event: events.CustomEvent) -> None:
+        try:
+            event.callback(self, event)
+        except Exception as e:
+            logger.exception(
+                "Unknown error {0} in _process_custom_event {1}".format(
+                    e, event.callback
+                )
+            )
+        for market in self.markets:
+            self._process_market_orders(market)
+
     def _process_close_market(self, event: events.CloseMarketEvent) -> None:
         market = self.markets.markets.get(event.event.market_id)
         market.close_market()
         market.blotter.process_closed_market(event.event)
 
-        # todo update balance
-        # todo get cleared orders
-        # todo log closed market event
+        for strategy in self.strategies:
+            if strategy.check_market(market, event.event):
+                strategy.process_closed_market(market, event.event)
+
+        self.cleared_market_queue.put(market.market_id)
+        self.log_control(event)
+
+    def _process_cleared_orders(self, event):
+        # todo update blotter?
+        logger.info(
+            "Market closed and cleared", extra={"market_id": event.event.market_id},
+        )
+
+    def _process_cleared_markets(self, event: events.ClearedMarketsEvent):
+        # todo update blotter?
+        for cleared_market in event.event.orders:
+            logger.info(
+                "Market level cleared",
+                extra={
+                    "market_id": cleared_market.market_id,
+                    "profit": cleared_market.profit,
+                    "bet_count": cleared_market.bet_count,
+                },
+            )
 
     def _process_end_flumine(self) -> None:
         for strategy in self.strategies:
@@ -206,7 +258,7 @@ class BaseFlumine:
         self.simulated_execution.shutdown()
         self.betfair_execution.shutdown()
         # shutdown logging controls
-        # todo self.log_control(events.EventType.TERMINATOR)
+        self.log_control(events.TerminationEvent(None))
         for c in self._logging_controls:
             if c.is_alive():
                 c.join()
diff --git a/flumine/controls/loggingcontrols.py b/flumine/controls/loggingcontrols.py
new file mode 100644
index 0000000..ddf2789
--- /dev/null
+++ b/flumine/controls/loggingcontrols.py
@@ -0,0 +1,116 @@
+import logging
+import queue
+from threading import Thread
+
+from .. import config
+from ..events import events
+from ..events.events import EventType
+
+logger = logging.getLogger(__name__)
+
+
+class LoggingControl(Thread):
+    """
+    Base Logging Control instance, used to output logging of
+    trades, orders, new markets and market closure. Can be
+    used to send data to a database or API for further analysis.
+    """
+
+    # todo handle/store errors for retry
+
+    NAME = "LOGGING_CONTROL"
+
+    def __init__(self, daemon: bool = True):
+        Thread.__init__(self, daemon=daemon, name=self.NAME)
+        self.logging_queue = queue.Queue()
+        self.cache = []
+
+    def run(self) -> None:
+        logger.info("Starting logging control %s" % self.NAME)
+        while True:
+            event = self.logging_queue.get()
+            if event is None:
+                logger.info("Shutting down logging control %s" % self.NAME)
+                break
+            else:
+                self.process_event(event)
+
+    def process_config(self, config: config):
+        logger.debug("process_config: %s" % config)
+
+    def process_event(self, event: events.BaseEvent):
+        if event.EVENT_TYPE == EventType.STRATEGY:
+            self._process_strategy(event)
+
+        elif event.EVENT_TYPE == EventType.MARKET:
+            self._process_market(event)
+
+        elif event.EVENT_TYPE == EventType.TRADE:
+            self._process_trade(event)
+
+        elif event.EVENT_TYPE == EventType.ORDER:
+            self._process_order(event)
+
+        elif event.EVENT_TYPE == EventType.BALANCE:
+            self._process_balance(event)
+
+        elif event.EVENT_TYPE == EventType.CLEARED_ORDERS:
+            self._process_cleared_orders(event)
+
+        elif event.EVENT_TYPE == EventType.CLEARED_ORDERS_META:
+            self._process_cleared_orders_meta(event)  # todo
+
+        elif event.EVENT_TYPE == EventType.CLEARED_MARKETS:
+            self._process_cleared_markets(event)
+
+        elif event.EVENT_TYPE == EventType.CLOSE_MARKET:
+            self._process_closed_market(event)
+
+        elif event.EVENT_TYPE == EventType.CUSTOM_EVENT:
+            self._process_custom_event(event)
+
+        elif event.EVENT_TYPE == EventType.NEW_DAY:
+            self._process_new_day(event)
+
+        elif event.EVENT_TYPE == EventType.TERMINATOR:
+            self._process_end_flumine(event)
+            self.logging_queue.put(None)
+
+        else:
+            logger.error("Unwanted item in logging control: {0}".format(event))
+
+    def _process_strategy(self, event):
+        logger.debug("process_strategy: %s" % event)
+
+    def _process_market(self, event):
+        logger.debug("process_market: %s" % event)
+
+    def _process_trade(self, event):
+        logger.debug("process_trade: %s" % event)
+
+    def _process_order(self, event):
+        logger.debug("process_order: %s" % event)
+
+    def _process_balance(self, event):
+        logger.debug("process_balance: %s" % event)
+
+    def _process_cleared_orders(self, event):
+        logger.debug("process_cleared_orders: %s" % event)
+
+    def _process_cleared_orders_meta(self, event):
+        logger.debug("process_cleared_orders_meta: %s" % event)
+
+    def _process_cleared_markets(self, event):
+        logger.debug("process_cleared_markets: %s" % event)
+
+    def _process_closed_market(self, event):
+        logger.debug("process_closed_market: %s" % event)
+
+    def _process_custom_event(self, event):
+        logger.debug("process_custom_event: %s" % event)
+
+    def _process_new_day(self, event):
+        logger.debug("process_new_day: %s" % event)
+
+    def _process_end_flumine(self, event):
+        logger.debug("_process_end_flumine: %s" % event)
diff --git a/flumine/events/events.py b/flumine/events/events.py
index cc0b5c5..f728c00 100644
--- a/flumine/events/events.py
+++ b/flumine/events/events.py
@@ -11,17 +11,21 @@ class EventType(Enum):
     CURRENT_ORDERS = "CurrentOrders"
     CLEARED_MARKETS = "ClearedMarkets"
     CLEARED_ORDERS = "ClearedOrders"
+    CLEARED_ORDERS_META = "ClearedOrders metadata"
+    BALANCE = "Balance"
     # flumine objects
+    STRATEGY = "Strategy"
+    MARKET = "Market"
+    TRADE = "Trade"
+    ORDER = "Order"
     ORDER_PACKAGE = "Order package"
     CLOSE_MARKET = "Closed market"
-    STRATEGY_RESET = "Strategy reset"
     CUSTOM_EVENT = "Custom event"
     NEW_DAY = "New day"
 
 
 class QueueType(Enum):
     HANDLER = "Handler queue"
-    ACCOUNT = "Account queue"
     LOGGING = "Logging queue"
 
 
@@ -41,6 +45,9 @@ class BaseEvent:  # todo __slots__?
         return "<{0} [{1}]>".format(self.EVENT_TYPE.name, self.QUEUE_TYPE.name)
 
 
+# HANDLER
+
+
 class MarketCatalogueEvent(BaseEvent):
     EVENT_TYPE = EventType.MARKET_CATALOGUE
     QUEUE_TYPE = QueueType.HANDLER
@@ -76,16 +83,56 @@ class CloseMarketEvent(BaseEvent):
     QUEUE_TYPE = QueueType.HANDLER
 
 
-class StrategyResetEvent(BaseEvent):
-    EVENT_TYPE = EventType.STRATEGY_RESET
-    QUEUE_TYPE = QueueType.HANDLER
-
-
 class CustomEvent(BaseEvent):
     EVENT_TYPE = EventType.CUSTOM_EVENT
     QUEUE_TYPE = QueueType.HANDLER
 
+    def __init__(self, event, callback, *args, **kwargs):
+        super(CustomEvent, self).__init__(event)
+        self.callback = callback
+
 
 class NewDayEvent(BaseEvent):
     EVENT_TYPE = EventType.NEW_DAY
     QUEUE_TYPE = QueueType.HANDLER
+
+
+# LOGGING
+
+
+class ClearedOrdersMetaEvent(BaseEvent):
+    EVENT_TYPE = EventType.CLEARED_ORDERS_META
+    QUEUE_TYPE = QueueType.LOGGING
+
+
+class BalanceEvent(BaseEvent):
+    EVENT_TYPE = EventType.BALANCE
+    QUEUE_TYPE = QueueType.LOGGING
+
+
+class StrategyEvent(BaseEvent):
+    EVENT_TYPE = EventType.STRATEGY
+    QUEUE_TYPE = QueueType.LOGGING
+
+
+class MarketEvent(BaseEvent):
+    EVENT_TYPE = EventType.MARKET
+    QUEUE_TYPE = QueueType.LOGGING
+
+
+class TradeEvent(BaseEvent):
+    EVENT_TYPE = EventType.TRADE
+    QUEUE_TYPE = QueueType.LOGGING
+
+
+class OrderEvent(BaseEvent):
+    EVENT_TYPE = EventType.ORDER
+    QUEUE_TYPE = QueueType.LOGGING
+
+
+# both
+
+
+class TerminationEvent(BaseEvent):
+    EVENT_TYPE = EventType.TERMINATOR
+    QUEUE_TYPE = QueueType.HANDLER
diff --git a/flumine/flumine.py b/flumine/flumine.py
index a15f67b..d053d74 100644
--- a/flumine/flumine.py
+++ b/flumine/flumine.py
@@ -15,7 +15,7 @@ class Flumine(BaseFlumine):
         with self:
             while True:
                 event = self.handler_queue.get()
-                if event == EventType.TERMINATOR:
+                if event.EVENT_TYPE == EventType.TERMINATOR:
                     self._process_end_flumine()
                     break
 
@@ -35,19 +35,16 @@ class Flumine(BaseFlumine):
                     self._process_order_package(event)
 
                 elif event.EVENT_TYPE == EventType.CLEARED_MARKETS:
-                    logger.info(event)
+                    self._process_cleared_markets(event)
 
                 elif event.EVENT_TYPE == EventType.CLEARED_ORDERS:
-                    logger.info(event)
+                    self._process_cleared_orders(event)
 
                 elif event.EVENT_TYPE == EventType.CLOSE_MARKET:
                     self._process_close_market(event)
 
-                elif event.EVENT_TYPE == EventType.STRATEGY_RESET:
-                    logger.info(event)
-
                 elif event.EVENT_TYPE == EventType.CUSTOM_EVENT:
-                    logger.info(event)
+                    self._process_custom_event(event)
 
                 elif event.EVENT_TYPE == EventType.NEW_DAY:
                     logger.info(event)
@@ -64,6 +61,15 @@ class Flumine(BaseFlumine):
                 func_args=(self.client,),
             )
         )
+        self.add_worker(
+            worker.BackgroundWorker(
+                start_delay=5,  # wait for login
+                interval=60,
+                function=worker.poll_account_balance,
+                name="poll_account_balance",
+                func_args=(self, self.client),
+            )
+        )
         self.add_worker(
             worker.BackgroundWorker(
                 start_delay=5,  # wait for streams to populate
@@ -73,6 +79,14 @@ class Flumine(BaseFlumine):
                 func_args=(self.client, self.markets, self.handler_queue),
             )
         )
+        self.add_worker(
+            worker.BackgroundWorker(
+                interval=10,  # restart
+                function=worker.poll_cleared_orders,
+                name="poll_cleared_orders",
+                func_args=(self, self.client),
+            )
+        )
 
     def __repr__(self) -> str:
         return "<Flumine>"
diff --git a/flumine/markets/market.py b/flumine/markets/market.py
index ad2cd7b..ef3a419 100644
--- a/flumine/markets/market.py
+++ b/flumine/markets/market.py
@@ -3,9 +3,7 @@ import logging
 from betfairlightweight.resources.bettingresources import MarketBook, MarketCatalogue
 
 from .blotter import Blotter
-from .middleware import SimulatedMiddleware
-from .. import config
-from ..order.order import OrderStatus
+from ..events import events
 
 logger = logging.getLogger(__name__)
 
@@ -13,33 +11,21 @@ logger = logging.getLogger(__name__)
 class Market:
     def __init__(
         self,
+        flumine,
         market_id: str,
         market_book: MarketBook,
         market_catalogue: MarketCatalogue = None,
     ):
+        self.flumine = flumine
         self.market_id = market_id
         self.closed = False
         self.market_book = market_book
         self.market_catalogue = market_catalogue
-        self._simulated_middleware = SimulatedMiddleware()
-        self._middleware = []
+        self.context = {"simulated": {}}  # data store (raceCard / scores etc)
         self.blotter = Blotter(self)
 
-        if config.simulated:
-            self._middleware.append(self._simulated_middleware)
-
     def __call__(self, market_book: MarketBook):
         self.market_book = market_book
-        # process middleware
-        for middleware in self._middleware:
-            middleware(self.market_catalogue, self.market_book)  # todo error handling?
-        # process simulated orders
-        for order in self.blotter:
-            if order.simulated and order.status == OrderStatus.EXECUTABLE:
-                runner_analytics = self._simulated_middleware.runners.get(
-                    (order.selection_id, order.handicap)
-                )
-                order.simulated(self.market_book, runner_analytics)
 
     def open_market(self) -> None:
         self.closed = False
@@ -52,7 +38,8 @@ class Market:
         order.place()
         if order.id not in self.blotter:
             self.blotter[order.id] = order
-            # todo log trade?
+            self.flumine.log_control(events.TradeEvent(order.trade))  # todo dupes?
+            self.flumine.log_control(events.OrderEvent(order))
         else:
             return  # retry attempt so ignore?
         if execute:  # handles replaceOrder
@@ -70,6 +57,11 @@ class Market:
         order.replace(new_price)
         self.blotter.pending_replace.append(order)
 
+    @property
+    def event_id(self) -> str:
+        if self.market_book:
+            return self.market_book.market_definition.event_id
+
     @property
     def seconds_to_start(self):
         return (self.market_start_datetime - datetime.datetime.utcnow()).total_seconds()
diff --git a/flumine/markets/middleware.py b/flumine/markets/middleware.py
index 88cf218..fb01511 100644
--- a/flumine/markets/middleware.py
+++ b/flumine/markets/middleware.py
@@ -1,21 +1,14 @@
-import datetime
 import logging
-from betfairlightweight.resources.bettingresources import (
-    MarketBook,
-    MarketCatalogue,
-    RunnerBook,
-)
+from collections import defaultdict
+from betfairlightweight.resources.bettingresources import MarketBook, RunnerBook
 
-from .blotter import Blotter
-from .. import config
+from ..order.order import OrderStatus
 
 logger = logging.getLogger(__name__)
 
 
 class Middleware:
-    def __call__(
-        self, market_catalogue: MarketCatalogue, market_book: MarketBook
-    ) -> None:
+    def __call__(self, market) -> None:
         raise NotImplementedError
 
 
@@ -28,20 +21,33 @@ class SimulatedMiddleware:
     """
 
     def __init__(self):
-        self.runners = {}  # {(selectionId, handicap): RunnerAnalytics}
+        # {marketId: {(selectionId, handicap): RunnerAnalytics}}
+        self.markets = defaultdict(dict)
 
-    def __call__(
-        self, market_catalogue: MarketCatalogue, market_book: MarketBook
-    ) -> None:
-        for runner in market_book.runners:
+    def __call__(self, market) -> None:
+        market_analytics = self.markets[market.market_id]
+        for runner in market.market_book.runners:
             if runner.status == "ACTIVE":
-                self._process_runner(runner)
+                self._process_runner(market_analytics, runner)
+        market.context["simulated"] = market_analytics
+        # process simulated orders
+        self._process_simulated_orders(market, market_analytics)
 
-    def _process_runner(self, runner: RunnerBook) -> None:
+    @staticmethod
+    def _process_simulated_orders(market, market_analytics: dict) -> None:
+        for order in market.blotter:
+            if order.simulated and order.status == OrderStatus.EXECUTABLE:
+                runner_analytics = market_analytics.get(
+                    (order.selection_id, order.handicap)
+                )
+                order.simulated(market.market_book, runner_analytics)
+
+    @staticmethod
+    def _process_runner(market_analytics: dict, runner: RunnerBook) -> None:
         try:
-            runner_analytics = self.runners[(runner.selection_id, runner.handicap)]
+            runner_analytics = market_analytics[(runner.selection_id, runner.handicap)]
         except KeyError:
-            runner_analytics = self.runners[
+            runner_analytics = market_analytics[
                 (runner.selection_id, runner.handicap)
             ] = RunnerAnalytics(runner)
         runner_analytics(runner)
@@ -56,6 +62,7 @@ class RunnerAnalytics:
     def __call__(self, runner: RunnerBook):
         self.traded = self._calculate_traded(runner)
         self._traded_volume = runner.ex.traded_volume
+        self._runner = runner
 
     def _calculate_traded(self, runner: RunnerBook) -> dict:
         if self._traded_volume == {}:
diff --git a/flumine/order/trade.py b/flumine/order/trade.py
index c291d23..79e2000 100644
--- a/flumine/order/trade.py
+++ b/flumine/order/trade.py
@@ -1,5 +1,6 @@
 import uuid
 import logging
+import collections
 from enum import Enum
 from typing import Union, Type
 from betfairlightweight.resources.bettingresources import CurrentOrder
@@ -24,6 +25,7 @@ class Trade:
         selection_id: int,
         handicap: float,
         strategy: BaseStrategy,
+        notes: collections.OrderedDict = None,
         fill_kill=None,
         offset=None,
         green=None,
@@ -34,6 +36,9 @@ class Trade:
         self.selection_id = selection_id
         self.handicap = handicap
         self.strategy = strategy
+        self.notes = (
+            notes if notes else collections.OrderedDict()
+        )  # trade notes (e.g. triggers/market state)
         self.fill_kill = fill_kill
         self.offset = offset
         self.green = green
@@ -108,6 +113,10 @@ class Trade:
         self.orders.append(order)
         return order
 
+    @property
+    def notes_str(self) -> str:
+        return ",".join(str(x) for x in self.notes.values())
+
     @property
     def info(self) -> dict:
         return {
@@ -115,4 +124,5 @@ class Trade:
             "strategy": self.strategy,
             "status": self.status,
             "orders": [o.id for o in self.orders],
+            "notes": self.notes_str,
         }
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index 17bba5a..7f41ade 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -95,6 +95,10 @@ class BaseStrategy:
         # process list of Order objects for strategy and Market
         return
 
+    def process_closed_market(self, market: Market, market_book: MarketBook) -> None:
+        # process marketBook after closure
+        return
+
     def finish(self) -> None:
         # called before flumine ends
         return
diff --git a/flumine/worker.py b/flumine/worker.py
index ef1f283..4c1a859 100644
--- a/flumine/worker.py
+++ b/flumine/worker.py
@@ -4,7 +4,8 @@ import logging
 import queue
 from betfairlightweight import BetfairError, filters
 
-from .events.events import MarketCatalogueEvent
+from . import config
+from .events import events
 from .utils import chunks
 
 logger = logging.getLogger(__name__)
@@ -18,6 +19,7 @@ class BackgroundWorker(threading.Thread):
         func_args: tuple = None,
         func_kwargs: dict = None,
         start_delay: int = 0,
+        context: dict = None,
         **kwargs
     ):
         threading.Thread.__init__(self, daemon=True, **kwargs)
@@ -26,25 +28,42 @@ class BackgroundWorker(threading.Thread):
         self.func_args = func_args if func_args is not None else []
         self.func_kwargs = func_kwargs if func_kwargs is not None else {}
         self.start_delay = start_delay
+        self.context = context or {}
 
     def run(self) -> None:
         time.sleep(self.start_delay)
+        logger.info(
+            "BackgroundWorker {0} starting".format(self.name),
+            extra={
+                "worker_name": self.name,
+                "function": self.function,
+                "context": self.context,
+            },
+        )
         while self.is_alive():
             logger.debug(
                 "BackgroundWorker {0} executing".format(self.name),
-                extra={"name": self.name, "function": self.function},
+                extra={
+                    "worker_name": self.name,
+                    "function": self.function,
+                    "context": self.context,
+                },
             )
             try:
-                self.function(*self.func_args, **self.func_kwargs)
+                self.function(self.context, *self.func_args, **self.func_kwargs)
             except Exception as e:
                 logger.error(
                     "Error in BackgroundWorker {0}: {1}".format(self.name, e),
-                    extra={"worker_name": self.name, "function": self.function},
+                    extra={
+                        "worker_name": self.name,
+                        "function": self.function,
+                        "context": self.context,
+                    },
                 )
             time.sleep(self.interval)
 
 
-def keep_alive(client) -> None:
+def keep_alive(context: dict, client) -> None:
     """ Attempt keep alive if required or
     login if keep alive failed
     """
@@ -70,7 +89,9 @@ def keep_alive(client) -> None:
         )
 
 
-def poll_market_catalogue(client, markets, handler_queue: queue.Queue) -> None:
+def poll_market_catalogue(
+    context: dict, client, markets, handler_queue: queue.Queue
+) -> None:
     live_markets = list(markets.markets.keys())
     for market_ids in chunks(live_markets, 100):
         try:
@@ -96,4 +117,90 @@ def poll_market_catalogue(client, markets, handler_queue: queue.Queue) -> None:
             continue
 
         if market_catalogues:
-            handler_queue.put(MarketCatalogueEvent(market_catalogues))
+            handler_queue.put(events.MarketCatalogueEvent(market_catalogues))
+
+
+def poll_account_balance(context: dict, flumine, client) -> None:
+    client.update_account_details()
+    if client.account_funds:
+        flumine.log_control(events.BalanceEvent(client.account_funds))
+
+
+def poll_cleared_orders(context: dict, flumine, client) -> None:
+    processed_orders, processed_markets, market_cache = [], [], []
+    while True:
+        market_id = flumine.cleared_market_queue.get()
+        if market_id not in processed_orders:
+            if _get_cleared_orders(flumine, client.betting_client, market_id):
+                processed_orders.append(market_id)
+            else:
+                time.sleep(10)
+                flumine.cleared_market_queue.put(market_id)  # try again
+
+        if market_id not in processed_markets:
+            time.sleep(32)  # takes ~30s for orders to be aggregated to market level
+            if _get_cleared_market(flumine, client.betting_client, market_id):
+                processed_markets.append(market_id)
+            else:
+                if market_cache.count(market_id) > 7:  # give up after 7 attempts
+                    market_cache = [m for m in market_cache if m != market_id]
+                else:
+                    market_cache.append(market_id)
+                    flumine.cleared_market_queue.put(market_id)  # try again
+
+
+def _get_cleared_orders(flumine, betting_client, market_id: str) -> bool:
+    from_record = 0
+    while True:
+        try:
+            cleared_orders = betting_client.betting.list_cleared_orders(
+                bet_status="SETTLED",
+                from_record=from_record,
+                market_ids=[market_id],
+                customer_strategy_refs=[config.hostname],
+            )
+        except BetfairError as e:
+            logger.error(
+                "poll_cleared_orders error",
+                extra={"trading_function": "list_cleared_orders", "response": e},
+                exc_info=True,
+            )
+            return False
+
+        logger.info(
+            "{0}: {1} cleared orders found, more available: {2}".format(
+                market_id, len(cleared_orders.orders), cleared_orders.more_available
+            )
+        )
+        cleared_orders.market_id = market_id
+        flumine.handler_queue.put(events.ClearedOrdersEvent(cleared_orders))
+        flumine.log_control(events.ClearedOrdersEvent(cleared_orders))
+
+        from_record += len(cleared_orders.orders)
+        if not cleared_orders.more_available:
+            break
+    return True
+
+
+def _get_cleared_market(flumine, betting_client, market_id: str) -> bool:
+    try:
+        cleared_markets = betting_client.betting.list_cleared_orders(
+            bet_status="SETTLED",
+            market_ids=[market_id],
+            customer_strategy_refs=[config.hostname],
+            group_by="MARKET",
+        )
+    except BetfairError as e:
+        logger.error(
+            "_get_cleared_markets error",
+            extra={"trading_function": "list_cleared_orders", "response": e},
+            exc_info=True,
+        )
+        return False
+
+    if cleared_markets.orders:
+        flumine.log_control(events.ClearedMarketsEvent(cleared_markets))
+        flumine.handler_queue.put(events.ClearedMarketsEvent(cleared_markets))
+        return True
+    else:
+        return False
diff --git a/setup.py b/setup.py
index 0c6296e..3b36ce3 100644
--- a/setup.py
+++ b/setup.py
@@ -13,7 +13,9 @@ with open(os.path.join(here, "flumine", "__version__.py"), "r") as f:
 setup(
     name=about["__title__"],
     version=about["__version__"],
-    packages=find_packages(exclude=["examples", "examples.strategies"]),
+    packages=find_packages(
+        exclude=["examples", "examples.strategies", "examples.workers"]
+    ),
     package_dir={"flumine": "flumine"},
     install_requires=INSTALL_REQUIRES,
     url=about["__url__"],


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index b369610..223304d 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -10,6 +10,7 @@ from .utils import PendingPackages
 from ..markets.market import Market
 from ..order.orderpackage import OrderPackageType
 from ..order import process
+from ..markets.middleware import SimulatedMiddleware
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +27,7 @@ class FlumineBacktest(BaseFlumine):
     def __init__(self, client):
         super(FlumineBacktest, self).__init__(client)
         self._pending_packages = PendingPackages()
+        self._market_middleware.append(SimulatedMiddleware())
 
     def run(self) -> None:
         if self.client.EXCHANGE != ExchangeType.SIMULATED:
@@ -60,6 +62,7 @@ class FlumineBacktest(BaseFlumine):
             self._unpatch_datetime()
 
     def _process_market_books(self, event: events.MarketBookEvent) -> None:
+        # todo DRY!
         for market_book in event.event:
             market_id = market_book.market_id
             config.current_time = market_book.publish_time
@@ -72,12 +75,17 @@ class FlumineBacktest(BaseFlumine):
                 continue
 
             market = self.markets.markets.get(market_id)
-            if not market:
+            if market is None:
                 market = self._add_live_market(market_id, market_book)
+                self.log_control(events.MarketEvent(market))
 
             # process market
             market(market_book)
 
+            # process middleware
+            for middleware in self._market_middleware:
+                middleware(market)  # todo err handling?
+
             for strategy in self.strategies:
                 if strategy.check_market(market, market_book):
                     strategy.process_market_book(market, market_book)
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index 1f2262d..7322daa 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -13,6 +13,7 @@ class BaseFlumineTest(unittest.TestCase):
         self.assertFalse(self.base_flumine.BACKTEST)
         self.assertEqual(self.base_flumine.client, self.mock_client)
         self.assertFalse(self.base_flumine._running)
+        self.assertEqual(self.base_flumine._market_middleware, [])
         self.assertEqual(self.base_flumine._logging_controls, [])
         self.assertEqual(len(self.base_flumine._trading_controls), 2)
         self.assertEqual(self.base_flumine._workers, [])
@@ -21,11 +22,14 @@ class BaseFlumineTest(unittest.TestCase):
         with self.assertRaises(NotImplementedError):
             self.base_flumine.run()
 
-    def test_add_strategy(self):
+    @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()
         self.base_flumine.add_strategy(mock_strategy)
         self.assertEqual(len(self.base_flumine.strategies), 1)
         self.assertEqual(len(self.base_flumine.streams), 1)
+        mock_log_control.assert_called_with(mock_events.StrategyEvent(mock_strategy))
 
     def test_add_worker(self):
         mock_worker = mock.Mock()
@@ -46,6 +50,23 @@ class BaseFlumineTest(unittest.TestCase):
         self.base_flumine.add_trading_control(mock_control)
         self.assertEqual(len(self.base_flumine._trading_controls), 3)
 
+    def test_add_market_middleware(self):
+        mock_middleware = mock.Mock()
+        self.base_flumine.add_market_middleware(mock_middleware)
+        self.assertEqual(len(self.base_flumine._market_middleware), 1)
+
+    def test_add_logging_control(self):
+        mock_control = mock.Mock()
+        self.base_flumine.add_logging_control(mock_control)
+        self.assertEqual(len(self.base_flumine._logging_controls), 1)
+
+    def test_log_control(self):
+        mock_control = mock.Mock()
+        self.base_flumine._logging_controls.append(mock_control)
+        mock_event = mock.Mock()
+        self.base_flumine.log_control(mock_event)
+        mock_control.logging_queue.put.assert_called_with(mock_event)
+
     def test__add_default_workers(self):
         self.base_flumine._add_default_workers()
         self.assertEqual(len(self.base_flumine._workers), 0)
@@ -117,6 +138,22 @@ class BaseFlumineTest(unittest.TestCase):
         mock_event.event = (12, 12345, {})
         self.base_flumine._process_raw_data(mock_event)
 
+    @mock.patch("flumine.baseflumine.events")
+    @mock.patch("flumine.baseflumine.BaseFlumine.log_control")
+    def test__process_market_catalogue(self, mock_log_control, mock_events):
+        mock_market = mock.Mock()
+        mock_market.market_catalogue = None
+        mock_markets = mock.Mock()
+        mock_markets.markets = {"1.23": mock_market}
+        self.base_flumine.markets = mock_markets
+        mock_market_catalogue = mock.Mock()
+        mock_market_catalogue.market_id = "1.23"
+        mock_event = mock.Mock()
+        mock_event.event = [mock_market_catalogue]
+        self.base_flumine._process_market_catalogues(mock_event)
+        self.assertEqual(mock_market.market_catalogue, mock_market_catalogue)
+        mock_log_control.assert_called_with(mock_events.MarketEvent(mock_market))
+
     def test__process_current_orders(self):
         mock_event = mock.Mock()
         mock_current_orders = mock.Mock()
@@ -124,7 +161,19 @@ class BaseFlumineTest(unittest.TestCase):
         mock_event.event = [mock_current_orders]
         self.base_flumine._process_current_orders(mock_event)
 
-    def test__process_close_market(self):
+    @mock.patch("flumine.baseflumine.BaseFlumine._process_market_orders")
+    def test__process_custom_event(self, mock__process_market_orders):
+        mock_market = mock.Mock()
+        self.base_flumine.markets = [mock_market]
+        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.patch("flumine.baseflumine.BaseFlumine.log_control")
+    def test__process_close_market(self, mock_log_control):
+        mock_strategy = mock.Mock()
+        self.base_flumine.strategies = [mock_strategy]
         mock_market = mock.Mock()
         mock_markets = mock.Mock()
         mock_markets.markets = {"1.23": mock_market}
@@ -135,11 +184,31 @@ class BaseFlumineTest(unittest.TestCase):
         self.base_flumine._process_close_market(mock_event)
         mock_market.close_market.assert_called_with()
         mock_market.blotter.process_closed_market.assert_called_with(mock_market_book)
+        mock_strategy.check_market.assert_called_with(mock_market, mock_market_book)
+        mock_strategy.process_closed_market.assert_called_with(
+            mock_market, mock_market_book
+        )
+        mock_log_control.assert_called_with(mock_event)
+        self.assertEqual(
+            self.base_flumine.cleared_market_queue.get(), mock_market.market_id
+        )
+
+    def test__process_cleared_orders(self):
+        mock_event = mock.Mock()
+        mock_event.event.orders = []
+        self.base_flumine._process_cleared_orders(mock_event)
+
+    def test__process_cleared_markets(self):
+        mock_event = mock.Mock()
+        mock_event.event.orders = []
+        self.base_flumine._process_cleared_markets(mock_event)
 
     def test__process_end_flumine(self):
         self.base_flumine._process_end_flumine()
 
-    def test_enter_exit(self):
+    @mock.patch("flumine.baseflumine.events")
+    @mock.patch("flumine.baseflumine.BaseFlumine.log_control")
+    def test_enter_exit(self, mock_log_control, mock_events):
         control = mock.Mock()
         self.base_flumine._logging_controls = [control]
         self.base_flumine.simulated_execution = mock.Mock()
@@ -153,3 +222,4 @@ class BaseFlumineTest(unittest.TestCase):
         self.base_flumine.simulated_execution.shutdown.assert_called_with()
         self.base_flumine.betfair_execution.shutdown.assert_called_with()
         control.start.assert_called_with()
+        mock_log_control.assert_called_with(mock_events.TerminationEvent(None))
diff --git a/tests/test_flumine.py b/tests/test_flumine.py
index b110702..8ec419e 100644
--- a/tests/test_flumine.py
+++ b/tests/test_flumine.py
@@ -11,6 +11,9 @@ class FlumineTest(unittest.TestCase):
         self.mock_trading = mock.Mock()
         self.flumine = Flumine(self.mock_trading)
 
+    @mock.patch("flumine.flumine.Flumine._process_custom_event")
+    @mock.patch("flumine.flumine.Flumine._process_cleared_orders")
+    @mock.patch("flumine.flumine.Flumine._process_cleared_markets")
     @mock.patch("flumine.flumine.Flumine._process_close_market")
     @mock.patch("flumine.flumine.Flumine._process_order_package")
     @mock.patch("flumine.flumine.Flumine._process_current_orders")
@@ -27,6 +30,9 @@ class FlumineTest(unittest.TestCase):
         mock__process_current_orders,
         mock__process_order_package,
         mock__process_close_market,
+        mock__process_cleared_markets,
+        mock__process_cleared_orders,
+        mock__process_custom_event,
     ):
         mock_events = [
             events.MarketCatalogueEvent(None),
@@ -37,10 +43,9 @@ class FlumineTest(unittest.TestCase):
             events.ClearedMarketsEvent(None),
             events.ClearedOrdersEvent(None),
             events.CloseMarketEvent(None),
-            events.StrategyResetEvent(None),
-            events.CustomEvent(None),
+            events.CustomEvent(None, None),
             events.NewDayEvent(None),
-            events.EventType.TERMINATOR,
+            events.TerminationEvent(None),
         ]
         for i in mock_events:
             self.flumine.handler_queue.put(i)
@@ -53,10 +58,13 @@ class FlumineTest(unittest.TestCase):
         mock__process_current_orders.assert_called_with(mock_events[3])
         mock__process_order_package.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])
 
     def test__add_default_workers(self):
         self.flumine._add_default_workers()
-        self.assertEqual(len(self.flumine._workers), 2)
+        self.assertEqual(len(self.flumine._workers), 4)
 
     def test_str(self):
         assert str(self.flumine) == "<Flumine>"
diff --git a/tests/test_loggingcontrols.py b/tests/test_loggingcontrols.py
new file mode 100644
index 0000000..c3a9c01
--- /dev/null
+++ b/tests/test_loggingcontrols.py
@@ -0,0 +1,151 @@
+import unittest
+from unittest import mock
+from queue import Queue
+
+from flumine.controls.loggingcontrols import LoggingControl, EventType
+
+
+class TestLoggingControl(unittest.TestCase):
+    def setUp(self):
+        self.logging_control = LoggingControl()
+
+    def test_init(self):
+        self.assertIsInstance(self.logging_control.logging_queue, Queue)
+        self.assertEqual(self.logging_control.cache, [])
+        self.assertEqual(self.logging_control.NAME, "LOGGING_CONTROL")
+
+    def test_run(self):
+        self.logging_control.logging_queue.put(None)
+        self.logging_control.run()
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_strategy")
+    def test_process_event_strategy(self, mock_process_strategy):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.STRATEGY
+        self.logging_control.process_event(mock_event)
+        mock_process_strategy.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_market")
+    def test_process_event_market(self, _process_market):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.MARKET
+        self.logging_control.process_event(mock_event)
+        _process_market.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_trade")
+    def test_process_event_trade(self, _process_trade):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.TRADE
+        self.logging_control.process_event(mock_event)
+        _process_trade.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_order")
+    def test_process_event_order(self, _process_order):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.ORDER
+        self.logging_control.process_event(mock_event)
+        _process_order.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_balance")
+    def test_process_event_balance(self, _process_balance):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.BALANCE
+        self.logging_control.process_event(mock_event)
+        _process_balance.assert_called_with(mock_event)
+
+    @mock.patch(
+        "flumine.controls.loggingcontrols.LoggingControl._process_cleared_orders"
+    )
+    def test_process_event_cleared(self, _process_cleared_orders):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.CLEARED_ORDERS
+        self.logging_control.process_event(mock_event)
+        _process_cleared_orders.assert_called_with(mock_event)
+
+    @mock.patch(
+        "flumine.controls.loggingcontrols.LoggingControl._process_cleared_orders_meta"
+    )
+    def test_process_event_cleared_orders_meta(self, _process_cleared_orders_meta):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.CLEARED_ORDERS_META
+        self.logging_control.process_event(mock_event)
+        _process_cleared_orders_meta.assert_called_with(mock_event)
+
+    @mock.patch(
+        "flumine.controls.loggingcontrols.LoggingControl._process_cleared_markets"
+    )
+    def test_process_event_cleared_markets(self, _process_cleared_markets):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.CLEARED_MARKETS
+        self.logging_control.process_event(mock_event)
+        _process_cleared_markets.assert_called_with(mock_event)
+
+    @mock.patch(
+        "flumine.controls.loggingcontrols.LoggingControl._process_closed_market"
+    )
+    def test_process_event_closed(self, _closed_market):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.CLOSE_MARKET
+        self.logging_control.process_event(mock_event)
+        _closed_market.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_custom_event")
+    def test_process_event_custom_event(self, _closed_market):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.CUSTOM_EVENT
+        self.logging_control.process_event(mock_event)
+        _closed_market.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_new_day")
+    def test_process_event_process_day(self, _end_flumine):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.NEW_DAY
+        self.logging_control.process_event(mock_event)
+        _end_flumine.assert_called_with(mock_event)
+
+    @mock.patch("flumine.controls.loggingcontrols.LoggingControl._process_end_flumine")
+    def test_process_event_end(self, _end_flumine):
+        mock_event = mock.Mock()
+        mock_event.EVENT_TYPE = EventType.TERMINATOR
+        self.logging_control.process_event(mock_event)
+        _end_flumine.assert_called_with(mock_event)
+        self.assertIsNone(self.logging_control.logging_queue.get())
+
+    def test_process_config(self):
+        self.logging_control.process_config(None)
+
+    def test_process_strategy(self):
+        self.logging_control._process_strategy(None)
+
+    def test_process_market(self):
+        self.logging_control._process_market(None)
+
+    def test_process_trade(self):
+        self.logging_control._process_trade(None)
+
+    def test_process_order(self):
+        self.logging_control._process_order(None)
+
+    def test_process_balance(self):
+        self.logging_control._process_balance(None)
+
+    def test_process_cleared_orders(self):
+        self.logging_control._process_cleared_orders(None)
+
+    def test_process_cleared_orders_meta(self):
+        self.logging_control._process_cleared_orders_meta(None)
+
+    def test_process_cleared_markets(self):
+        self.logging_control._process_cleared_markets(None)
+
+    def test_process_closed_market(self):
+        self.logging_control._process_closed_market(None)
+
+    def test_process_custom_event(self):
+        self.logging_control._process_custom_event(None)
+
+    def test_process_new_day(self):
+        self.logging_control._process_new_day(None)
+
+    def test_process_end_flumine(self):
+        self.logging_control._process_end_flumine(None)
diff --git a/tests/test_markets.py b/tests/test_markets.py
index 33d2c23..4e15c0e 100644
--- a/tests/test_markets.py
+++ b/tests/test_markets.py
@@ -3,7 +3,7 @@ import datetime
 from unittest import mock
 
 from flumine.markets.markets import Markets
-from flumine.markets.market import Market, OrderStatus
+from flumine.markets.market import Market
 
 
 class MarketsTest(unittest.TestCase):
@@ -87,34 +87,28 @@ class MarketsTest(unittest.TestCase):
 
 class MarketTest(unittest.TestCase):
     def setUp(self) -> None:
+        self.mock_flumine = mock.Mock()
         self.mock_market_book = mock.Mock()
         self.mock_market_catalogue = mock.Mock()
-        self.market = Market("1.234", self.mock_market_book, self.mock_market_catalogue)
+        self.market = Market(
+            self.mock_flumine,
+            "1.234",
+            self.mock_market_book,
+            self.mock_market_catalogue,
+        )
 
     def test_init(self):
+        self.assertEqual(self.market.flumine, self.mock_flumine)
         self.assertEqual(self.market.market_id, "1.234")
         self.assertFalse(self.market.closed)
         self.assertEqual(self.market.market_book, self.mock_market_book)
         self.assertEqual(self.market.market_catalogue, self.mock_market_catalogue)
-        self.assertEqual(self.market._middleware, [])
+        self.assertEqual(self.market.context, {"simulated": {}})
 
     def test_call(self):
-        mock_middleware = mock.Mock()
-        self.market._middleware = [mock_middleware]
-        mock_order = mock.Mock()
-        mock_order.status = OrderStatus.EXECUTABLE
-        mock_order_two = mock.Mock()
-        mock_order_two.status = OrderStatus.PENDING
-        mock_order_three = mock.Mock()
-        mock_order_three.status = OrderStatus.EXECUTABLE
-        mock_order_three.simulated = False
-        self.market.blotter = [mock_order, mock_order_two, mock_order_three]
         mock_market_book = mock.Mock()
         self.market(mock_market_book)
         self.assertEqual(self.market.market_book, mock_market_book)
-        mock_middleware.assert_called_with(self.mock_market_catalogue, mock_market_book)
-        mock_order.simulated.assert_called_with(mock_market_book, None)
-        mock_order_two.simulated.assert_not_called()
 
     def test_open_market(self):
         self.market.open_market()
@@ -124,19 +118,23 @@ class MarketTest(unittest.TestCase):
         self.market.close_market()
         self.assertTrue(self.market.closed)
 
-    def test_place_order(self):
+    @mock.patch("flumine.markets.market.events")
+    def test_place_order(self, mock_events):
         mock_order = mock.Mock()
         mock_order.id = "123"
         self.market.place_order(mock_order)
         mock_order.place.assert_called_with()
         self.assertEqual(self.market.blotter.pending_place, [mock_order])
+        self.mock_flumine.log_control.assert_called_with(mock_events.OrderEvent())
 
-    def test_place_order_not_executed(self):
+    @mock.patch("flumine.markets.market.events")
+    def test_place_order_not_executed(self, mock_events):
         mock_order = mock.Mock()
         mock_order.id = "123"
         self.market.place_order(mock_order, execute=False)
         mock_order.place.assert_called_with()
         self.assertEqual(self.market.blotter.pending_place, [])
+        self.mock_flumine.log_control.assert_called_with(mock_events.OrderEvent())
 
     def test_place_order_retry(self):
         mock_order = mock.Mock()
@@ -171,6 +169,13 @@ class MarketTest(unittest.TestCase):
         mock_order.replace.assert_called_with(1.01)
         self.assertEqual(mock_blotter.pending_replace, [mock_order])
 
+    def test_event_id(self):
+        mock_market_book = mock.Mock()
+        self.market.market_book = mock_market_book
+        self.assertEqual(
+            self.market.event_id, mock_market_book.market_definition.event_id
+        )
+
     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 9388ad9..fdd417a 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -1,7 +1,12 @@
 import unittest
 from unittest import mock
 
-from flumine.markets.middleware import Middleware, SimulatedMiddleware, RunnerAnalytics
+from flumine.markets.middleware import (
+    Middleware,
+    SimulatedMiddleware,
+    RunnerAnalytics,
+    OrderStatus,
+)
 
 
 class MiddlewareTest(unittest.TestCase):
@@ -10,7 +15,7 @@ class MiddlewareTest(unittest.TestCase):
 
     def test_call(self):
         with self.assertRaises(NotImplementedError):
-            self.middleware(None, None)
+            self.middleware(None)
 
 
 class SimulatedMiddlewareTest(unittest.TestCase):
@@ -18,25 +23,49 @@ class SimulatedMiddlewareTest(unittest.TestCase):
         self.middleware = SimulatedMiddleware()
 
     def test_init(self):
-        self.assertEqual(self.middleware.runners, {})
+        self.assertEqual(self.middleware.markets, {})
 
+    @mock.patch(
+        "flumine.markets.middleware.SimulatedMiddleware._process_simulated_orders"
+    )
     @mock.patch("flumine.markets.middleware.SimulatedMiddleware._process_runner")
-    def test_call(self, mock__process_runner):
-        mock_market_catalogue = mock.Mock()
+    def test_call(self, mock__process_runner, mock__process_simulated_orders):
+        mock_market = mock.Mock(context={})
         mock_market_book = mock.Mock()
         mock_runner = mock.Mock()
         mock_runner.status = "ACTIVE"
         mock_market_book.runners = [mock_runner]
-        self.middleware(mock_market_catalogue, mock_market_book)
-        mock__process_runner.assert_called_with(mock_runner)
+        mock_market.market_book = mock_market_book
+        self.middleware(mock_market)
+        mock__process_runner.assert_called_with({}, mock_runner)
+        self.assertEqual(mock_market.context, {"simulated": {}})
+        mock__process_simulated_orders.assert_called_with(mock_market, {})
+
+    def test__process_simulated_orders(self):
+        mock_market_book = mock.Mock()
+        mock_order = mock.Mock()
+        mock_order.status = OrderStatus.EXECUTABLE
+        mock_market = mock.Mock()
+        mock_order_two = mock.Mock()
+        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_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)
+        mock_order.simulated.assert_called_with(mock_market_book, "test")
+        mock_order_two.simulated.assert_not_called()
 
     @mock.patch("flumine.markets.middleware.RunnerAnalytics")
     def test__process_runner(self, mock_runner_analytics):
+        market_analytics = {}
         mock_runner = mock.Mock()
-        self.middleware._process_runner(mock_runner)
-        self.assertEqual(len(self.middleware.runners), 1)
-        self.middleware._process_runner(mock_runner)
-        self.assertEqual(len(self.middleware.runners), 1)
+        self.middleware._process_runner(market_analytics, mock_runner)
+        self.assertEqual(len(market_analytics), 1)
+        self.middleware._process_runner(market_analytics, mock_runner)
+        self.assertEqual(len(market_analytics), 1)
         mock_runner_analytics.assert_called_with(mock_runner)
 
 
@@ -59,6 +88,7 @@ class RunnerAnalyticsTest(unittest.TestCase):
             self.runner_analytics._traded_volume, mock_runner.ex.traded_volume
         )
         self.assertEqual(self.runner_analytics.traded, mock__calculate_traded())
+        self.assertEqual(self.runner_analytics._runner, mock_runner)
 
     def test__calculate_traded_dict_empty(self):
         mock_runner = mock.Mock()
diff --git a/tests/test_strategy.py b/tests/test_strategy.py
index 165cf39..763363a 100644
--- a/tests/test_strategy.py
+++ b/tests/test_strategy.py
@@ -122,6 +122,9 @@ class BaseStrategyTest(unittest.TestCase):
     def test_process_orders(self):
         self.strategy.process_orders(None, None)
 
+    def test_process_closed_market(self):
+        self.strategy.process_closed_market(None, None)
+
     def test_finish(self):
         self.strategy.finish()
 
diff --git a/tests/test_trade.py b/tests/test_trade.py
index aa2d7eb..9226763 100644
--- a/tests/test_trade.py
+++ b/tests/test_trade.py
@@ -1,3 +1,4 @@
+import collections
 import unittest
 from unittest import mock
 
@@ -11,11 +12,13 @@ class TradeTest(unittest.TestCase):
         self.mock_offset = mock.Mock()
         self.mock_green = mock.Mock()
         self.mock_stop = mock.Mock()
+        self.notes = collections.OrderedDict({"trigger": 123})
         self.trade = Trade(
             "1.234",
             567,
             1.0,
             self.mock_strategy,
+            self.notes,
             self.mock_fill_kill,
             self.mock_offset,
             self.mock_green,
@@ -27,6 +30,7 @@ class TradeTest(unittest.TestCase):
         self.assertEqual(self.trade.selection_id, 567)
         self.assertEqual(self.trade.handicap, 1.0)
         self.assertEqual(self.trade.strategy, self.mock_strategy)
+        self.assertEqual(self.trade.notes, self.notes)
         self.assertEqual(self.trade.fill_kill, self.mock_fill_kill)
         self.assertEqual(self.trade.offset, self.mock_offset)
         self.assertEqual(self.trade.green, self.mock_green)
@@ -93,6 +97,12 @@ class TradeTest(unittest.TestCase):
         with self.assertRaises(NotImplementedError):
             self.trade.create_order_from_current(mock_current_order, "12345")
 
+    def test_notes_str(self):
+        self.trade.notes = collections.OrderedDict({"1": 1, 2: "2", 3: 3, 4: "four"})
+        # self.assertEqual(self.trade.notes_str, "1,2,3,four")
+        self.trade.notes = collections.OrderedDict()
+        self.assertEqual(self.trade.notes_str, "")
+
     def test_info(self):
         self.assertEqual(
             self.trade.info,
@@ -101,5 +111,6 @@ class TradeTest(unittest.TestCase):
                 "orders": [],
                 "status": TradeStatus.LIVE,
                 "strategy": self.mock_strategy,
+                "notes": "123",
             },
         )
diff --git a/tests/test_worker.py b/tests/test_worker.py
index a684ca5..480622d 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -1,3 +1,4 @@
+import logging
 import unittest
 from unittest import mock
 from betfairlightweight import BetfairError
@@ -9,7 +10,7 @@ class BackgroundWorkerTest(unittest.TestCase):
     def setUp(self):
         self.mock_function = mock.Mock()
         self.worker = worker.BackgroundWorker(
-            123, self.mock_function, (1, 2), {"hello": "world"}, 5
+            123, self.mock_function, (1, 2), {"hello": "world"}, 5, {1: 2}
         )
 
     def test_init(self):
@@ -18,55 +19,66 @@ class BackgroundWorkerTest(unittest.TestCase):
         self.assertEqual(self.worker.func_args, (1, 2))
         self.assertEqual(self.worker.func_kwargs, {"hello": "world"})
         self.assertEqual(self.worker.start_delay, 5)
+        self.assertEqual(self.worker.context, {1: 2})
 
     # def test_run(self):
     #     self.worker.run()
 
 
 class WorkersTest(unittest.TestCase):
+    def setUp(self) -> None:
+        logging.disable(logging.CRITICAL)
+
     def test_keep_alive(self):
+        mock_context = mock.Mock()
         mock_client = mock.Mock()
         mock_client.betting_client.session_token = None
-        worker.keep_alive(mock_client)
+        worker.keep_alive(mock_context, mock_client)
         mock_client.login.assert_called_with()
 
         mock_client.betting_client.session_token = 1
         mock_client.betting_client.session_expired = True
-        worker.keep_alive(mock_client)
+        worker.keep_alive(mock_context, mock_client)
         mock_client.keep_alive.assert_called_with()
 
     def test_keep_alive_failure(self):
+        mock_context = mock.Mock()
         mock_client = mock.Mock()
         mock_client.betting_client.session_token = None
         mock_response = mock.Mock()
         mock_response.status = "FAILURE"
         mock_client.betting_client.keep_alive.return_value = mock_response
-        worker.keep_alive(mock_client)
+        worker.keep_alive(mock_context, mock_client)
         mock_client.login.assert_called_with()
 
     def test_keep_alive_error(self):
+        mock_context = mock.Mock()
         mock_client = mock.Mock()
         mock_client.betting_client.session_token = None
         mock_client.betting_client.keep_alive.side_effect = BetfairError
-        worker.keep_alive(mock_client)
+        worker.keep_alive(mock_context, mock_client)
         mock_client.login.assert_called_with()
 
     def test_keep_alive_ka_error(self):
+        mock_context = mock.Mock()
         mock_client = mock.Mock()
         mock_client.betting_client.session_token = 1
         mock_client.betting_client.session_expired = True
         mock_client.betting_client.keep_alive.side_effect = BetfairError
-        worker.keep_alive(mock_client)
+        worker.keep_alive(mock_context, mock_client)
         mock_client.keep_alive.assert_called_with()
 
-    @mock.patch("flumine.worker.MarketCatalogueEvent")
-    def test_poll_market_catalogue(self, mock_event):
+    @mock.patch("flumine.worker.events")
+    def test_poll_market_catalogue(self, mock_events):
+        mock_context = mock.Mock()
         mock_client = mock.Mock()
         mock_markets = mock.Mock()
         mock_markets.markets = {"1.234": None, "5.678": None}
         mock_handler_queue = mock.Mock()
 
-        worker.poll_market_catalogue(mock_client, mock_markets, mock_handler_queue)
+        worker.poll_market_catalogue(
+            mock_context, mock_client, mock_markets, mock_handler_queue
+        )
         mock_client.betting_client.betting.list_market_catalogue.assert_called_with(
             filter={"marketIds": list(mock_markets.markets.keys())},
             market_projection=[
@@ -80,4 +92,126 @@ class WorkersTest(unittest.TestCase):
             ],
             max_results=100,
         )
-        mock_handler_queue.put.assert_called_with(mock_event())
+        mock_handler_queue.put.assert_called_with(mock_events.MarketCatalogueEvent())
+
+    @mock.patch("flumine.worker.events")
+    def test_poll_account_balance(self, mock_events):
+        mock_context = mock.Mock()
+        mock_client = mock.Mock()
+        mock_client.account_funds = {1: 2}
+        mock_flumine = mock.Mock()
+        worker.poll_account_balance(mock_context, mock_flumine, mock_client)
+        mock_client.update_account_details.assert_called_with()
+        mock_flumine.log_control.assert_called_with(
+            mock_events.BalanceEvent(mock_client.account_funds)
+        )
+
+    # @mock.patch("flumine.worker._get_cleared_market")
+    # @mock.patch("flumine.worker._get_cleared_orders")
+    # def test_poll_cleared_orders(self, mock__get_cleared_orders, mock__get_cleared_market):
+    #     mock_flumine = mock.Mock()
+    #     mock_client = mock.Mock()
+    #
+    #     worker.poll_cleared_orders(mock_flumine, mock_client)
+    #
+
+    @mock.patch("flumine.worker.config")
+    @mock.patch("flumine.worker.events")
+    def test__get_cleared_orders(self, mock_events, mock_config):
+        mock_flumine = mock.Mock()
+        mock_betting_client = mock.Mock()
+        mock_cleared_orders = mock.Mock()
+        mock_cleared_orders.orders = []
+        mock_cleared_orders.more_available = False
+        mock_betting_client.betting.list_cleared_orders.return_value = (
+            mock_cleared_orders
+        )
+
+        self.assertTrue(
+            worker._get_cleared_orders(mock_flumine, mock_betting_client, "1.23")
+        )
+        mock_betting_client.betting.list_cleared_orders.assert_called_with(
+            bet_status="SETTLED",
+            market_ids=["1.23"],
+            from_record=0,
+            customer_strategy_refs=[mock_config.hostname],
+        )
+        mock_flumine.log_control.assert_called_with(mock_events.ClearedOrdersEvent())
+        mock_flumine.handler_queue.put.assert_called_with(
+            mock_events.ClearedOrdersEvent()
+        )
+
+    @mock.patch("flumine.worker.config")
+    @mock.patch("flumine.worker.events")
+    def test__get_cleared_orders_error(self, mock_events, mock_config):
+        mock_flumine = mock.Mock()
+        mock_betting_client = mock.Mock()
+        mock_betting_client.betting.list_cleared_orders.side_effect = BetfairError
+        self.assertFalse(
+            worker._get_cleared_orders(mock_flumine, mock_betting_client, "1.23")
+        )
+        mock_betting_client.betting.list_cleared_orders.assert_called_with(
+            bet_status="SETTLED",
+            market_ids=["1.23"],
+            from_record=0,
+            customer_strategy_refs=[mock_config.hostname],
+        )
+
+    @mock.patch("flumine.worker.config")
+    @mock.patch("flumine.worker.events")
+    def test__get_cleared_market(self, mock_events, mock_config):
+        mock_flumine = mock.Mock()
+        mock_betting_client = mock.Mock()
+        mock_cleared_markets = mock.Mock()
+        mock_cleared_markets.orders = [1]
+        mock_betting_client.betting.list_cleared_orders.return_value = (
+            mock_cleared_markets
+        )
+        self.assertTrue(
+            worker._get_cleared_market(mock_flumine, mock_betting_client, "1.23")
+        )
+        mock_betting_client.betting.list_cleared_orders.assert_called_with(
+            bet_status="SETTLED",
+            market_ids=["1.23"],
+            group_by="MARKET",
+            customer_strategy_refs=[mock_config.hostname],
+        )
+        mock_flumine.log_control.assert_called_with(mock_events.ClearedMarketsEvent())
+        mock_flumine.handler_queue.put.assert_called_with(
+            mock_events.ClearedMarketsEvent()
+        )
+
+    @mock.patch("flumine.worker.config")
+    def test__get_cleared_market_no_orders(self, mock_config):
+        mock_flumine = mock.Mock()
+        mock_betting_client = mock.Mock()
+        mock_cleared_markets = mock.Mock()
+        mock_cleared_markets.orders = []
+        mock_betting_client.betting.list_cleared_orders.return_value = (
+            mock_cleared_markets
+        )
+        self.assertFalse(
+            worker._get_cleared_market(mock_flumine, mock_betting_client, "1.23")
+        )
+        mock_betting_client.betting.list_cleared_orders.assert_called_with(
+            bet_status="SETTLED",
+            market_ids=["1.23"],
+            group_by="MARKET",
+            customer_strategy_refs=[mock_config.hostname],
+        )
+
+    @mock.patch("flumine.worker.config")
+    @mock.patch("flumine.worker.events")
+    def test__get_cleared_market_error(self, mock_events, mock_config):
+        mock_flumine = mock.Mock()
+        mock_betting_client = mock.Mock()
+        mock_betting_client.betting.list_cleared_orders.side_effect = BetfairError
+        self.assertFalse(
+            worker._get_cleared_market(mock_flumine, mock_betting_client, "1.23")
+        )
+        mock_betting_client.betting.list_cleared_orders.assert_called_with(
+            bet_status="SETTLED",
+            market_ids=["1.23"],
+            group_by="MARKET",
+            customer_strategy_refs=[mock_config.hostname],
+        )