liuyu/qlib
1
0
Fork 0
mirror of https://github.com/microsoft/qlib.git synced 2026-07-17 09:24:34 +08:00
Code Activity

Merge branch 'main' into dnn_drop

Browse Source
This commit is contained in:
bxdd
2020-11-26 23:04:34 -06:00
committed by GitHub
parent a144a9c3c6 c8355f9f18
commit c31bf61278
105 changed files with 6034 additions and 2725 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
qlib/contrib/data/handler.py
View File

@@ -10,6 +10,28 @@ from inspect import getfullargspec
import copy
def check_transform_proc(proc_l, fit_start_time, fit_end_time):
new_l = []
for p in proc_l:
if not isinstance(p, Processor):
klass, pkwargs = get_cls_kwargs(p, processor_module)
args = getfullargspec(klass).args
if "fit_start_time" in args and "fit_end_time" in args:
assert (
fit_start_time is not None and fit_end_time is not None
), "Make sure `fit_start_time` and `fit_end_time` are not None."
pkwargs.update(
{
"fit_start_time": fit_start_time,
"fit_end_time": fit_end_time,
}
)
new_l.append({"class": klass.__name__, "kwargs": pkwargs})
else:
new_l.append(p)
return new_l
class ALPHA360_Denoise(DataHandlerLP):
def __init__(self, instruments="csi500", start_time=None, end_time=None, fit_start_time=None, fit_end_time=None):
data_loader = {
@@ -83,28 +105,42 @@ class ALPHA360_Denoise(DataHandlerLP):
return fields, names
_DEFAULT_LEARN_PROCESSORS = [
{"class": "DropnaLabel"},
{"class": "CSZScoreNorm", "kwargs": {"fields_group": "label"}},
]
_DEFAULT_INFER_PROCESSORS = [
{"class": "ProcessInf", "kwargs": {}},
{"class": "ZScoreNorm", "kwargs": {}},
{"class": "Fillna", "kwargs": {}},
]
class ALPHA360(DataHandlerLP):
def __init__(self, instruments="csi500", start_time=None, end_time=None, fit_start_time=None, fit_end_time=None):
def __init__(
self,
instruments="csi500",
start_time=None,
end_time=None,
infer_processors=_DEFAULT_INFER_PROCESSORS,
learn_processors=_DEFAULT_LEARN_PROCESSORS,
fit_start_time=None,
fit_end_time=None,
**kwargs,
):
infer_processors = check_transform_proc(infer_processors, fit_start_time, fit_end_time)
learn_processors = check_transform_proc(learn_processors, fit_start_time, fit_end_time)
data_loader = {
"class": "QlibDataLoader",
"kwargs": {
"config": {
"feature": self.get_feature_config(),
"label": self.get_label_config(),
"label": kwargs.get("label", self.get_label_config()),
},
},
}
learn_processors = [
{"class": "DropnaLabel", "kwargs": {"fields_group": "label"}},
{"class": "CSZScoreNorm", "kwargs": {"fields_group": "label"}},
]
infer_processors = [
{"class": "ProcessInf", "kwargs": {}},
{"class": "ZscoreNorm", "kwargs": {"fit_start_time": fit_start_time, "fit_end_time": fit_end_time}},
{"class": "Fillna", "kwargs": {}},
]
super().__init__(
instruments,
start_time,
@@ -168,39 +204,19 @@ class Alpha158(DataHandlerLP):
start_time=None,
end_time=None,
infer_processors=[],
learn_processors=["DropnaLabel", {"class": "CSZScoreNorm", "kwargs": {"fields_group": "label"}}],
learn_processors=_DEFAULT_LEARN_PROCESSORS,
fit_start_time=None,
fit_end_time=None,
process_type=DataHandlerLP.PTYPE_A
**kwargs,
):
def check_transform_proc(proc_l):
new_l = []
for p in proc_l:
if not isinstance(p, Processor):
klass, pkwargs = get_cls_kwargs(p, processor_module)
args = getfullargspec(klass).args
if "fit_start_time" in args and "fit_end_time" in args:
assert (
fit_start_time is not None and fit_end_time is not None
), "Make sure `fit_start_time` and `fit_end_time` are not None."
pkwargs.update(
{
"fit_start_time": fit_start_time,
"fit_end_time": fit_end_time,
}
)
new_l.append({"class": klass.__name__, "kwargs": pkwargs})
else:
new_l.append(p)
return new_l
infer_processors = check_transform_proc(infer_processors)
learn_processors = check_transform_proc(learn_processors)
infer_processors = check_transform_proc(infer_processors, fit_start_time, fit_end_time)
learn_processors = check_transform_proc(learn_processors, fit_start_time, fit_end_time)
data_loader = {
"class": "QlibDataLoader",
"kwargs": {
"config": {"feature": self.get_feature_config(), "label": self.get_label_config()},
"config": {"feature": self.get_feature_config(), "label": kwargs.get("label", self.get_label_config())},
},
}
super().__init__(

0
qlib/contrib/estimator/__init__.py
View File

176
qlib/contrib/estimator/config.py
View File

@@ -1,176 +0,0 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import yaml
import copy
import os
import json
import tempfile
from pathlib import Path
from ...config import REG_CN
class EstimatorConfigManager(object):
def __init__(self, config_path):
if not config_path:
raise ValueError("Config path is invalid.")
self.config_path = config_path
with open(config_path) as fp:
config = yaml.load(fp, Loader=yaml.FullLoader)
self.config = copy.deepcopy(config)
self.ex_config = ExperimentConfig(config.get("experiment", dict()), self)
self.data_config = DataConfig(config.get("data", dict()), self)
self.model_config = ModelConfig(config.get("model", dict()), self)
self.trainer_config = TrainerConfig(config.get("trainer", dict()), self)
self.strategy_config = StrategyConfig(config.get("strategy", dict()), self)
self.backtest_config = BacktestConfig(config.get("backtest", dict()), self)
self.qlib_data_config = QlibDataConfig(config.get("qlib_data", dict()), self)
# If the start_date and end_date are not given in data_config, they will be referred from the trainer_config.
handler_start_date = self.data_config.handler_parameters.get("start_date", None)
handler_end_date = self.data_config.handler_parameters.get("end_date", None)
if handler_start_date is None:
self.data_config.handler_parameters["start_date"] = self.trainer_config.parameters["train_start_date"]
if handler_end_date is None:
self.data_config.handler_parameters["end_date"] = self.trainer_config.parameters["test_end_date"]
class ExperimentConfig(object):
TRAIN_MODE = "train"
TEST_MODE = "test"
OBSERVER_FILE_STORAGE = "file_storage"
OBSERVER_MONGO = "mongo"
def __init__(self, config, CONFIG_MANAGER):
"""__init__
:param config: The config dict for experiment
:param CONFIG_MANAGER: The estimator config manager
"""
self.name = config.get("name", "test_experiment")
# The dir of the result of all the experiments
self.global_dir = config.get("dir", os.path.dirname(CONFIG_MANAGER.config_path))
# The dir of the result of current experiment
self.ex_dir = os.path.join(self.global_dir, self.name)
if not os.path.exists(self.ex_dir):
os.makedirs(self.ex_dir)
self.tmp_run_dir = tempfile.mkdtemp(dir=self.ex_dir)
self.mode = config.get("mode", ExperimentConfig.TRAIN_MODE)
self.sacred_dir = os.path.join(self.ex_dir, "sacred")
self.observer_type = config.get("observer_type", ExperimentConfig.OBSERVER_FILE_STORAGE)
self.mongo_url = config.get("mongo_url", None)
self.db_name = config.get("db_name", None)
self.finetune = config.get("finetune", False)
# The path of the experiment id of the experiment
self.exp_info_path = config.get("exp_info_path", os.path.join(self.ex_dir, "exp_info.json"))
exp_info_dir = Path(self.exp_info_path).parent
exp_info_dir.mkdir(parents=True, exist_ok=True)
# Test mode config
loader_args = config.get("loader", dict())
if self.mode == ExperimentConfig.TEST_MODE or self.finetune:
loader_exp_info_path = loader_args.get("exp_info_path", None)
self.loader_model_index = loader_args.get("model_index", None)
if (loader_exp_info_path is not None) and (os.path.exists(loader_exp_info_path)):
with open(loader_exp_info_path) as fp:
loader_dict = json.load(fp)
for k, v in loader_dict.items():
setattr(self, "loader_{}".format(k), v)
# Check loader experiment id
assert hasattr(self, "loader_id"), "If mode is test or finetune is True, loader must contain id."
else:
self.loader_id = loader_args.get("id", None)
if self.loader_id is None:
raise ValueError("If mode is test or finetune is True, loader must contain id.")
self.loader_observer_type = loader_args.get("observer_type", self.observer_type)
self.loader_name = loader_args.get("name", self.name)
self.loader_dir = loader_args.get("dir", self.global_dir)
self.loader_mongo_url = loader_args.get("mongo_url", self.mongo_url)
self.loader_db_name = loader_args.get("db_name", self.db_name)
class DataConfig(object):
def __init__(self, config, CONFIG_MANAGER):
"""__init__
:param config: The config dict for data
:param CONFIG_MANAGER: The estimator config manager
"""
self.handler_module_path = config.get("module_path", "qlib.contrib.data.handler")
self.handler_class = config.get("class", "ALPHA360")
self.handler_parameters = config.get("args", dict())
self.handler_filter = config.get("filter", dict())
# Update provider uri.
class ModelConfig(object):
def __init__(self, config, CONFIG_MANAGER):
"""__init__
:param config: The config dict for model
:param CONFIG_MANAGER: The estimator config manager
"""
self.model_class = config.get("class", "Model")
self.model_module_path = config.get("module_path", "qlib.model")
self.save_dir = os.path.join(CONFIG_MANAGER.ex_config.tmp_run_dir, "model")
self.save_path = config.get("save_path", os.path.join(self.save_dir, "model.bin"))
self.parameters = config.get("args", dict())
# Make dir if need.
if not os.path.exists(self.save_dir):
os.makedirs(self.save_dir)
class TrainerConfig(object):
def __init__(self, config, CONFIG_MANAGER):
"""__init__
:param config: The config dict for trainer
:param CONFIG_MANAGER: The estimator config manager
"""
self.trainer_class = config.get("class", "StaticTrainer")
self.trainer_module_path = config.get("module_path", "qlib.contrib.estimator.trainer")
self.parameters = config.get("args", dict())
class StrategyConfig(object):
def __init__(self, config, CONFIG_MANAGER):
"""__init__
:param config: The config dict for strategy
:param CONFIG_MANAGER: The estimator config manager
"""
self.strategy_class = config.get("class", "TopkDropoutStrategy")
self.strategy_module_path = config.get("module_path", "qlib.contrib.strategy.strategy")
self.parameters = config.get("args", dict())
class BacktestConfig(object):
def __init__(self, config, CONFIG_MANAGE):
"""__init__
:param config: The config dict for strategy
:param CONFIG_MANAGE: The estimator config manager
"""
self.normal_backtest_parameters = config.get("normal_backtest_args", dict())
self.long_short_backtest_parameters = config.get("long_short_backtest_args", dict())
class QlibDataConfig(object):
def __init__(self, config, CONFIG_MANAGE):
"""__init__
:param config: The config dict for qlib_client
:param CONFIG_MANAGE: The estimator config manager
"""
self.provider_uri = config.pop("provider_uri", "~/.qlib/qlib_data/cn_data")
self.auto_mount = config.pop("auto_mount", False)
self.mount_path = config.pop("mount_path", "~/.qlib/qlib_data/cn_data")
self.region = config.pop("region", REG_CN)
self.args = config

328
qlib/contrib/estimator/estimator.py
View File

@@ -1,328 +0,0 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# coding=utf-8
import pandas as pd
import os
import copy
import json
import yaml
import pickle
import qlib
from ..evaluate import risk_analysis
from ..evaluate import backtest as normal_backtest
from ..evaluate import long_short_backtest
from .config import ExperimentConfig
from .fetcher import create_fetcher_with_config
from ...log import get_module_logger, TimeInspector
from ...utils import get_module_by_module_path, compare_dict_value
class Estimator(object):
def __init__(self, config_manager, sacred_ex):
# Set logger.
self.logger = get_module_logger("Estimator")
# 1. Set config manager.
self.config_manager = config_manager
# 2. Set configs.
self.ex_config = config_manager.ex_config
self.data_config = config_manager.data_config
self.model_config = config_manager.model_config
self.trainer_config = config_manager.trainer_config
self.strategy_config = config_manager.strategy_config
self.backtest_config = config_manager.backtest_config
# If experiment.mode is test or experiment.finetune is True, load the experimental results in the loader
if self.ex_config.mode == self.ex_config.TEST_MODE or self.ex_config.finetune:
self.compare_config_with_config_manger(self.config_manager)
# 3. Set sacred_experiment.
self.ex = sacred_ex
# 4. Init data handler.
self.data_handler = None
self._init_data_handler()
# 5. Init trainer.
self.trainer = None
self._init_trainer()
# 6. Init strategy.
self.strategy = None
self._init_strategy()
def _init_data_handler(self):
handler_module = get_module_by_module_path(self.data_config.handler_module_path)
# Set market
market = self.data_config.handler_filter.get("market", None)
if market is None:
if "market" in self.data_config.handler_parameters:
self.logger.warning(
"Warning: The market in data.args section is deprecated. "
"It only works when market is not set in data.filter section. "
"It will be overridden by market in the data.filter section."
)
market = self.data_config.handler_parameters["market"]
else:
market = "csi500"
self.data_config.handler_parameters["market"] = market
data_filter_list = []
handler_filters = self.data_config.handler_filter.get("filter_pipeline", list())
for h_filter in handler_filters:
filter_module_path = h_filter.get("module_path", "qlib.data.filter")
filter_class_name = h_filter.get("class", "")
filter_parameters = h_filter.get("args", {})
filter_module = get_module_by_module_path(filter_module_path)
filter_class = getattr(filter_module, filter_class_name)
data_filter = filter_class(**filter_parameters)
data_filter_list.append(data_filter)
self.data_config.handler_parameters["data_filter_list"] = data_filter_list
handler_class = getattr(handler_module, self.data_config.handler_class)
self.data_handler = handler_class(**self.data_config.handler_parameters)
def _init_trainer(self):
model_module = get_module_by_module_path(self.model_config.model_module_path)
trainer_module = get_module_by_module_path(self.trainer_config.trainer_module_path)
model_class = getattr(model_module, self.model_config.model_class)
trainer_class = getattr(trainer_module, self.trainer_config.trainer_class)
self.trainer = trainer_class(
model_class,
self.model_config.save_path,
self.model_config.parameters,
self.data_handler,
self.ex,
**self.trainer_config.parameters
)
def _init_strategy(self):
module = get_module_by_module_path(self.strategy_config.strategy_module_path)
strategy_class = getattr(module, self.strategy_config.strategy_class)
self.strategy = strategy_class(**self.strategy_config.parameters)
def run(self):
if self.ex_config.mode == ExperimentConfig.TRAIN_MODE:
self.trainer.train()
elif self.ex_config.mode == ExperimentConfig.TEST_MODE:
self.trainer.load()
else:
raise ValueError("unexpected mode: %s" % self.ex_config.mode)
analysis = self.backtest()
print(analysis)
self.logger.info(
"experiment id: {}, experiment name: {}".format(self.ex.experiment.current_run._id, self.ex_config.name)
)
# Remove temp dir
# shutil.rmtree(self.ex_config.tmp_run_dir)
def backtest(self):
TimeInspector.set_time_mark()
# 1. Get pred and prediction score of model(s).
pred = self.trainer.get_test_score()
try:
performance = self.trainer.get_test_performance()
except NotImplementedError:
performance = None
# 2. Normal Backtest.
report_normal, positions_normal = self._normal_backtest(pred)
# 3. Long-Short Backtest.
# Deprecated
# long_short_reports = self._long_short_backtest(pred)
# 4. Analyze
analysis_df = self._analyze(report_normal)
# 5. Save.
self._save_backtest_result(
pred,
analysis_df,
positions_normal,
report_normal,
# long_short_reports,
performance,
)
return analysis_df
def _normal_backtest(self, pred):
TimeInspector.set_time_mark()
if "account" not in self.backtest_config.normal_backtest_parameters:
if "account" in self.strategy_config.parameters:
self.logger.warning(
"Warning: The account in strategy section is deprecated. "
"It only works when account is not set in backtest section. "
"It will be overridden by account in the backtest section."
)
self.backtest_config.normal_backtest_parameters["account"] = self.strategy_config.parameters["account"]
report_normal, positions_normal = normal_backtest(
pred, strategy=self.strategy, **self.backtest_config.normal_backtest_parameters
)
TimeInspector.log_cost_time("Finished normal backtest.")
return report_normal, positions_normal
def _long_short_backtest(self, pred):
TimeInspector.set_time_mark()
long_short_reports = long_short_backtest(pred, **self.backtest_config.long_short_backtest_parameters)
TimeInspector.log_cost_time("Finished long-short backtest.")
return long_short_reports
@staticmethod
def _analyze(report_normal):
TimeInspector.set_time_mark()
analysis = dict()
# analysis["pred_long"] = risk_analysis(long_short_reports["long"])
# analysis["pred_short"] = risk_analysis(long_short_reports["short"])
# analysis["pred_long_short"] = risk_analysis(long_short_reports["long_short"])
analysis["excess_return_without_cost"] = risk_analysis(report_normal["return"] - report_normal["bench"])
analysis["excess_return_with_cost"] = risk_analysis(
report_normal["return"] - report_normal["bench"] - report_normal["cost"]
)
analysis_df = pd.concat(analysis) # type: pd.DataFrame
TimeInspector.log_cost_time(
"Finished generating analysis," " average turnover is: {0:.4f}.".format(report_normal["turnover"].mean())
)
return analysis_df
def _save_backtest_result(self, pred, analysis, positions, report_normal, performance):
# 1. Result dir.
result_dir = os.path.join(self.config_manager.ex_config.tmp_run_dir, "result")
if not os.path.exists(result_dir):
os.makedirs(result_dir)
self.ex.add_info(
"task_config",
json.loads(json.dumps(self.config_manager.config, default=str)),
)
# 2. Pred.
TimeInspector.set_time_mark()
pred_pkl_path = os.path.join(result_dir, "pred.pkl")
pred.to_pickle(pred_pkl_path)
self.ex.add_artifact(pred_pkl_path)
TimeInspector.log_cost_time("Finished saving pred.pkl to: {}".format(pred_pkl_path))
# 3. Ana.
TimeInspector.set_time_mark()
analysis_pkl_path = os.path.join(result_dir, "analysis.pkl")
analysis.to_pickle(analysis_pkl_path)
self.ex.add_artifact(analysis_pkl_path)
TimeInspector.log_cost_time("Finished saving analysis.pkl to: {}".format(analysis_pkl_path))
# 4. Pos.
TimeInspector.set_time_mark()
positions_pkl_path = os.path.join(result_dir, "positions.pkl")
with open(positions_pkl_path, "wb") as fp:
pickle.dump(positions, fp)
self.ex.add_artifact(positions_pkl_path)
TimeInspector.log_cost_time("Finished saving positions.pkl to: {}".format(positions_pkl_path))
# 5. Report normal.
TimeInspector.set_time_mark()
report_normal_pkl_path = os.path.join(result_dir, "report_normal.pkl")
report_normal.to_pickle(report_normal_pkl_path)
self.ex.add_artifact(report_normal_pkl_path)
TimeInspector.log_cost_time("Finished saving report_normal.pkl to: {}".format(report_normal_pkl_path))
# 6. Report long short.
# Deprecated
# for k, name in zip(
# ["long", "short", "long_short"],
# ["report_long.pkl", "report_short.pkl", "report_long_short.pkl"],
# ):
# TimeInspector.set_time_mark()
# pkl_path = os.path.join(result_dir, name)
# long_short_reports[k].to_pickle(pkl_path)
# self.ex.add_artifact(pkl_path)
# TimeInspector.log_cost_time("Finished saving {} to: {}".format(name, pkl_path))
# 7. Origin test label.
TimeInspector.set_time_mark()
label_pkl_path = os.path.join(result_dir, "label.pkl")
self.data_handler.get_origin_test_label_with_date(
self.trainer_config.parameters["test_start_date"],
self.trainer_config.parameters["test_end_date"],
).to_pickle(label_pkl_path)
self.ex.add_artifact(label_pkl_path)
TimeInspector.log_cost_time("Finished saving label.pkl to: {}".format(label_pkl_path))
# 8. Experiment info, save the model(s) performance here.
TimeInspector.set_time_mark()
cur_ex_id = self.ex.experiment.current_run._id
exp_info = {
"id": cur_ex_id,
"name": self.ex_config.name,
"performance": performance,
"observer_type": self.ex_config.observer_type,
}
if self.ex_config.observer_type == ExperimentConfig.OBSERVER_MONGO:
exp_info.update(
{
"mongo_url": self.ex_config.mongo_url,
"db_name": self.ex_config.db_name,
}
)
else:
exp_info.update({"dir": self.ex_config.global_dir})
with open(self.ex_config.exp_info_path, "w") as fp:
json.dump(exp_info, fp, indent=4, sort_keys=True)
self.ex.add_artifact(self.ex_config.exp_info_path)
TimeInspector.log_cost_time("Finished saving ex_info to: {}".format(self.ex_config.exp_info_path))
@staticmethod
def compare_config_with_config_manger(config_manager):
"""Compare loader model args and current config with ConfigManage
:param config_manager: ConfigManager
:return:
"""
fetcher = create_fetcher_with_config(config_manager, load_form_loader=True)
loader_mode_config = fetcher.get_experiment(
exp_name=config_manager.ex_config.loader_name,
exp_id=config_manager.ex_config.loader_id,
fields=["task_config"],
)["task_config"]
with open(config_manager.config_path) as fp:
current_config = yaml.load(fp.read())
current_config = json.loads(json.dumps(current_config, default=str))
logger = get_module_logger("Estimator")
loader_mode_config = copy.deepcopy(loader_mode_config)
current_config = copy.deepcopy(current_config)
# Require test_mode_config.test_start_date <= current_config.test_start_date
loader_trainer_args = loader_mode_config.get("trainer", {}).get("args", {})
cur_trainer_args = current_config.get("trainer", {}).get("args", {})
loader_start_date = loader_trainer_args.pop("test_start_date")
cur_test_start_date = cur_trainer_args.pop("test_start_date")
assert (
loader_start_date <= cur_test_start_date
), "Require: loader_mode_config.test_start_date <= current_config.test_start_date"
# TODO: For the user's own extended `Trainer`, the support is not very good
if "RollingTrainer" == current_config.get("trainer", {}).get("class", None):
loader_period = loader_trainer_args.pop("rolling_period")
cur_period = cur_trainer_args.pop("rolling_period")
assert (
loader_period == cur_period
), "Require: loader_mode_config.rolling_period == current_config.rolling_period"
compare_section = ["trainer", "model", "data"]
for section in compare_section:
changes = compare_dict_value(loader_mode_config.get(section, {}), current_config.get(section, {}))
if changes:
logger.warning("Warning: Loader mode config and current config, `{}` are different:\n".format(section))

290
qlib/contrib/estimator/fetcher.py
View File

@@ -1,290 +0,0 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# coding=utf-8
import copy
import json
import yaml
import pickle
import gridfs
import pymongo
from pathlib import Path
from abc import abstractmethod
from .config import EstimatorConfigManager, ExperimentConfig
class Fetcher(object):
"""Sacred Experiments Fetcher"""
@abstractmethod
def _get_experiment(self, exp_name, exp_id):
"""Get experiment basic info with experiment and experiment id
:param exp_name: experiment name
:param exp_id: experiment id
:return: dict
Must contain keys: _id, experiment, info, stop_time.
Here is an example below for FileFetcher.
exp = {
'_id': exp_id, # experiment id
'path': path, # experiment result path
'experiment': {'name': exp_name}, # experiment
'info': info, # experiment config info
'stop_time': run.get('stop_time', None) # The time the experiment ended
}
"""
pass
@abstractmethod
def _list_experiments(self, exp_name=None):
"""Get experiment basic info list with experiment name
:param exp_name: experiment name
:return: list
"""
pass
@abstractmethod
def _iter_artifacts(self, experiment):
"""Get information about the data in the experiment results
:param experiment: `self._get_experiment` method result
:return: iterable
Each element contains two elements.
first element : data name
second element : data uri
"""
pass
@abstractmethod
def _load_data(self, uri):
"""Load data with uri
:param uri: data uri
:return: bytes
"""
pass
@staticmethod
def model_dict_to_buffer_list(model_dict):
"""
:param model_dict:
:return:
"""
model_list = []
is_static_model = False
if len(model_dict) == 1 and list(model_dict.keys())[0] == "model.bin":
is_static_model = True
model_list.append(list(model_dict.values())[0])
else:
sep = "model.bin_"
model_ids = list(map(lambda x: int(x.split(sep)[1]), model_dict.keys()))
min_id, max_id = min(model_ids), max(model_ids)
for i in range(min_id, max_id + 1):
model_key = sep + str(i)
model = model_dict.get(model_key, None)
if model is None:
print(
"WARNING: In Fetcher, {} is missing when the get model is in the get_experiment function.".format(
model_key
)
)
break
else:
model_list.append(model)
if is_static_model:
return model_list[0]
return model_list
def get_experiments(self, exp_name=None):
"""Get experiments with name.
:param exp_name: str
If `exp_name` is set to None, then all experiments will return.
:return: dict
Experiments info dict(Including experiment id and task_config to run the
experiment). Here is an example below.
{
'a_experiment': [
{
'id': '1',
'task_config': {...}
},
...
]
...
}
"""
res = dict()
for ex in self._list_experiments(exp_name):
name = ex["experiment"]["name"]
tmp = {
"id": ex["_id"],
"task_config": ex["info"].get("task_config", {}),
"ex_run_stop_time": ex.get("stop_time", None),
}
res.setdefault(name, []).append(tmp)
return res
def get_experiment(self, exp_name, exp_id, fields=None):
"""
:param exp_name:
:param exp_id:
:param fields: list
Experiment result fields, if fields is None, will get all fields.
Currently supported fields:
['model', 'analysis', 'positions', 'report_normal', 'pred', 'task_config', 'label']
:return: dict
"""
fields = copy.copy(fields)
ex = self._get_experiment(exp_name, exp_id)
results = dict()
model_dict = dict()
for name, uri in self._iter_artifacts(ex):
# When saving, use `sacred.experiment.add_artifact(filename)` , so `name` is os.path.basename(filename)
prefix = name.split(".")[0]
if fields and prefix not in fields:
continue
data = self._load_data(uri)
if prefix == "model":
model_dict[name] = data
else:
results[prefix] = pickle.loads(data)
# Sort model
if model_dict:
results["model"] = self.model_dict_to_buffer_list(model_dict)
# Info
results["task_config"] = ex["info"].get("task_config", {})
return results
def estimator_config_to_dict(self, exp_name, exp_id):
"""Save configuration to file
:param exp_name:
:param exp_id:
:return: config dict
"""
return self.get_experiment(exp_name, exp_id, fields=["task_config"])["task_config"]
class FileFetcher(Fetcher):
"""File Fetcher"""
def __init__(self, experiments_dir):
self.experiments_dir = Path(experiments_dir)
def _get_experiment(self, exp_name, exp_id):
path = self.experiments_dir / exp_name / "sacred" / str(exp_id)
info_path = path / "info.json"
run_path = path / "run.json"
if info_path.exists():
with info_path.open("r") as f:
info = json.load(f)
else:
info = {}
if run_path.exists():
with run_path.open("r") as f:
run = json.load(f)
else:
run = {}
exp = {
"_id": exp_id,
"path": path,
"experiment": {"name": exp_name},
"info": info,
"stop_time": run.get("stop_time", None),
}
return exp
def _list_experiments(self, exp_name=None):
runs = []
for path in self.experiments_dir.glob("{}/sacred/[!_]*".format(exp_name or "*")):
exp_name, exp_id = path.parents[1].name, path.name
runs.append(self._get_experiment(exp_name, exp_id))
return runs
def _iter_artifacts(self, experiment):
if experiment is None:
return []
for fname in experiment["path"].iterdir():
if fname.suffix == ".pkl" or ".bin" in fname.suffix:
name, uri = fname.name, str(fname)
yield name, uri
def _load_data(self, uri):
with open(uri, "rb") as f:
data = f.read()
return data
class MongoFetcher(Fetcher):
"""MongoDB Fetcher"""
def __init__(self, mongo_url, db_name):
self.mongo_url = mongo_url
self.db_name = db_name
self.client = None
self.db = None
self.runs = None
self.fs = None
self._setup_mongo_client()
def _setup_mongo_client(self):
self.client = pymongo.MongoClient(self.mongo_url)
self.db = self.client[self.db_name]
self.runs = self.db.runs
self.fs = gridfs.GridFS(self.db)
def _get_experiment(self, exp_name, exp_id):
return self.runs.find_one({"_id": exp_id})
def _list_experiments(self, exp_name=None):
if exp_name is None:
return self.runs.find()
return self.runs.find({"experiment.name": exp_name})
def _iter_artifacts(self, experiment):
if experiment is None:
return []
for artifact in experiment.get("artifacts", []):
name, uri = artifact["name"], artifact["file_id"]
yield name, uri
def _load_data(self, uri):
data = self.fs.get(uri).read()
return data
def create_fetcher_with_config(config_manager: EstimatorConfigManager, load_form_loader: bool = False):
"""Create fetcher with loader config
:param config_manager:
:param load_form_loader
:return:
"""
flag = ""
if load_form_loader:
flag = "loader_"
if config_manager.ex_config.observer_type == ExperimentConfig.OBSERVER_FILE_STORAGE:
return FileFetcher(eval("config_manager.ex_config.{}_dir".format("loader" if load_form_loader else "global")))
elif config_manager.ex_config.observer_type == ExperimentConfig.OBSERVER_MONGO:
return MongoFetcher(
mongo_url=eval("config_manager.ex_config.{}mongo_url".format(flag)),
db_name=eval("config_manager.ex_config.{}db_name".format(flag)),
)
else:
return NotImplementedError("Unkown Backend")

115
qlib/contrib/estimator/launcher.py
View File

@@ -1,115 +0,0 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
import argparse
import importlib
from ... import init
from .config import EstimatorConfigManager
from ...log import get_module_logger
from sacred import Experiment
from sacred.observers import FileStorageObserver
from sacred.observers import MongoObserver
args_parser = argparse.ArgumentParser(prog="estimator")
args_parser.add_argument(
"-c",
"--config_path",
required=True,
type=str,
help="json config path indicates where to load config.",
)
args = args_parser.parse_args()
class SacredExperiment(object):
def __init__(
self,
experiment_name,
experiment_dir,
observer_type="file_storage",
mongo_url=None,
db_name=None,
):
"""__init__
:param experiment_name: The name of the experiments.
:param experiment_dir: The directory to store all the results of the experiments(This is for file_storage).
:param observer_type: The observer to record the results: the `file_storage` or `mongo`
:param mongo_url: The mongo url(for mongo observer)
:param db_name: The mongo url(for mongo observer)
"""
self.experiment_name = experiment_name
self.experiment = Experiment(self.experiment_name)
self.experiment_dir = experiment_dir
self.experiment.logger = get_module_logger("Sacred")
self.observer_type = observer_type
self.mongo_db_url = mongo_url
self.mongo_db_name = db_name
self._setup_experiment()
def _setup_experiment(self):
if self.observer_type == "file_storage":
file_storage_observer = FileStorageObserver.create(basedir=self.experiment_dir)
self.experiment.observers.append(file_storage_observer)
elif self.observer_type == "mongo":
mongo_observer = MongoObserver.create(url=self.mongo_db_url, db_name=self.mongo_db_name)
self.experiment.observers.append(mongo_observer)
else:
raise NotImplementedError("Unsupported observer type: {}".format(self.observer_type))
def add_artifact(self, filename):
self.experiment.add_artifact(filename)
def add_info(self, key, value):
self.experiment.info[key] = value
def main_wrapper(self, func):
return self.experiment.main(func)
def config_wrapper(self, func):
return self.experiment.config(func)
CONFIG_MANAGER = EstimatorConfigManager(args.config_path)
ex = SacredExperiment(
CONFIG_MANAGER.ex_config.name,
CONFIG_MANAGER.ex_config.sacred_dir,
observer_type=CONFIG_MANAGER.ex_config.observer_type,
mongo_url=CONFIG_MANAGER.ex_config.mongo_url,
db_name=CONFIG_MANAGER.ex_config.db_name,
)
# qlib init
init(
provider_uri=CONFIG_MANAGER.qlib_data_config.provider_uri,
mount_path=CONFIG_MANAGER.qlib_data_config.mount_path,
auto_mount=CONFIG_MANAGER.qlib_data_config.auto_mount,
region=CONFIG_MANAGER.qlib_data_config.region,
**CONFIG_MANAGER.qlib_data_config.args
)
@ex.main_wrapper
def _main():
# 1. Get estimator class.
estimator_class = getattr(
importlib.import_module(".estimator", package="qlib.contrib.estimator"),
"Estimator",
)
# 2. Init estimator.
estimator = estimator_class(CONFIG_MANAGER, ex)
estimator.run()
def run():
ex.experiment.run()
if __name__ == "__main__":
run()

317
qlib/contrib/estimator/trainer.py
View File

@@ -1,317 +0,0 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# coding=utf-8
from abc import abstractmethod
import pandas as pd
import numpy as np
from scipy.stats import pearsonr
from ...log import get_module_logger, TimeInspector
from ...data.dataset.handler import DataHandlerLP
from .launcher import CONFIG_MANAGER
from .fetcher import create_fetcher_with_config
from ...utils import drop_nan_by_y_index, transform_end_date
class BaseTrainer(object):
def __init__(self, model_class, model_save_path, model_args, data_handler: DataHandlerLP, sacred_ex, **kwargs):
# 1. Model.
self.model_class = model_class
self.model_save_path = model_save_path
self.model_args = model_args
# 2. Data handler.
self.data_handler = data_handler
# 3. Sacred ex.
self.ex = sacred_ex
# 4. Logger.
self.logger = get_module_logger("Trainer")
# 5. Data time
self.train_start_date = kwargs.get("train_start_date", None)
self.train_end_date = kwargs.get("train_end_date", None)
self.validate_start_date = kwargs.get("validate_start_date", None)
self.validate_end_date = kwargs.get("validate_end_date", None)
self.test_start_date = kwargs.get("test_start_date", None)
self.test_end_date = transform_end_date(kwargs.get("test_end_date", None))
@abstractmethod
def train(self):
"""
Implement this method indicating how to train a model.
"""
pass
@abstractmethod
def load(self):
"""
Implement this method indicating how to restore a model and the data.
"""
pass
@abstractmethod
def get_test_pred(self):
"""
Implement this method indicating how to get prediction result(s) from a model.
"""
pass
def get_test_performance(self):
"""
Implement this method indicating how to get the performance of the model.
"""
raise NotImplementedError(f"Please implement `get_test_performance`")
def get_test_score(self):
"""
Override this method to transfer the predict result(s) into the score of the stock.
Note: If this is a multi-label training, you need to transfer predict labels into one score.
Or you can just use the result of `get_test_pred()` (you can also process the result) if this is one label training.
We use the first column of the result of `get_test_pred()` as default method (regard it as one label training).
"""
pred = self.get_test_pred()
pred_score = pd.DataFrame(index=pred.index)
pred_score["score"] = pred.iloc(axis=1)[0]
return pred_score
class StaticTrainer(BaseTrainer):
def __init__(self, model_class, model_save_path, model_args, data_handler, sacred_ex, **kwargs):
super(StaticTrainer, self).__init__(model_class, model_save_path, model_args, data_handler, sacred_ex, **kwargs)
self.model = None
split_data = self.data_handler.get_split_data(
self.train_start_date,
self.train_end_date,
self.validate_start_date,
self.validate_end_date,
self.test_start_date,
self.test_end_date,
)
(
self.x_train,
self.y_train,
self.x_validate,
self.y_validate,
self.x_test,
self.y_test,
) = split_data
def train(self):
TimeInspector.set_time_mark()
model = self.model_class(**self.model_args)
if CONFIG_MANAGER.ex_config.finetune:
fetcher = create_fetcher_with_config(CONFIG_MANAGER, load_form_loader=True)
loader_model = fetcher.get_experiment(
exp_name=CONFIG_MANAGER.ex_config.loader_name,
exp_id=CONFIG_MANAGER.ex_config.loader_id,
fields=["model"],
)["model"]
if isinstance(loader_model, list):
model_index = (
-1
if CONFIG_MANAGER.ex_config.loader_model_index is None
else CONFIG_MANAGER.ex_config.loader_model_index
)
loader_model = loader_model[model_index]
model.load(loader_model)
model.finetune(self.x_train, self.y_train, self.x_validate, self.y_validate)
else:
model.fit(self.x_train, self.y_train, self.x_validate, self.y_validate)
model.save(self.model_save_path)
self.ex.add_artifact(self.model_save_path)
self.model = model
TimeInspector.log_cost_time("Finished training model.")
def load(self):
model = self.model_class(**self.model_args)
# Load model
fetcher = create_fetcher_with_config(CONFIG_MANAGER, load_form_loader=True)
loader_model = fetcher.get_experiment(
exp_name=CONFIG_MANAGER.ex_config.loader_name,
exp_id=CONFIG_MANAGER.ex_config.loader_id,
fields=["model"],
)["model"]
if isinstance(loader_model, list):
model_index = (
-1
if CONFIG_MANAGER.ex_config.loader_model_index is None
else CONFIG_MANAGER.ex_config.loader_model_index
)
loader_model = loader_model[model_index]
model.load(loader_model)
# Save model, after load, if you don't save the model, the result of this experiment will be no model
model.save(self.model_save_path)
self.ex.add_artifact(self.model_save_path)
self.model = model
def get_test_pred(self):
pred = self.model.predict(self.x_test)
pred = pd.DataFrame(pred, index=self.x_test.index, columns=self.y_test.columns)
return pred
def get_test_performance(self):
try:
model_score = self.model.score(self.x_test, self.y_test)
except NotImplementedError:
model_score = None
# Remove rows from x, y and w, which contain Nan in any columns in y_test.
x_test, y_test, __ = drop_nan_by_y_index(self.x_test, self.y_test)
pred_test = self.model.predict(x_test)
model_pearsonr = pearsonr(np.ravel(pred_test), np.ravel(y_test.values))[0]
performance = {"model_score": model_score, "model_pearsonr": model_pearsonr}
return performance
class RollingTrainer(BaseTrainer):
def __init__(self, model_class, model_save_path, model_args, data_handler, sacred_ex, **kwargs):
super(RollingTrainer, self).__init__(
model_class, model_save_path, model_args, data_handler, sacred_ex, **kwargs
)
self.rolling_period = kwargs.get("rolling_period", 60)
self.models = []
self.rolling_data = []
self.all_x_test = []
self.all_y_test = []
for data in self.data_handler.get_rolling_data(
self.train_start_date,
self.train_end_date,
self.validate_start_date,
self.validate_end_date,
self.test_start_date,
self.test_end_date,
self.rolling_period,
):
self.rolling_data.append(data)
__, __, __, __, x_test, y_test = data
self.all_x_test.append(x_test)
self.all_y_test.append(y_test)
def train(self):
# 1. Get total data parts.
# total_data_parts = self.data_handler.total_data_parts
# self.logger.warning('Total numbers of model are: {}, start training models...'.format(total_data_parts))
if CONFIG_MANAGER.ex_config.finetune:
fetcher = create_fetcher_with_config(CONFIG_MANAGER, load_form_loader=True)
loader_model = fetcher.get_experiment(
exp_name=CONFIG_MANAGER.ex_config.loader_name,
exp_id=CONFIG_MANAGER.ex_config.loader_id,
fields=["model"],
)["model"]
loader_model_index = CONFIG_MANAGER.ex_config.loader_model_index
previous_model_path = ""
# 2. Rolling train.
for (
index,
(x_train, y_train, x_validate, y_validate, x_test, y_test),
) in enumerate(self.rolling_data):
TimeInspector.set_time_mark()
model = self.model_class(**self.model_args)
if CONFIG_MANAGER.ex_config.finetune:
# Finetune model
if loader_model_index is None and isinstance(loader_model, list):
try:
model.load(loader_model[index])
except IndexError:
# Load model by previous_model_path
with open(previous_model_path, "rb") as fp:
model.load(fp)
model.finetune(x_train, y_train, x_validate, y_validate)
else:
if index == 0:
loader_model = (
loader_model[loader_model_index] if isinstance(loader_model, list) else loader_model
)
model.load(loader_model)
else:
with open(previous_model_path, "rb") as fp:
model.load(fp)
model.finetune(x_train, y_train, x_validate, y_validate)
else:
model.fit(x_train, y_train, x_validate, y_validate)
model_save_path = "{}_{}".format(self.model_save_path, index)
model.save(model_save_path)
previous_model_path = model_save_path
self.ex.add_artifact(model_save_path)
self.models.append(model)
TimeInspector.log_cost_time("Finished training model: {}.".format(index + 1))
def load(self):
"""
Load the data and the model
"""
fetcher = create_fetcher_with_config(CONFIG_MANAGER, load_form_loader=True)
loader_model = fetcher.get_experiment(
exp_name=CONFIG_MANAGER.ex_config.loader_name,
exp_id=CONFIG_MANAGER.ex_config.loader_id,
fields=["model"],
)["model"]
for index in range(len(self.all_x_test)):
model = self.model_class(**self.model_args)
model.load(loader_model[index])
# Save model
model_save_path = "{}_{}".format(self.model_save_path, index)
model.save(model_save_path)
self.ex.add_artifact(model_save_path)
self.models.append(model)
def get_test_pred(self):
"""
Predict the score on test data with the models.
Please ensure the models and data are loaded before call this score.
:return: the predicted scores for the pred
"""
pred_df_list = []
y_test_columns = self.all_y_test[0].columns
# Start iteration.
for model, x_test in zip(self.models, self.all_x_test):
pred = model.predict(x_test)
pred_df = pd.DataFrame(pred, index=x_test.index, columns=y_test_columns)
pred_df_list.append(pred_df)
return pd.concat(pred_df_list)
def get_test_performance(self):
"""
Get the performances of the models
:return: the performances of models
"""
pred_test_list = []
y_test_list = []
scorer = self.models[0]._scorer
for model, x_test, y_test in zip(self.models, self.all_x_test, self.all_y_test):
# Remove rows from x, y and w, which contain Nan in any columns in y_test.
x_test, y_test, __ = drop_nan_by_y_index(x_test, y_test)
pred_test_list.append(model.predict(x_test))
y_test_list.append(np.squeeze(y_test.values))
pred_test_array = np.concatenate(pred_test_list, axis=0)
y_test_array = np.concatenate(y_test_list, axis=0)
model_score = scorer(y_test_array, pred_test_array)
model_pearsonr = pearsonr(np.ravel(y_test_array), np.ravel(pred_test_array))[0]
performance = {"model_score": model_score, "model_pearsonr": model_pearsonr}
return performance

86
qlib/contrib/evaluate.py
View File

@@ -26,9 +26,9 @@ def risk_analysis(r, N=252):
Parameters
----------
r : pandas.Series
daily return series
daily return series.
N: int
scaler for annualizing information_ratio (day: 250, week: 50, month: 12)
scaler for annualizing information_ratio (day: 250, week: 50, month: 12).
"""
mean = r.mean()
std = r.std(ddof=1)
@@ -61,7 +61,7 @@ def get_strategy(
----------
strategy : Strategy()
strategy used in backtest
strategy used in backtest.
topk : int (Default value: 50)
top-N stocks to buy.
margin : int or float(Default value: 0.5)
@@ -73,14 +73,14 @@ def get_strategy(
sell_limit = pred_in_a_day.count() * margin
buffer margin, in single score_mode, continue holding stock if it is in nlargest(sell_limit)
sell_limit should be no less than topk
buffer margin, in single score_mode, continue holding stock if it is in nlargest(sell_limit).
sell_limit should be no less than topk.
n_drop : int
number of stocks to be replaced in each trading date
number of stocks to be replaced in each trading date.
risk_degree: float
0-1, 0.95 for example, use 95% money to trade
0-1, 0.95 for example, use 95% money to trade.
str_type: 'amount', 'weight' or 'dropout'
strategy type: TopkAmountStrategy ,TopkWeightStrategy or TopkDropoutStrategy
strategy type: TopkAmountStrategy ,TopkWeightStrategy or TopkDropoutStrategy.
Returns
-------
@@ -126,21 +126,21 @@ def get_exchange(
----------
# exchange related arguments
exchange: Exchange()
exchange: Exchange().
subscribe_fields: list
subscribe fields
subscribe fields.
open_cost : float
open transaction cost
open transaction cost.
close_cost : float
close transaction cost
close transaction cost.
min_cost : float
min transaction cost
min transaction cost.
trade_unit : int
100 for China A
100 for China A.
deal_price: str
dealing price type: 'close', 'open', 'vwap'
dealing price type: 'close', 'open', 'vwap'.
limit_threshold : float
limit move 0.1 (10%) for example, long and short with same limit
limit move 0.1 (10%) for example, long and short with same limit.
extract_codes: bool
will we pass the codes extracted from the pred to the exchange.
NOTE: This will be faster with offline qlib.
@@ -193,20 +193,20 @@ def backtest(pred, account=1e9, shift=1, benchmark="SH000905", verbose=True, **k
- **backtest workflow related or commmon arguments**
pred : pandas.DataFrame
predict should has <datetime, instrument> index and one `score` column
predict should has <datetime, instrument> index and one `score` column.
account : float
init account value
init account value.
shift : int
whether to shift prediction by one day
whether to shift prediction by one day.
benchmark : str
benchmark code, default is SH000905 CSI 500
benchmark code, default is SH000905 CSI 500.
verbose : bool
whether to print log
whether to print log.
- **strategy related arguments**
strategy : Strategy()
strategy used in backtest
strategy used in backtest.
topk : int (Default value: 50)
top-N stocks to buy.
margin : int or float(Default value: 0.5)
@@ -218,33 +218,33 @@ def backtest(pred, account=1e9, shift=1, benchmark="SH000905", verbose=True, **k
sell_limit = pred_in_a_day.count() * margin
buffer margin, in single score_mode, continue holding stock if it is in nlargest(sell_limit)
sell_limit should be no less than topk
buffer margin, in single score_mode, continue holding stock if it is in nlargest(sell_limit).
sell_limit should be no less than topk.
n_drop : int
number of stocks to be replaced in each trading date
number of stocks to be replaced in each trading date.
risk_degree: float
0-1, 0.95 for example, use 95% money to trade
0-1, 0.95 for example, use 95% money to trade.
str_type: 'amount', 'weight' or 'dropout'
strategy type: TopkAmountStrategy ,TopkWeightStrategy or TopkDropoutStrategy
strategy type: TopkAmountStrategy ,TopkWeightStrategy or TopkDropoutStrategy.
- **exchange related arguments**
exchange: Exchange()
pass the exchange for speeding up.
subscribe_fields: list
subscribe fields
subscribe fields.
open_cost : float
open transaction cost. The default value is 0.002(0.2%).
close_cost : float
close transaction cost. The default value is 0.002(0.2%).
min_cost : float
min transaction cost
min transaction cost.
trade_unit : int
100 for China A
100 for China A.
deal_price: str
dealing price type: 'close', 'open', 'vwap'
dealing price type: 'close', 'open', 'vwap'.
limit_threshold : float
limit move 0.1 (10%) for example, long and short with same limit
limit move 0.1 (10%) for example, long and short with same limit.
extract_codes: bool
will we pass the codes extracted from the pred to the exchange.
@@ -291,17 +291,17 @@ def long_short_backtest(
"""
A backtest for long-short strategy
:param pred: The trading signal produced on day `T`
:param topk: The short topk securities and long topk securities
:param deal_price: The price to deal the trading
:param pred: The trading signal produced on day `T`.
:param topk: The short topk securities and long topk securities.
:param deal_price: The price to deal the trading.
:param shift: Whether to shift prediction by one day. The trading day will be T+1 if shift==1.
:param open_cost: open transaction cost
:param close_cost: close transaction cost
:param trade_unit: 100 for China A
:param limit_threshold: limit move 0.1 (10%) for example, long and short with same limit
:param min_cost: min transaction cost
:param subscribe_fields: subscribe fields
:param extract_codes: bool
:param open_cost: open transaction cost.
:param close_cost: close transaction cost.
:param trade_unit: 100 for China A.
:param limit_threshold: limit move 0.1 (10%) for example, long and short with same limit.
:param min_cost: min transaction cost.
:param subscribe_fields: subscribe fields.
:param extract_codes: bool.
will we pass the codes extracted from the pred to the exchange.
NOTE: This will be faster with offline qlib.
:return: The result of backtest, it is represented by a dict.

12
qlib/contrib/model/catboost_model.py
View File

@@ -1,3 +1,15 @@
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import numpy as np
import pandas as pd
from catboost import Pool, CatBoost

349
qlib/contrib/model/pytorch_alstm.py Normal file
View File

@@ -0,0 +1,349 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
from __future__ import division
from __future__ import print_function
import os
import numpy as np
import pandas as pd
import copy
from sklearn.metrics import roc_auc_score, mean_squared_error
import logging
from ...utils import unpack_archive_with_buffer, save_multiple_parts_file, create_save_path, drop_nan_by_y_index
from ...log import get_module_logger, TimeInspector
import torch
import torch.nn as nn
import torch.optim as optim
from ...model.base import Model
from ...data.dataset import DatasetH
from ...data.dataset.handler import DataHandlerLP
class ALSTM(Model):
"""ALSTM Model
Parameters
----------
d_feat : int
input dimension for each time step
metric: str
the evaluate metric used in early stop
optimizer : str
optimizer name
GPU : str
the GPU ID(s) used for training
"""
def __init__(
self,
d_feat=6,
hidden_size=64,
num_layers=2,
dropout=0.0,
n_epochs=200,
lr=0.001,
metric="",
batch_size=2000,
early_stop=20,
loss="mse",
optimizer="adam",
GPU="0",
seed=0,
**kwargs
):
# Set logger.
self.logger = get_module_logger("ALSTM")
self.logger.info("ALSTM pytorch version...")
# set hyper-parameters.
self.d_feat = d_feat
self.hidden_size = hidden_size
self.num_layers = num_layers
self.dropout = dropout
self.n_epochs = n_epochs
self.lr = lr
self.metric = metric
self.batch_size = batch_size
self.early_stop = early_stop
self.optimizer = optimizer.lower()
self.loss = loss
self.visible_GPU = GPU
self.use_gpu = torch.cuda.is_available()
self.seed = seed
self.logger.info(
"ALSTM parameters setting:"
"\nd_feat : {}"
"\nhidden_size : {}"
"\nnum_layers : {}"
"\ndropout : {}"
"\nn_epochs : {}"
"\nlr : {}"
"\nmetric : {}"
"\nbatch_size : {}"
"\nearly_stop : {}"
"\noptimizer : {}"
"\nloss_type : {}"
"\nvisible_GPU : {}"
"\nuse_GPU : {}"
"\nseed : {}".format(
d_feat,
hidden_size,
num_layers,
dropout,
n_epochs,
lr,
metric,
batch_size,
early_stop,
optimizer.lower(),
loss,
GPU,
self.use_gpu,
seed,
)
)
self.ALSTM_model = ALSTMModel(
d_feat=self.d_feat, hidden_size=self.hidden_size, num_layers=self.num_layers, dropout=self.dropout
)
if optimizer.lower() == "adam":
self.train_optimizer = optim.Adam(self.ALSTM_model.parameters(), lr=self.lr)
elif optimizer.lower() == "gd":
self.train_optimizer = optim.SGD(self.ALSTM_model.parameters(), lr=self.lr)
else:
raise NotImplementedError("optimizer {} is not supported!".format(optimizer))
self._fitted = False
if self.use_gpu:
self.ALSTM_model.cuda()
# set the visible GPU
if self.visible_GPU:
os.environ["CUDA_VISIBLE_DEVICES"] = self.visible_GPU
def mse(self, pred, label):
loss = (pred - label) ** 2
return torch.mean(loss)
def loss_fn(self, pred, label):
mask = ~torch.isnan(label)
if self.loss == "mse":
return self.mse(pred[mask], label[mask])
raise ValueError("unknown loss `%s`" % self.loss)
def metric_fn(self, pred, label):
mask = torch.isfinite(label)
if self.metric == "" or self.metric == "loss": # use loss
return -self.loss_fn(pred[mask], label[mask])
raise ValueError("unknown metric `%s`" % self.metric)
def train_epoch(self, x_train, y_train):
x_train_values = x_train.values
y_train_values = np.squeeze(y_train.values)
self.ALSTM_model.train()
indices = np.arange(len(x_train_values))
np.random.shuffle(indices)
for i in range(len(indices))[:: self.batch_size]:
if len(indices) - i < self.batch_size:
break
feature = torch.from_numpy(x_train_values[indices[i : i + self.batch_size]]).float()
label = torch.from_numpy(y_train_values[indices[i : i + self.batch_size]]).float()
if self.use_gpu:
feature = feature.cuda()
label = label.cuda()
pred = self.ALSTM_model(feature)
loss = self.loss_fn(pred, label)
self.train_optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_value_(self.ALSTM_model.parameters(), 3.0)
self.train_optimizer.step()
def test_epoch(self, data_x, data_y):
# prepare training data
x_values = data_x.values
y_values = np.squeeze(data_y.values)
self.ALSTM_model.eval()
scores = []
losses = []
indices = np.arange(len(x_values))
for i in range(len(indices))[:: self.batch_size]:
if len(indices) - i < self.batch_size:
break
feature = torch.from_numpy(x_values[indices[i : i + self.batch_size]]).float()
label = torch.from_numpy(y_values[indices[i : i + self.batch_size]]).float()
if self.use_gpu:
feature = feature.cuda()
label = label.cuda()
pred = self.ALSTM_model(feature)
loss = self.loss_fn(pred, label)
losses.append(loss.item())
score = self.metric_fn(pred, label)
scores.append(score.item())
return np.mean(losses), np.mean(scores)
def fit(
self,
dataset: DatasetH,
evals_result=dict(),
verbose=True,
save_path=None,
):
df_train, df_valid, df_test = dataset.prepare(
["train", "valid", "test"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
)
x_train, y_train = df_train["feature"], df_train["label"]
x_valid, y_valid = df_valid["feature"], df_valid["label"]
if save_path == None:
save_path = create_save_path(save_path)
stop_steps = 0
train_loss = 0
best_score = -np.inf
best_epoch = 0
evals_result["train"] = []
evals_result["valid"] = []
# train
self.logger.info("training...")
self._fitted = True
for step in range(self.n_epochs):
self.logger.info("Epoch%d:", step)
self.logger.info("training...")
self.train_epoch(x_train, y_train)
self.logger.info("evaluating...")
train_loss, train_score = self.test_epoch(x_train, y_train)
val_loss, val_score = self.test_epoch(x_valid, y_valid)
self.logger.info("train %.6f, valid %.6f" % (train_score, val_score))
evals_result["train"].append(train_score)
evals_result["valid"].append(val_score)
if val_score > best_score:
best_score = val_score
stop_steps = 0
best_epoch = step
best_param = copy.deepcopy(self.ALSTM_model.state_dict())
else:
stop_steps += 1
if stop_steps >= self.early_stop:
self.logger.info("early stop")
break
self.logger.info("best score: %.6lf @ %d" % (best_score, best_epoch))
self.ALSTM_model.load_state_dict(best_param)
torch.save(best_param, save_path)
if self.use_gpu:
torch.cuda.empty_cache()
def predict(self, dataset):
if not self._fitted:
raise ValueError("model is not fitted yet!")
x_test = dataset.prepare("test", col_set="feature")
index = x_test.index
self.ALSTM_model.eval()
x_values = x_test.values
sample_num = x_values.shape[0]
preds = []
for begin in range(sample_num)[:: self.batch_size]:
if sample_num - begin < self.batch_size:
end = sample_num
else:
end = begin + self.batch_size
x_batch = torch.from_numpy(x_values[begin:end]).float()
if self.use_gpu:
x_batch = x_batch.cuda()
with torch.no_grad():
if self.use_gpu:
pred = self.ALSTM_model(x_batch).detach().cpu().numpy()
else:
pred = self.ALSTM_model(x_batch).detach().numpy()
preds.append(pred)
return pd.Series(np.concatenate(preds), index=index)
class ALSTMModel(nn.Module):
def __init__(self, d_feat=6, hidden_size=64, num_layers=2, dropout=0.0, rnn_type="GRU"):
super().__init__()
self.hid_size = hidden_size
self.input_size = d_feat
self.dropout = dropout
self.rnn_type = rnn_type
self.rnn_layer = num_layers
self._build_model()
def _build_model(self):
try:
klass = getattr(nn, self.rnn_type.upper())
except:
raise ValueError("unknown rnn_type `%s`" % self.rnn_type)
self.net = nn.Sequential()
self.net.add_module("fc_in", nn.Linear(in_features=self.input_size, out_features=self.hid_size))
self.net.add_module("act", nn.Tanh())
self.rnn = klass(
input_size=self.hid_size,
hidden_size=self.hid_size,
num_layers=self.rnn_layer,
batch_first=True,
dropout=self.dropout,
)
self.fc_out = nn.Linear(in_features=self.hid_size * 2, out_features=1)
self.att_net = nn.Sequential()
self.att_net.add_module("att_fc_in", nn.Linear(in_features=self.hid_size, out_features=int(self.hid_size / 2)))
self.att_net.add_module("att_dropout", torch.nn.Dropout(self.dropout))
self.att_net.add_module("att_act", nn.Tanh())
self.att_net.add_module("att_fc_out", nn.Linear(in_features=int(self.hid_size / 2), out_features=1, bias=False))
self.att_net.add_module("att_softmax", nn.Softmax(dim=1))
def forward(self, inputs):
# inputs: [batch_size, input_size*input_day]
inputs = inputs.view(len(inputs), self.input_size, -1)
inputs = inputs.permute(0, 2, 1) # [batch, input_size, seq_len] -> [batch, seq_len, input_size]
rnn_out, _ = self.rnn(self.net(inputs)) # [batch, seq_len, num_directions * hidden_size]
attention_score = self.att_net(rnn_out) # [batch, seq_len, 1]
out_att = torch.mul(rnn_out, attention_score)
out_att = torch.sum(out_att, dim=1)
out = self.fc_out(
torch.cat((rnn_out[:, -1, :], out_att), dim=1)
) # [batch, seq_len, num_directions * hidden_size] -> [batch, 1]
return out[..., 0]

100
qlib/contrib/model/pytorch_gats.py
View File

@@ -9,10 +9,8 @@ import os
import numpy as np
import pandas as pd
import copy
from sklearn.metrics import roc_auc_score, mean_squared_error
import logging
from ...utils import unpack_archive_with_buffer, save_multiple_parts_file, create_save_path, drop_nan_by_y_index
from ...log import get_module_logger, TimeInspector
from ...utils import create_save_path
from ...log import get_module_logger
import torch
import torch.nn as nn
@@ -28,14 +26,12 @@ class GAT(Model):
Parameters
----------
input_dim : int
input dimension
output_dim : int
output dimension
layers : tuple
layer sizes
lr : float
learning rate
d_feat : int
input dimensions for each time step
metric : str
the evaluate metric used in early stop
optimizer : str
optimizer name
GPU : str
@@ -50,8 +46,7 @@ class GAT(Model):
dropout=0.0,
n_epochs=200,
lr=0.001,
metric="IC",
batch_size=2000,
metric="",
early_stop=20,
loss="mse",
base_model="GRU",
@@ -73,7 +68,6 @@ class GAT(Model):
self.n_epochs = n_epochs
self.lr = lr
self.metric = metric
self.batch_size = batch_size
self.early_stop = early_stop
self.optimizer = optimizer.lower()
self.loss = loss
@@ -92,7 +86,6 @@ class GAT(Model):
"\nn_epochs : {}"
"\nlr : {}"
"\nmetric : {}"
"\nbatch_size : {}"
"\nearly_stop : {}"
"\noptimizer : {}"
"\nloss_type : {}"
@@ -108,7 +101,6 @@ class GAT(Model):
n_epochs,
lr,
metric,
batch_size,
early_stop,
optimizer.lower(),
loss,
@@ -120,10 +112,6 @@ class GAT(Model):
)
)
if loss not in {"mse", "binary"}:
raise NotImplementedError("loss {} is not supported!".format(loss))
self._scorer = mean_squared_error if loss == "mse" else roc_auc_score
self.GAT_model = GATModel(
d_feat=self.d_feat,
hidden_size=self.hidden_size,
@@ -160,34 +148,37 @@ class GAT(Model):
def metric_fn(self, pred, label):
mask = torch.isfinite(label)
if self.metric == "IC":
return self.cal_ic(pred[mask], label[mask])
if self.metric == "" or self.metric == "loss": # use loss
return -self.loss_fn(pred[mask], label[mask])
raise ValueError("unknown metric `%s`" % self.metric)
def cal_ic(self, pred, label):
return torch.mean(pred * label)
def get_daily_inter(self, df, shuffle=False):
# organize the train data into daily inter as daily batches
daily_count = df.groupby(level=0).size().values
daily_index = np.roll(np.cumsum(daily_count), 1)
daily_index[0] = 0
if shuffle:
# shuffle the daily inter data
daily_shuffle = list(zip(daily_index, daily_count))
np.random.shuffle(daily_shuffle)
daily_index, daily_count = zip(*daily_shuffle)
return daily_index, daily_count
def train_epoch(self, x_train, y_train):
x_train_values = x_train.values
y_train_values = np.squeeze(y_train.values) * 100
y_train_values = np.squeeze(y_train.values)
self.GAT_model.train()
indices = np.arange(len(x_train_values))
np.random.shuffle(indices)
# organize the train data into daily inter as daily batches
daily_index, daily_count = self.get_daily_inter(x_train, shuffle=True)
for i in range(len(indices))[:: self.batch_size]:
if len(indices) - i < self.batch_size:
break
feature = torch.from_numpy(x_train_values[indices[i : i + self.batch_size]]).float()
label = torch.from_numpy(y_train_values[indices[i : i + self.batch_size]]).float()
for idx, count in zip(daily_index, daily_count):
batch = slice(idx, idx + count)
feature = torch.from_numpy(x_train_values[batch]).float()
label = torch.from_numpy(y_train_values[batch]).float()
if self.use_gpu:
feature = feature.cuda()
@@ -212,16 +203,13 @@ class GAT(Model):
scores = []
losses = []
indices = np.arange(len(x_values))
np.random.shuffle(indices)
# organize the test data into daily inter as daily batches
daily_index, daily_count = self.get_daily_inter(data_x, shuffle=False)
for i in range(len(indices))[:: self.batch_size]:
if len(indices) - i < self.batch_size:
break
feature = torch.from_numpy(x_values[indices[i : i + self.batch_size]]).float()
label = torch.from_numpy(y_values[indices[i : i + self.batch_size]]).float()
for idx, count in zip(daily_index, daily_count):
batch = slice(idx, idx + count)
feature = torch.from_numpy(x_values[batch]).float()
label = torch.from_numpy(y_values[batch]).float()
if self.use_gpu:
feature = feature.cuda()
@@ -254,7 +242,6 @@ class GAT(Model):
if save_path == None:
save_path = create_save_path(save_path)
stop_steps = 0
train_loss = 0
best_score = -np.inf
best_epoch = 0
evals_result["train"] = []
@@ -265,12 +252,14 @@ class GAT(Model):
self.logger.info("Loading pretrained model...")
if self.base_model == "LSTM":
from ...contrib.model.pytorch_lstm import LSTMModel
pretrained_model = LSTMModel()
pretrained_model.load_state_dict(torch.load('benchmarks/LSTM/model_lstm_csi300.pkl'))
pretrained_model.load_state_dict(torch.load("benchmarks/LSTM/model_lstm_csi300.pkl"))
elif self.base_model == "GRU":
from ...contrib.model.pytorch_gru import GRUModel
pretrained_model = GRUModel()
pretrained_model.load_state_dict(torch.load('benchmarks/GRU/model_gru_csi300.pkl'))
pretrained_model.load_state_dict(torch.load("benchmarks/GRU/model_gru_csi300.pkl"))
model_dict = self.GAT_model.state_dict()
pretrained_dict = {k: v for k, v in pretrained_model.state_dict().items() if k in model_dict}
model_dict.update(pretrained_dict)
@@ -319,17 +308,14 @@ class GAT(Model):
index = x_test.index
self.GAT_model.eval()
x_values = x_test.values
sample_num = x_values.shape[0]
preds = []
for begin in range(sample_num)[:: self.batch_size]:
# organize the data into daily inter as daily batches
daily_index, daily_count = self.get_daily_inter(x_test, shuffle=False)
if sample_num - begin < self.batch_size:
end = sample_num
else:
end = begin + self.batch_size
x_batch = torch.from_numpy(x_values[begin:end]).float()
for idx, count in zip(daily_index, daily_count):
batch = slice(idx, idx + count)
x_batch = torch.from_numpy(x_values[batch]).float()
if self.use_gpu:
x_batch = x_batch.cuda()
@@ -375,7 +361,6 @@ class GATModel(nn.Module):
self.fc_out = nn.Linear(hidden_size, 1)
self.leaky_relu = nn.LeakyReLU()
self.softmax = nn.Softmax(dim=1)
self.d_feat = d_feat
def cal_convariance(self, x, y): # the 2nd dimension of x and y are the same
@@ -394,12 +379,7 @@ class GATModel(nn.Module):
out, _ = self.rnn(x)
hidden = out[:, -1, :]
hidden = self.bn1(hidden)
gamma = self.cal_convariance(hidden, hidden)
# gamma = hidden.mm(torch.t(hidden))
# gamma = self.leaky_relu(gamma)
# gamma = self.softmax(gamma)
# gamma = gamma * (torch.ones(x.shape[0], x.shape[0]).to(device) - torch.diag(torch.ones(x.shape[0])).to(device))
output = gamma.mm(hidden)
output = self.fc(output)
output = self.bn2(output)

27
qlib/contrib/model/pytorch_gru.py
View File

@@ -28,14 +28,10 @@ class GRU(Model):
Parameters
----------
input_dim : int
input dimension
output_dim : int
output dimension
layers : tuple
layer sizes
lr : float
learning rate
d_feat : int
input dimension for each time step
metric: str
the evaluate metric used in early stop
optimizer : str
optimizer name
GPU : str
@@ -50,7 +46,7 @@ class GRU(Model):
dropout=0.0,
n_epochs=200,
lr=0.001,
metric="IC",
metric="",
batch_size=2000,
early_stop=20,
loss="mse",
@@ -112,10 +108,6 @@ class GRU(Model):
)
)
if loss not in {"mse", "binary"}:
raise NotImplementedError("loss {} is not supported!".format(loss))
self._scorer = mean_squared_error if loss == "mse" else roc_auc_score
self.gru_model = GRUModel(
d_feat=self.d_feat, hidden_size=self.hidden_size, num_layers=self.num_layers, dropout=self.dropout
)
@@ -148,21 +140,16 @@ class GRU(Model):
def metric_fn(self, pred, label):
mask = torch.isfinite(label)
if self.metric == "IC":
return self.cal_ic(pred[mask], label[mask])
if self.metric == "" or self.metric == "loss": # use loss
return -self.loss_fn(pred[mask], label[mask])
raise ValueError("unknown metric `%s`" % self.metric)
def cal_ic(self, pred, label):
return torch.mean(pred * label)
def train_epoch(self, x_train, y_train):
x_train_values = x_train.values
y_train_values = np.squeeze(y_train.values) * 100
y_train_values = np.squeeze(y_train.values)
self.gru_model.train()
@@ -201,7 +188,6 @@ class GRU(Model):
losses = []
indices = np.arange(len(x_values))
np.random.shuffle(indices)
for i in range(len(indices))[:: self.batch_size]:
@@ -251,7 +237,6 @@ class GRU(Model):
# train
self.logger.info("training...")
self._fitted = True
# return
for step in range(self.n_epochs):
self.logger.info("Epoch%d:", step)

491
qlib/contrib/model/pytorch_hats.py Normal file
View File

@@ -0,0 +1,491 @@
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import division
from __future__ import print_function
import os
import numpy as np
import pandas as pd
import copy
from ...utils import create_save_path
from ...log import get_module_logger
import torch
import torch.nn as nn
import torch.optim as optim
from ...model.base import Model
from ...data.dataset import DatasetH
from ...data.dataset.handler import DataHandlerLP
class HATS(Model):
"""HATS Model
Parameters
----------
d_feat : int
input dimension for each time step
metric: str
the evaluate metric used in early stop
optimizer : str
optimizer name
GPU : str
the GPU ID(s) used for training
"""
def __init__(
self,
d_feat=6,
hidden_size=64,
num_layers=2,
dropout=0.5,
n_epochs=200,
lr=0.01,
metric="",
early_stop=20,
loss="mse",
base_model="GRU",
with_pretrain=True,
optimizer="adam",
GPU="0",
seed=0,
**kwargs
):
# Set logger.
self.logger = get_module_logger("HATS")
self.logger.info("HATS pytorch version...")
# set hyper-parameters.
self.d_feat = d_feat
self.hidden_size = hidden_size
self.num_layers = num_layers
self.dropout = dropout
self.n_epochs = n_epochs
self.lr = lr
self.metric = metric
self.early_stop = early_stop
self.optimizer = optimizer.lower()
self.loss = loss
self.base_model = base_model
self.with_pretrain = with_pretrain
self.visible_GPU = GPU
self.use_gpu = torch.cuda.is_available()
self.seed = seed
self.logger.info(
"HATS parameters setting:"
"\nd_feat : {}"
"\nhidden_size : {}"
"\nnum_layers : {}"
"\ndropout : {}"
"\nn_epochs : {}"
"\nlr : {}"
"\nmetric : {}"
"\nearly_stop : {}"
"\noptimizer : {}"
"\nloss_type : {}"
"\nbase_model : {}"
"\nwith_pretrain : {}"
"\nvisible_GPU : {}"
"\nuse_GPU : {}"
"\nseed : {}".format(
d_feat,
hidden_size,
num_layers,
dropout,
n_epochs,
lr,
metric,
early_stop,
optimizer.lower(),
loss,
base_model,
with_pretrain,
GPU,
self.use_gpu,
seed,
)
)
self.HATS_model = HATSModel(
d_feat=self.d_feat,
hidden_size=self.hidden_size,
num_layers=self.num_layers,
dropout=self.dropout,
base_model=self.base_model,
)
if optimizer.lower() == "adam":
self.train_optimizer = optim.Adam(self.HATS_model.parameters(), lr=self.lr)
elif optimizer.lower() == "gd":
self.train_optimizer = optim.SGD(self.HATS_model.parameters(), lr=self.lr)
else:
raise NotImplementedError("optimizer {} is not supported!".format(optimizer))
self._fitted = False
if self.use_gpu:
self.HATS_model.cuda()
# set the visible GPU
if self.visible_GPU:
os.environ["CUDA_VISIBLE_DEVICES"] = self.visible_GPU
def mse(self, pred, label):
loss = (pred - label) ** 2
return torch.mean(loss)
def loss_fn(self, pred, label):
mask = ~torch.isnan(label)
if self.loss == "mse":
return self.mse(pred[mask], label[mask])
raise ValueError("unknown loss `%s`" % self.loss)
def metric_fn(self, pred, label):
mask = torch.isfinite(label)
if self.metric == "" or self.metric == "loss": # use loss
return -self.loss_fn(pred[mask], label[mask])
raise ValueError("unknown metric `%s`" % self.metric)
def get_daily_inter(self, df, shuffle=False):
# organize the train data into daily inter as daily batches
daily_count = df.groupby(level=0).size().values
daily_index = np.roll(np.cumsum(daily_count), 1)
daily_index[0] = 0
if shuffle:
# shuffle the daily inter data
daily_shuffle = list(zip(daily_index, daily_count))
np.random.shuffle(daily_shuffle)
daily_index, daily_count = zip(*daily_shuffle)
return daily_index, daily_count
def train_epoch(self, x_train, y_train):
x_train_values = x_train.values
y_train_values = np.squeeze(y_train.values)
self.HATS_model.train()
# organize the train data into daily inter as daily batches
daily_index, daily_count = self.get_daily_inter(x_train, shuffle=True)
for idx, count in zip(daily_index, daily_count):
batch = slice(idx, idx + count)
feature = torch.from_numpy(x_train_values[batch]).float()
label = torch.from_numpy(y_train_values[batch]).float()
if self.use_gpu:
feature = feature.cuda()
label = label.cuda()
pred = self.HATS_model(feature)
loss = self.loss_fn(pred, label)
self.train_optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_value_(self.HATS_model.parameters(), 3.0)
self.train_optimizer.step()
def test_epoch(self, data_x, data_y):
# prepare testing data
x_values = data_x.values
y_values = np.squeeze(data_y.values)
self.HATS_model.eval()
scores = []
losses = []
# organize the test data into daily inter as daily batches
daily_index, daily_count = self.get_daily_inter(data_x, shuffle=False)
for idx, count in zip(daily_index, daily_count):
batch = slice(idx, idx + count)
feature = torch.from_numpy(x_values[batch]).float()
label = torch.from_numpy(y_values[batch]).float()
if self.use_gpu:
feature = feature.cuda()
label = label.cuda()
pred = self.HATS_model(feature)
loss = self.loss_fn(pred, label)
losses.append(loss.item())
score = self.metric_fn(pred, label)
scores.append(score.item())
return np.mean(losses), np.mean(scores)
def fit(
self,
dataset: DatasetH,
evals_result=dict(),
verbose=True,
save_path=None,
):
df_train, df_valid, df_test = dataset.prepare(
["train", "valid", "test"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
)
x_train, y_train = df_train["feature"], df_train["label"]
x_valid, y_valid = df_valid["feature"], df_valid["label"]
if save_path == None:
save_path = create_save_path(save_path)
stop_steps = 0
best_score = -np.inf
best_epoch = 0
evals_result["train"] = []
evals_result["valid"] = []
# load pretrained base_model
if self.with_pretrain:
self.logger.info("Loading pretrained model...")
if self.base_model == "LSTM":
from ...contrib.model.pytorch_lstm import LSTMModel
pretrained_model = LSTMModel()
pretrained_model.load_state_dict(torch.load("benchmarks/LSTM/model_lstm_csi300.pkl"))
elif self.base_model == "GRU":
from ...contrib.model.pytorch_gru import GRUModel
pretrained_model = GRUModel()
pretrained_model.load_state_dict(torch.load("benchmarks/GRU/model_gru_csi300.pkl"))
model_dict = self.HATS_model.state_dict()
pretrained_dict = {k: v for k, v in pretrained_model.state_dict().items() if k in model_dict}
model_dict.update(pretrained_dict)
self.HATS_model.load_state_dict(model_dict)
self.logger.info("Loading pretrained model Done...")
# train
self.logger.info("training...")
self._fitted = True
for step in range(self.n_epochs):
self.logger.info("Epoch%d:", step)
self.logger.info("training...")
self.train_epoch(x_train, y_train)
self.logger.info("evaluating...")
train_loss, train_score = self.test_epoch(x_train, y_train)
val_loss, val_score = self.test_epoch(x_valid, y_valid)
self.logger.info("train %.6f, valid %.6f" % (train_score, val_score))
evals_result["train"].append(train_score)
evals_result["valid"].append(val_score)
if val_score > best_score:
best_score = val_score
stop_steps = 0
best_epoch = step
best_param = copy.deepcopy(self.HATS_model.state_dict())
else:
stop_steps += 1
if stop_steps >= self.early_stop:
self.logger.info("early stop")
break
self.logger.info("best score: %.6lf @ %d" % (best_score, best_epoch))
self.HATS_model.load_state_dict(best_param)
torch.save(best_param, save_path)
if self.use_gpu:
torch.cuda.empty_cache()
def predict(self, dataset):
if not self._fitted:
raise ValueError("model is not fitted yet!")
x_test = dataset.prepare("test", col_set="feature")
index = x_test.index
self.HATS_model.eval()
x_values = x_test.values
sample_num = x_values.shape[0]
preds = []
# organize the data into daily inter as daily batches
daily_index, daily_count = self.get_daily_inter(x_test, shuffle=False)
for idx, count in zip(daily_index, daily_count):
batch = slice(idx, idx + count)
x_batch = torch.from_numpy(x_values[batch]).float()
if self.use_gpu:
x_batch = x_batch.cuda()
with torch.no_grad():
if self.use_gpu:
pred = self.HATS_model(x_batch).detach().cpu().numpy()
else:
pred = self.HATS_model(x_batch).detach().numpy()
preds.append(pred)
return pd.Series(np.concatenate(preds), index=index)
class HATSModel(nn.Module):
def __init__(self, d_feat=6, hidden_size=64, num_layers=2, dropout=0.0, base_model="GRU"):
super().__init__()
if base_model == "GRU":
self.model = nn.GRU(
input_size=d_feat,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout,
)
elif base_model == "LSTM":
self.model = nn.LSTM(
input_size=d_feat,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout,
)
else:
raise ValueError("unknown base model name `%s`" % base_model)
self.hidden_size = hidden_size
self.bn1 = nn.BatchNorm1d(num_features=hidden_size, track_running_stats=False)
self.fc = nn.Linear(hidden_size, hidden_size)
self.bn2 = nn.BatchNorm1d(num_features=hidden_size, track_running_stats=False)
self.fc_out = nn.Linear(hidden_size, 1)
self.leaky_relu = nn.LeakyReLU()
self.softmax = nn.Softmax(dim=1)
self.d_feat = d_feat
num_head_att = [1] * num_layers
hidden_dim = [hidden_size] * num_layers
dims = [d_feat] + [d * nh for (d, nh) in zip(hidden_dim, num_head_att[:-1])] + [num_head_att[-1]]
in_dims = dims[:-1]
out_dims = [d // nh for (d, nh) in zip(dims[1:], num_head_att)]
self.attn = nn.ModuleList(
[GraphAttention(i, o, nh, dropout) for (i, o, nh) in zip(in_dims, out_dims, num_head_att)]
)
self.bns = nn.ModuleList([nn.BatchNorm1d(dim) for dim in dims[1:-1]])
self.dropout = nn.Dropout(dropout)
self.elu = nn.ELU()
def forward(self, x):
x = x.reshape(len(x), self.d_feat, -1) # [N, F, T]
x = x.permute(0, 2, 1) # [N, T, F]
out, _ = self.model(x)
hidden = out[:, -1, :]
hidden = self.bn1(hidden)
attention = GraphAttention.cal_attention(hidden, hidden)
output = attention.mm(hidden)
output = self.fc(output)
output = self.bn2(output)
output = self.leaky_relu(output)
return self.fc_out(output).squeeze()
class GraphAttention(nn.Module):
def __init__(self, input_dim, output_dim, num_heads, dropout=0.5):
super().__init__()
"""
Parameters
----------
input_dim : int
Dimension of input node features.
output_dim : int
Dimension of output node features.
num_heads : list of ints
Number of attention heads in each hidden layer and output layer. Must be non empty. Note that len(num_heads) = len(hidden_dims)+1.
dropout : float
Dropout rate. Default: 0.5.
"""
self.input_dim = input_dim
self.output_dim = output_dim
self.num_heads = num_heads
self.fcs = nn.ModuleList([nn.Linear(input_dim, output_dim) for _ in range(num_heads)])
self.a = nn.ModuleList([nn.Linear(2 * output_dim, 1) for _ in range(num_heads)])
self.dropout = nn.Dropout(dropout)
self.softmax = nn.Softmax(dim=0)
self.leakyrelu = nn.LeakyReLU()
def forward(self, features, nodes, mappings, rows):
"""
Parameters
----------
features : torch.Tensor
An (n' x input_dim) tensor of input node features.
nodes : list of numpy array
nodes[i] is an array of the nodes in the ith layer of the
computation graph.
mappings : list of dictionary
mappings[i] is a dictionary mappings node v (labelled 0 to |V|-1)
in nodes[i] to its position in nodes[i]. For example,
if nodes[i] = [2,5], then mappings[i][2] = 0 and
mappings[i][5] = 1.
rows : numpy array
rows[i] is an array of neighbors of node i.
Returns
-------
out : torch.Tensor
An (len(node_layers[-1]) x output_dim) tensor of output node features.
"""
nprime = features.shape[0]
rows = [np.array([mappings[v] for v in row], dtype=np.int64) for row in rows]
sum_degs = np.hstack(([0], np.cumsum([len(row) for row in rows])))
mapped_nodes = [mappings[v] for v in nodes]
indices = torch.LongTensor([[v, c] for (v, row) in zip(mapped_nodes, rows) for c in row]).t()
out = []
for k in range(self.num_heads):
h = self.fcs[k](features)
nbr_h = torch.cat(tuple([h[row] for row in rows]), dim=0)
self_h = torch.cat(
tuple([h[mappings[nodes[i]]].repeat(len(row), 1) for (i, row) in enumerate(rows)]), dim=0
)
cat_h = torch.cat((self_h, nbr_h), dim=1)
e = self.leakyrelu(self.a[k](cat_h))
alpha = [self.softmax(e[lo:hi]) for (lo, hi) in zip(sum_degs, sum_degs[1:])]
alpha = torch.cat(tuple(alpha), dim=0)
alpha = alpha.squeeze(1)
alpha = self.dropout(alpha)
adj = torch.sparse.FloatTensor(indices, alpha, torch.Size([nprime, nprime]))
out.append(torch.sparse.mm(adj, h)[mapped_nodes])
return out
@staticmethod
def cal_attention(x, y):
att_x = torch.mean(x, dim=1).reshape(-1, 1)
att_y = torch.mean(y, dim=1).reshape(-1, 1)
att = att_x.mm(torch.t(att_y))
return (
torch.mean(
x.reshape(x.shape[0], 1, x.shape[1]).repeat(1, y.shape[0], 1)
* y.reshape(1, y.shape[0], y.shape[1]).repeat(x.shape[0], 1, 1),
dim=2,
)
- att
)

27
qlib/contrib/model/pytorch_lstm.py
View File

@@ -28,14 +28,10 @@ class LSTM(Model):
Parameters
----------
input_dim : int
input dimension
output_dim : int
output dimension
layers : tuple
layer sizes
lr : float
learning rate
d_feat : int
input dimension for each time step
metric: str
the evaluate metric used in early stop
optimizer : str
optimizer name
GPU : str
@@ -50,7 +46,7 @@ class LSTM(Model):
dropout=0.0,
n_epochs=200,
lr=0.001,
metric="IC",
metric="",
batch_size=2000,
early_stop=20,
loss="mse",
@@ -112,10 +108,6 @@ class LSTM(Model):
)
)
if loss not in {"mse", "binary"}:
raise NotImplementedError("loss {} is not supported!".format(loss))
self._scorer = mean_squared_error if loss == "mse" else roc_auc_score
self.lstm_model = LSTMModel(
d_feat=self.d_feat, hidden_size=self.hidden_size, num_layers=self.num_layers, dropout=self.dropout
)
@@ -148,21 +140,16 @@ class LSTM(Model):
def metric_fn(self, pred, label):
mask = torch.isfinite(label)
if self.metric == "IC":
return self.cal_ic(pred[mask], label[mask])
if self.metric == "" or self.metric == "loss": # use loss
return -self.loss_fn(pred[mask], label[mask])
raise ValueError("unknown metric `%s`" % self.metric)
def cal_ic(self, pred, label):
return torch.mean(pred * label)
def train_epoch(self, x_train, y_train):
x_train_values = x_train.values
y_train_values = np.squeeze(y_train.values) * 100
y_train_values = np.squeeze(y_train.values)
self.lstm_model.train()
@@ -201,7 +188,6 @@ class LSTM(Model):
losses = []
indices = np.arange(len(x_values))
np.random.shuffle(indices)
for i in range(len(indices))[:: self.batch_size]:
@@ -251,7 +237,6 @@ class LSTM(Model):
# train
self.logger.info("training...")
self._fitted = True
# return
for step in range(self.n_epochs):
self.logger.info("Epoch%d:", step)

280
qlib/contrib/model/pytorch_sfm.py
View File

@@ -1,5 +1,15 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import division
from __future__ import print_function
@@ -90,10 +100,7 @@ class SFM_Model(nn.Module):
x_c = torch.matmul(x * B_W[0], self.W_c) + self.b_c
x_o = torch.matmul(x * B_W[0], self.W_o) + self.b_o
i = self.inner_activation(
x_i + torch.matmul(h_tm1 * B_U[0], self.U_i)
) # not sure whether I am doing in the right unsquuze
i = self.inner_activation(x_i + torch.matmul(h_tm1 * B_U[0], self.U_i))
ste = self.inner_activation(x_ste + torch.matmul(h_tm1 * B_U[0], self.U_ste))
fre = self.inner_activation(x_fre + torch.matmul(h_tm1 * B_U[0], self.U_fre))
@@ -173,10 +180,6 @@ class SFM(Model):
output dimension
lr : float
learning rate
lr_decay : float
learning rate decay
lr_decay_steps : int
learning rate decay steps
optimizer : str
optimizer name
GPU : str
@@ -193,12 +196,11 @@ class SFM(Model):
dropout_U=0.0,
n_epochs=200,
lr=0.001,
metric="",
batch_size=2000,
early_stop=20,
eval_steps=5,
loss="mse",
lr_decay=0.96,
lr_decay_steps=100,
optimizer="gd",
GPU="0",
seed=0,
@@ -217,13 +219,12 @@ class SFM(Model):
self.dropout_U = dropout_U
self.n_epochs = n_epochs
self.lr = lr
self.metric = metric
self.batch_size = batch_size
self.early_stop = early_stop
self.eval_steps = eval_steps
self.lr_decay = lr_decay
self.lr_decay_steps = lr_decay_steps
self.optimizer = optimizer.lower()
self.loss_type = loss
self.loss = loss
self.device = "cuda:%d" % (GPU) if torch.cuda.is_available() else "cpu"
self.use_gpu = torch.cuda.is_available()
self.seed = seed
@@ -232,16 +233,16 @@ class SFM(Model):
"SFM parameters setting:"
"\nd_feat : {}"
"\nhidden_size : {}"
"\noutput_size : {}"
"\nfrequency_dimension : {}"
"\ndropout_W: {}"
"\ndropout_U: {}"
"\nn_epochs : {}"
"\nlr : {}"
"\nmetric : {}"
"\nbatch_size : {}"
"\nearly_stop : {}"
"\neval_steps : {}"
"\nlr_decay : {}"
"\nlr_decay_steps : {}"
"\noptimizer : {}"
"\nloss_type : {}"
"\nvisible_GPU : {}"
@@ -249,16 +250,16 @@ class SFM(Model):
"\nseed : {}".format(
d_feat,
hidden_size,
output_dim,
freq_dim,
dropout_W,
dropout_U,
n_epochs,
lr,
metric,
batch_size,
early_stop,
eval_steps,
lr_decay,
lr_decay_steps,
optimizer.lower(),
loss,
GPU,
@@ -267,10 +268,6 @@ class SFM(Model):
)
)
if loss not in {"mse", "binary"}:
raise NotImplementedError("loss {} is not supported!".format(loss))
self._scorer = mean_squared_error if loss == "mse" else roc_auc_score
self.sfm_model = SFM_Model(
d_feat=self.d_feat,
output_dim=self.output_dim,
@@ -287,24 +284,72 @@ class SFM(Model):
else:
raise NotImplementedError("optimizer {} is not supported!".format(optimizer))
# Reduce learning rate when loss has stopped decrease
self.scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
self.train_optimizer,
mode="min",
factor=0.5,
patience=10,
verbose=True,
threshold=0.0001,
threshold_mode="rel",
cooldown=0,
min_lr=0.00001,
eps=1e-08,
)
self._fitted = False
self.sfm_model.to(self.device)
def fit(self, dataset: DatasetH, evals_result=dict(), verbose=True, save_path=None, **kwargs):
def test_epoch(self, data_x, data_y):
# prepare training data
x_values = data_x.values
y_values = np.squeeze(data_y.values)
self.sfm_model.eval()
scores = []
losses = []
indices = np.arange(len(x_values))
for i in range(len(indices))[:: self.batch_size]:
if len(indices) - i < self.batch_size:
break
feature = torch.from_numpy(x_values[indices[i : i + self.batch_size]]).float().to(self.device)
label = torch.from_numpy(y_values[indices[i : i + self.batch_size]]).float().to(self.device)
pred = self.sfm_model(feature)
loss = self.loss_fn(pred, label)
losses.append(loss.item())
score = self.metric_fn(pred, label)
scores.append(score.item())
return np.mean(losses), np.mean(scores)
def train_epoch(self, x_train, y_train):
x_train_values = x_train.values
y_train_values = np.squeeze(y_train.values)
self.sfm_model.train()
indices = np.arange(len(x_train_values))
np.random.shuffle(indices)
for i in range(len(indices))[:: self.batch_size]:
if len(indices) - i < self.batch_size:
break
feature = torch.from_numpy(x_train_values[indices[i : i + self.batch_size]]).float().to(self.device)
label = torch.from_numpy(y_train_values[indices[i : i + self.batch_size]]).float().to(self.device)
pred = self.sfm_model(feature)
loss = self.loss_fn(pred, label)
self.train_optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_value_(self.sfm_model.parameters(), 3.0)
self.train_optimizer.step()
def fit(
self,
dataset: DatasetH,
evals_result=dict(),
verbose=True,
save_path=None,
):
df_train, df_valid = dataset.prepare(
["train", "valid"], col_set=["feature", "label"], data_key=DataHandlerLP.DK_L
@@ -312,10 +357,10 @@ class SFM(Model):
x_train, y_train = df_train["feature"], df_train["label"]
x_valid, y_valid = df_valid["feature"], df_valid["label"]
save_path = create_save_path(save_path)
stop_steps = 0
train_loss = 0
best_loss = np.inf
best_score = -np.inf
best_epoch = 0
evals_result["train"] = []
evals_result["valid"] = []
@@ -323,90 +368,51 @@ class SFM(Model):
self.logger.info("training...")
self._fitted = True
# prepare training data
x_train_values = torch.from_numpy(x_train.values).float()
y_train_values = torch.from_numpy(np.squeeze(y_train.values)).float()
train_num = y_train_values.shape[0]
# prepare validation data
x_val_auto = torch.from_numpy(x_valid.values).float()
y_val_auto = torch.from_numpy(np.squeeze(y_valid.values)).float()
x_val_auto = x_val_auto.to(self.device)
y_val_auto = y_val_auto.to(self.device)
for step in range(self.n_epochs):
if stop_steps >= self.early_stop:
if verbose:
self.logger.info("\tearly stop")
break
loss = AverageMeter()
self.sfm_model.train()
self.train_optimizer.zero_grad()
self.logger.info("Epoch%d:", step)
self.logger.info("training...")
self.train_epoch(x_train, y_train)
self.logger.info("evaluating...")
train_loss, train_score = self.test_epoch(x_train, y_train)
val_loss, val_score = self.test_epoch(x_valid, y_valid)
self.logger.info("train %.6f, valid %.6f" % (train_score, val_score))
evals_result["train"].append(train_score)
evals_result["valid"].append(val_score)
choice = np.random.choice(train_num, self.batch_size)
x_batch_auto = x_train_values[choice]
y_batch_auto = y_train_values[choice]
x_batch_auto = x_batch_auto.to(self.device)
y_batch_auto = y_batch_auto.to(self.device)
# forward
preds = self.sfm_model(x_batch_auto)
cur_loss = self.get_loss(preds, y_batch_auto, self.loss_type)
cur_loss.backward()
self.train_optimizer.step()
loss.update(cur_loss.item())
# validation
train_loss += loss.val
# print(loss.val)
if step and step % self.eval_steps == 0:
if val_score > best_score:
best_score = val_score
stop_steps = 0
best_epoch = step
best_param = copy.deepcopy(self.sfm_model.state_dict())
else:
stop_steps += 1
train_loss /= self.eval_steps
with torch.no_grad():
self.sfm_model.eval()
loss_val = AverageMeter()
# forward
preds = self.sfm_model(x_val_auto)
cur_loss_val = self.get_loss(preds, y_val_auto, self.loss_type)
loss_val.update(cur_loss_val.item())
if verbose:
self.logger.info(
"[Epoch {}]: train_loss {:.6f}, valid_loss {:.6f}".format(step, train_loss, loss_val.val)
)
evals_result["train"].append(train_loss)
evals_result["valid"].append(loss_val.val)
if loss_val.val < best_loss:
if verbose:
self.logger.info(
"\tvalid loss update from {:.6f} to {:.6f}, save checkpoint.".format(
best_loss, loss_val.val
)
)
best_loss = loss_val.val
stop_steps = 0
torch.save(self.sfm_model.state_dict(), save_path)
train_loss = 0
# update learning rate
self.scheduler.step(cur_loss_val)
if stop_steps >= self.early_stop:
self.logger.info("early stop")
break
self.logger.info("best score: %.6lf @ %d" % (best_score, best_epoch))
if self.device != "cpu":
torch.cuda.empty_cache()
def get_loss(self, pred, target, loss_type):
if loss_type == "mse":
sqr_loss = (pred - target) ** 2
loss = sqr_loss.mean()
return loss
elif loss_type == "binary":
loss = nn.BCELoss()
return loss(pred, target)
else:
raise NotImplementedError("loss {} is not supported!".format(loss_type))
def mse(self, pred, label):
loss = (pred - label) ** 2
return torch.mean(loss)
def loss_fn(self, pred, label):
mask = ~torch.isnan(label)
if self.loss == "mse":
return self.mse(pred[mask], label[mask])
raise ValueError("unknown loss `%s`" % self.loss)
def metric_fn(self, pred, label):
mask = torch.isfinite(label)
if self.metric == "" or self.metric == "loss": # use loss
return -self.loss_fn(pred[mask], label[mask])
raise ValueError("unknown metric `%s`" % self.metric)
def predict(self, dataset):
if not self._fitted:
@@ -414,34 +420,28 @@ class SFM(Model):
x_test = dataset.prepare("test", col_set="feature")
index = x_test.index
x_test = torch.from_numpy(x_test.values).float()
x_test = x_test.to(self.device)
self.sfm_model.eval()
x_values = x_test.values
sample_num = x_values.shape[0]
preds = []
with torch.no_grad():
if self.device != "cpu":
preds = self.sfm_model(x_test).detach().cpu().numpy()
for begin in range(sample_num)[:: self.batch_size]:
if sample_num - begin < self.batch_size:
end = sample_num
else:
preds = self.sfm_model(x_test).detach().numpy()
return pd.Series(preds, index=index)
end = begin + self.batch_size
def save(self, filename, **kwargs):
with save_multiple_parts_file(filename) as model_dir:
model_path = os.path.join(model_dir, os.path.split(model_dir)[-1])
# Save model
torch.save(self.sfm_model.state_dict(), model_path)
x_batch = torch.from_numpy(x_values[begin:end]).float()
def load(self, buffer, **kwargs):
with unpack_archive_with_buffer(buffer) as model_dir:
# Get model name
_model_name = os.path.splitext(list(filter(lambda x: x.startswith("model.bin"), os.listdir(model_dir)))[0])[
0
]
_model_path = os.path.join(model_dir, _model_name)
# Load model
self.sfm_model.load_state_dict(torch.load(_model_path))
self._fitted = True
if self.device != "cpu":
x_batch = x_batch.to(self.device)
with torch.no_grad():
pred = self.sfm_model(x_batch).detach().cpu().numpy()
preds.append(pred)
return pd.Series(np.concatenate(preds), index=index)
class AverageMeter(object):

19
qlib/contrib/model/xgboost.py
View File

@@ -1,5 +1,14 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import numpy as np
import pandas as pd
@@ -13,10 +22,8 @@ from ...data.dataset.handler import DataHandlerLP
class XGBModel(Model):
"""XGBModel Model"""
def __init__(self, obj="mse", **kwargs):
if obj not in {"mse", "binary"}:
raise NotImplementedError
self._params = {"obj": obj}
def __init__(self, **kwargs):
self._params = {}
self._params.update(kwargs)
self.model = None

16
qlib/contrib/report/analysis_model/analysis_model_performance.py
View File

@@ -252,7 +252,7 @@ def model_performance_graph(
"""Model performance
:param pred_label: index is **pd.MultiIndex**, index name is **[instrument, datetime]**; columns names is **[score,
label]**. It is usually same as the label of model training(e.g. "Ref($close, -2)/Ref($close, -1) - 1")
label]**. It is usually same as the label of model training(e.g. "Ref($close, -2)/Ref($close, -1) - 1").
.. code-block:: python
@@ -266,13 +266,13 @@ def model_performance_graph(
:param lag: `pred.groupby(level='instrument')['score'].shift(lag)`. It will be only used in the auto-correlation computing.
:param N: group number, default 5
:param reverse: if `True`, `pred['score'] *= -1`
:param rank: if **True**, calculate rank ic
:param graph_names: graph names; default ['cumulative_return', 'pred_ic', 'pred_autocorr', 'pred_turnover']
:param show_notebook: whether to display graphics in notebook, the default is `True`
:param show_nature_day: whether to display the abscissa of non-trading day
:return: if show_notebook is True, display in notebook; else return `plotly.graph_objs.Figure` list
:param N: group number, default 5.
:param reverse: if `True`, `pred['score'] *= -1`.
:param rank: if **True**, calculate rank ic.
:param graph_names: graph names; default ['cumulative_return', 'pred_ic', 'pred_autocorr', 'pred_turnover'].
:param show_notebook: whether to display graphics in notebook, the default is `True`.
:param show_nature_day: whether to display the abscissa of non-trading day.
:return: if show_notebook is True, display in notebook; else return `plotly.graph_objs.Figure` list.
"""
figure_list = []
for graph_name in graph_names:

6
qlib/contrib/report/analysis_position/cumulative_return.py
View File

@@ -218,10 +218,10 @@ def cumulative_return_graph(
Graph desc:
- Axis X: Trading day
- Axis X: Trading day.
- Axis Y:
- Above axis Y: `(((Ref($close, -1)/$close - 1) * weight).sum() / weight.sum()).cumsum()`
- Below axis Y: Daily weight sum
- Above axis Y: `(((Ref($close, -1)/$close - 1) * weight).sum() / weight.sum()).cumsum()`.
- Below axis Y: Daily weight sum.
- In the **sell** graph, `y < 0` stands for profit; in other cases, `y > 0` stands for profit.
- In the **buy_minus_sell** graph, the **y** value of the **weight** graph at the bottom is `buy_weight + sell_weight`.
- In each graph, the **red line** in the histogram on the right represents the average.

6
qlib/contrib/report/analysis_position/rank_label.py
View File

@@ -97,9 +97,9 @@ def rank_label_graph(
qcr.rank_label_graph(positions, features_df, pred_df_dates.min(), pred_df_dates.max())
:param position: position data; **qlib.contrib.backtest.backtest.backtest** result
:param position: position data; **qlib.contrib.backtest.backtest.backtest** result.
:param label_data: **D.features** result; index is **pd.MultiIndex**, index name is **[instrument, datetime]**; columns names is **[label]**.
**The label T is the change from T to T+1**, it is recommended to use ``close``, example: `D.features(D.instruments('csi500'), ['Ref($close, -1)/$close-1'])`
**The label T is the change from T to T+1**, it is recommended to use ``close``, example: `D.features(D.instruments('csi500'), ['Ref($close, -1)/$close-1'])`.
.. code-block:: python
@@ -115,7 +115,7 @@ def rank_label_graph(
:param start_date: start date
:param end_date: end_date
:param show_notebook: **True** or **False**. If True, show graph in notebook, else return figures
:param show_notebook: **True** or **False**. If True, show graph in notebook, else return figures.
:return:
"""
position = copy.deepcopy(position)

6
qlib/contrib/report/analysis_position/report.py
View File

@@ -186,7 +186,7 @@ def report_graph(report_df: pd.DataFrame, show_notebook: bool = True) -> [list,
qcr.report_graph(report_normal_df)
:param report_df: **df.index.name** must be **date**, **df.columns** must contain **return**, **turnover**, **cost**, **bench**
:param report_df: **df.index.name** must be **date**, **df.columns** must contain **return**, **turnover**, **cost**, **bench**.
.. code-block:: python
@@ -200,8 +200,8 @@ def report_graph(report_df: pd.DataFrame, show_notebook: bool = True) -> [list,
2017-01-10 -0.000416 0.000440 -0.003350 0.208396
:param show_notebook: whether to display graphics in notebook, the default is **True**
:return: if show_notebook is True, display in notebook; else return **plotly.graph_objs.Figure** list
:param show_notebook: whether to display graphics in notebook, the default is **True**.
:return: if show_notebook is True, display in notebook; else return **plotly.graph_objs.Figure** list.
"""
report_df = report_df.copy()
fig_list = _report_figure(report_df)

6
qlib/contrib/report/analysis_position/risk_analysis.py
View File

@@ -218,7 +218,7 @@ def risk_analysis_graph(
max_drawdown -0.088263
:param report_normal_df: **df.index.name** must be **date**, df.columns must contain **return**, **turnover**, **cost**, **bench**
:param report_normal_df: **df.index.name** must be **date**, df.columns must contain **return**, **turnover**, **cost**, **bench**.
.. code-block:: python
@@ -232,7 +232,7 @@ def risk_analysis_graph(
2017-01-10 -0.000416 0.000440 -0.003350 0.208396
:param report_long_short_df: **df.index.name** must be **date**, df.columns contain **long**, **short**, **long_short**
:param report_long_short_df: **df.index.name** must be **date**, df.columns contain **long**, **short**, **long_short**.
.. code-block:: python
@@ -246,7 +246,7 @@ def risk_analysis_graph(
2017-01-10 0.000824 -0.001944 -0.001120
:param show_notebook: Whether to display graphics in a notebook, default **True**
:param show_notebook: Whether to display graphics in a notebook, default **True**.
If True, show graph in notebook
If False, return graph figure
:return:

6
qlib/contrib/report/analysis_position/score_ic.py
View File

@@ -36,7 +36,7 @@ def score_ic_graph(pred_label: pd.DataFrame, show_notebook: bool = True) -> [lis
analysis_position.score_ic_graph(pred_label)
:param pred_label: index is **pd.MultiIndex**, index name is **[instrument, datetime]**; columns names is **[score, label]**
:param pred_label: index is **pd.MultiIndex**, index name is **[instrument, datetime]**; columns names is **[score, label]**.
.. code-block:: python
@@ -49,8 +49,8 @@ def score_ic_graph(pred_label: pd.DataFrame, show_notebook: bool = True) -> [lis
2017-12-15 -0.102778 -0.102778
:param show_notebook: whether to display graphics in notebook, the default is **True**
:return: if show_notebook is True, display in notebook; else return **plotly.graph_objs.Figure** list
:param show_notebook: whether to display graphics in notebook, the default is **True**.
:return: if show_notebook is True, display in notebook; else return **plotly.graph_objs.Figure** list.
"""
_ic_df = _get_score_ic(pred_label)
# FIXME: support HIGH-FREQ

70
qlib/contrib/strategy/strategy.py
View File

@@ -31,16 +31,16 @@ class BaseStrategy:
Parameters
-----------
score_series : pd.Seires
stock_id , score
stock_id , score.
current : Position()
current state of position
DO NOT directly change the state of current
current state of position.
DO NOT directly change the state of current.
trade_exchange : Exchange()
trade exchange
trade exchange.
pred_date : pd.Timestamp
predict date
predict date.
trade_date : pd.Timestamp
trade date
trade date.
"""
pass
@@ -49,11 +49,11 @@ class BaseStrategy:
Parameters
-----------
score_series : pd.Series
stock_id , score
stock_id , score.
pred_date : pd.Timestamp
oredict date
oredict date.
trade_date : pd.Timestamp
trade date
trade date.
"""
pass
@@ -67,7 +67,7 @@ class BaseStrategy:
"""
This method only be used in 'online' module, it will generate the *args to initial the strategy.
:param
mode : model used in 'online' module
mode : model used in 'online' module.
"""
return {}
@@ -82,7 +82,7 @@ class StrategyWrapper:
def __init__(self, inner_strategy):
"""__init__
:param inner_strategy: set the inner strategy
:param inner_strategy: set the inner strategy.
"""
self.inner_strategy = inner_strategy
@@ -99,9 +99,9 @@ class AdjustTimer:
Responsible for timing of position adjusting
This is designed as multiple inheritance mechanism due to:
- the is_adjust may need access to the internel state of a strategy
- the is_adjust may need access to the internel state of a strategy.
- it can be reguard as a enhancement to the existing strategy
- it can be reguard as a enhancement to the existing strategy.
"""
# adjust position in each trade date
@@ -146,12 +146,12 @@ class WeightStrategyBase(BaseStrategy, AdjustTimer):
Parameters
-----------
score : pd.Series
pred score for this trade date, index is stock_id, contain 'score' column
pred score for this trade date, index is stock_id, contain 'score' column.
current : Position()
current position
current position.
trade_exchange : Exchange()
trade_date : pd.Timestamp
trade date
trade date.
"""
raise NotImplementedError()
@@ -160,13 +160,13 @@ class WeightStrategyBase(BaseStrategy, AdjustTimer):
Parameters
-----------
score_series : pd.Seires
stock_id , score
stock_id , score.
current : Position()
current of account
current of account.
trade_exchange : Exchange()
exchange
exchange.
trade_date : pd.Timestamp
date
date.
"""
# judge if to adjust
if not self.is_adjust(trade_date):
@@ -206,26 +206,26 @@ class TopkDropoutStrategy(BaseStrategy, ListAdjustTimer):
Parameters
-----------
topk : int
The number of stocks in the portfolio
the number of stocks in the portfolio.
n_drop : int
number of stocks to be replaced in each trading date
number of stocks to be replaced in each trading date.
method_sell : str
dropout method_sell, random/bottom
dropout method_sell, random/bottom.
method_buy : str
dropout method_buy, random/top
dropout method_buy, random/top.
risk_degree : float
position percentage of total value
position percentage of total value.
thresh : int
minimun holding days since last buy singal of the stock
minimun holding days since last buy singal of the stock.
hold_thresh : int
minimum holding days
before sell stock , will check current.get_stock_count(order.stock_id) >= self.thresh
before sell stock , will check current.get_stock_count(order.stock_id) >= self.thresh.
only_tradable : bool
will the strategy only consider the tradable stock when buying and selling.
if only_tradable:
strategy will make buy sell decision without checking the tradable state of the stock
strategy will make buy sell decision without checking the tradable state of the stock.
else:
strategy will make decision with the tradable state of the stock info and avoid buy and sell them
strategy will make decision with the tradable state of the stock info and avoid buy and sell them.
"""
super(TopkDropoutStrategy, self).__init__()
ListAdjustTimer.__init__(self, kwargs.get("adjust_dates", None))
@@ -245,7 +245,7 @@ class TopkDropoutStrategy(BaseStrategy, ListAdjustTimer):
def get_risk_degree(self, date):
"""get_risk_degree
Return the proportion of your total value you will used in investment.
Dynamically risk_degree will result in Market timing
Dynamically risk_degree will result in Market timing.
"""
# It will use 95% amoutn of your total value by default
return self.risk_degree
@@ -257,15 +257,15 @@ class TopkDropoutStrategy(BaseStrategy, ListAdjustTimer):
Parameters
-----------
score_series : pd.Series
stock_id , score
stock_id , score.
current : Position()
current of account
current of account.
trade_exchange : Exchange()
exchange
exchange.
pred_date : pd.Timestamp
predict date
predict date.
trade_date : pd.Timestamp
trade date
trade date.
"""
if not self.is_adjust(trade_date):
return []