498 lines
19 KiB
Python
498 lines
19 KiB
Python
import logging
| |
import dateparser
| |
import pytz
| |
from datetime import datetime, timezone, timedelta
| |
from typing import Union, List
| |
import json
| |
from decimal import *
| |
| |
import numpy as np
| |
import pandas as pd
| |
from pandas.tseries.frequencies import to_offset
| |
| |
from sklearn import metrics
| |
| |
from apscheduler.triggers.cron import CronTrigger
| |
| |
from common.gen_features import *
| |
| |
#
| |
# Decimals
| |
#
| |
| |
def to_decimal(value):
| |
"""Convert to a decimal with the required precision. The value can be string, float or decimal."""
| |
# Possible cases: string, 4.1-e7, float like 0.1999999999999 (=0.2), Decimal('4.1E-7')
| |
| |
# App.config["trade"]["symbol_info"]["baseAssetPrecision"]
| |
| |
n = 8
| |
rr = Decimal(1) / (Decimal(10) ** n) # Result: 0.00000001
| |
ret = Decimal(str(value)).quantize(rr, rounding=ROUND_DOWN)
| |
return ret
| |
| |
def round_str(value, digits):
| |
rr = Decimal(1) / (Decimal(10) ** digits) # Result for 8 digits: 0.00000001
| |
ret = Decimal(str(value)).quantize(rr, rounding=ROUND_HALF_UP)
| |
return f"{ret:.{digits}f}"
| |
| |
def round_down_str(value, digits):
| |
rr = Decimal(1) / (Decimal(10) ** digits) # Result for 8 digits: 0.00000001
| |
ret = Decimal(str(value)).quantize(rr, rounding=ROUND_DOWN)
| |
return f"{ret:.{digits}f}"
| |
| |
#
| |
# Interval arithmetic (pandas)
| |
#
| |
| |
def pandas_get_interval(freq: str, timestamp: int=None):
| |
"""
| |
Find a discrete interval for the given timestamp and return its start and end.
| |
| |
:param freq: pandas frequency
| |
:param timestamp: milliseconds (13 digits)
| |
:return: triple (start, end)
| |
"""
| |
if not timestamp:
| |
timestamp = int(datetime.now(timezone.utc).timestamp()) # seconds (10 digits)
| |
# Alternatively: timestamp = int(datetime.utcnow().replace(tzinfo=pytz.utc).timestamp())
| |
elif isinstance(timestamp, datetime):
| |
timestamp = int(timestamp.replace(tzinfo=pytz.utc).timestamp())
| |
elif isinstance(timestamp, int):
| |
pass
| |
else:
| |
ValueError(f"Error converting timestamp {timestamp} to millis. Unknown type {type(timestamp)} ")
| |
| |
# Interval length for the given frequency
| |
interval_length_sec = pandas_interval_length_ms(freq) / 1000
| |
| |
start = (timestamp // interval_length_sec) * interval_length_sec
| |
end = start + interval_length_sec
| |
| |
return int(start*1000), int(end*1000)
| |
| |
def pandas_interval_length_ms(freq: str):
| |
return int(freq_to_timedelta(freq).total_seconds() * 1000)
| |
| |
def get_interval_count_from_start_dt(freq: str, start_dt):
| |
"""How many whole intervals are from the specified start datetime and now."""
| |
interval_length_td = freq_to_timedelta(freq)
| |
now = datetime.now(timezone.utc)
| |
interval_count = (now - start_dt) // interval_length_td # How many whole intervals
| |
return interval_count + 2
| |
| |
def get_start_dt_for_interval_count(freq: str, interval_count: int):
| |
"""Start datetime for the specified number of whole intervals back. Result is not aligned with the reaster."""
| |
interval_length_td = freq_to_timedelta(freq)
| |
period_length_td = interval_length_td * (interval_count + 1)
| |
now = datetime.now(timezone.utc)
| |
start_dt = (now - period_length_td)
| |
return start_dt
| |
| |
def freq_to_timedelta(freq: str):
| |
"""
| |
Given pandas freq string, return its duration (timedelta).
| |
| |
Offsets: https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#offset-aliases
| |
| |
to_offset("10min").delta -- works w/o number but does not work with B. delta is deprecated
| |
pd.Timedelta(days=1) or pd.Timedelta("h") -- does not work w/o number prefix like "h"
| |
datetime.timedelta(minutes=120) -- manually
| |
"""
| |
if freq.endswith("B"): # Replace by D
| |
freq = freq[:-1] + "D"
| |
| |
return pd.Timedelta(to_offset(freq)).to_pytimedelta() # or _as_pd_timedelta
| |
| |
#
| |
# Date and time
| |
#
| |
| |
def freq_to_CronTrigger(freq: str):
| |
# Add small time after interval end to get a complete interval from the server
| |
if freq.endswith("min"):
| |
if freq[:-3] == "1":
| |
trigger = CronTrigger(minute="*", second="1", timezone="UTC")
| |
else:
| |
trigger = CronTrigger(minute="*/" + freq[:-3], second="1", timezone="UTC")
| |
elif freq.endswith("h"):
| |
if freq[:-1] == "1":
| |
trigger = CronTrigger(hour="*", minute="0", second="20", timezone="UTC")
| |
else:
| |
trigger = CronTrigger(hour="*/" + freq[:-1], minute="0", second="20", timezone="UTC")
| |
elif freq.endswith("D"):
| |
if freq[:-1] == "1":
| |
trigger = CronTrigger(day="*", hour="0", minute="1", second="0", timezone="UTC")
| |
else:
| |
trigger = CronTrigger(day="*/" + freq[:-1], hour="0", minute="1", second="0", timezone="UTC")
| |
elif freq.endswith("W"):
| |
if freq[:-1] == "1":
| |
trigger = CronTrigger(week="*", day_of_week="1", hour="1", minute="0", second="0", timezone="UTC")
| |
else:
| |
trigger = CronTrigger(day="*/" + freq[:-1], day_of_week="1", hour="1", minute="0", second="0", timezone="UTC")
| |
elif freq.endswith("MS"):
| |
if freq[:-2] == "1":
| |
trigger = CronTrigger(month="*", day="1", hour="1", minute="0", second="0", timezone="UTC")
| |
else:
| |
trigger = CronTrigger(month="*/" + freq[:-1], day="1", hour="1", minute="0", second="0", timezone="UTC")
| |
else:
| |
raise ValueError(f"Cannot convert frequency '{freq}' to cron.")
| |
| |
return trigger
| |
| |
def now_timestamp():
| |
"""
| |
"""
| |
return int(datetime.now(timezone.utc).timestamp() * 1000)
| |
| |
def find_index(df: pd.DataFrame, date_str: str, column_name: str = "timestamp"):
| |
"""
| |
Return index of the record with the specified datetime string.
| |
| |
:return: row id in the input data frame which can be then used in iloc function
| |
:rtype: int
| |
"""
| |
d = dateparser.parse(date_str)
| |
try:
| |
res = df[df[column_name] == d]
| |
except TypeError: # "Cannot compare tz-naive and tz-aware datetime-like objects"
| |
# Change timezone (set UTC timezone or reset timezone)
| |
if d.tzinfo is None or d.tzinfo.utcoffset(d) is None:
| |
d = d.replace(tzinfo=pytz.utc)
| |
else:
| |
d = d.replace(tzinfo=None)
| |
| |
# Repeat
| |
res = df[df[column_name] == d]
| |
| |
if res is None or len(res) == 0:
| |
raise ValueError(f"Cannot find date '{date_str}' in the column '{column_name}'. Either it does not exist or wrong format")
| |
| |
id = res.index[0]
| |
| |
return id
| |
| |
def notnull_tail_rows(df):
| |
"""Maximum number of tail rows without nulls."""
| |
| |
nan_df = df.isnull()
| |
nan_cols = nan_df.any() # Series with columns having at least one NaN
| |
nan_cols = nan_cols[nan_cols].index.to_list()
| |
if len(nan_cols) == 0:
| |
return len(df)
| |
| |
# Indexes of last NaN for all columns and then their minimum
| |
tail_rows = nan_df[nan_cols].values[::-1].argmax(axis=0).min()
| |
| |
return tail_rows
| |
| |
#
| |
# System etc.
| |
#
| |
| |
def resolve_generator_name(gen_name: str):
| |
"""
| |
Resolve the specified name to a function reference.
| |
Fully qualified name consists of module name and function name separated by a colon,
| |
for example: 'mod1.mod2.mod3:my_func'.
| |
| |
Example: fn = resolve_generator_name("common.gen_features_topbot:generate_labels_topbot3")
| |
"""
| |
| |
mod_and_func = gen_name.split(':', 1)
| |
mod_name = mod_and_func[0] if len(mod_and_func) > 1 else None
| |
func_name = mod_and_func[-1]
| |
| |
if not mod_name:
| |
return None
| |
| |
try:
| |
mod = importlib.import_module(mod_name)
| |
except Exception as e:
| |
return None
| |
if mod is None:
| |
return None
| |
| |
try:
| |
func = getattr(mod, func_name)
| |
except AttributeError as e:
| |
return None
| |
| |
return func
| |
| |
#
| |
# Data processing
| |
#
| |
| |
def double_columns(df, shifts: List[int]):
| |
"""
| |
Use previous rows as features appended to this row. This allows us to move history to the current time.
| |
One limitation is that this function will duplicate *all* features and only using the explicitly specified list of offsets.
| |
"""
| |
if not shifts:
| |
return df
| |
df_list = [df.shift(shift) for shift in shifts]
| |
df_list.insert(0, df)
| |
max_shift = max(shifts)
| |
| |
# Shift and add same columns
| |
df_out = pd.concat(df_list, axis=1) # keys=('A', 'B')
| |
| |
return df_out
| |
| |
def append_rows(df, new_df):
| |
"""
| |
Append all rows from the second new data frame to the first old data frame by overwriting existing rows if any.
| |
| |
Notes:
| |
- The rows are appended individually (row-by-row in the loop). If the second (new) data frame is large then this operation might be inefficient.
| |
- No validity check are performed like column overlaps or resulting index gaps.
| |
- If it is necessary to retain continuous index then additional validations have to be done (before or after this operation).
| |
"""
| |
for idx, row in new_df.iterrows():
| |
df.loc[idx] = row
| |
return df
| |
| |
def append_df_drop_concat(df, new_df):
| |
| |
# Drop explicitly last rows in the overlap range
| |
#df_wo_overlap = df.drop(new_df.index[:], errors="ignore") # ignore errors in case of missing indexes (for new rows). Set in_place if necessary
| |
| |
# Drop explicitly last rows in the overlap range
| |
df_wo_overlap = df[:new_df.index[0]] # Select only records till the first index in the new frame. Assume the rows in the new frame are ordered
| |
df_wo_overlap = df_wo_overlap.iloc[:-1] # Remove last element because slicing above includes the range right side
| |
| |
# Drop explicitly last rows in the overlap range
| |
#last_idx = df.index.get_loc(new_df.index[0])
| |
#df_wo_overlap = df.iloc[:last_idx]
| |
| |
df3 = pd.concat([df_wo_overlap, new_df]) # Append new rows with overlap removed above
| |
| |
return df3
| |
| |
def append_df_combine_update(df, new_df):
| |
# The first old frame has priority and its values will be always retained if available (non-null).
| |
# Only if an old value is null or a new row was appended, the new values from new frame is used.
| |
# Update null (and only null) elements with values in the same location in other (if any, that is, null is not used for overwriting).
| |
df2 = df.combine_first(new_df)
| |
# Yet, we want to have new values overwrite even old non-null values so enforce complete overwriting (inplace)
| |
df2.update(new_df)
| |
return df2
| |
| |
def merge_data_sources(data_sources: list, time_column: str, freq: str, merge_interpolate: bool):
| |
"""
| |
Create one data frame by merging multiple source data frames on the specified time column by generating a common time raster.
| |
| |
:param data_sources: list of dicts where each dict describes one data source and stores its data in df
| |
:param time_column: column name with timestamps
| |
:param freq: pandas frequency for the common time raster like 1min, 1h etc.
| |
:param merge_interpolate: if True then the missing values will be interpolated
| |
:return: data frame with all data merged on timestamps
| |
"""
| |
for ds in data_sources:
| |
df = ds.get("df")
| |
| |
if time_column in df.columns:
| |
df = df.set_index(time_column)
| |
elif df.index.name == time_column:
| |
pass
| |
else:
| |
print(f"ERROR: Timestamp column is absent.")
| |
return
| |
| |
# Add prefix if not already there
| |
if ds['column_prefix']:
| |
#df = df.add_prefix(ds['column_prefix']+"_")
| |
df.columns = [
| |
ds['column_prefix']+"_"+col if not col.startswith(ds['column_prefix']+"_") else col
| |
for col in df.columns
| |
]
| |
| |
ds["start"] = df.first_valid_index() # df.index[0]
| |
ds["end"] = df.last_valid_index() # df.index[-1]
| |
| |
ds["df"] = df
| |
| |
#
| |
# Create common (main) index and empty data frame
| |
#
| |
range_start = min([ds["start"] for ds in data_sources])
| |
range_end = min([ds["end"] for ds in data_sources])
| |
| |
# Generate a discrete time raster according to the (pandas) frequency parameter
| |
index = pd.date_range(range_start, range_end, freq=freq)
| |
| |
df_out = pd.DataFrame(index=index)
| |
df_out.index.name = time_column
| |
df_out.insert(0, time_column, df_out.index) # Repeat index as a new column
| |
| |
for ds in data_sources:
| |
# Note that timestamps must have the same semantics, for example, start of kline (and not end of kline)
| |
# If different data sets have different semantics for timestamps, then data must be shifted accordingly
| |
df_out = df_out.join(ds["df"])
| |
| |
# Interpolate numeric columns
| |
if merge_interpolate:
| |
num_columns = df_out.select_dtypes((float, int)).columns.tolist()
| |
for col in num_columns:
| |
df_out[col] = df_out[col].interpolate()
| |
| |
return df_out
| |
| |
#
| |
# Analysis
| |
#
| |
| |
def compute_scores(y_true, y_hat):
| |
"""Compute several scores and return them as dict."""
| |
y_true = y_true.astype(int)
| |
y_hat_class = np.where(y_hat.values > 0.5, 1, 0)
| |
| |
try:
| |
auc = metrics.roc_auc_score(y_true, y_hat.fillna(value=0))
| |
except ValueError:
| |
auc = 0.0 # Only one class is present (if dataset is too small, e.g,. when debugging) or Nulls in predictions
| |
| |
try:
| |
ap = metrics.average_precision_score(y_true, y_hat.fillna(value=0))
| |
except ValueError:
| |
ap = 0.0 # Only one class is present (if dataset is too small, e.g,. when debugging) or Nulls in predictions
| |
| |
f1 = metrics.f1_score(y_true, y_hat_class)
| |
precision = metrics.precision_score(y_true, y_hat_class)
| |
recall = metrics.recall_score(y_true, y_hat_class)
| |
| |
scores = dict(auc=auc, ap=ap, f1=f1, precision=precision, recall=recall,)
| |
| |
scores = {key: round(float(value), 3) for (key, value) in scores.items()}
| |
| |
return scores
| |
| |
def compute_scores_regression(y_true, y_hat):
| |
"""Compute regression scores. Input columns must have numeric data type."""
| |
| |
try:
| |
mae = metrics.mean_absolute_error(y_true, y_hat)
| |
except ValueError:
| |
mae = np.nan
| |
| |
try:
| |
mape = metrics.mean_absolute_percentage_error(y_true, y_hat)
| |
except ValueError:
| |
mape = np.nan
| |
| |
try:
| |
r2 = metrics.r2_score(y_true, y_hat)
| |
except ValueError:
| |
r2 = np.nan
| |
| |
#
| |
# How good it is in predicting the sign (increase of decrease)
| |
#
| |
| |
y_true_class = np.where(y_true.values > 0.0, +1, -1)
| |
y_hat_class = np.where(y_hat.values > 0.0, +1, -1)
| |
| |
try:
| |
auc = metrics.roc_auc_score(y_true_class, y_hat_class)
| |
except ValueError:
| |
auc = 0.0 # Only one class is present (if dataset is too small, e.g,. when debugging) or Nulls in predictions
| |
| |
try:
| |
ap = metrics.average_precision_score(y_true_class, y_hat_class)
| |
except ValueError:
| |
ap = 0.0 # Only one class is present (if dataset is too small, e.g,. when debugging) or Nulls in predictions
| |
| |
f1 = metrics.f1_score(y_true_class, y_hat_class)
| |
precision = metrics.precision_score(y_true_class, y_hat_class)
| |
recall = metrics.recall_score(y_true_class, y_hat_class)
| |
| |
scores = dict(
| |
mae=mae, mape=mape, r2=r2,
| |
auc=auc, ap=ap, f1=f1, precision=precision, recall=recall,
| |
)
| |
| |
scores = {key: round(float(value), 3) for (key, value) in scores.items()}
| |
| |
return scores
| |
| |
def first_location_of_crossing_threshold(df, horizon, threshold, close_column_name, price_column_name):
| |
"""
| |
For each point, take its close price as a reference, and then find the distance
| |
(relative offset, index) to the _first_ future point with the price higher (or lower)
| |
than the reference price by the specified level.
| |
| |
If the location (relative index) is 0 then it is the next point. If location (index) is NaN,
| |
then the price does not cross the specified threshold within the horizon
| |
(or there is not enough data, e.g., at the end of the series). Therefore, this
| |
function can be used to find whether the price will cross the threshold at all
| |
during the specified horizon.
| |
| |
The function is somewhat similar to the tsfresh function first_location_of_maximum
| |
or minimum. The difference is that this function does not search for local maximum
| |
(which can vary) but rather finds the first point with the fixed specified increase
| |
(or decrease).
| |
| |
Horizon specifies how many future points are considered (excluding the current point).
| |
| |
If the specified threshold is positive then the function is searching for increase.
| |
Otherwise, if it is negative, then the functions will search for decrease.
| |
The threshold is specified as percentage. For example, 2% means that the function
| |
will find relative offset (distance from this point) where the price increases by 2%
| |
relative to the current price. This will be done for all point.
| |
| |
:param df: dataframe with the specified columns
| |
:param horizon: how many future data rows to consider when searching for the specified price increase or decrease
| |
:param threshold: percentage of increase (if positive) or decrease (if negative)
| |
:param close_column_name: column for reference (initial) prices
| |
:param price_column_name: columns to search for the specified increase or decrease.
| |
Frequently, it is high column for price increase and low column for price decrease.
| |
But it can be also the reference column (for example, close price) for both cases if we ignore high and low peaks..
| |
:return: A series with integer (or None) values. One integer value represent the offset from the current value
| |
where the price increases (or decreases) by the specified percentage. None means that the price does not reach
| |
the specified level within the horizon.
| |
"""
| |
| |
def fn_high(x):
| |
if len(x) < 2:
| |
return np.nan
| |
p = x[0, 0] # Reference price
| |
p_threshold = p*(1+(threshold/100.0)) # Cross line
| |
idx = np.argmax(x[1:, 1] > p_threshold) # First index where price crosses the threshold
| |
| |
# If all False, then index is 0 (first element of constant series) and we are not able to distinguish it from first element being True
| |
# If index is 0 and first element False (under threshold) then NaN (not exceeds)
| |
if idx == 0 and x[1, 1] <= p_threshold:
| |
return np.nan
| |
return idx
| |
| |
def fn_low(x):
| |
if len(x) < 2:
| |
return np.nan
| |
p = x[0, 0] # Reference price
| |
p_threshold = p*(1+(threshold/100.0)) # Cross line
| |
idx = np.argmax(x[1:, 1] < p_threshold) # First index where price crosses the threshold
| |
| |
# If all False, then index is 0 (first element of constant series) and we are not able to distinguish it from first element being True
| |
# If index is 0 and first element False (under threshold) then NaN (not exceeds)
| |
if idx == 0 and x[1, 1] >= p_threshold:
| |
return np.nan
| |
return idx
| |
| |
# Window df will include the current row as well as horizon of past rows with 0 index starting from the oldest row and last index with the current row
| |
rl = df[[close_column_name, price_column_name]].rolling(horizon + 1, min_periods=(horizon // 2), method='table')
| |
| |
if threshold > 0:
| |
df_out = rl.apply(fn_high, raw=True, engine='numba')
| |
elif threshold < 0:
| |
df_out = rl.apply(fn_low, raw=True, engine='numba')
| |
else:
| |
raise ValueError(f"Threshold cannot be zero.")
| |
| |
# Because rolling apply processes past records while we need future records
| |
df_out = df_out.shift(-horizon)
| |
| |
# For some unknown reason (bug?), rolling apply (with table and numba) returns several columns rather than one column
| |
out_column = df_out.iloc[:, 0]
| |
| |
return out_column
|