mirror of
https://github.com/microsoft/qlib.git
synced 2026-07-14 08:16:54 +08:00
Merge branch 'main' of https://github.com/you-n-g/qlib into main
This commit is contained in:
meng-ustc
@@ -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
|
||||
@@ -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))
|
||||
@@ -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")
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -34,14 +34,14 @@ class CatBoostModel(Model):
|
||||
def fit(
|
||||
self,
|
||||
dataset: DatasetH,
|
||||
num_boost_round = 1000,
|
||||
early_stopping_rounds = 50,
|
||||
verbose_eval = 20,
|
||||
evals_result = dict(),
|
||||
num_boost_round=1000,
|
||||
early_stopping_rounds=50,
|
||||
verbose_eval=20,
|
||||
evals_result=dict(),
|
||||
**kwargs
|
||||
):
|
||||
df_train, df_valid = dataset.prepare(
|
||||
["train", "valid"], col_set = ["feature", "label"], data_key = DataHandlerLP.DK_L
|
||||
["train", "valid"], 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"]
|
||||
@@ -52,8 +52,8 @@ class CatBoostModel(Model):
|
||||
else:
|
||||
raise ValueError("CatBoost doesn't support multi-label training")
|
||||
|
||||
train_pool = Pool(data = x_train, label = y_train_1d)
|
||||
valid_pool = Pool(data = x_valid, label = y_valid_1d)
|
||||
train_pool = Pool(data=x_train, label=y_train_1d)
|
||||
valid_pool = Pool(data=x_valid, label=y_valid_1d)
|
||||
|
||||
# Initialize the catboost model
|
||||
self._params["iterations"] = num_boost_round
|
||||
@@ -63,7 +63,7 @@ class CatBoostModel(Model):
|
||||
self.model = CatBoost(self._params, **kwargs)
|
||||
|
||||
# train the model
|
||||
self.model.fit(train_pool, eval_set = valid_pool, use_best_model = True, **kwargs)
|
||||
self.model.fit(train_pool, eval_set=valid_pool, use_best_model=True, **kwargs)
|
||||
|
||||
evals_result = self.model.get_evals_result()
|
||||
evals_result["train"] = list(evals_result["learn"].values())[0]
|
||||
@@ -72,8 +72,8 @@ class CatBoostModel(Model):
|
||||
def predict(self, dataset):
|
||||
if self.model is None:
|
||||
raise ValueError("model is not fitted yet!")
|
||||
x_test = dataset.prepare("test", col_set = "feature")
|
||||
return pd.Series(self.model.predict(x_test.values), index = x_test.index)
|
||||
x_test = dataset.prepare("test", col_set="feature")
|
||||
return pd.Series(self.model.predict(x_test.values), index=x_test.index)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -111,7 +111,7 @@ class GAT(Model):
|
||||
seed,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
self.GAT_model = GATModel(
|
||||
d_feat=self.d_feat,
|
||||
hidden_size=self.hidden_size,
|
||||
|
||||
@@ -264,10 +264,12 @@ class HATS(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"))
|
||||
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()
|
||||
@@ -462,7 +464,9 @@ class GraphAttention(nn.Module):
|
||||
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)
|
||||
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))
|
||||
|
||||
@@ -31,6 +31,7 @@ from ...model.base import Model
|
||||
from ...data.dataset import DatasetH
|
||||
from ...data.dataset.handler import DataHandlerLP
|
||||
|
||||
|
||||
class SFM_Model(nn.Module):
|
||||
def __init__(self, d_feat=6, output_dim=1, freq_dim=10, hidden_size=64, dropout_W=0.0, dropout_U=0.0, device="cpu"):
|
||||
super().__init__()
|
||||
@@ -75,13 +76,13 @@ class SFM_Model(nn.Module):
|
||||
self.states = []
|
||||
|
||||
def forward(self, input):
|
||||
input = input.reshape(len(input), self.input_dim, -1) # [N, F, T]
|
||||
input = input.permute(0, 2, 1) # [N, T, F]
|
||||
input = input.reshape(len(input), self.input_dim, -1) # [N, F, T]
|
||||
input = input.permute(0, 2, 1) # [N, T, F]
|
||||
time_step = input.shape[1]
|
||||
|
||||
|
||||
for ts in range(time_step):
|
||||
x = input[:, ts,:]
|
||||
if len(self.states)==0: #hasn't initialized yet
|
||||
x = input[:, ts, :]
|
||||
if len(self.states) == 0: # hasn't initialized yet
|
||||
self.init_states(x)
|
||||
self.get_constants(x)
|
||||
p_tm1 = self.states[0]
|
||||
@@ -98,64 +99,65 @@ class SFM_Model(nn.Module):
|
||||
x_fre = torch.matmul(x * B_W[0], self.W_fre) + self.b_fre
|
||||
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)
|
||||
) # not sure whether I am doing in the right unsquuze
|
||||
|
||||
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))
|
||||
|
||||
ste = torch.reshape(ste, (-1, self.hidden_dim, 1))
|
||||
fre = torch.reshape(fre, (-1, 1, self.freq_dim))
|
||||
|
||||
|
||||
f = ste * fre
|
||||
|
||||
|
||||
c = i * self.activation(x_c + torch.matmul(h_tm1 * B_U[0], self.U_c))
|
||||
|
||||
time = time_tm1 + 1
|
||||
|
||||
omega = torch.tensor(2 * np.pi) * time * frequency
|
||||
|
||||
re = torch.cos(omega)
|
||||
re = torch.cos(omega)
|
||||
im = torch.sin(omega)
|
||||
|
||||
|
||||
c = torch.reshape(c, (-1, self.hidden_dim, 1))
|
||||
|
||||
S_re = f * S_re_tm1 + c * re
|
||||
S_im = f * S_im_tm1 + c * im
|
||||
|
||||
|
||||
A = torch.square(S_re) + torch.square(S_im)
|
||||
|
||||
A = torch.reshape(A, (-1, self.freq_dim)).float()
|
||||
A_a = torch.matmul(A * B_U[0], self.U_a)
|
||||
A_a = torch.reshape(A_a, (-1, self.hidden_dim))
|
||||
a = self.activation(A_a + self.b_a)
|
||||
|
||||
|
||||
o = self.inner_activation(x_o + torch.matmul(h_tm1 * B_U[0], self.U_o))
|
||||
|
||||
h = o * a
|
||||
p = torch.matmul(h, self.W_p) + self.b_p
|
||||
|
||||
self.states = [p, h, S_re, S_im, time, None, None, None]
|
||||
self.states = []
|
||||
self.states = []
|
||||
return self.fc_out(p).squeeze()
|
||||
|
||||
def init_states(self, x):
|
||||
reducer_f = torch.zeros((self.hidden_dim, self.freq_dim)).to(self.device)
|
||||
reducer_p = torch.zeros((self.hidden_dim, self.output_dim)).to(self.device)
|
||||
|
||||
|
||||
init_state_h = torch.zeros(self.hidden_dim).to(self.device)
|
||||
init_state_p = torch.matmul(init_state_h, reducer_p)
|
||||
|
||||
|
||||
init_state = torch.zeros_like(init_state_h).to(self.device)
|
||||
init_freq = torch.matmul(init_state_h, reducer_f)
|
||||
|
||||
init_state = torch.reshape(init_state, (-1, self.hidden_dim, 1))
|
||||
init_freq = torch.reshape(init_freq, (-1, 1, self.freq_dim))
|
||||
|
||||
|
||||
init_state_S_re = init_state * init_freq
|
||||
init_state_S_im = init_state * init_freq
|
||||
|
||||
|
||||
init_state_time = torch.tensor(0).to(self.device)
|
||||
|
||||
self.states = [init_state_p, init_state_h, init_state_S_re, init_state_S_im, init_state_time, None, None, None]
|
||||
@@ -201,7 +203,7 @@ class SFM(Model):
|
||||
dropout_U=0.0,
|
||||
n_epochs=200,
|
||||
lr=0.001,
|
||||
metric = "",
|
||||
metric="",
|
||||
batch_size=2000,
|
||||
early_stop=20,
|
||||
eval_steps=5,
|
||||
@@ -234,7 +236,7 @@ class SFM(Model):
|
||||
self.lr_decay_steps = lr_decay_steps
|
||||
self.optimizer = optimizer.lower()
|
||||
self.loss = loss
|
||||
self.device = "cuda:%d"%(GPU) if torch.cuda.is_available() else "cpu"
|
||||
self.device = "cuda:%d" % (GPU) if torch.cuda.is_available() else "cpu"
|
||||
self.use_gpu = torch.cuda.is_available()
|
||||
self.seed = seed
|
||||
|
||||
@@ -243,7 +245,7 @@ class SFM(Model):
|
||||
"\nd_feat : {}"
|
||||
"\nhidden_size : {}"
|
||||
"\noutput_size : {}"
|
||||
"\nfrequency_dimension : {}"
|
||||
"\nfrequency_dimension : {}"
|
||||
"\ndropout_W: {}"
|
||||
"\ndropout_U: {}"
|
||||
"\nn_epochs : {}"
|
||||
@@ -286,14 +288,14 @@ class SFM(Model):
|
||||
self._scorer = mean_squared_error if loss == "mse" else roc_auc_score
|
||||
|
||||
self.sfm_model = SFM_Model(
|
||||
d_feat=self.d_feat,
|
||||
d_feat=self.d_feat,
|
||||
output_dim=self.output_dim,
|
||||
hidden_size=self.hidden_size,
|
||||
freq_dim=self.freq_dim,
|
||||
dropout_W=self.dropout_W,
|
||||
dropout_U=self.dropout_U,
|
||||
device=self.device
|
||||
)
|
||||
hidden_size=self.hidden_size,
|
||||
freq_dim=self.freq_dim,
|
||||
dropout_W=self.dropout_W,
|
||||
dropout_U=self.dropout_U,
|
||||
device=self.device,
|
||||
)
|
||||
if optimizer.lower() == "adam":
|
||||
self.train_optimizer = optim.Adam(self.sfm_model.parameters(), lr=self.lr)
|
||||
elif optimizer.lower() == "gd":
|
||||
@@ -414,7 +416,7 @@ class SFM(Model):
|
||||
def mse(self, pred, label):
|
||||
loss = (pred - label) ** 2
|
||||
return torch.mean(loss)
|
||||
|
||||
|
||||
def loss_fn(self, pred, label):
|
||||
mask = ~torch.isnan(label)
|
||||
|
||||
@@ -422,7 +424,7 @@ class SFM(Model):
|
||||
return self.mse(pred[mask], label[mask])
|
||||
|
||||
raise ValueError("unknown loss `%s`" % self.loss)
|
||||
|
||||
|
||||
def metric_fn(self, pred, label):
|
||||
|
||||
mask = torch.isfinite(label)
|
||||
@@ -436,6 +438,7 @@ class SFM(Model):
|
||||
|
||||
def cal_ic(self, pred, label):
|
||||
return torch.mean(pred * label)
|
||||
|
||||
def predict(self, dataset):
|
||||
if not self._fitted:
|
||||
raise ValueError("model is not fitted yet!")
|
||||
@@ -447,7 +450,7 @@ class SFM(Model):
|
||||
sample_num = x_values.shape[0]
|
||||
preds = []
|
||||
|
||||
for begin in range(sample_num)[::self.batch_size]:
|
||||
for begin in range(sample_num)[:: self.batch_size]:
|
||||
if sample_num - begin < self.batch_size:
|
||||
end = sample_num
|
||||
else:
|
||||
@@ -457,16 +460,18 @@ class SFM(Model):
|
||||
|
||||
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):
|
||||
"""Computes and stores the average and current value"""
|
||||
|
||||
def __init__(self):
|
||||
self.reset()
|
||||
|
||||
|
||||
@@ -30,15 +30,15 @@ class XGBModel(Model):
|
||||
def fit(
|
||||
self,
|
||||
dataset: DatasetH,
|
||||
num_boost_round = 1000,
|
||||
early_stopping_rounds = 50,
|
||||
verbose_eval = 20,
|
||||
evals_result = dict(),
|
||||
num_boost_round=1000,
|
||||
early_stopping_rounds=50,
|
||||
verbose_eval=20,
|
||||
evals_result=dict(),
|
||||
**kwargs
|
||||
):
|
||||
|
||||
df_train, df_valid = dataset.prepare(
|
||||
["train", "valid"], col_set = ["feature", "label"], data_key = DataHandlerLP.DK_L
|
||||
["train", "valid"], 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"]
|
||||
@@ -49,16 +49,16 @@ class XGBModel(Model):
|
||||
else:
|
||||
raise ValueError("XGBoost doesn't support multi-label training")
|
||||
|
||||
dtrain = xgb.DMatrix(x_train.values, label = y_train_1d)
|
||||
dvalid = xgb.DMatrix(x_valid.values, label = y_valid_1d)
|
||||
dtrain = xgb.DMatrix(x_train.values, label=y_train_1d)
|
||||
dvalid = xgb.DMatrix(x_valid.values, label=y_valid_1d)
|
||||
self.model = xgb.train(
|
||||
self._params,
|
||||
dtrain = dtrain,
|
||||
num_boost_round = num_boost_round,
|
||||
evals = [(dtrain, "train"), (dvalid, "valid")],
|
||||
early_stopping_rounds = early_stopping_rounds,
|
||||
verbose_eval = verbose_eval,
|
||||
evals_result = evals_result,
|
||||
dtrain=dtrain,
|
||||
num_boost_round=num_boost_round,
|
||||
evals=[(dtrain, "train"), (dvalid, "valid")],
|
||||
early_stopping_rounds=early_stopping_rounds,
|
||||
verbose_eval=verbose_eval,
|
||||
evals_result=evals_result,
|
||||
**kwargs
|
||||
)
|
||||
evals_result["train"] = list(evals_result["train"].values())[0]
|
||||
@@ -67,5 +67,5 @@ class XGBModel(Model):
|
||||
def predict(self, dataset):
|
||||
if self.model is None:
|
||||
raise ValueError("model is not fitted yet!")
|
||||
x_test = dataset.prepare("test", col_set = "feature")
|
||||
return pd.Series(self.model.predict(xgb.DMatrix(x_test.values)), index = x_test.index)
|
||||
x_test = dataset.prepare("test", col_set="feature")
|
||||
return pd.Series(self.model.predict(xgb.DMatrix(x_test.values)), index=x_test.index)
|
||||
|
||||
Reference in New Issue
Block a user