Improve Cache behavior when adding more recent data

RELEASES.md

@@ -4,6 +4,7 @@ Released on TBD (UTC).
 
 ### Enhancements
 None
+- Improved `Cache` behavior when adding more recent quotes, trades, or bars (now adds to cache)
 
 ### Internal Improvements
 - Improve live engines error logging (will now log all exceptions rather than just `RuntimeError`)

nautilus_trader/cache/cache.pyx

@@ -1223,14 +1223,12 @@ cdef class Cache(CacheFacade):
             # The instrument_id was not registered
             cached_ticks = deque(maxlen=self.tick_capacity)
             self._quote_ticks[instrument_id] = cached_ticks
-        elif len(cached_ticks) > 0:
-            # Currently the simple solution for multiple consumers requesting
-            # ticks at system spool up is just to add only if the cache is empty.
-            self._log.debug("Cache already contains ticks")
-            return
 
         cdef QuoteTick tick
         for tick in ticks:
+            if cached_ticks and tick.ts_event <= cached_ticks[0].ts_event:
+                # Only add more recent data to cache
+                continue
             cached_ticks.appendleft(tick)
 
     cpdef void add_trade_ticks(self, list ticks):
@@ -1257,17 +1255,14 @@ cdef class Cache(CacheFacade):
         cached_ticks = self._trade_ticks.get(instrument_id)
 
         if not cached_ticks:
-            # The instrument_id was not registered
             cached_ticks = deque(maxlen=self.tick_capacity)
             self._trade_ticks[instrument_id] = cached_ticks
-        elif len(cached_ticks) > 0:
-            # Currently the simple solution for multiple consumers requesting
-            # ticks at system spool up is just to add only if the cache is empty.
-            self._log.debug("Cache already contains ticks")
-            return
 
         cdef TradeTick tick
         for tick in ticks:
+            if cached_ticks and tick.ts_event <= cached_ticks[0].ts_event:
+                # Only add more recent data to cache
+                continue
             cached_ticks.appendleft(tick)
 
     cpdef void add_bars(self, list bars):
@@ -1294,17 +1289,14 @@ cdef class Cache(CacheFacade):
         cached_bars = self._bars.get(bar_type)
 
         if not cached_bars:
-            # The instrument_id was not registered
             cached_bars = deque(maxlen=self.bar_capacity)
             self._bars[bar_type] = cached_bars
-        elif len(cached_bars) > 0:
-            # Currently the simple solution for multiple consumers requesting
-            # bars at system spool up is just to add only if the cache is empty.
-            self._log.debug("Cache already contains bars")
-            return
 
         cdef Bar bar
         for bar in bars:
+            if cached_bars and bar.ts_event <= cached_bars[0].ts_event:
+                # Only add more recent data to cache
+                continue
             cached_bars.appendleft(bar)
 
         bar = bars[-1]

nautilus_trader/test_kit/stubs/data.py

@@ -192,16 +192,16 @@ def bartype_adabtc_binance_1min_last() -> BarType:
         return BarType(TestIdStubs.adabtc_binance_id(), TestDataStubs.bar_spec_1min_last())
 
     @staticmethod
-    def bar_5decimal() -> Bar:
+    def bar_5decimal(ts_event=0, ts_init=0) -> Bar:
         return Bar(
             bar_type=TestDataStubs.bartype_audusd_1min_bid(),
             open=Price.from_str("1.00002"),
             high=Price.from_str("1.00004"),
             low=Price.from_str("1.00001"),
             close=Price.from_str("1.00003"),
             volume=Quantity.from_int(1_000_000),
-            ts_event=0,
-            ts_init=0,
+            ts_event=ts_event,
+            ts_init=ts_init,
         )
 
     @staticmethod

tests/unit_tests/cache/test_data.py

@@ -240,7 +240,7 @@ def test_quote_ticks_when_one_tick_returns_expected_list(self):
         # Assert
         assert result == [tick]
 
-    def test_add_quote_ticks_when_already_ticks_does_not_add(self):
+    def test_add_quote_ticks_when_identical_ticks_does_not_add(self):
         # Arrange
         tick = TestDataStubs.quote_tick()
 
@@ -253,6 +253,20 @@ def test_add_quote_ticks_when_already_ticks_does_not_add(self):
         # Assert
         assert result == [tick]
 
+    def test_add_quote_ticks_when_older_quotes(self):
+        # Arrange
+        tick1 = TestDataStubs.quote_tick()
+        self.cache.add_quote_tick(tick1)
+
+        tick2 = TestDataStubs.quote_tick(ts_event=1, ts_init=1)
+
+        # Act
+        self.cache.add_quote_ticks([tick2])
+        result = self.cache.quote_ticks(tick1.instrument_id)
+
+        # Assert
+        assert result == [tick2, tick1]
+
     def test_trade_ticks_when_one_tick_returns_expected_list(self):
         # Arrange
         tick = TestDataStubs.trade_tick()
@@ -265,7 +279,7 @@ def test_trade_ticks_when_one_tick_returns_expected_list(self):
         # Assert
         assert result == [tick]
 
-    def test_add_trade_ticks_when_already_ticks_does_not_add(self):
+    def test_add_trade_ticks_when_identical_ticks_does_not_add(self):
         # Arrange
         tick = TestDataStubs.trade_tick()
 
@@ -278,6 +292,21 @@ def test_add_trade_ticks_when_already_ticks_does_not_add(self):
         # Assert
         assert result == [tick]
 
+    def test_add_trade_ticks_when_older_trades(self):
+        # Arrange
+        tick1 = TestDataStubs.trade_tick()
+        self.cache.add_trade_tick(tick1)
+
+        tick2 = TestDataStubs.trade_tick(ts_event=1, ts_init=1)
+        self.cache.add_trade_tick(tick2)
+
+        # Act
+        self.cache.add_trade_ticks([tick1])
+        result = self.cache.trade_ticks(tick1.instrument_id)
+
+        # Assert
+        assert result == [tick2, tick1]
+
     def test_bars_when_one_bar_returns_expected_list(self):
         # Arrange
         bar = TestDataStubs.bar_5decimal()
@@ -290,7 +319,7 @@ def test_bars_when_one_bar_returns_expected_list(self):
         # Assert
         assert result == [bar]
 
-    def test_add_bars_when_already_bars_does_not_add(self):
+    def test_add_bars_when_already_identical_bar_does_not_add(self):
         # Arrange
         bar = TestDataStubs.bar_5decimal()
 
@@ -303,6 +332,21 @@ def test_add_bars_when_already_bars_does_not_add(self):
         # Assert
         assert result == [bar]
 
+    def test_add_bars_when_older_cached_bars(self):
+        # Arrange
+        bar1 = TestDataStubs.bar_5decimal()
+        self.cache.add_bar(bar1)
+
+        bar2 = TestDataStubs.bar_5decimal(ts_event=1)
+        self.cache.add_bar(bar2)
+
+        # Act
+        self.cache.add_bars([bar2])
+        result = self.cache.bars(bar1.bar_type)
+
+        # Assert
+        assert result == [bar2, bar1]
+
     def test_instrument_when_no_instrument_returns_none(self):
         # Arrange, Act
         result = self.cache.instrument(AUDUSD_SIM.id)