diff --git a/freqtrade/exchange/binance.py b/freqtrade/exchange/binance.py index e9c44a41e..18ba2aa50 100644 --- a/freqtrade/exchange/binance.py +++ b/freqtrade/exchange/binance.py @@ -11,7 +11,11 @@ from freqtrade.constants import DEFAULT_DATAFRAME_COLUMNS from freqtrade.enums import CandleType, MarginMode, PriceType, TradingMode from freqtrade.exceptions import DDosProtection, OperationalException, TemporaryError from freqtrade.exchange import Exchange -from freqtrade.exchange.binance_public_data import concat_safe, download_archive_ohlcv +from freqtrade.exchange.binance_public_data import ( + concat_safe, + download_archive_ohlcv, + download_archive_trades, +) from freqtrade.exchange.common import retrier from freqtrade.exchange.exchange_types import FtHas, Tickers from freqtrade.exchange.exchange_utils_timeframe import timeframe_to_msecs @@ -379,3 +383,48 @@ class Binance(Exchange): if not t: return [], "0" return t, from_id + + async def _async_get_trade_history_id( + self, pair: str, until: int, since: int, from_id: str | None = None + ) -> tuple[str, list[list]]: + logger.info(f"Fetching trades from Binance, {from_id=}, {since=}, {until=}") + + if not self._config["exchange"].get("only_from_ccxt", False): + if from_id is None or not since: + trades = await self._api_async.fetch_trades( + pair, + params={ + self._trades_pagination_arg: "0", + }, + limit=5, + ) + listing_date: int = trades[0]["timestamp"] + since = max(since, listing_date) + + _, res = await download_archive_trades( + CandleType.SPOT, + pair, + since_ms=since, + until_ms=until, + markets=self.markets, + ) + + if not res: + end_time = since + end_id = from_id + else: + end_time = res[-1][0] + end_id = res[-1][1] + + if end_time and end_time >= until: + return pair, res + else: + _, res2 = await super()._async_get_trade_history_id( + pair, until=until, since=end_time, from_id=end_id + ) + res.extend(res2) + return pair, res + + return await super()._async_get_trade_history_id( + pair, until=until, since=since, from_id=from_id + ) diff --git a/freqtrade/exchange/binance_public_data.py b/freqtrade/exchange/binance_public_data.py index 42eda09bf..efa376ebc 100644 --- a/freqtrade/exchange/binance_public_data.py +++ b/freqtrade/exchange/binance_public_data.py @@ -1,5 +1,6 @@ """ Fetch daily-archived OHLCV data from https://data.binance.vision/ +Documentation can be found in https://github.com/binance/binance-public-data """ import asyncio @@ -10,9 +11,11 @@ from io import BytesIO from typing import Any import aiohttp +import numpy as np import pandas as pd from pandas import DataFrame +from freqtrade.constants import DEFAULT_TRADES_COLUMNS from freqtrade.enums import CandleType from freqtrade.misc import chunks from freqtrade.util.datetime_helpers import dt_from_ts, dt_now @@ -212,6 +215,20 @@ def binance_vision_ohlcv_zip_url( return url +def binance_vision_trades_zip_url(symbol: str, candle_type: CandleType, date: date) -> str: + """ + example urls: + https://data.binance.vision/data/spot/daily/aggTrades/BTCUSDT/BTCUSDT-aggTrades-2023-10-27.zip + https://data.binance.vision/data/futures/um/daily/aggTrades/BTCUSDT/BTCUSDT-aggTrades-2023-10-27.zip + """ + asset_type_url_segment = candle_type_to_url_segment(candle_type) + url = ( + f"https://data.binance.vision/data/{asset_type_url_segment}/daily/aggTrades/{symbol}" + f"/{symbol}-aggTrades-{date.strftime('%Y-%m-%d')}.zip" + ) + return url + + async def get_daily_ohlcv( symbol: str, timeframe: str, @@ -280,3 +297,203 @@ async def get_daily_ohlcv( if isinstance(e, Http404) or retry > retry_count: logger.debug(f"Failed to get data from {url}: {e}") raise + + +async def download_archive_trades( + candle_type: CandleType, + pair: str, + *, + since_ms: int, + until_ms: int | None, + markets: dict[str, Any], + stop_on_404: bool = True, +) -> tuple[str, list[list]]: + try: + symbol = markets[pair]["id"] + + last_available_date = dt_now() - timedelta(days=2) + + start = dt_from_ts(since_ms) + end = dt_from_ts(until_ms) if until_ms else dt_now() + end = min(end, last_available_date) + if start >= end: + return pair, [] + result_list = await _download_archive_trades( + symbol, pair, candle_type, start, end, stop_on_404 + ) + return pair, result_list + + except Exception as e: + logger.warning( + "An exception occurred during fast trades download from Binance, falling back to " + "the slower REST API, this can take a lot more time.", + exc_info=e, + ) + return pair, [] + + +def parse_trades_from_zip(csvf): + # https://github.com/binance/binance-public-data/issues/283 + first_byte = csvf.read(1)[0] + if chr(first_byte).isdigit(): + # spot + header = None + names = [ + "id", + "price", + "amount", + "first_trade_id", + "last_trade_id", + "timestamp", + "is_buyer_maker", + "is_best_match", + ] + else: + # futures + header = 0 + names = [ + "id", + "price", + "amount", + "first_trade_id", + "last_trade_id", + "timestamp", + "is_buyer_maker", + ] + csvf.seek(0) + + df = pd.read_csv( + csvf, + names=names, + header=header, + ) + df.loc[:, "cost"] = df["price"] * df["amount"] + # Side is reversed intentionally + # based on ccxt parseTrade logic. + df.loc[:, "side"] = np.where(df["is_buyer_maker"], "sell", "buy") + df.loc[:, "type"] = None + # Convert timestamp to ms + df.loc[:, "timestamp"] = np.where( + df["timestamp"] > 10000000000000, + df["timestamp"] // 1000, + df["timestamp"], + ) + return df.loc[:, DEFAULT_TRADES_COLUMNS].to_records(index=False).tolist() + + +async def get_daily_trades( + symbol: str, + candle_type: CandleType, + date: date, + session: aiohttp.ClientSession, + retry_count: int = 3, + retry_delay: float = 0.0, +) -> list[list]: + """ + Get daily OHLCV from https://data.binance.vision + See https://github.com/binance/binance-public-data + + :symbol: binance symbol name, e.g. BTCUSDT + :candle_type: SPOT or FUTURES + :date: the returned DataFrame will cover the entire day of `date` in UTC + :session: an aiohttp.ClientSession instance + :retry_count: times to retry before returning the exceptions + :retry_delay: the time to wait before every retry + :return: a list containing trades in DEFAULT_TRADES_COLUMNS format + """ + + url = binance_vision_trades_zip_url(symbol, candle_type, date) + + logger.debug(f"download trades data from binance: {url}") + + retry = 0 + while True: + if retry > 0: + sleep_secs = retry * retry_delay + logger.debug( + f"[{retry}/{retry_count}] retry to download {url} after {sleep_secs} seconds" + ) + await asyncio.sleep(sleep_secs) + try: + async with session.get(url) as resp: + if resp.status == 200: + content = await resp.read() + logger.debug(f"Successfully downloaded {url}") + with zipfile.ZipFile(BytesIO(content)) as zipf: + with zipf.open(zipf.namelist()[0]) as csvf: + return parse_trades_from_zip(csvf) + elif resp.status == 404: + logger.debug(f"Failed to download {url}") + raise Http404(f"404: {url}", date, url) + else: + raise BadHttpStatus(f"{resp.status} - {resp.reason}") + except Exception as e: + logger.info("download Daily_trades raised: %s", e) + retry += 1 + if isinstance(e, Http404) or retry > retry_count: + logger.debug(f"Failed to get data from {url}: {e}") + raise + + +async def _download_archive_trades( + symbol: str, + pair: str, + candle_type: CandleType, + start: date, + end: date, + stop_on_404: bool, +) -> list[list]: + # daily dataframes, `None` indicates missing data in that day (when `stop_on_404` is False) + results: list[list] = [] + # the current day being processing, starting at 1. + current_day = 0 + + connector = aiohttp.TCPConnector(limit=100) + async with aiohttp.ClientSession(connector=connector, trust_env=True) as session: + # the HTTP connections has been throttled by TCPConnector + for dates in chunks(list(date_range(start, end)), 30): + tasks = [ + asyncio.create_task(get_daily_trades(symbol, candle_type, date, session)) + for date in dates + ] + for task in tasks: + current_day += 1 + try: + result = await task + except Http404 as e: + if stop_on_404: + logger.debug(f"Failed to download {e.url} due to 404.") + + # A 404 error on the first day indicates missing data + # on https://data.binance.vision, we provide the warning and the advice. + # https://github.com/freqtrade/freqtrade/blob/acc53065e5fa7ab5197073276306dc9dc3adbfa3/tests/exchange_online/test_binance_compare_ohlcv.py#L7 + if current_day == 1: + logger.warning( + f"Fast download is unavailable due to missing data: " + f"{e.url}. Falling back to the slower REST API, " + "which may take more time." + ) + if pair in ["BTC/USDT:USDT", "ETH/USDT:USDT", "BCH/USDT:USDT"]: + logger.warning( + f"To avoid the delay, you can first download {pair} using " + "`--timerange -20200101`, and then download the " + "remaining data with `--timerange 20200101-`." + ) + else: + logger.warning( + f"Binance fast download for {pair} stopped at {e.date} due to " + f"missing data: {e.url}, falling back to rest API for the " + "remaining data, this can take more time." + ) + await cancel_and_await_tasks(tasks[tasks.index(task) + 1 :]) + return results + except BaseException as e: + logger.warning(f"An exception raised: : {e}") + # Directly return the existing data, do not allow the gap within the data + await cancel_and_await_tasks(tasks[tasks.index(task) + 1 :]) + return results + else: + # Happy case + results.extend(result) + + return results diff --git a/tests/exchange/test_binance.py b/tests/exchange/test_binance.py index fc683e056..998952b33 100644 --- a/tests/exchange/test_binance.py +++ b/tests/exchange/test_binance.py @@ -6,6 +6,7 @@ import ccxt import pandas as pd import pytest +from freqtrade.data.converter.trade_converter import trades_dict_to_list from freqtrade.enums import CandleType, MarginMode, TradingMode from freqtrade.exceptions import DependencyException, InvalidOrderException, OperationalException from freqtrade.exchange.exchange_utils_timeframe import timeframe_to_seconds @@ -1002,6 +1003,7 @@ def test_get_maintenance_ratio_and_amt_binance( async def test__async_get_trade_history_id_binance(default_conf_usdt, mocker, fetch_trades_result): + default_conf_usdt["exchange"]["only_from_ccxt"] = True exchange = get_patched_exchange(mocker, default_conf_usdt, exchange="binance") async def mock_get_trade_hist(pair, *args, **kwargs): @@ -1056,3 +1058,53 @@ async def test__async_get_trade_history_id_binance(default_conf_usdt, mocker, fe # Clean up event loop to avoid warnings exchange.close() + + +async def test__async_get_trade_history_id_binance_fast( + default_conf_usdt, mocker, fetch_trades_result +): + default_conf_usdt["exchange"]["only_from_ccxt"] = False + exchange = get_patched_exchange(mocker, default_conf_usdt, exchange="binance") + + async def mock_get_trade_hist(pair, *args, **kwargs): + if "since" in kwargs: + pass + # older than initial call + # if kwargs["since"] < 1565798399752: + # return [] + # else: + # # Don't expect to get here + # raise ValueError("Unexpected call") + # # return fetch_trades_result[:-2] + elif kwargs.get("params", {}).get(exchange._trades_pagination_arg) == "0": + # Return first 3 + return fetch_trades_result[:-2] + # elif kwargs.get("params", {}).get(exchange._trades_pagination_arg) in ( + # fetch_trades_result[-3]["id"], + # 1565798399752, + # ): + # # Return 2 + # return fetch_trades_result[-3:-1] + # else: + # # Return last 2 + # return fetch_trades_result[-2:] + + pair = "ETH/BTC" + mocker.patch( + "freqtrade.exchange.binance.download_archive_trades", + return_value=(pair, trades_dict_to_list(fetch_trades_result[-2:])), + ) + + exchange._api_async.fetch_trades = MagicMock(side_effect=mock_get_trade_hist) + + ret = await exchange._async_get_trade_history( + pair, + since=fetch_trades_result[0]["timestamp"], + until=fetch_trades_result[-1]["timestamp"] - 1, + ) + + assert ret[0] == pair + assert isinstance(ret[1], list) + + # Clean up event loop to avoid warnings + exchange.close() diff --git a/tests/exchange/test_binance_public_data.py b/tests/exchange/test_binance_public_data.py index 655a0c874..ab299321b 100644 --- a/tests/exchange/test_binance_public_data.py +++ b/tests/exchange/test_binance_public_data.py @@ -14,11 +14,15 @@ from freqtrade.enums import CandleType from freqtrade.exchange.binance_public_data import ( BadHttpStatus, Http404, + binance_vision_trades_zip_url, binance_vision_zip_name, download_archive_ohlcv, + download_archive_trades, get_daily_ohlcv, + get_daily_trades, ) from freqtrade.util.datetime_helpers import dt_ts, dt_utc +from ft_client.test_client.test_rest_client import log_has_re @pytest.fixture(scope="module") @@ -337,3 +341,156 @@ async def test_get_daily_ohlcv(mocker, testdatadir): with pytest.raises(zipfile.BadZipFile): df = await get_daily_ohlcv(symbol, timeframe, CandleType.SPOT, date, session) assert get.call_count == 4 # 1 + 3 default retries + + +async def test_download_archive_trades(mocker, caplog): + pair = "BTC/USDT" + + since_ms = dt_ts(dt_utc(2020, 1, 1)) + until_ms = dt_ts(dt_utc(2020, 1, 2)) + markets = {"BTC/USDT": {"id": "BTCUSDT"}, "BTC/USDT:USDT": {"id": "BTCUSDT"}} + + mocker.patch("freqtrade.exchange.binance_public_data.get_daily_trades", return_value=[[2, 3]]) + + pair1, res = await download_archive_trades( + CandleType.SPOT, pair, since_ms=since_ms, until_ms=until_ms, markets=markets + ) + assert pair1 == pair + assert res == [[2, 3], [2, 3]] + + mocker.patch( + "freqtrade.exchange.binance_public_data.get_daily_trades", + side_effect=Http404("xxx", dt_utc(2020, 1, 1), "http://example.com/something"), + ) + + pair1, res = await download_archive_trades( + CandleType.SPOT, pair, since_ms=since_ms, until_ms=until_ms, markets=markets + ) + + assert pair1 == pair + assert res == [] + # exit on day 1 + assert log_has_re("Fast download is unavailable", caplog) + + # Test fail on day 2 + caplog.clear() + mocker.patch( + "freqtrade.exchange.binance_public_data.get_daily_trades", + side_effect=[ + [[2, 3]], + [[2, 3]], + Http404("xxx", dt_utc(2020, 1, 2), "http://example.com/something"), + [[2, 3]], + ], + ) + # Download 3 days + until_ms = dt_ts(dt_utc(2020, 1, 3)) + + pair1, res = await download_archive_trades( + CandleType.SPOT, pair, since_ms=since_ms, until_ms=until_ms, markets=markets + ) + + assert pair1 == pair + assert res == [[2, 3], [2, 3]] + assert log_has_re(r"Binance fast download .*stopped", caplog) + + +async def test_download_archive_trades_exception(mocker, caplog): + pair = "BTC/USDT" + + since_ms = dt_ts(dt_utc(2020, 1, 1)) + until_ms = dt_ts(dt_utc(2020, 1, 2)) + + markets = {"BTC/USDT": {"id": "BTCUSDT"}, "BTC/USDT:USDT": {"id": "BTCUSDT"}} + mocker.patch( + "freqtrade.exchange.binance_public_data.aiohttp.ClientSession.get", side_effect=RuntimeError + ) + + pair1, res = await download_archive_trades( + CandleType.SPOT, pair, since_ms=since_ms, until_ms=until_ms, markets=markets + ) + + assert pair1 == pair + assert res == [] + mocker.patch( + "freqtrade.exchange.binance_public_data._download_archive_trades", side_effect=RuntimeError + ) + + await download_archive_trades( + CandleType.SPOT, pair, since_ms=since_ms, until_ms=until_ms, markets=markets + ) + assert pair1 == pair + assert res == [] + assert log_has_re("An exception occurred during fast trades download", caplog) + + +async def test_binance_vision_trades_zip_url(): + url = binance_vision_trades_zip_url("BTCUSDT", CandleType.SPOT, dt_utc(2023, 10, 27)) + assert ( + url == "https://data.binance.vision/data/spot/daily/aggTrades/" + "BTCUSDT/BTCUSDT-aggTrades-2023-10-27.zip" + ) + + url = binance_vision_trades_zip_url("BTCUSDT", CandleType.FUTURES, dt_utc(2023, 10, 28)) + assert ( + url == "https://data.binance.vision/data/futures/um/daily/aggTrades/" + "BTCUSDT/BTCUSDT-aggTrades-2023-10-28.zip" + ) + + +async def test_get_daily_trades(mocker, testdatadir): + symbol = "PEPEUSDT" + symbol_futures = "APEUSDT" + date = dt_utc(2024, 10, 28).date() + first_date = 1729987202368 + last_date = 1730073596350 + + async with aiohttp.ClientSession() as session: + spot_path = ( + testdatadir / "binance/binance_public_data/spot-PEPEUSDT-aggTrades-2024-10-27.zip" + ) + get = mocker.patch( + "freqtrade.exchange.binance_public_data.aiohttp.ClientSession.get", + return_value=MockResponse(spot_path.read_bytes(), 200), + ) + res = await get_daily_trades(symbol, CandleType.SPOT, date, session) + assert get.call_count == 1 + assert res[0][0] == first_date + assert res[-1][0] == last_date + + futures_path = ( + testdatadir / "binance/binance_public_data/futures-APEUSDT-aggTrades-2024-10-18.zip" + ) + get = mocker.patch( + "freqtrade.exchange.binance_public_data.aiohttp.ClientSession.get", + return_value=MockResponse(futures_path.read_bytes(), 200), + ) + res_fut = await get_daily_trades(symbol_futures, CandleType.FUTURES, date, session) + assert get.call_count == 1 + assert res_fut[0][0] == 1729209603958 + assert res_fut[-1][0] == 1729295981272 + + get = mocker.patch( + "freqtrade.exchange.binance_public_data.aiohttp.ClientSession.get", + return_value=MockResponse(b"", 404), + ) + with pytest.raises(Http404): + await get_daily_trades(symbol, CandleType.SPOT, date, session, retry_delay=0) + assert get.call_count == 1 + + get = mocker.patch( + "freqtrade.exchange.binance_public_data.aiohttp.ClientSession.get", + return_value=MockResponse(b"", 500), + ) + mocker.patch("asyncio.sleep") + with pytest.raises(BadHttpStatus): + await get_daily_trades(symbol, CandleType.SPOT, date, session) + assert get.call_count == 4 # 1 + 3 default retries + + get = mocker.patch( + "freqtrade.exchange.binance_public_data.aiohttp.ClientSession.get", + return_value=MockResponse(b"nop", 200), + ) + with pytest.raises(zipfile.BadZipFile): + await get_daily_trades(symbol, CandleType.SPOT, date, session) + assert get.call_count == 4 # 1 + 3 default retries diff --git a/tests/exchange/test_exchange.py b/tests/exchange/test_exchange.py index 58abb141e..0ed8e97e9 100644 --- a/tests/exchange/test_exchange.py +++ b/tests/exchange/test_exchange.py @@ -2373,6 +2373,8 @@ def test_refresh_latest_trades( caplog.set_level(logging.DEBUG) use_trades_conf = default_conf use_trades_conf["exchange"]["use_public_trades"] = True + use_trades_conf["exchange"]["only_from_ccxt"] = True + use_trades_conf["datadir"] = tmp_path use_trades_conf["orderflow"] = {"max_candles": 1500} exchange = get_patched_exchange(mocker, use_trades_conf) @@ -3365,6 +3367,7 @@ async def test__async_fetch_trades_contract_size( async def test__async_get_trade_history_id( default_conf, mocker, exchange_name, fetch_trades_result ): + default_conf["exchange"]["only_from_ccxt"] = True exchange = get_patched_exchange(mocker, default_conf, exchange=exchange_name) if exchange._trades_pagination != "id": exchange.close() diff --git a/tests/testdata/binance/binance_public_data/futures-APEUSDT-aggTrades-2024-10-18.zip b/tests/testdata/binance/binance_public_data/futures-APEUSDT-aggTrades-2024-10-18.zip new file mode 100644 index 000000000..ea734e4ca Binary files /dev/null and b/tests/testdata/binance/binance_public_data/futures-APEUSDT-aggTrades-2024-10-18.zip differ diff --git a/tests/testdata/binance/binance_public_data/spot-PEPEUSDT-aggTrades-2024-10-27.zip b/tests/testdata/binance/binance_public_data/spot-PEPEUSDT-aggTrades-2024-10-27.zip new file mode 100644 index 000000000..7cec1e299 Binary files /dev/null and b/tests/testdata/binance/binance_public_data/spot-PEPEUSDT-aggTrades-2024-10-27.zip differ