fix: add end_str to aggregate_trade_iter to bound iteration

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__

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:

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

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