mirror of
https://github.com/webclinic017/drift.git
synced 2026-07-28 11:17:47 +00:00
Mark Aron Szulyovszky
95573eb9dd
* feat(Data): add option to predict 3 classes * feat(Evaluation): added ability to evaluate 3 class predictions * chore(Config): set sensible config for regression models * feat(Data): added option to use balanced or imbalanced three-class data * feat(Evaluate): correctly track "no_of_samples" now that we have three classes * chore(Sweep): remove probably not useful scaler values from sweep
101 lines
4.6 KiB
Python
101 lines
4.6 KiB
Python
from typing import Literal
|
|
from sklearn.metrics import mean_absolute_error, accuracy_score, r2_score, f1_score, precision_score, recall_score
|
|
from quantstats.stats import expected_return, sharpe, skew, sortino
|
|
from utils.helpers import get_first_valid_return_index
|
|
import pandas as pd
|
|
import numpy as np
|
|
|
|
def backtest(returns: pd.Series, signal: pd.Series, transaction_cost = 0.00) -> pd.Series:
|
|
delta_pos = signal.diff(1).abs().fillna(0.)
|
|
costs = transaction_cost * delta_pos
|
|
return (signal * returns) - costs
|
|
|
|
|
|
def __preprocess(target_returns: pd.Series, y_pred: pd.Series, y_true: pd.Series, method: Literal['classification', 'regression'], no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced']) -> pd.DataFrame:
|
|
y_pred.name = 'y_pred'
|
|
target_returns.name = 'target_returns'
|
|
df = pd.concat([y_pred, target_returns],axis=1).dropna()
|
|
|
|
def categorize_binary(x): return 1 if x > 0 else -1
|
|
def categorize_threeway(x): return 0 if x == 0 else 1 if x > 0 else -1
|
|
categorize = categorize_binary if no_of_classes == 'two' else categorize_threeway
|
|
# make sure that we evaluate binary/three-way predictions even if the model is a regression
|
|
if method == 'regression':
|
|
df['sign_pred'] = df.y_pred.apply(categorize)
|
|
df['sign_true'] = df.target_returns.apply(categorize)
|
|
else:
|
|
df['sign_pred'] = df.y_pred.apply(categorize)
|
|
df['sign_true'] = y_true
|
|
|
|
df['result'] = backtest(df.target_returns, df.sign_pred)
|
|
|
|
return df
|
|
|
|
def evaluate_predictions(
|
|
model_name: str,
|
|
target_returns: pd.Series,
|
|
y_pred: pd.Series,
|
|
y_true: pd.Series,
|
|
method: Literal['classification', 'regression'],
|
|
no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'],
|
|
) -> pd.Series:
|
|
# ignore the predictions until we see a non-zero returns (and definitely skip the first sliding_window_size)
|
|
first_nonzero_return = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(y_pred))
|
|
evaluate_from = first_nonzero_return + 1
|
|
|
|
target_returns = pd.Series(target_returns[evaluate_from:])
|
|
if method == 'regression':
|
|
# if there are lots of zeros in the ground truth returns, probably something is wrong, but we can tolerate a couple of days of missing data.
|
|
is_zero = target_returns[target_returns == 0]
|
|
assert len(is_zero) < 15
|
|
|
|
y_pred = pd.Series(y_pred[evaluate_from:])
|
|
|
|
df = __preprocess(target_returns, y_pred, y_true, method, no_of_classes)
|
|
|
|
scorecard = pd.Series()
|
|
if method == 'regression':
|
|
scorecard.loc['RSQ'] = r2_score(df.target_returns, df.y_pred)
|
|
scorecard.loc['MAE'] = mean_absolute_error(df.target_returns, df.y_pred)
|
|
elif method == 'classification':
|
|
scorecard.loc['RSQ'] = 0.
|
|
scorecard.loc['MAE Matrix'] = 0.
|
|
sign_true = df.sign_true.astype(int)
|
|
sign_pred = df.sign_pred.astype(int)
|
|
|
|
def count_non_zero(series: pd.Series) -> int:
|
|
return len(series[series != 0])
|
|
scorecard.loc['no_of_samples'] = count_non_zero(y_pred[evaluate_from:])
|
|
scorecard.loc['sharpe'] = sharpe(df.result)
|
|
scorecard.loc['sortino'] = sortino(df.result)
|
|
scorecard.loc['skew'] = skew(df.result)
|
|
|
|
labels = [1, -1] if no_of_classes == 'two' else [1, -1, 0]
|
|
avg_type = 'weighted' if no_of_classes == 'two' else 'macro'
|
|
|
|
scorecard.loc['accuracy'] = accuracy_score(sign_true, sign_pred) * 100
|
|
scorecard.loc['recall'] = recall_score(sign_true, sign_pred, labels = labels, average=avg_type)
|
|
scorecard.loc['precision'] = precision_score(sign_true, sign_pred, labels = labels, average=avg_type)
|
|
scorecard.loc['f1_score'] = f1_score(sign_true, sign_pred, labels = labels, average=avg_type)
|
|
scorecard.loc['edge'] = df.result.mean()
|
|
scorecard.loc['noise'] = df.y_pred.diff().abs().mean()
|
|
scorecard.loc['edge_to_noise'] = scorecard.loc['edge'] / scorecard.loc['noise']
|
|
|
|
for index, row in sign_true.value_counts().iteritems():
|
|
scorecard.loc['sign_true_ratio_' + str(index)] = row / len(sign_true)
|
|
|
|
for index, row in sign_pred.value_counts().iteritems():
|
|
scorecard.loc['sign_pred_ratio_' + str(index)] = row / len(sign_pred)
|
|
|
|
|
|
if method == 'regression':
|
|
scorecard.loc['edge_to_mae'] = scorecard.loc['edge'] / scorecard.loc['MAE']
|
|
elif method == 'classification':
|
|
scorecard.loc['edge_to_mae'] = 0.
|
|
|
|
scorecard = scorecard.round(3)
|
|
print("Model name: ", model_name)
|
|
print(scorecard)
|
|
return scorecard
|
|
|