mirror of
https://github.com/microsoft/qlib.git
synced 2026-07-14 16:26:55 +08:00
Add files via upload
This commit is contained in:
Wendi Li
@@ -1,249 +1,291 @@
|
|||||||
# Copyright (c) Microsoft Corporation.
|
# Copyright (c) Microsoft Corporation.
|
||||||
# Licensed under the MIT License.
|
# Licensed under the MIT License.
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
import tensorflow.compat.v1 as tf
|
import tensorflow.compat.v1 as tf
|
||||||
import data_formatters.base
|
import data_formatters.base
|
||||||
import expt_settings.configs
|
import expt_settings.configs
|
||||||
import libs.hyperparam_opt
|
import libs.hyperparam_opt
|
||||||
import libs.tft_model
|
import libs.tft_model
|
||||||
import libs.utils as utils
|
import libs.utils as utils
|
||||||
import os
|
import os
|
||||||
import datetime as dte
|
import datetime as dte
|
||||||
|
|
||||||
|
|
||||||
from qlib.model.base import ModelFT
|
from qlib.model.base import ModelFT
|
||||||
from qlib.data.dataset import DatasetH
|
from qlib.data.dataset import DatasetH
|
||||||
from qlib.data.dataset.handler import DataHandlerLP
|
from qlib.data.dataset.handler import DataHandlerLP
|
||||||
|
|
||||||
|
|
||||||
# To register new datasets, please add them here.
|
# To register new datasets, please add them here.
|
||||||
ALLOW_DATASET = ["Alpha158"]
|
ALLOW_DATASET = ["Alpha158", "Alpha360"]
|
||||||
DATASET_SETTING = {
|
# To register new datasets, please add their configurations here.
|
||||||
"Alpha158": {
|
DATASET_SETTING = {
|
||||||
"feature_col": ["RESI5", "WVMA5", "RSQR5", "KLEN", "RSQR10", "CORR5", "CORD5", "CORR10", "ROC60", "RESI10"],
|
"Alpha158": {
|
||||||
"label_col": ["LABEL0"],
|
"feature_col": [
|
||||||
},
|
"RESI5",
|
||||||
}
|
"WVMA5",
|
||||||
# To register new datasets, please add their configurations here.
|
"RSQR5",
|
||||||
|
"KLEN",
|
||||||
|
"RSQR10",
|
||||||
def get_shifted_label(data_df, shifts=5, col_shift="LABEL0"):
|
"CORR5",
|
||||||
return data_df[[col_shift]].groupby("instrument").apply(lambda df: df.shift(shifts))
|
"CORD5",
|
||||||
|
"CORR10",
|
||||||
|
"ROC60",
|
||||||
def fill_test_na(test_df):
|
"RESI10",
|
||||||
test_df_res = test_df.copy()
|
"VSTD5",
|
||||||
feature_cols = ~test_df_res.columns.str.contains("label", case=False)
|
"RSQR60",
|
||||||
test_feature_fna = test_df_res.loc[:, feature_cols].groupby("datetime").apply(lambda df: df.fillna(df.mean()))
|
"CORR60",
|
||||||
test_df_res.loc[:, feature_cols] = test_feature_fna
|
"WVMA60",
|
||||||
return test_df_res
|
"STD5",
|
||||||
|
"RSQR20",
|
||||||
|
"CORD60",
|
||||||
def process_qlib_data(df, dataset, fillna=False):
|
"CORD10",
|
||||||
"""Prepare data to fit the TFT model.
|
"CORR20",
|
||||||
|
"KLOW",
|
||||||
Args:
|
],
|
||||||
df: Original DataFrame.
|
"label_col": "LABEL0",
|
||||||
fillna: Whether to fill the data with the mean values.
|
},
|
||||||
|
"Alpha360": {
|
||||||
Returns:
|
"feature_col": [
|
||||||
Transformed DataFrame.
|
"HIGH0",
|
||||||
|
"LOW0",
|
||||||
"""
|
"OPEN0",
|
||||||
# Several features selected manually
|
"CLOSE1",
|
||||||
feature_col = DATASET_SETTING[dataset]["feature_col"]
|
"HIGH1",
|
||||||
label_col = DATASET_SETTING[dataset]["label_col"]
|
"VOLUME1",
|
||||||
temp_df = df.loc[:, feature_col + label_col]
|
"LOW1",
|
||||||
if fillna:
|
"VOLUME3",
|
||||||
temp_df = fill_test_na(temp_df)
|
"OPEN1",
|
||||||
temp_df = temp_df.swaplevel()
|
"VOLUME4",
|
||||||
temp_df = temp_df.sort_index()
|
"CLOSE2",
|
||||||
temp_df = temp_df.reset_index(level=0)
|
"CLOSE4",
|
||||||
dates = pd.to_datetime(temp_df.index)
|
"VOLUME5",
|
||||||
temp_df["date"] = dates
|
"LOW2",
|
||||||
temp_df["day_of_week"] = dates.dayofweek
|
"CLOSE3",
|
||||||
temp_df["month"] = dates.month
|
"VOLUME2",
|
||||||
temp_df["year"] = dates.year
|
"HIGH2",
|
||||||
temp_df["const"] = 1.0
|
"LOW4",
|
||||||
return temp_df
|
"VOLUME8",
|
||||||
|
"VOLUME11",
|
||||||
|
],
|
||||||
def process_predicted(df, col_name):
|
"label_col": "LABEL0",
|
||||||
"""Transform the TFT predicted data into Qlib format.
|
},
|
||||||
|
}
|
||||||
Args:
|
|
||||||
df: Original DataFrame.
|
|
||||||
fillna: New column name.
|
def get_shifted_label(data_df, shifts=5, col_shift="LABEL0"):
|
||||||
|
return data_df[[col_shift]].groupby("instrument").apply(lambda df: df.shift(shifts))
|
||||||
Returns:
|
|
||||||
Transformed DataFrame.
|
|
||||||
|
def fill_test_na(test_df):
|
||||||
"""
|
test_df_res = test_df.copy()
|
||||||
df_res = df.copy()
|
feature_cols = ~test_df_res.columns.str.contains("label", case=False)
|
||||||
df_res = df_res.rename(columns={"forecast_time": "datetime", "identifier": "instrument", "t+4": col_name})
|
test_feature_fna = test_df_res.loc[:, feature_cols].groupby("datetime").apply(lambda df: df.fillna(df.mean()))
|
||||||
df_res = df_res.set_index(["datetime", "instrument"]).sort_index()
|
test_df_res.loc[:, feature_cols] = test_feature_fna
|
||||||
df_res = df_res[[col_name]]
|
return test_df_res
|
||||||
return df_res
|
|
||||||
|
|
||||||
|
def process_qlib_data(df, dataset, fillna=False):
|
||||||
def format_score(forecast_df, col_name="pred", label_shift=5):
|
"""Prepare data to fit the TFT model.
|
||||||
pred = process_predicted(forecast_df, col_name=col_name)
|
|
||||||
pred = get_shifted_label(pred, shifts=-label_shift, col_shift=col_name)
|
Args:
|
||||||
pred = pred.dropna()[col_name]
|
df: Original DataFrame.
|
||||||
return pred
|
fillna: Whether to fill the data with the mean values.
|
||||||
|
|
||||||
|
Returns:
|
||||||
def transform_df(df, col_name="LABEL0"):
|
Transformed DataFrame.
|
||||||
df_res = df["feature"]
|
|
||||||
df_res[col_name] = df["label"]
|
"""
|
||||||
return df_res
|
# Several features selected manually
|
||||||
|
feature_col = DATASET_SETTING[dataset]["feature_col"]
|
||||||
|
label_col = [DATASET_SETTING[dataset]["label_col"]]
|
||||||
class TFTModel(ModelFT):
|
temp_df = df.loc[:, feature_col + label_col]
|
||||||
"""TFT Model"""
|
if fillna:
|
||||||
|
temp_df = fill_test_na(temp_df)
|
||||||
def __init__(self, **kwargs):
|
temp_df = temp_df.swaplevel()
|
||||||
self.model = None
|
temp_df = temp_df.sort_index()
|
||||||
|
temp_df = temp_df.reset_index(level=0)
|
||||||
def _prepare_data(self, dataset: DatasetH):
|
dates = pd.to_datetime(temp_df.index)
|
||||||
df_train, df_valid = dataset.prepare(
|
temp_df["date"] = dates
|
||||||
["train", "valid"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
|
temp_df["day_of_week"] = dates.dayofweek
|
||||||
)
|
temp_df["month"] = dates.month
|
||||||
return transform_df(df_train), transform_df(df_valid)
|
temp_df["year"] = dates.year
|
||||||
|
temp_df["const"] = 1.0
|
||||||
def fit(
|
return temp_df
|
||||||
self,
|
|
||||||
dataset: DatasetH,
|
|
||||||
DATASET="Alpha158",
|
def process_predicted(df, col_name):
|
||||||
MODEL_FOLDER="qlib_alpha158_model",
|
"""Transform the TFT predicted data into Qlib format.
|
||||||
LABEL_COL="LABEL0",
|
|
||||||
LABEL_SHIFT=5,
|
Args:
|
||||||
USE_GPU_ID=0,
|
df: Original DataFrame.
|
||||||
**kwargs
|
fillna: New column name.
|
||||||
):
|
|
||||||
|
Returns:
|
||||||
if DATASET not in ALLOW_DATASET:
|
Transformed DataFrame.
|
||||||
raise AssertionError("The dataset is not supported, please make a new formatter to fit this dataset")
|
|
||||||
|
"""
|
||||||
dtrain, dvalid = self._prepare_data(dataset)
|
df_res = df.copy()
|
||||||
dtrain.loc[:, LABEL_COL] = get_shifted_label(dtrain, shifts=LABEL_SHIFT, col_shift=LABEL_COL)
|
df_res = df_res.rename(columns={"forecast_time": "datetime", "identifier": "instrument", "t+4": col_name})
|
||||||
dvalid.loc[:, LABEL_COL] = get_shifted_label(dvalid, shifts=LABEL_SHIFT, col_shift=LABEL_COL)
|
df_res = df_res.set_index(["datetime", "instrument"]).sort_index()
|
||||||
|
df_res = df_res[[col_name]]
|
||||||
train = process_qlib_data(dtrain, DATASET, fillna=True).dropna()
|
return df_res
|
||||||
valid = process_qlib_data(dvalid, DATASET, fillna=True).dropna()
|
|
||||||
|
|
||||||
ExperimentConfig = expt_settings.configs.ExperimentConfig
|
def format_score(forecast_df, col_name="pred", label_shift=5):
|
||||||
config = ExperimentConfig(DATASET)
|
pred = process_predicted(forecast_df, col_name=col_name)
|
||||||
self.data_formatter = config.make_data_formatter()
|
pred = get_shifted_label(pred, shifts=-label_shift, col_shift=col_name)
|
||||||
self.model_folder = MODEL_FOLDER
|
pred = pred.dropna()[col_name]
|
||||||
self.gpu_id = USE_GPU_ID
|
return pred
|
||||||
self.label_shift = LABEL_SHIFT
|
|
||||||
self.expt_name = DATASET
|
|
||||||
self.label_col = LABEL_COL
|
def transform_df(df, col_name="LABEL0"):
|
||||||
|
df_res = df["feature"]
|
||||||
use_gpu = (True, self.gpu_id)
|
df_res[col_name] = df["label"]
|
||||||
# ===========================Training Process===========================
|
return df_res
|
||||||
ModelClass = libs.tft_model.TemporalFusionTransformer
|
|
||||||
if not isinstance(self.data_formatter, data_formatters.base.GenericDataFormatter):
|
|
||||||
raise ValueError(
|
class TFTModel(ModelFT):
|
||||||
"Data formatters should inherit from"
|
"""TFT Model"""
|
||||||
+ "AbstractDataFormatter! Type={}".format(type(self.data_formatter))
|
|
||||||
)
|
def __init__(self, **kwargs):
|
||||||
|
self.model = None
|
||||||
default_keras_session = tf.keras.backend.get_session()
|
self.params = {"DATASET": "Alpha158", "label_shift": 5}
|
||||||
|
self.params.update(kwargs)
|
||||||
if use_gpu[0]:
|
|
||||||
self.tf_config = utils.get_default_tensorflow_config(tf_device="gpu", gpu_id=use_gpu[1])
|
def _prepare_data(self, dataset: DatasetH):
|
||||||
else:
|
df_train, df_valid = dataset.prepare(
|
||||||
self.tf_config = utils.get_default_tensorflow_config(tf_device="cpu")
|
["train", "valid"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
|
||||||
|
)
|
||||||
self.data_formatter.set_scalers(train)
|
return transform_df(df_train), transform_df(df_valid)
|
||||||
|
|
||||||
# Sets up default params
|
def fit(self, dataset: DatasetH, MODEL_FOLDER="qlib_tft_model", USE_GPU_ID=0, **kwargs):
|
||||||
fixed_params = self.data_formatter.get_experiment_params()
|
DATASET = self.params["DATASET"]
|
||||||
params = self.data_formatter.get_default_model_params()
|
LABEL_SHIFT = self.params["label_shift"]
|
||||||
|
LABEL_COL = DATASET_SETTING[DATASET]["label_col"]
|
||||||
# Wendi: 合并调优的参数和非调优的参数
|
|
||||||
params = {**params, **fixed_params}
|
if DATASET not in ALLOW_DATASET:
|
||||||
|
raise AssertionError("The dataset is not supported, please make a new formatter to fit this dataset")
|
||||||
if not os.path.exists(self.model_folder):
|
|
||||||
os.makedirs(self.model_folder)
|
dtrain, dvalid = self._prepare_data(dataset)
|
||||||
params["model_folder"] = self.model_folder
|
dtrain.loc[:, LABEL_COL] = get_shifted_label(dtrain, shifts=LABEL_SHIFT, col_shift=LABEL_COL)
|
||||||
|
dvalid.loc[:, LABEL_COL] = get_shifted_label(dvalid, shifts=LABEL_SHIFT, col_shift=LABEL_COL)
|
||||||
print("*** Begin training ***")
|
|
||||||
best_loss = np.Inf
|
train = process_qlib_data(dtrain, DATASET, fillna=True).dropna()
|
||||||
|
valid = process_qlib_data(dvalid, DATASET, fillna=True).dropna()
|
||||||
tf.reset_default_graph()
|
|
||||||
|
ExperimentConfig = expt_settings.configs.ExperimentConfig
|
||||||
self.tf_graph = tf.Graph()
|
config = ExperimentConfig(DATASET)
|
||||||
with self.tf_graph.as_default():
|
self.data_formatter = config.make_data_formatter()
|
||||||
self.sess = tf.Session(config=self.tf_config)
|
self.model_folder = MODEL_FOLDER
|
||||||
tf.keras.backend.set_session(self.sess)
|
self.gpu_id = USE_GPU_ID
|
||||||
self.model = ModelClass(params, use_cudnn=use_gpu[0])
|
self.label_shift = LABEL_SHIFT
|
||||||
self.sess.run(tf.global_variables_initializer())
|
self.expt_name = DATASET
|
||||||
self.model.fit(train_df=train, valid_df=valid)
|
self.label_col = LABEL_COL
|
||||||
print("*** Finished training ***")
|
|
||||||
saved_model_dir = self.model_folder + "/" + "saved_model"
|
use_gpu = (True, self.gpu_id)
|
||||||
if not os.path.exists(saved_model_dir):
|
# ===========================Training Process===========================
|
||||||
os.makedirs(saved_model_dir)
|
ModelClass = libs.tft_model.TemporalFusionTransformer
|
||||||
self.model.save(saved_model_dir)
|
if not isinstance(self.data_formatter, data_formatters.base.GenericDataFormatter):
|
||||||
|
raise ValueError(
|
||||||
def extract_numerical_data(data):
|
"Data formatters should inherit from"
|
||||||
"""Strips out forecast time and identifier columns."""
|
+ "AbstractDataFormatter! Type={}".format(type(self.data_formatter))
|
||||||
return data[[col for col in data.columns if col not in {"forecast_time", "identifier"}]]
|
)
|
||||||
|
|
||||||
# p50_loss = utils.numpy_normalised_quantile_loss(
|
default_keras_session = tf.keras.backend.get_session()
|
||||||
# extract_numerical_data(targets), extract_numerical_data(p50_forecast),
|
|
||||||
# 0.5)
|
if use_gpu[0]:
|
||||||
# p90_loss = utils.numpy_normalised_quantile_loss(
|
self.tf_config = utils.get_default_tensorflow_config(tf_device="gpu", gpu_id=use_gpu[1])
|
||||||
# extract_numerical_data(targets), extract_numerical_data(p90_forecast),
|
else:
|
||||||
# 0.9)
|
self.tf_config = utils.get_default_tensorflow_config(tf_device="cpu")
|
||||||
tf.keras.backend.set_session(default_keras_session)
|
|
||||||
print("Training completed.".format(dte.datetime.now()))
|
self.data_formatter.set_scalers(train)
|
||||||
# ===========================Training Process===========================
|
|
||||||
|
# Sets up default params
|
||||||
def predict(self, dataset):
|
fixed_params = self.data_formatter.get_experiment_params()
|
||||||
if self.model is None:
|
params = self.data_formatter.get_default_model_params()
|
||||||
raise ValueError("model is not fitted yet!")
|
|
||||||
d_test = dataset.prepare("test", col_set=["feature", "label"])
|
# Wendi: 合并调优的参数和非调优的参数
|
||||||
d_test = transform_df(d_test)
|
params = {**params, **fixed_params}
|
||||||
d_test.loc[:, self.label_col] = get_shifted_label(d_test, shifts=self.label_shift, col_shift=self.label_col)
|
|
||||||
test = process_qlib_data(d_test, self.expt_name, fillna=True).dropna()
|
if not os.path.exists(self.model_folder):
|
||||||
|
os.makedirs(self.model_folder)
|
||||||
use_gpu = (True, self.gpu_id)
|
params["model_folder"] = self.model_folder
|
||||||
# ===========================Predicting Process===========================
|
|
||||||
default_keras_session = tf.keras.backend.get_session()
|
print("*** Begin training ***")
|
||||||
|
best_loss = np.Inf
|
||||||
# Sets up default params
|
|
||||||
fixed_params = self.data_formatter.get_experiment_params()
|
tf.reset_default_graph()
|
||||||
params = self.data_formatter.get_default_model_params()
|
|
||||||
params = {**params, **fixed_params}
|
self.tf_graph = tf.Graph()
|
||||||
|
with self.tf_graph.as_default():
|
||||||
print("*** Begin predicting ***")
|
self.sess = tf.Session(config=self.tf_config)
|
||||||
tf.reset_default_graph()
|
tf.keras.backend.set_session(self.sess)
|
||||||
|
self.model = ModelClass(params, use_cudnn=use_gpu[0])
|
||||||
with self.tf_graph.as_default():
|
self.sess.run(tf.global_variables_initializer())
|
||||||
tf.keras.backend.set_session(self.sess)
|
self.model.fit(train_df=train, valid_df=valid)
|
||||||
output_map = self.model.predict(test, return_targets=True)
|
print("*** Finished training ***")
|
||||||
targets = self.data_formatter.format_predictions(output_map["targets"])
|
saved_model_dir = self.model_folder + "/" + "saved_model"
|
||||||
p50_forecast = self.data_formatter.format_predictions(output_map["p50"])
|
if not os.path.exists(saved_model_dir):
|
||||||
p90_forecast = self.data_formatter.format_predictions(output_map["p90"])
|
os.makedirs(saved_model_dir)
|
||||||
tf.keras.backend.set_session(default_keras_session)
|
self.model.save(saved_model_dir)
|
||||||
|
|
||||||
predict50 = format_score(p50_forecast, "pred", 1)
|
def extract_numerical_data(data):
|
||||||
predict90 = format_score(p90_forecast, "pred", 1)
|
"""Strips out forecast time and identifier columns."""
|
||||||
predict = (predict50 + predict90) / 2 # self.label_shift
|
return data[[col for col in data.columns if col not in {"forecast_time", "identifier"}]]
|
||||||
# ===========================Predicting Process===========================
|
|
||||||
return predict
|
# p50_loss = utils.numpy_normalised_quantile_loss(
|
||||||
|
# extract_numerical_data(targets), extract_numerical_data(p50_forecast),
|
||||||
def finetune(self, dataset: DatasetH):
|
# 0.5)
|
||||||
"""
|
# p90_loss = utils.numpy_normalised_quantile_loss(
|
||||||
finetune model
|
# extract_numerical_data(targets), extract_numerical_data(p90_forecast),
|
||||||
Parameters
|
# 0.9)
|
||||||
----------
|
tf.keras.backend.set_session(default_keras_session)
|
||||||
dataset : DatasetH
|
print("Training completed.".format(dte.datetime.now()))
|
||||||
dataset for finetuning
|
# ===========================Training Process===========================
|
||||||
"""
|
|
||||||
pass
|
def predict(self, dataset):
|
||||||
|
if self.model is None:
|
||||||
|
raise ValueError("model is not fitted yet!")
|
||||||
|
d_test = dataset.prepare("test", col_set=["feature", "label"])
|
||||||
|
d_test = transform_df(d_test)
|
||||||
|
d_test.loc[:, self.label_col] = get_shifted_label(d_test, shifts=self.label_shift, col_shift=self.label_col)
|
||||||
|
test = process_qlib_data(d_test, self.expt_name, fillna=True).dropna()
|
||||||
|
|
||||||
|
use_gpu = (True, self.gpu_id)
|
||||||
|
# ===========================Predicting Process===========================
|
||||||
|
default_keras_session = tf.keras.backend.get_session()
|
||||||
|
|
||||||
|
# Sets up default params
|
||||||
|
fixed_params = self.data_formatter.get_experiment_params()
|
||||||
|
params = self.data_formatter.get_default_model_params()
|
||||||
|
params = {**params, **fixed_params}
|
||||||
|
|
||||||
|
print("*** Begin predicting ***")
|
||||||
|
tf.reset_default_graph()
|
||||||
|
|
||||||
|
with self.tf_graph.as_default():
|
||||||
|
tf.keras.backend.set_session(self.sess)
|
||||||
|
output_map = self.model.predict(test, return_targets=True)
|
||||||
|
targets = self.data_formatter.format_predictions(output_map["targets"])
|
||||||
|
p50_forecast = self.data_formatter.format_predictions(output_map["p50"])
|
||||||
|
p90_forecast = self.data_formatter.format_predictions(output_map["p90"])
|
||||||
|
tf.keras.backend.set_session(default_keras_session)
|
||||||
|
|
||||||
|
predict50 = format_score(p50_forecast, "pred", 1)
|
||||||
|
predict90 = format_score(p90_forecast, "pred", 1)
|
||||||
|
predict = (predict50 + predict90) / 2 # self.label_shift
|
||||||
|
# ===========================Predicting Process===========================
|
||||||
|
return predict
|
||||||
|
|
||||||
|
def finetune(self, dataset: DatasetH):
|
||||||
|
"""
|
||||||
|
finetune model
|
||||||
|
Parameters
|
||||||
|
----------
|
||||||
|
dataset : DatasetH
|
||||||
|
dataset for finetuning
|
||||||
|
"""
|
||||||
|
pass
|
||||||
|
|||||||
Reference in New Issue
Block a user