mohammadian7/freqtrade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: further work on download-trades-archive

Browse Source
This commit is contained in:
Matthias
2025-01-26 11:18:27 +01:00
parent af3f9d5a75
commit 3746305b5a
1 changed files with 31 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
freqtrade/exchange/binance_public_data.py
+31 -21
View File
@@ -3,7 +3,6 @@ Fetch daily-archived OHLCV data from https://data.binance.vision/
""" """
import asyncio import asyncio
import csv
import logging import logging
import zipfile import zipfile
from datetime import date, timedelta from datetime import date, timedelta
@@ -11,9 +10,11 @@ from io import BytesIO
from typing import Any from typing import Any
import aiohttp import aiohttp
import numpy as np
import pandas as pd import pandas as pd
from pandas import DataFrame from pandas import DataFrame
from freqtrade.constants import DEFAULT_TRADES_COLUMNS
from freqtrade.enums import CandleType from freqtrade.enums import CandleType
from freqtrade.misc import chunks from freqtrade.misc import chunks
from freqtrade.util.datetime_helpers import dt_from_ts, dt_now from freqtrade.util.datetime_helpers import dt_from_ts, dt_now
@@ -316,7 +317,10 @@ async def download_archive_trades(
end = min(end, last_available_date) end = min(end, last_available_date)
if start >= end: if start >= end:
return DataFrame() return DataFrame()
await _download_archive_ohlcv(asset_type_url_segment, symbol, pair, start, end, stop_on_404) result_list = await _download_archive_trades(
asset_type_url_segment, symbol, pair, start, end, stop_on_404
)
return pair, result_list
except Exception as e: except Exception as e:
logger.warning( logger.warning(
@@ -371,12 +375,14 @@ async def get_daily_trades(
first_byte = csvf.read(1)[0] first_byte = csvf.read(1)[0]
if chr(first_byte).isdigit(): if chr(first_byte).isdigit():
# spot # spot
header = [ header = None
names = [
"id", "id",
"price", "price",
"qty", "amount",
"quote_qty", "first_trade_id",
"time", "last_trade_id",
"timestamp",
"is_buyer_maker", "is_buyer_maker",
"is_best_match", "is_best_match",
] ]
@@ -386,22 +392,27 @@ async def get_daily_trades(
names = [ names = [
"id", "id",
"price", "price",
"qty", "amount",
"quote_qty", "first_trade_id",
"time", "last_trade_id",
"timestamp",
"is_buyer_maker", "is_buyer_maker",
] ]
csvf.seek(0) csvf.seek(0)
df = pd.read_csv( df = pd.read_csv(
csvf, csvf,
usecols=[0, 1, 2, 3, 4, 5],
names=names, names=names,
header=header, header=header,
) )
df["cost"] = df["price"] * df["qty"] df["cost"] = df["price"] * df["amount"]
# Side is reversed intentionally
return df[].to_records(index=False).tolist() # based on ccxt parseTrade logic.
df["side"] = np.where(df["is_buyer_maker"], "sell", "buy")
df["type"] = None
if header is None:
df["timestamp"] = df["timestamp"] // 1000
return df[DEFAULT_TRADES_COLUMNS].to_records(index=False).tolist()
elif resp.status == 404: elif resp.status == 404:
logger.debug(f"Failed to download {url}") logger.debug(f"Failed to download {url}")
raise Http404(f"404: {url}", date, url) raise Http404(f"404: {url}", date, url)
@@ -423,7 +434,7 @@ async def _download_archive_trades(
stop_on_404: bool, stop_on_404: bool,
) -> list[list]: ) -> list[list]:
# daily dataframes, `None` indicates missing data in that day (when `stop_on_404` is False) # daily dataframes, `None` indicates missing data in that day (when `stop_on_404` is False)
result: list[list] = [] results: list[list] = []
# the current day being processing, starting at 1. # the current day being processing, starting at 1.
current_day = 0 current_day = 0
@@ -438,7 +449,7 @@ async def _download_archive_trades(
for task in tasks: for task in tasks:
current_day += 1 current_day += 1
try: try:
df = await task result = await task
except Http404 as e: except Http404 as e:
if stop_on_404: if stop_on_404:
logger.debug(f"Failed to download {e.url} due to 404.") logger.debug(f"Failed to download {e.url} due to 404.")
@@ -465,14 +476,13 @@ async def _download_archive_trades(
"remaining data, this can take more time." "remaining data, this can take more time."
) )
await cancel_and_await_tasks(tasks[tasks.index(task) + 1 :]) await cancel_and_await_tasks(tasks[tasks.index(task) + 1 :])
return concat_safe(dfs) return results
else:
dfs.append(None)
except BaseException as e: except BaseException as e:
logger.warning(f"An exception raised: : {e}") logger.warning(f"An exception raised: : {e}")
# Directly return the existing data, do not allow the gap within the data # Directly return the existing data, do not allow the gap within the data
await cancel_and_await_tasks(tasks[tasks.index(task) + 1 :]) await cancel_and_await_tasks(tasks[tasks.index(task) + 1 :])
return concat_safe(dfs) return results
else: else:
dfs.append(df) # Happy case
return concat_safe(dfs) results.extend(result)
return results