405 lines
16 KiB
Python
405 lines
16 KiB
Python
from pathlib import Path
| |
from typing import Union
| |
import json
| |
import pickle
| |
from datetime import datetime, date, timedelta
| |
import queue
| |
| |
import numpy as np
| |
import pandas as pd
| |
| |
from service.App import *
| |
| |
from common.utils import *
| |
from common.classifiers import *
| |
from common.model_store import *
| |
from common.generators import generate_feature_set
| |
from common.generators import predict_feature_set
| |
| |
from outputs.notifier_trades import load_last_transaction
| |
| |
from scripts.merge import *
| |
from scripts.features import *
| |
| |
import logging
| |
log = logging.getLogger('analyzer')
| |
| |
| |
class Analyzer:
| |
"""
| |
In-memory database which represents the current state of the (trading) environment including its history.
| |
| |
Properties of klines:
| |
- "timestamp" is a left border of the interval like "2017-08-17 04:00:00"
| |
- "close_time" is a right border of the interval in ms (last millisecond) like "1502942459999" equivalent to "2017-08-17 04:00::59.999"
| |
"""
| |
| |
def __init__(self, config):
| |
"""
| |
Create a new operation object using its definition.
| |
| |
:param config: Initialization parameters defining what is in the database including its persistent parameters and schema
| |
"""
| |
| |
self.config = config
| |
| |
#
| |
# Data state
| |
#
| |
| |
# Klines are stored as a dict of lists. Key is a symbol and the list is a list of latest kline records
| |
# One kline record is a list of values (not dict) as returned by API: open time, open, high, low, close, volume etc.
| |
self.klines = {}
| |
| |
self.queue = queue.Queue()
| |
| |
#
| |
# Load models
| |
#
| |
symbol = App.config["symbol"]
| |
data_path = Path(App.config["data_folder"]) / symbol
| |
| |
model_path = Path(App.config["model_folder"])
| |
if not model_path.is_absolute():
| |
model_path = data_path / model_path
| |
model_path = model_path.resolve()
| |
| |
self.models = load_models_for_generators(App.config, model_path)
| |
| |
# Load latest transaction and (simulated) trade state
| |
App.transaction = load_last_transaction()
| |
| |
#
| |
# Data state operations
| |
#
| |
| |
def get_klines_count(self, symbol):
| |
return len(self.klines.get(symbol, []))
| |
| |
def get_last_kline(self, symbol):
| |
if self.get_klines_count(symbol) > 0:
| |
return self.klines.get(symbol)[-1]
| |
else:
| |
return None
| |
| |
def get_last_kline_ts(self, symbol):
| |
"""Open time of the last kline. It is simultaneously kline id. Add 1m if the end is needed."""
| |
last_kline = self.get_last_kline(symbol=symbol)
| |
if not last_kline:
| |
return 0
| |
last_kline_ts = last_kline[0]
| |
return last_kline_ts
| |
| |
def get_missing_klines_count(self, symbol):
| |
"""
| |
The number of complete discrete intervals between the last available kline and current timestamp.
| |
The interval length is determined by the frequency parameter.
| |
"""
| |
last_kline_ts = self.get_last_kline_ts(symbol)
| |
if not last_kline_ts:
| |
return App.config["features_horizon"]
| |
| |
freq = App.config["freq"]
| |
now = datetime.utcnow()
| |
last_kline = datetime.utcfromtimestamp(last_kline_ts // 1000)
| |
interval_length = pd.Timedelta(freq).to_pytimedelta()
| |
intervals_count = (now-last_kline) // interval_length
| |
| |
intervals_count += 2
| |
return intervals_count
| |
| |
def store_klines(self, data: dict):
| |
"""
| |
Store latest klines for the specified symbols.
| |
Existing klines for the symbol and timestamp will be overwritten.
| |
| |
:param data: Dict of lists with symbol as a key, and list of klines for this symbol as a value.
| |
Example: { 'BTCUSDT': [ [], [], [] ] }
| |
:type dict:
| |
"""
| |
now_ts = now_timestamp()
| |
freq = App.config["freq"]
| |
interval_length_ms = pandas_interval_length_ms(freq)
| |
| |
for symbol, klines in data.items():
| |
# If symbol does not exist then create
| |
klines_data = self.klines.get(symbol)
| |
if klines_data is None:
| |
self.klines[symbol] = []
| |
klines_data = self.klines.get(symbol)
| |
| |
ts = klines[0][0] # Very first timestamp of the new data
| |
| |
# Find kline with this or younger timestamp in the database
| |
# same_kline = next((x for x in klines_data if x[0] == ts), None)
| |
existing_indexes = [i for i, x in enumerate(klines_data) if x[0] >= ts]
| |
#print(f"===>>> Existing tss: {[x[0] for x in klines_data]}")
| |
#print(f"===>>> New tss: {[x[0] for x in klines]}")
| |
#print(f"===>>> {symbol} Overlap {len(existing_indexes)}. Existing Indexes: {existing_indexes}")
| |
if existing_indexes: # If there is overlap with new klines
| |
start = min(existing_indexes)
| |
num_deleted = len(klines_data) - start
| |
del klines_data[start:] # Delete starting from the first kline in new data (which will be added below)
| |
if len(klines) < num_deleted: # It is expected that we add same or more klines than deleted
| |
log.error("More klines is deleted by new klines added, than we actually add. Something woring with timestamps and storage logic.")
| |
| |
# Append new klines
| |
klines_data.extend(klines)
| |
| |
# Remove too old klines
| |
kline_window = App.config["features_horizon"]
| |
to_delete = len(klines_data) - kline_window
| |
if to_delete > 0:
| |
del klines_data[:to_delete]
| |
| |
# Check validity. It has to be an ordered time series with certain frequency
| |
for i, kline in enumerate(self.klines.get(symbol)):
| |
ts = kline[0]
| |
if i > 0:
| |
if ts - prev_ts != interval_length_ms:
| |
log.error("Wrong sequence of klines. They are expected to be a regular time series with 1m frequency.")
| |
prev_ts = kline[0]
| |
| |
# Debug message about the last received kline end and current ts (which must be less than 1m - rather small delay)
| |
log.debug(f"Stored klines. Total {len(klines_data)} in db. Last kline end: {self.get_last_kline_ts(symbol)+interval_length_ms}. Current time: {now_ts}")
| |
| |
def store_depth(self, depths: list, freq):
| |
"""
| |
Persistently store order books from the input list. Each entry is one response from order book request for one symbol.
| |
Currently the order books are directly stored in a file (for this symbol) and not in this object.
| |
| |
:param depths: List of dicts where each dict is an order book with such fields as 'asks', 'bids' and 'symbol' (symbol is added after loading).
| |
:type list:
| |
"""
| |
| |
# File name like TRADE_HOME/COLLECT/DEPTH/depth-BTCUSDT-5s.txt
| |
TRADE_DATA = "." # TODO: We need to read it from the environment. It could be data dir or docker volume.
| |
# BASE_DIR = Path(__file__).resolve().parent.parent
| |
# BASE_DIR = Path.cwd()
| |
| |
for depth in depths:
| |
# TODO: The result might be an exception or some other object denoting bad return (timeout, cancelled etc.)
| |
| |
symbol = depth["symbol"]
| |
| |
path = Path(TRADE_DATA).joinpath(App.config["collector"]["folder"])
| |
path = path.joinpath(App.config["collector"]["depth"]["folder"])
| |
path.mkdir(parents=True, exist_ok=True) # Ensure that dir exists
| |
| |
file_name = f"depth-{symbol}-{freq}"
| |
file = Path(path, file_name).with_suffix(".txt")
| |
| |
# Append to the file (create if it does not exist)
| |
json_line = json.dumps(depth)
| |
with open(file, 'a+') as f:
| |
f.write(json_line + "\n")
| |
| |
def store_queue(self):
| |
"""
| |
Persistently store the queue data to one or more files corresponding to the stream (event) type, symbol (and frequency).
| |
| |
:return:
| |
"""
| |
#
| |
# Get all the data from the queue
| |
#
| |
events = {}
| |
item = None
| |
while True:
| |
try:
| |
item = self.queue.get_nowait()
| |
except queue.Empty as ee:
| |
break
| |
except:
| |
break
| |
| |
if item is None:
| |
break
| |
| |
c = item.get("e") # Channel
| |
if not events.get(c): # Insert if does not exit
| |
events[c] = {}
| |
symbols = events[c]
| |
| |
s = item.get("s") # Symbol
| |
if not symbols.get(s): # Insert if does not exit
| |
symbols[s] = []
| |
data = symbols[s]
| |
| |
data.append(item)
| |
| |
self.queue.task_done() # TODO: Do we really need this?
| |
| |
# File name like TRADE_HOME/COLLECT/DEPTH/depth-BTCUSDT-5s.txt
| |
TRADE_DATA = "." # TODO: We need to read it from the environment. It could be data dir or docker volume.
| |
# BASE_DIR = Path(__file__).resolve().parent.parent
| |
# BASE_DIR = Path.cwd()
| |
| |
path = Path(TRADE_DATA).joinpath(App.config["collector"]["folder"])
| |
path = path.joinpath(App.config["collector"]["stream"]["folder"])
| |
path.mkdir(parents=True, exist_ok=True) # Ensure that dir exists
| |
| |
now = datetime.utcnow()
| |
#rotate_suffix = f"{now:%Y}{now:%m}{now:%d}" # Daily files
| |
rotate_suffix = f"{now:%Y}{now:%m}" # Monthly files
| |
| |
#
| |
# Get all the data from the queue and store in file
| |
#
| |
for c, symbols in events.items():
| |
for s, data in symbols.items():
| |
file_name = f"{c}-{s}-{rotate_suffix}"
| |
file = Path(path, file_name).with_suffix(".txt")
| |
| |
# Append to the file (create if it does not exist)
| |
data = [json.dumps(event) for event in data]
| |
data_str = "\n".join(data)
| |
with open(file, 'a+') as f:
| |
f.write(data_str + "\n")
| |
| |
#
| |
# Analysis (features, predictions, signals etc.)
| |
#
| |
| |
def analyze(self, ignore_last_rows=False):
| |
"""
| |
1. Convert klines to df
| |
2. Derive (compute) features (use same function as for model training)
| |
3. Derive (predict) labels by applying models trained for each label
| |
4. Generate buy/sell signals by applying rule models trained for best overall trade performance
| |
"""
| |
symbol = App.config["symbol"]
| |
| |
# Features, predictions, signals etc. have to be computed only for these last rows (for performance reasons)
| |
last_rows = App.config["features_last_rows"]
| |
| |
last_kline_ts = self.get_last_kline_ts(symbol)
| |
last_kline_ts_str = str(pd.to_datetime(last_kline_ts, unit='ms'))
| |
| |
log.info(f"Analyze {symbol}. Last kline timestamp: {last_kline_ts_str}")
| |
| |
#
| |
# Convert source data (klines) into data frames for each data source
| |
#
| |
data_sources = App.config.get("data_sources", [])
| |
for ds in data_sources:
| |
if ds.get("file") == "klines":
| |
try:
| |
klines = self.klines.get(ds.get("folder"))
| |
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']
| |
if df.isnull().any().any():
| |
null_columns = {k: v for k, v in df.isnull().any().to_dict().items() if v}
| |
log.warning(f"Null in source data found. Columns with Null: {null_columns}")
| |
# TODO: We might receive empty strings or 0s in numeric data - how can we detect them?
| |
# TODO: Check that timestamps in 'close_time' are strictly consecutive
| |
except Exception as e:
| |
log.error(f"Error in klines_to_df method: {e}. Length klines: {len(klines)}")
| |
return
| |
else:
| |
log.error("Unknown data sources. Currently only 'klines' is supported. Check 'data_sources' in config, key 'file'")
| |
return
| |
ds["df"] = df
| |
| |
#
| |
# 1.
| |
# MERGE multiple dfs in one df with prefixes and common regular time index
| |
#
| |
df = merge_data_sources(data_sources)
| |
| |
#
| |
# 2.
| |
# Generate all necessary derived features (NaNs are possible due to limited history)
| |
#
| |
feature_sets = App.config.get("feature_sets", [])
| |
feature_columns = []
| |
for fs in feature_sets:
| |
df, feats = generate_feature_set(df, fs, last_rows=last_rows if not ignore_last_rows else 0)
| |
feature_columns.extend(feats)
| |
| |
# Shorten the data frame. Only several last rows will be needed and not the whole data context
| |
if not ignore_last_rows:
| |
df = df.iloc[-last_rows:]
| |
| |
features = App.config["train_features"]
| |
# Exclude rows with at least one NaN
| |
tail_rows = notnull_tail_rows(df[features])
| |
df = df.tail(tail_rows)
| |
| |
#
| |
# 3.
| |
# Apply ML models and generate score columns
| |
#
| |
| |
# Select row for which to do predictions
| |
predict_df = df[features]
| |
if predict_df.isnull().any().any():
| |
null_columns = {k: v for k, v in predict_df.isnull().any().to_dict().items() if v}
| |
log.error(f"Null in predict_df found. Columns with Null: {null_columns}")
| |
return
| |
| |
train_feature_sets = App.config.get("train_feature_sets", [])
| |
score_df = pd.DataFrame(index=predict_df.index)
| |
train_feature_columns = []
| |
for fs in train_feature_sets:
| |
fs_df, feats, _ = predict_feature_set(predict_df, fs, App.config, self.models)
| |
score_df = pd.concat([score_df, fs_df], axis=1)
| |
train_feature_columns.extend(feats)
| |
| |
# Attach all predicted features to the main data frame
| |
df = pd.concat([df, score_df], axis=1)
| |
| |
#
| |
# 4.
| |
# Signals
| |
#
| |
signal_sets = App.config.get("signal_sets", [])
| |
signal_columns = []
| |
for fs in signal_sets:
| |
df, feats = generate_feature_set(df, fs, last_rows=last_rows if not ignore_last_rows else 0)
| |
signal_columns.extend(feats)
| |
| |
#
| |
# Append the new rows to the main data frame with all previously computed data
| |
#
| |
| |
# Log signal values
| |
row = df.iloc[-1] # Last row stores the latest values we need
| |
scores = ", ".join([f"{x}={row[x]:+.3f}" if isinstance(row[x], float) else f"{x}={str(row[x])}" for x in signal_columns])
| |
log.info(f"Analyze finished. Close: {int(row['close']):,} Signals: {scores}")
| |
| |
if App.df is None or len(App.df) == 0:
| |
App.df = df
| |
return
| |
| |
# Test if newly retrieved and computed values are equal to the previous ones
| |
check_row_count = 3 # These last rows must be correctly computed (particularly, have enough history in case of aggregation)
| |
num_cols = df.select_dtypes((float, int)).columns.tolist()
| |
# Loop over several last newly computed data rows
| |
# Skip last row because it should not exist, and before the last row because its kline is frequently updated after retrieval
| |
for r in range(2, check_row_count):
| |
idx = df.index[-r-1]
| |
| |
if idx not in App.df.index:
| |
continue
| |
| |
# Compare all numeric values of the previously retrieved and newly retrieved rows for the same time
| |
old_row = App.df[num_cols].loc[idx]
| |
new_row = df[num_cols].loc[idx]
| |
comp_idx = np.isclose(old_row, new_row)
| |
if not np.all(comp_idx):
| |
log.warning(f"Newly computed row is not equal to the previously computed row for '{idx}'. NEW: {new_row[~comp_idx].to_dict()}. OLD: {old_row[~comp_idx].to_dict()}")
| |
| |
# Append new rows to the main data frame
| |
App.df = df.tail(check_row_count).combine_first(App.df)
| |
| |
# Remove too old rows
| |
features_horizon = App.config["features_horizon"]
| |
if len(App.df) > features_horizon + 15:
| |
App.df = App.df.tail(features_horizon)
| |
| |
| |
if __name__ == "__main__":
| |
pass
|