From 60f2cd61170b72cc4956c57ccaf7acf81ec86816 Mon Sep 17 00:00:00 2001 From: Tiya Agarwal Date: Thu, 20 Aug 2026 13:51:17 +0530 Subject: [PATCH] fix: add end_str to aggregate_trade_iter to bound iteration aggregate_trade_iter has no way to stop before "now", so collecting trades over a long historical range runs out of memory (#497). Adds an optional end_str parameter, following the same convention as get_historical_klines_generator's start_str/end_str pair: the iterator now stops once it has yielded the last trade at or before end_str. Fixed in both Client and AsyncClient, which share this logic. Co-Authored-By: Claude Sonnet 5 --- binance/async_client.py | 43 ++++++++++++++++++++----- binance/client.py | 52 +++++++++++++++++++++++------- tests/test_async_client.py | 24 ++++++++++++++ tests/test_client.py | 65 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 165 insertions(+), 19 deletions(-) diff --git a/binance/async_client.py b/binance/async_client.py index 1e765c0b..74cbe745 100644 --- a/binance/async_client.py +++ b/binance/async_client.py @@ -400,12 +400,23 @@ async def get_aggregate_trades(self, **params) -> Dict: get_aggregate_trades.__doc__ = Client.get_aggregate_trades.__doc__ - async def aggregate_trade_iter(self, symbol, start_str=None, last_id=None): + async def aggregate_trade_iter( + self, symbol, start_str=None, last_id=None, end_str=None + ): if start_str is not None and last_id is not None: raise ValueError( "start_time and last_id may not be simultaneously specified." ) + end_ts = convert_ts_str(end_str) + + def cutoff(trades): + """Drop any trades at/after end_ts; report whether that happened.""" + if end_ts is None: + return trades, False + kept = [t for t in trades if t[self.AGG_TIME] <= end_ts] + return kept, len(kept) < len(trades) + # If there's no last_id, get one. if last_id is None: # Without a last_id, we actually need the first trade. Normally, @@ -417,23 +428,34 @@ async def aggregate_trade_iter(self, symbol, start_str=None, last_id=None): # or equal than an hour and the result set should contain at # least one trade. start_ts = convert_ts_str(start_str) + if end_ts is not None and start_ts > end_ts: + return # If the resulting set is empty (i.e. no trades in that interval) # then we just move forward hour by hour until we find at least one - # trade or reach present moment + # trade or reach present moment (or end_ts, if given) while True: - end_ts = start_ts + (60 * 60 * 1000) + window_end_ts = start_ts + (60 * 60 * 1000) + if end_ts is not None: + window_end_ts = min(window_end_ts, end_ts) trades = await self.get_aggregate_trades( - symbol=symbol, startTime=start_ts, endTime=end_ts + symbol=symbol, startTime=start_ts, endTime=window_end_ts ) if len(trades) > 0: break - # If we reach present moment and find no trades then there is - # nothing to iterate, so we're done - if end_ts > int(time.time() * 1000): + # If we reach present moment (or end_ts) and find no trades + # then there is nothing to iterate, so we're done + if window_end_ts > int(time.time() * 1000) or ( + end_ts is not None and window_end_ts >= end_ts + ): return - start_ts = end_ts + start_ts = window_end_ts + trades, reached_end = cutoff(trades) + if not trades: + return for t in trades: yield t + if reached_end: + return last_id = trades[-1][self.AGG_ID] while True: @@ -449,8 +471,13 @@ async def aggregate_trade_iter(self, symbol, start_str=None, last_id=None): trades = trades[1:] if len(trades) == 0: return + trades, reached_end = cutoff(trades) + if not trades: + return for t in trades: yield t + if reached_end: + return last_id = trades[-1][self.AGG_ID] aggregate_trade_iter.__doc__ = Client.aggregate_trade_iter.__doc__ diff --git a/binance/client.py b/binance/client.py index 9d631b11..e2d0829c 100755 --- a/binance/client.py +++ b/binance/client.py @@ -601,9 +601,11 @@ def get_aggregate_trades(self, **params) -> Dict: """ return self._get("aggTrades", data=params) - def aggregate_trade_iter(self, symbol: str, start_str=None, last_id=None): + def aggregate_trade_iter( + self, symbol: str, start_str=None, last_id=None, end_str=None + ): """Iterate over aggregate trade data from (start_time or last_id) to - the end of the history so far. + the end of the history so far, or to end_str if specified. If start_time is specified, start with the first trade after start_time. Meant to initialise a local cache of trade data. @@ -626,17 +628,29 @@ def aggregate_trade_iter(self, symbol: str, start_str=None, last_id=None): :type start_str: str|int :param last_id: aggregate trade ID of the last known aggregate trade. Not a regular trade ID. See https://binance-docs.github.io/apidocs/spot/en/#compressed-aggregate-trades-list + :type last_id: int + :param end_str: optional - end date string in UTC format or timestamp in milliseconds. The iterator + will stop after yielding the last trade occurring at or before this time. Defaults to None, which + preserves the previous behaviour of iterating to the end of the history so far. + :type end_str: str|int :returns: an iterator of JSON objects, one per trade. The format of each object is identical to Client.aggregate_trades(). - - :type last_id: int """ if start_str is not None and last_id is not None: raise ValueError( "start_time and last_id may not be simultaneously specified." ) + end_ts = convert_ts_str(end_str) + + def cutoff(trades): + """Drop any trades at/after end_ts; report whether that happened.""" + if end_ts is None: + return trades, False + kept = [t for t in trades if t[self.AGG_TIME] <= end_ts] + return kept, len(kept) < len(trades) + # If there's no last_id, get one. if last_id is None: # Without a last_id, we actually need the first trade. Normally, @@ -648,23 +662,34 @@ def aggregate_trade_iter(self, symbol: str, start_str=None, last_id=None): # or equal than an hour and the result set should contain at # least one trade. start_ts = convert_ts_str(start_str) + if end_ts is not None and start_ts > end_ts: + return # If the resulting set is empty (i.e. no trades in that interval) # then we just move forward hour by hour until we find at least one - # trade or reach present moment + # trade or reach present moment (or end_ts, if given) while True: - end_ts = start_ts + (60 * 60 * 1000) + window_end_ts = start_ts + (60 * 60 * 1000) + if end_ts is not None: + window_end_ts = min(window_end_ts, end_ts) trades = self.get_aggregate_trades( - symbol=symbol, startTime=start_ts, endTime=end_ts + symbol=symbol, startTime=start_ts, endTime=window_end_ts ) if len(trades) > 0: break - # If we reach present moment and find no trades then there is - # nothing to iterate, so we're done - if end_ts > int(time.time() * 1000): + # If we reach present moment (or end_ts) and find no trades + # then there is nothing to iterate, so we're done + if window_end_ts > int(time.time() * 1000) or ( + end_ts is not None and window_end_ts >= end_ts + ): return - start_ts = end_ts + start_ts = window_end_ts + trades, reached_end = cutoff(trades) + if not trades: + return for t in trades: yield t + if reached_end: + return last_id = trades[-1][self.AGG_ID] while True: @@ -680,8 +705,13 @@ def aggregate_trade_iter(self, symbol: str, start_str=None, last_id=None): trades = trades[1:] if len(trades) == 0: return + trades, reached_end = cutoff(trades) + if not trades: + return for t in trades: yield t + if reached_end: + return last_id = trades[-1][self.AGG_ID] def get_ui_klines(self, **params) -> Dict: diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 4e526769..3aec560a 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -1,5 +1,6 @@ import pytest import sys +from unittest.mock import AsyncMock from binance.async_client import AsyncClient from .conftest import proxy, api_key, api_secret, testnet @@ -314,3 +315,26 @@ async def test_handle_response(clientAsync): mock_response._body = b"error message" with pytest.raises(BinanceAPIException): await clientAsync._handle_response(mock_response) + + +async def test_aggregate_trade_iter_end_str_with_last_id_spans_pages_async(): + """Async counterpart of the sync end_str test (see test_client.py) -- + AsyncClient() mirrors Client's pagination logic and had the same + unbounded-iteration bug.""" + client = AsyncClient(api_key, api_secret, {"proxies": {}}, testnet=testnet) + client.get_aggregate_trades = AsyncMock( + side_effect=[ + [{"a": 100, "T": 900}, {"a": 101, "T": 1000}, {"a": 102, "T": 2000}], + [{"a": 102, "T": 2000}, {"a": 103, "T": 2600}, {"a": 104, "T": 3000}], + ] + ) + + result = [ + t + async for t in client.aggregate_trade_iter( + symbol="BTCUSDT", last_id=100, end_str=2500 + ) + ] + + assert [t["a"] for t in result] == [101, 102] + assert client.get_aggregate_trades.await_count == 2 diff --git a/tests/test_client.py b/tests/test_client.py index 0932acd3..8e43b48f 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,4 +1,5 @@ import sys +from unittest.mock import MagicMock import pytest from binance.client import Client from binance.exceptions import BinanceAPIException, BinanceRequestException @@ -281,3 +282,67 @@ def test_handle_response(client): ) with pytest.raises(BinanceAPIException): client._handle_response(mock_error_response) + + +def _no_network_client(): + # aggregate_trade_iter's pagination logic doesn't need a live connection; + # ping=False avoids the network round-trip Client() otherwise makes. + return Client( + api_key, api_secret, {"proxies": proxies}, testnet=testnet, ping=False + ) + + +def test_aggregate_trade_iter_end_str_stops_at_boundary(): + """Regression test for #497: without end_str the iterator has no way to + stop, so a long-running collection can grow unbounded. With end_str set, + it should yield only trades at/before that time and stop without + fetching further pages.""" + client = _no_network_client() + client.get_aggregate_trades = MagicMock( + return_value=[ + {"a": 1, "T": 1000}, + {"a": 2, "T": 2000}, + {"a": 3, "T": 3000}, + ] + ) + + result = list( + client.aggregate_trade_iter(symbol="BTCUSDT", start_str=500, end_str=2000) + ) + + assert [t["a"] for t in result] == [1, 2] + client.get_aggregate_trades.assert_called_once() + + +def test_aggregate_trade_iter_end_str_with_last_id_spans_pages(): + client = _no_network_client() + client.get_aggregate_trades = MagicMock( + side_effect=[ + # first page: fromId=100 echoes id 100, then two new trades + [{"a": 100, "T": 900}, {"a": 101, "T": 1000}, {"a": 102, "T": 2000}], + # second page: fromId=102 echoes id 102, then trades past end_ts + [{"a": 102, "T": 2000}, {"a": 103, "T": 2600}, {"a": 104, "T": 3000}], + ] + ) + + result = list( + client.aggregate_trade_iter(symbol="BTCUSDT", last_id=100, end_str=2500) + ) + + assert [t["a"] for t in result] == [101, 102] + assert client.get_aggregate_trades.call_count == 2 + + +def test_aggregate_trade_iter_without_end_str_keeps_old_behavior(): + client = _no_network_client() + client.get_aggregate_trades = MagicMock( + side_effect=[ + [{"a": 5, "T": 100}, {"a": 6, "T": 200}], + [{"a": 6, "T": 200}], # no new trades after the echoed id -> stop + ] + ) + + result = list(client.aggregate_trade_iter(symbol="BTCUSDT", last_id=5)) + + assert [t["a"] for t in result] == [6] + assert client.get_aggregate_trades.call_count == 2