Repository: liampauling/flumine
Pull request #162 (2020-05-14 07:52:34)

base logging control added
events updated


Patch:
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/flumine/baseflumine.py b/flumine/baseflumine.py
index 0c064ac..9140677 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -15,6 +15,7 @@ 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,6 +37,7 @@ class BaseFlumine:
 
         # queues
         self.handler_queue = queue.Queue()
+        self.cleared_market_queue = queue.Queue()
 
         # all markets
         self.markets = Markets()
@@ -69,9 +71,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 +86,14 @@ class BaseFlumine:
         logger.info("Adding trading control {0}".format(trading_control.NAME))
         self._trading_controls.append(trading_control(self, **kwargs))
 
+    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
 
@@ -128,7 +138,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 +155,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:
@@ -169,9 +179,26 @@ class BaseFlumine:
             if strategy.check_market(market, event.event):
                 strategy.process_closed_market(market, event.event)
 
-        # todo update balance
-        # todo get cleared orders
-        # todo log closed market 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:
@@ -210,7 +237,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..80599ed 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,11 +83,6 @@ 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
@@ -89,3 +91,44 @@ class CustomEvent(BaseEvent):
 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..b43c20b 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,17 +35,14 @@ 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)
 
@@ -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..f62c884 100644
--- a/flumine/markets/market.py
+++ b/flumine/markets/market.py
@@ -6,6 +6,7 @@ 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,10 +14,12 @@ 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
@@ -52,7 +55,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
diff --git a/flumine/worker.py b/flumine/worker.py
index ef1f283..322747e 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__)
@@ -29,10 +30,14 @@ class BackgroundWorker(threading.Thread):
 
     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},
+        )
         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},
             )
             try:
                 self.function(*self.func_args, **self.func_kwargs)
@@ -96,4 +101,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(flumine, client) -> None:
+    client.update_account_details()
+    if client.account_funds:
+        flumine.log_control(events.BalanceEvent(client.account_funds))
+
+
+def poll_cleared_orders(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


Test patch:
diff --git a/flumine/backtest/backtest.py b/flumine/backtest/backtest.py
index b369610..0322198 100644
--- a/flumine/backtest/backtest.py
+++ b/flumine/backtest/backtest.py
@@ -72,8 +72,9 @@ 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)
diff --git a/tests/test_baseflumine.py b/tests/test_baseflumine.py
index e2d0857..a947ada 100644
--- a/tests/test_baseflumine.py
+++ b/tests/test_baseflumine.py
@@ -21,11 +21,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 +49,18 @@ class BaseFlumineTest(unittest.TestCase):
         self.base_flumine.add_trading_control(mock_control)
         self.assertEqual(len(self.base_flumine._trading_controls), 3)
 
+    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 +132,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 +155,8 @@ 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.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()
@@ -141,11 +173,27 @@ class BaseFlumineTest(unittest.TestCase):
         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()
@@ -159,3 +207,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..1f67c67 100644
--- a/tests/test_flumine.py
+++ b/tests/test_flumine.py
@@ -11,6 +11,8 @@ class FlumineTest(unittest.TestCase):
         self.mock_trading = mock.Mock()
         self.flumine = Flumine(self.mock_trading)
 
+    @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 +29,8 @@ 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_events = [
             events.MarketCatalogueEvent(None),
@@ -37,10 +41,9 @@ class FlumineTest(unittest.TestCase):
             events.ClearedMarketsEvent(None),
             events.ClearedOrdersEvent(None),
             events.CloseMarketEvent(None),
-            events.StrategyResetEvent(None),
             events.CustomEvent(None),
             events.NewDayEvent(None),
-            events.EventType.TERMINATOR,
+            events.TerminationEvent(None),
         ]
         for i in mock_events:
             self.flumine.handler_queue.put(i)
@@ -53,10 +56,12 @@ 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])
 
     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..c416d1a 100644
--- a/tests/test_markets.py
+++ b/tests/test_markets.py
@@ -87,11 +87,18 @@ 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)
@@ -124,19 +131,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()
diff --git a/tests/test_worker.py b/tests/test_worker.py
index a684ca5..0b123d2 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
@@ -24,6 +25,9 @@ class BackgroundWorkerTest(unittest.TestCase):
 
 
 class WorkersTest(unittest.TestCase):
+    def setUp(self) -> None:
+        logging.disable(logging.CRITICAL)
+
     def test_keep_alive(self):
         mock_client = mock.Mock()
         mock_client.betting_client.session_token = None
@@ -59,8 +63,8 @@ class WorkersTest(unittest.TestCase):
         worker.keep_alive(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_client = mock.Mock()
         mock_markets = mock.Mock()
         mock_markets.markets = {"1.234": None, "5.678": None}
@@ -80,4 +84,125 @@ 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_client = mock.Mock()
+        mock_client.account_funds = {1: 2}
+        mock_flumine = mock.Mock()
+        worker.poll_account_balance(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],
+        )