refactor(Evaluate): print out accuracy, f1, etc. for the final & meta predictions, separated out evaluation step (#228)

* refactor(Evaluate): print out accuracy, f1, etc. for the final & meta predictions, separated out evaluation step

* fix(Linter): ran

* fix(Tests): syntax change

* fix(Inference): runs now again

* fix(Linter): ran

utils/evaluate.py

@@ -1,16 +1,15 @@
-from typing import Literal, Callable
+from typing import Callable
 from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
 from quantstats.stats import skew, sortino
 from utils.metrics import probabilistic_sharpe_ratio, sharpe_ratio
 from utils.helpers import get_first_valid_return_index
 import pandas as pd
-import numpy as np
 from data_loader.types import ForwardReturnSeries, ySeries
 from training.types import Stats, WeightsSeries
 
 
 def backtest(
-    returns: pd.Series, signal: pd.Series, transaction_cost=0.002
+    returns: pd.Series, signal: pd.Series, transaction_cost: float
 ) -> pd.Series:
     delta_pos = signal.diff(1).abs().fillna(0.0)
     costs = transaction_cost * delta_pos
@@ -21,19 +20,18 @@ def __preprocess(
     forward_returns: ForwardReturnSeries,
     y_pred: pd.Series,
     y_true: pd.Series,
-    no_of_classes: Literal["two", "three-balanced", "three-imbalanced"],
-    discretize: bool,
+    discretize_func: Callable,
+    transaction_costs: float,
 ) -> pd.DataFrame:
     y_pred.name = "y_pred"
     forward_returns.name = "forward_returns"
     df = pd.concat([y_pred, forward_returns], axis=1).dropna()
 
-    discretize_func = get_discretize_function(no_of_classes)
     # make sure that we evaluate binary/three-way predictions even if the model is a regression
-    df["sign_pred"] = df.y_pred.apply(discretize_func) if discretize else df.y_pred
+    df["sign_pred"] = df.y_pred.apply(discretize_func)
     df["sign_true"] = y_true
 
-    df["result"] = backtest(df.forward_returns, df.sign_pred)
+    df["result"] = backtest(df.forward_returns, df.y_pred, transaction_costs)
 
     return df
 
@@ -42,8 +40,9 @@ def evaluate_predictions(
     forward_returns: ForwardReturnSeries,
     y_pred: WeightsSeries,
     y_true: ySeries,
-    no_of_classes: Literal["two", "three-balanced", "three-imbalanced"],
-    discretize: bool = False,
+    discretize_func: Callable,
+    labels: list[int],
+    transaction_costs: float,
 ) -> Stats:
     # ignore the predictions until we see a non-zero returns (and definitely skip the first sliding_window_size)
     evaluate_from = max(
@@ -54,7 +53,9 @@ def evaluate_predictions(
     forward_returns = pd.Series(forward_returns[evaluate_from:])
     y_pred = pd.Series(y_pred[evaluate_from:])
 
-    df = __preprocess(forward_returns, y_pred, y_true, no_of_classes, discretize)
+    df = __preprocess(
+        forward_returns, y_pred, y_true, discretize_func, transaction_costs
+    )
 
     scorecard = dict()
 
@@ -73,60 +74,27 @@ def evaluate_predictions(
     scorecard["sortino"] = sortino(df.result)
     scorecard["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"
+    avg_type = "weighted" if len(labels) == 2 else "macro"
 
-    if discretize == True:
-        scorecard["accuracy"] = accuracy_score(df.sign_true, df.sign_pred) * 100
-        scorecard["recall"] = recall_score(
-            df.sign_true, df.sign_pred, labels=labels, average=avg_type
-        )
-        scorecard["precision"] = precision_score(
-            df.sign_true, df.sign_pred, labels=labels, average=avg_type
-        )
-        scorecard["f1_score"] = f1_score(
-            df.sign_true, df.sign_pred, labels=labels, average=avg_type
-        )
+    scorecard["accuracy"] = accuracy_score(df.sign_true, df.sign_pred) * 100
+    scorecard["recall"] = recall_score(
+        df.sign_true, df.sign_pred, labels=labels, average=avg_type
+    )
+    scorecard["precision"] = precision_score(
+        df.sign_true, df.sign_pred, labels=labels, average=avg_type
+    )
+    scorecard["f1_score"] = f1_score(
+        df.sign_true, df.sign_pred, labels=labels, average=avg_type
+    )
     scorecard["edge"] = df.result.mean()
     scorecard["noise"] = df.y_pred.diff().abs().mean()
     scorecard["edge_to_noise"] = scorecard["edge"] / (scorecard["noise"] + 0.00001)
 
-    if discretize == True:
-        for index, row in df.sign_true.value_counts().iteritems():
-            scorecard["sign_true_ratio_" + str(index)] = row / len(df.sign_true)
+    for index, row in df.sign_true.value_counts().iteritems():
+        scorecard["sign_true_ratio_" + str(index)] = row / len(df.sign_true)
 
-        for index, row in df.sign_pred.value_counts().iteritems():
-            scorecard["sign_pred_ratio_" + str(index)] = row / len(df.sign_pred)
+    for index, row in df.sign_pred.value_counts().iteritems():
+        scorecard["sign_pred_ratio_" + str(index)] = row / len(df.sign_pred)
 
     scorecard = {k: round(float(v), 3) for k, v in scorecard.items()}
     return scorecard
-
-
-def __discretize_binary(x):
-    return 1 if x > 0 else -1
-
-
-def __discretize_threeway(x):
-    return 0 if x == 0 else 1 if x > 0 else -1
-
-
-def discretize_threeway_threshold(threshold: float) -> Callable:
-    def discretize(current_value):
-        lower_threshold = -threshold
-        upper_threshold = threshold
-        if np.isnan(current_value):
-            return np.nan
-        elif current_value <= lower_threshold:
-            return -1
-        elif current_value > lower_threshold and current_value < upper_threshold:
-            return 0
-        else:
-            return 1
-
-    return discretize
-
-
-def get_discretize_function(
-    no_of_classes: Literal["two", "three-balanced", "three-imbalanced"]
-) -> Callable:
-    return __discretize_binary if no_of_classes == "two" else __discretize_threeway