83 lines
3.4 KiB
Python
83 lines
3.4 KiB
Python
 | from datetime import datetime, timezone, timedelta
| ||
import pandas as pd
| |||
from binance.helpers import date_to_milliseconds, interval_to_milliseconds
| |||
| |||
def binance_freq_from_pandas(freq: str) -> str:
| |||
"""
| |||
Map pandas frequency to binance API frequency
| |||
| |||
:param freq: pandas frequency https://pandas.pydata.org/docs/user_guide/timeseries.html#timeseries-offset-aliases
| |||
:return: binance frequency https://developers.binance.com/docs/derivatives/coin-margined-futures/market-data/Kline-Candlestick-Data
| |||
"""
| |||
if freq.endswith("min"): # Binance: 1m, 3m, 5m, 15m, 30m
| |||
freq = freq.replace("min", "m")
| |||
elif freq.endswith("D"):
| |||
freq = freq.replace("D", "d") # Binance: 1d, 3d
| |||
elif freq.endswith("W"):
| |||
freq = freq.replace("W", "w")
| |||
elif freq == "BMS":
| |||
freq = freq.replace("BMS", "M")
| |||
| |||
if len(freq) == 1:
| |||
freq = "1" + freq
| |||
| |||
if not (2 <= len(freq) <= 3) or not freq[:-1].isdigit() or freq[-1] not in ["m", "h", "d", "w", "M"]:
| |||
raise ValueError(f"Not supported Binance frequency {freq}. It should be one or two digits followed by a character.")
| |||
| |||
return freq
| |||
| |||
def binance_get_interval(freq: str, timestamp: int=None):
| |||
"""
| |||
Return a triple of interval start (including), end (excluding) in milliseconds for the specified timestamp or now
| |||
| |||
INFO:
| |||
https://github.com/sammchardy/python-binance/blob/master/binance/helpers.py
| |||
interval_to_milliseconds(interval) - binance freq string (like 1m) to millis
| |||
| |||
:param freq: binance frequency https://developers.binance.com/docs/derivatives/coin-margined-futures/market-data/Kline-Candlestick-Data
| |||
:return: tuple of start (inclusive) and end (exclusive) of the interval in millis
| |||
:rtype: (int, int)
| |||
"""
| |||
if not timestamp:
| |||
timestamp = datetime.now(timezone.utc)
| |||
elif isinstance(timestamp, int):
| |||
timestamp = pd.to_datetime(timestamp, unit='ms', utc=True).to_pydatetime()
| |||
| |||
# Although in 3.6 (at least), datetime.timestamp() assumes a timezone naive (tzinfo=None) datetime is in UTC
| |||
timestamp = timestamp.replace(microsecond=0, tzinfo=timezone.utc)
| |||
| |||
if freq == "1s":
| |||
start = timestamp.timestamp()
| |||
end = timestamp + timedelta(seconds=1)
| |||
end = end.timestamp()
| |||
elif freq == "5s":
| |||
reference_timestamp = timestamp.replace(second=0)
| |||
now_duration = timestamp - reference_timestamp
| |||
| |||
freq_duration = timedelta(seconds=5)
| |||
| |||
full_intervals_no = now_duration.total_seconds() // freq_duration.total_seconds()
| |||
| |||
start = reference_timestamp + freq_duration * full_intervals_no
| |||
end = start + freq_duration
| |||
| |||
start = start.timestamp()
| |||
end = end.timestamp()
| |||
elif freq == "1m":
| |||
timestamp = timestamp.replace(second=0)
| |||
start = timestamp.timestamp()
| |||
end = timestamp + timedelta(minutes=1)
| |||
end = end.timestamp()
| |||
elif freq == "5m":
| |||
# Here we need to find 1 h border (or 1 day border) by removing minutes
| |||
# Then divide (now-1hourstart) by 5 min interval length by finding 5 min border for now
| |||
print(f"Frequency 5m not implemented.")
| |||
elif freq == "1h":
| |||
timestamp = timestamp.replace(minute=0, second=0)
| |||
start = timestamp.timestamp()
| |||
end = timestamp + timedelta(hours=1)
| |||
end = end.timestamp()
| |||
else:
| |||
print(f"Unknown frequency.")
| |||
| |||
return int(start * 1000), int(end * 1000)
|