Continue Binance internal aggregation inference

RELEASES.md

@@ -11,6 +11,7 @@ This will be the final release with support for Python 3.9.
 - Added `Cache.is_order_pending_cancel_local(...)` (tracks local orders in cancel transition)
 - Added `BinanceTimeInForce.GTD` enum member (futures only)
 - Added Binance Futures support for GTD orders
+- Added Binance internal bar aggregation inference from aggregated trade ticks
 - Added `BinanceExecClientConfig.use_gtd` option (to remap to GTC and locally manage GTD orders)
 - Added package version check for `nautilus_ibapi`, thanks @rsmb7z
 - Added `RiskEngine` min/max instrument notional limit checks

nautilus_trader/adapters/binance/common/data.py

@@ -548,9 +548,6 @@ async def _request_bars(  # (too complex)
             )
             return
 
-        if limit == 0 or limit > 1000:
-            limit = 1000
-
         start_time_ms = None
         if start is not None:
             start_time_ms = secs_to_millis(start.timestamp())
@@ -559,7 +556,7 @@ async def _request_bars(  # (too complex)
         if end is not None:
             end_time_ms = secs_to_millis(end.timestamp())
 
-        if bar_type.is_externally_aggregated():
+        if bar_type.is_externally_aggregated() or bar_type.spec.is_time_aggregated():
             if not bar_type.spec.is_time_aggregated():
                 self._log.error(
                     f"Cannot request {bar_type}: only time bars are aggregated by Binance.",
@@ -580,20 +577,27 @@ async def _request_bars(  # (too complex)
                     "not supported.",
                 )
                 return
+
             bars = await self._http_market.request_binance_bars(
                 bar_type=bar_type,
                 interval=interval,
                 start_time=start_time_ms,
                 end_time=end_time_ms,
-                limit=limit,
+                limit=limit if limit > 0 else None,
                 ts_init=self._clock.timestamp_ns(),
             )
+
+            if bar_type.is_internally_aggregated():
+                self._log.info(
+                    "Inferred INTERNAL time bars from EXTERNAL time bars.",
+                    LogColor.BLUE,
+                )
         else:
             bars = await self._aggregate_internal_from_agg_trade_ticks(
                 bar_type=bar_type,
                 start_time_ms=start_time_ms,
                 end_time_ms=end_time_ms,
-                limit=limit,
+                limit=limit if limit > 0 else None,
             )
 
         partial: Bar = bars.pop()
@@ -613,18 +617,18 @@ async def _aggregate_internal_from_agg_trade_ticks(
             )
             return []
 
+        self._log.info("Requesting aggregated trade ticks to infer INTERNAL bars...", LogColor.BLUE)
+
         ticks = await self._http_market.request_agg_trade_ticks(
             instrument_id=instrument.id,
             start_time=start_time_ms,
             end_time=end_time_ms,
             ts_init=self._clock.timestamp_ns(),
+            limit=limit,
         )
 
         bars: list[Bar] = []
-        if bar_type.spec.is_time_aggregated():
-            self._log.error("Internally aggregating historical time bars not yet supported.")
-            return bars
-        elif bar_type.spec.aggregation == BarAggregation.TICK:
+        if bar_type.spec.aggregation == BarAggregation.TICK:
             aggregator = TickBarAggregator(
                 instrument=instrument,
                 bar_type=bar_type,
@@ -656,7 +660,7 @@ async def _aggregate_internal_from_agg_trade_ticks(
             aggregator.handle_trade_tick(tick)
 
         self._log.info(
-            f"Internally aggregated {len(ticks)} external trade ticks to {len(bars)} {bar_type} bars.",
+            f"Inferred {len(bars)} {bar_type} bars aggregated from {len(ticks)} trade ticks.",
             LogColor.BLUE,
         )
 

nautilus_trader/adapters/binance/http/market.py

@@ -14,6 +14,7 @@
 # -------------------------------------------------------------------------------------------------
 
 import sys
+import time
 from typing import Optional
 
 import msgspec
@@ -35,6 +36,7 @@
 from nautilus_trader.adapters.binance.http.client import BinanceHttpClient
 from nautilus_trader.adapters.binance.http.endpoint import BinanceHttpEndpoint
 from nautilus_trader.core.correctness import PyCondition
+from nautilus_trader.core.datetime import nanos_to_millis
 from nautilus_trader.core.nautilus_pyo3.network import HttpMethod
 from nautilus_trader.model.data import BarType
 from nautilus_trader.model.data import OrderBookDeltas
@@ -750,7 +752,7 @@ async def request_agg_trade_ticks(
         self,
         instrument_id: InstrumentId,
         ts_init: int,
-        limit: int = 1000,
+        limit: Optional[int] = 1000,
         start_time: Optional[int] = None,
         end_time: Optional[int] = None,
         from_id: Optional[int] = None,
@@ -765,6 +767,9 @@ async def request_agg_trade_ticks(
         ticks: list[TradeTick] = []
         next_start_time = start_time
 
+        if end_time is None:
+            end_time = sys.maxsize
+
         if from_id is not None and (start_time or end_time) is not None:
             raise RuntimeError(
                 "Cannot specify both fromId and startTime or endTime.",
@@ -806,10 +811,14 @@ def _calculate_next_end_time(start_time: int, end_time: int) -> tuple[int, bool]
                     ),
                 )
 
-            if len(response) < limit and interval_limited is False:
+            if limit and len(response) < limit and interval_limited is False:
                 # end loop regardless when limit is not hit
                 break
-            if start_time is None or end_time is None:
+            if (
+                start_time is None
+                or end_time is None
+                or next_end_time >= nanos_to_millis(time.time_ns())
+            ):
                 break
             else:
                 last = response[-1]

nautilus_trader/examples/strategies/ema_cross.py

@@ -15,6 +15,8 @@
 
 from decimal import Decimal
 
+import pandas as pd
+
 from nautilus_trader.common.enums import LogColor
 from nautilus_trader.config import StrategyConfig
 from nautilus_trader.core.correctness import PyCondition
@@ -129,7 +131,7 @@ def on_start(self) -> None:
         self.register_indicator_for_bars(self.bar_type, self.slow_ema)
 
         # Get historical data
-        self.request_bars(self.bar_type)
+        self.request_bars(self.bar_type, start=self._clock.utc_now() - pd.Timedelta(days=1))
         # self.request_quote_ticks(self.instrument_id)
         # self.request_trade_ticks(self.instrument_id)