194 lines
8.2 KiB
Python
194 lines
8.2 KiB
Python
import logging
|
|
from typing import Any, Dict, Tuple
|
|
|
|
import pandas as pd
|
|
from catboost import CatBoostRegressor, Pool
|
|
from pandas import DataFrame
|
|
|
|
from freqtrade.freqai.freqai_interface import IFreqaiModel
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CatboostPredictionModel(IFreqaiModel):
|
|
"""
|
|
User created prediction model. The class needs to override three necessary
|
|
functions, predict(), train(), fit(). The class inherits ModelHandler which
|
|
has its own DataHandler where data is held, saved, loaded, and managed.
|
|
"""
|
|
|
|
def make_labels(self, dataframe: DataFrame) -> DataFrame:
|
|
"""
|
|
User defines the labels here (target values).
|
|
:params:
|
|
:dataframe: the full dataframe for the present training period
|
|
"""
|
|
|
|
dataframe["s"] = (
|
|
dataframe["close"]
|
|
.shift(-self.feature_parameters["period"])
|
|
.rolling(self.feature_parameters["period"])
|
|
.mean()
|
|
/ dataframe["close"]
|
|
- 1
|
|
)
|
|
self.dh.data["s_mean"] = dataframe["s"].mean()
|
|
self.dh.data["s_std"] = dataframe["s"].std()
|
|
|
|
# logger.info("label mean", self.dh.data["s_mean"], "label std", self.dh.data["s_std"])
|
|
|
|
return dataframe["s"]
|
|
|
|
def train(self, unfiltered_dataframe: DataFrame, metadata: dict) -> Tuple[DataFrame, DataFrame]:
|
|
"""
|
|
Filter the training data and train a model to it. Train makes heavy use of the datahkitchen
|
|
for storing, saving, loading, and analyzing the data.
|
|
:params:
|
|
:unfiltered_dataframe: Full dataframe for the current training period
|
|
:metadata: pair metadata from strategy.
|
|
:returns:
|
|
:model: Trained model which can be used to inference (self.predict)
|
|
"""
|
|
logger.info("--------------------Starting training--------------------")
|
|
|
|
# create the full feature list based on user config info
|
|
self.dh.training_features_list = self.dh.find_features(unfiltered_dataframe)
|
|
unfiltered_labels = self.make_labels(unfiltered_dataframe)
|
|
# filter the features requested by user in the configuration file and elegantly handle NaNs
|
|
features_filtered, labels_filtered = self.dh.filter_features(
|
|
unfiltered_dataframe,
|
|
self.dh.training_features_list,
|
|
unfiltered_labels,
|
|
training_filter=True,
|
|
)
|
|
|
|
# split data into train/test data.
|
|
data_dictionary = self.dh.make_train_test_datasets(features_filtered, labels_filtered)
|
|
# standardize all data based on train_dataset only
|
|
data_dictionary = self.dh.standardize_data(data_dictionary)
|
|
|
|
# optional additional data cleaning/analysis
|
|
self.data_cleaning_train()
|
|
|
|
logger.info(f'Training model on {len(self.dh.training_features_list)} features')
|
|
logger.info(f'Training model on {len(data_dictionary["train_features"])} data points')
|
|
|
|
model = self.fit(data_dictionary)
|
|
|
|
logger.info(f'--------------------done training {metadata["pair"]}--------------------')
|
|
|
|
return model
|
|
|
|
def fit(self, data_dictionary: Dict) -> Any:
|
|
"""
|
|
User sets up the training and test data to fit their desired model here
|
|
:params:
|
|
:data_dictionary: the dictionary constructed by DataHandler to hold
|
|
all the training and test data/labels.
|
|
"""
|
|
|
|
train_data = Pool(
|
|
data=data_dictionary["train_features"],
|
|
label=data_dictionary["train_labels"],
|
|
weight=data_dictionary["train_weights"],
|
|
)
|
|
|
|
test_data = Pool(
|
|
data=data_dictionary["test_features"],
|
|
label=data_dictionary["test_labels"],
|
|
weight=data_dictionary["test_weights"],
|
|
)
|
|
|
|
model = CatBoostRegressor(
|
|
allow_writing_files=False,
|
|
verbose=100, early_stopping_rounds=400, **self.model_training_parameters
|
|
)
|
|
model.fit(X=train_data, eval_set=test_data)
|
|
|
|
return model
|
|
|
|
def predict(self, unfiltered_dataframe: DataFrame, metadata: dict) -> Tuple[DataFrame,
|
|
DataFrame]:
|
|
"""
|
|
Filter the prediction features data and predict with it.
|
|
:param: unfiltered_dataframe: Full dataframe for the current backtest period.
|
|
:return:
|
|
:predictions: np.array of predictions
|
|
:do_predict: np.array of 1s and 0s to indicate places where freqai needed to remove
|
|
data (NaNs) or felt uncertain about data (PCA and DI index)
|
|
"""
|
|
|
|
# logger.info("--------------------Starting prediction--------------------")
|
|
|
|
original_feature_list = self.dh.find_features(unfiltered_dataframe)
|
|
filtered_dataframe, _ = self.dh.filter_features(
|
|
unfiltered_dataframe, original_feature_list, training_filter=False
|
|
)
|
|
filtered_dataframe = self.dh.standardize_data_from_metadata(filtered_dataframe)
|
|
self.dh.data_dictionary["prediction_features"] = filtered_dataframe
|
|
|
|
# optional additional data cleaning/analysis
|
|
self.data_cleaning_predict(filtered_dataframe)
|
|
|
|
predictions = self.model.predict(self.dh.data_dictionary["prediction_features"])
|
|
|
|
# compute the non-standardized predictions
|
|
self.dh.predictions = (predictions + 1) * (self.dh.data["labels_max"] -
|
|
self.dh.data["labels_min"]) / 2 + self.dh.data[
|
|
"labels_min"]
|
|
|
|
# logger.info("--------------------Finished prediction--------------------")
|
|
|
|
return (self.dh.predictions, self.dh.do_predict)
|
|
|
|
def data_cleaning_train(self) -> None:
|
|
"""
|
|
User can add data analysis and cleaning here.
|
|
Any function inside this method should drop training data points from the filtered_dataframe
|
|
based on user decided logic. See FreqaiDataKitchen::remove_outliers() for an example
|
|
of how outlier data points are dropped from the dataframe used for training.
|
|
"""
|
|
if self.feature_parameters["principal_component_analysis"]:
|
|
self.dh.principal_component_analysis()
|
|
|
|
# if self.feature_parameters["determine_statistical_distributions"]:
|
|
# self.dh.determine_statistical_distributions()
|
|
# if self.feature_parameters["remove_outliers"]:
|
|
# self.dh.remove_outliers(predict=False)
|
|
|
|
if self.feature_parameters["use_SVM_to_remove_outliers"]:
|
|
self.dh.use_SVM_to_remove_outliers(predict=False)
|
|
if self.feature_parameters["DI_threshold"]:
|
|
self.dh.data["avg_mean_dist"] = self.dh.compute_distances()
|
|
|
|
def data_cleaning_predict(self, filtered_dataframe: DataFrame) -> None:
|
|
"""
|
|
User can add data analysis and cleaning here.
|
|
These functions each modify self.dh.do_predict, which is a dataframe with equal length
|
|
to the number of candles coming from and returning to the strategy. Inside do_predict,
|
|
1 allows prediction and < 0 signals to the strategy that the model is not confident in
|
|
the prediction.
|
|
See FreqaiDataKitchen::remove_outliers() for an example
|
|
of how the do_predict vector is modified. do_predict is ultimately passed back to strategy
|
|
for buy signals.
|
|
"""
|
|
if self.feature_parameters["principal_component_analysis"]:
|
|
pca_components = self.dh.pca.transform(filtered_dataframe)
|
|
self.dh.data_dictionary["prediction_features"] = pd.DataFrame(
|
|
data=pca_components,
|
|
columns=["PC" + str(i) for i in range(0, self.dh.data["n_kept_components"])],
|
|
index=filtered_dataframe.index,
|
|
)
|
|
|
|
# if self.feature_parameters["determine_statistical_distributions"]:
|
|
# self.dh.determine_statistical_distributions()
|
|
# if self.feature_parameters["remove_outliers"]:
|
|
# self.dh.remove_outliers(predict=True) # creates dropped index
|
|
|
|
if self.feature_parameters["use_SVM_to_remove_outliers"]:
|
|
self.dh.use_SVM_to_remove_outliers(predict=True)
|
|
|
|
if self.feature_parameters["DI_threshold"]:
|
|
self.dh.check_if_pred_in_training_spaces() # sets do_predict
|