better organize binance specific functions

collectors/collector_depth.py

@@ -115,7 +115,7 @@ async def request_depth(symbol, freq, limit):
     # Post-process
     #
 
-    depth['timestamp'] = get_interval(freq=freq, timestamp=requestTime)[0]
+    depth['timestamp'] = binance_get_interval(freq=freq, timestamp=requestTime)[0]
     depth['requestTime'] = requestTime
     depth['responseTime'] = responseTime
     depth['symbol'] = symbol

common/gen_features.py

@@ -696,39 +696,5 @@ def add_threshold_feature(df, column_name: str, thresholds: list, out_names: lis
     return out_names
 
 
-def klines_to_df(klines: list):
-    """
-    Convert a list of klines to a data frame.
-    """
-    columns = [
-        'timestamp',
-        'open', 'high', 'low', 'close', 'volume',
-        'close_time',
-        'quote_av', 'trades', 'tb_base_av', 'tb_quote_av',
-        'ignore'
-    ]
-
-    df = pd.DataFrame(klines, columns=columns)
-
-    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
-    df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
-
-    df["open"] = pd.to_numeric(df["open"])
-    df["high"] = pd.to_numeric(df["high"])
-    df["low"] = pd.to_numeric(df["low"])
-    df["close"] = pd.to_numeric(df["close"])
-    df["volume"] = pd.to_numeric(df["volume"])
-
-    df["quote_av"] = pd.to_numeric(df["quote_av"])
-    df["trades"] = pd.to_numeric(df["trades"])
-    df["tb_base_av"] = pd.to_numeric(df["tb_base_av"])
-    df["tb_quote_av"] = pd.to_numeric(df["tb_quote_av"])
-
-    if "timestamp" in df.columns:
-        df.set_index('timestamp', inplace=True)
-
-    return df
-
-
 if __name__ == "__main__":
     pass

common/utils.py

@@ -41,17 +41,106 @@ def round_down_str(value, digits):
 
 
 #
-# Date and time
+# Binance specific
 #
 
-def get_interval(freq: str, timestamp: int=None):
+def klines_to_df(klines, df):
+
+    data = pd.DataFrame(klines, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_av', 'trades', 'tb_base_av', 'tb_quote_av', 'ignore'])
+    data['timestamp'] = pd.to_datetime(data['timestamp'], unit='ms')
+    dtypes = {
+        'open': 'float64', 'high': 'float64', 'low': 'float64', 'close': 'float64', 'volume': 'float64',
+        'close_time': 'int64',
+        'quote_av': 'float64',
+        'trades': 'int64',
+        'tb_base_av': 'float64',
+        'tb_quote_av': 'float64',
+        'ignore': 'float64',
+    }
+    data = data.astype(dtypes)
+
+    if df is None or len(df) == 0:
+        df = data
+    else:
+        df = pd.concat([df, data])
+
+    # Drop duplicates
+    df = df.drop_duplicates(subset=["timestamp"], keep="last")
+    #df = df[~df.index.duplicated(keep='last')]  # alternatively, drop duplicates in index
+
+    df.set_index('timestamp', inplace=True)
+
+    return df
+
+
+def binance_klines_to_df(klines: list):
+    """
+    Convert a list of klines to a data frame.
+    """
+    columns = [
+        'timestamp',
+        'open', 'high', 'low', 'close', 'volume',
+        'close_time',
+        'quote_av', 'trades', 'tb_base_av', 'tb_quote_av',
+        'ignore'
+    ]
+
+    df = pd.DataFrame(klines, columns=columns)
+
+    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
+    df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
+
+    df["open"] = pd.to_numeric(df["open"])
+    df["high"] = pd.to_numeric(df["high"])
+    df["low"] = pd.to_numeric(df["low"])
+    df["close"] = pd.to_numeric(df["close"])
+    df["volume"] = pd.to_numeric(df["volume"])
+
+    df["quote_av"] = pd.to_numeric(df["quote_av"])
+    df["trades"] = pd.to_numeric(df["trades"])
+    df["tb_base_av"] = pd.to_numeric(df["tb_base_av"])
+    df["tb_quote_av"] = pd.to_numeric(df["tb_quote_av"])
+
+    if "timestamp" in df.columns:
+        df.set_index('timestamp', inplace=True)
+
+    return df
+
+
+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)
     """
@@ -100,6 +189,10 @@ def get_interval(freq: str, timestamp: int=None):
     return int(start * 1000), int(end * 1000)
 
 
+#
+# Date and time
+#
+
 def now_timestamp():
     """
     INFO:

service/analyzer.py

@@ -280,7 +280,7 @@ def analyze(self, ignore_last_rows=False):
             if ds.get("file") == "klines":
                 try:
                     klines = self.klines.get(ds.get("folder"))
-                    df = klines_to_df(klines)
+                    df = binance_klines_to_df(klines)
 
                     # Validate
                     source_columns = ['open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_av', 'trades', 'tb_base_av', 'tb_quote_av']

service/server.py

@@ -8,7 +8,7 @@
 
 from service.App import *
 from common.utils import *
-from service.collector import *
+from service.collector_binance import main_collector_task, data_provider_health_check, sync_data_collector_task
 from service.analyzer import *
 from service.notifier_trades import *
 from service.notifier_scores import *

service/trader.py

@@ -33,7 +33,7 @@ async def main_trader_task():
     It is a highest level task which is added to the event loop and executed normally every 1 minute and then it calls other tasks.
     """
     symbol = App.config["symbol"]
-    startTime, endTime = get_interval("1m")
+    startTime, endTime = binance_get_interval("1m")
     now_ts = now_timestamp()
 
     log.info(f"===> Start trade task. Timestamp {now_ts}. Interval [{startTime},{endTime}].")