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

update

Browse Source
This commit is contained in:
bxdd
2021-01-28 11:31:15 +00:00
parent 948b829ff4
commit f6dd006c35
12 changed files with 242 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
examples/high_freq/__init__.py → examples/highfreq/__init__.py
View File

107
examples/high_freq/highfreq_handler.py → examples/highfreq/highfreq_handler.py
View File

@@ -56,88 +56,44 @@ class HighFreqHandler(DataHandlerLP):
template_if = "If(IsNull({1}), {0}, {1})"
template_paused = "Select(Or(IsNull($paused), Eq($paused, 0.0)), {0})"
# template_paused="{0}"
template_fillnan = "FFillNan({0})"
template_fillnan = "BFillNan(FFillNan({0}))"
# Because there is no vwap field in the yahoo data, a method similar to Simpson integration is used to approximate vwap
simpson_vwap = "($open + 2*$high + 2*$low + $close)/6"
fields += [
"{0}/Ref(DayLast({1}), 240)".format(
def get_04_price_feature(price_field):
"""Get 0~4 column price feature ops"""
feature_ops = "{0}/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format("$open"),
template_paused.format(price_field),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += [
"{0}/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format("$high"),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += [
"{0}/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format("$low"),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += ["{0}/Ref(DayLast({0}), 240)".format(template_fillnan.format(template_paused.format("$close")))]
fields += [
"{0}/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format(simpson_vwap),
),
template_fillnan.format(template_paused.format("$close")),
)
]
return feature_ops
fields += [get_04_price_feature("$open")]
fields += [get_04_price_feature("$high")]
fields += [get_04_price_feature("$low")]
fields += [get_04_price_feature("$close")]
fields += [get_04_price_feature(simpson_vwap)]
names += ["$open", "$high", "$low", "$close", "$vwap"]
fields += [
"Ref({0}, 240)/Ref(DayLast({1}), 240)".format(
def get_59_price_feature(price_field):
"""Get 5~9 column price feature ops"""
feature_ops = "Ref({0}, 240)/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format("$open"),
template_paused.format(price_field),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += [
"Ref({0}, 240)/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format("$high"),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += [
"Ref({0}, 240)/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format("$low"),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += [
"Ref({0}, 240)/Ref(DayLast({0}), 240)".format(template_fillnan.format(template_paused.format("$close")))
]
return feature_ops
fields += [
"Ref({0}, 240)/Ref(DayLast({1}), 240)".format(
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format(simpson_vwap),
),
template_fillnan.format(template_paused.format("$close")),
)
]
fields += [get_59_price_feature("$open")]
fields += [get_59_price_feature("$high")]
fields += [get_59_price_feature("$low")]
fields += [get_59_price_feature("$close")]
fields += [get_59_price_feature(simpson_vwap)]
names += ["$open_1", "$high_1", "$low_1", "$close_1", "$vwap_1"]
fields += [
@@ -197,19 +153,20 @@ class HighFreqBacktestHandler(DataHandler):
template_if = "If(IsNull({1}), {0}, {1})"
template_paused = "Select(Or(IsNull($paused), Eq($paused, 0.0)), {0})"
# template_paused="{0}"
template_fillnan = "FFillNan({0})"
template_fillnan = "BFillNan(FFillNan({0}))"
# Because there is no vwap field in the yahoo data, a method similar to Simpson integration is used to approximate vwap
simpson_vwap = "($open + 2*$high + 2*$low + $close)/6"
# fields += [
# template_fillnan.format(template_paused.format("$close")),
# ]
fields += [
template_fillnan.format(template_paused.format("$close")),
]
names += ["$close0"]
fields += [
template_if.format(
template_fillnan.format(template_paused.format("$close")),
template_paused.format(simpson_vwap),
)
]
names += ["$vwap_0"]
names += ["$vwap0"]
fields += [
"If(IsNull({0}), 0, If(Or(Gt({1}, Mul(1.001, {3})), Lt({1}, Mul(0.999, {2}))), 0, {0}))".format(
template_paused.format("$volume"),
@@ -218,6 +175,6 @@ class HighFreqBacktestHandler(DataHandler):
template_paused.format("$high"),
)
]
names += ["$volume_0"]
names += ["$volume0"]
return fields, names

42
examples/high_freq/highfreq_ops.py → examples/highfreq/highfreq_ops.py
View File

@@ -3,51 +3,61 @@ import pandas as pd
import importlib
from qlib.data.ops import ElemOperator, PairOperator
from qlib.config import C
from qlib.data.cache import H
from qlib.data.data import Cal
class DayFirst(ElemOperator):
def __init__(self, feature):
super(DayFirst, self).__init__(feature, "day_first")
def _load_internal(self, instrument, start_index, end_index, freq):
_calendar = Cal.get_calendar_day(freq=freq)[0]
series = self.feature.load(instrument, start_index, end_index, freq)
return series.groupby(_calendar[series.index]).transform("first")
def get_calendar_day(freq="day", future=False):
flag = f"{freq}_future_{future}_day"
if flag in H["c"]:
_calendar = H["c"][flag]
else:
_calendar = np.array(list(map(lambda x: x.date(), Cal.load_calendar(freq, future))))
H["c"][flag] = _calendar
return _calendar
class DayLast(ElemOperator):
def __init__(self, feature):
super(DayLast, self).__init__(feature, "day_last")
super(DayLast, self).__init__(feature)
def _load_internal(self, instrument, start_index, end_index, freq):
_calendar = Cal.get_calendar_day(freq=freq)[0]
_calendar = get_calendar_day(freq=freq)
series = self.feature.load(instrument, start_index, end_index, freq)
return series.groupby(_calendar[series.index]).transform("last")
class FFillNan(ElemOperator):
def __init__(self, feature):
super(FFillNan, self).__init__(feature, "fill_nan")
super(FFillNan, self).__init__(feature)
def _load_internal(self, instrument, start_index, end_index, freq):
series = self.feature.load(instrument, start_index, end_index, freq)
return series.fillna(method="ffill")
class Date(ElemOperator):
class BFillNan(ElemOperator):
def __init__(self, feature):
super(Date, self).__init__(feature, "date")
super(BFillNan, self).__init__(feature)
def _load_internal(self, instrument, start_index, end_index, freq):
_calendar = Cal.get_calendar_day(freq=freq)[0]
series = self.feature.load(instrument, start_index, end_index, freq)
return series.fillna(method="bfill")
class Date(ElemOperator):
def __init__(self, feature):
super(Date, self).__init__(feature)
def _load_internal(self, instrument, start_index, end_index, freq):
_calendar = get_calendar_day(freq=freq)
series = self.feature.load(instrument, start_index, end_index, freq)
return pd.Series(_calendar[series.index], index=series.index)
class Select(PairOperator):
def __init__(self, condition, feature):
super(Select, self).__init__(condition, feature, "select")
super(Select, self).__init__(condition, feature)
def _load_internal(self, instrument, start_index, end_index, freq):
series_condition = self.feature_left.load(instrument, start_index, end_index, freq)
@@ -57,7 +67,7 @@ class Select(PairOperator):
class IsNull(ElemOperator):
def __init__(self, feature):
super(IsNull, self).__init__(feature, "isnull")
super(IsNull, self).__init__(feature)
def _load_internal(self, instrument, start_index, end_index, freq):
series = self.feature.load(instrument, start_index, end_index, freq)

32
examples/high_freq/highfreq_processor.py → examples/highfreq/highfreq_processor.py
View File

@@ -26,7 +26,7 @@ class HighFreqNorm(Processor):
if name == "volume":
part_values = np.log1p(part_values)
self.feature_med[name] = np.nanmedian(part_values)
part_values = part_values - self.feature_med[name] # mean, copy
part_values = part_values - self.feature_med[name]
self.feature_std[name] = np.nanmedian(np.absolute(part_values)) * 1.4826 + 1e-12
part_values = part_values / self.feature_std[name]
self.feature_vmax[name] = np.nanmax(part_values)
@@ -41,23 +41,27 @@ class HighFreqNorm(Processor):
}
for name, name_val in names.items():
part_values = df_values[:, name_val]
if name == "volume":
part_values[:] = np.log1p(part_values)
part_values -= self.feature_med[name]
part_values /= self.feature_std[name]
slice0 = part_values > 3.0
slice1 = part_values > 3.5
slice2 = part_values < -3.0
slice3 = part_values < -3.5
df_values[:, name_val] = np.log1p(df_values[:, name_val])
df_values[:, name_val] -= self.feature_med[name]
df_values[:, name_val] /= self.feature_std[name]
slice0 = df_values[:, name_val] > 3.0
slice1 = df_values[:, name_val] > 3.5
slice2 = df_values[:, name_val] < -3.0
slice3 = df_values[:, name_val] < -3.5
part_values[slice0] = 3.0 + (part_values[slice0] - 3.0) / (self.feature_vmax[name] - 3) * 0.5
part_values[slice1] = 3.5
part_values[slice2] = -3.0 - (part_values[slice2] + 3.0) / (self.feature_vmin[name] + 3) * 0.5
part_values[slice3] = -3.5
# print("start_call_feature_reshape")
df_values[:, name_val][slice0] = (
3.0 + (df_values[:, name_val][slice0] - 3.0) / (self.feature_vmax[name] - 3) * 0.5
)
df_values[:, name_val][slice1] = 3.5
df_values[:, name_val][slice2] = (
-3.0 - (df_values[:, name_val][slice2] + 3.0) / (self.feature_vmin[name] + 3) * 0.5
)
df_values[:, name_val][slice3] = -3.5
idx = df_features.index.droplevel("datetime").drop_duplicates()
idx.set_names(["instrument", "datetime"], inplace=True)
# Reshape is specifically for adapting to RL high-freq executor
feat = df_values[:, [0, 1, 2, 3, 4, 10]].reshape(-1, 6 * 240)
feat_1 = df_values[:, [5, 6, 7, 8, 9, 11]].reshape(-1, 6 * 240)
df_new_features = pd.DataFrame(

120
examples/high_freq/workflow.py → examples/highfreq/workflow.py
View File

@@ -2,13 +2,14 @@
# Licensed under the MIT License.
import sys
import fire
from pathlib import Path
import qlib
import pickle
import numpy as np
import pandas as pd
from qlib.config import REG_CN
from qlib.config import HIGH_FREQ_CONFIG
from qlib.contrib.model.gbdt import LGBModel
from qlib.contrib.data.handler import Alpha158
from qlib.contrib.strategy.strategy import TopkDropoutStrategy
@@ -23,42 +24,22 @@ from qlib.data.ops import Operators
from qlib.data.data import Cal
from qlib.utils import exists_qlib_data
from highfreq_ops import DayFirst, DayLast, FFillNan, Date, Select, IsNull
from highfreq_ops import get_calendar_day, DayLast, FFillNan, BFillNan, Date, Select, IsNull
if __name__ == "__main__":
# use yahoo_cn_1min data
provider_uri = "~/.qlib/qlib_data/yahoo_cn_1min"
if not exists_qlib_data(provider_uri):
print(f"Qlib data is not found in {provider_uri}")
sys.path.append(str(Path(__file__).resolve().parent.parent.parent.joinpath("scripts")))
from get_data import GetData
class HighfreqWorkflow(object):
GetData().qlib_data(target_dir=provider_uri, interval="1min", region=REG_CN)
qlib.init(
provider_uri=provider_uri,
custom_ops=[DayFirst, DayLast, FFillNan, Date, Select, IsNull],
redis_port=-1,
region=REG_CN,
auto_mount=False,
)
SPEC_CONF = {"custom_ops": [DayLast, FFillNan, BFillNan, Date, Select, IsNull], "expression_cache": None}
MARKET = "all"
BENCHMARK = "SH000300"
DROP_LOAD_DATASET = False # flag wether to test [drop and load dataset]
# start_time = "2019-01-01 00:00:00"
# end_time = "2019-12-31 15:00:00"
# train_end_time = "2019-05-31 15:00:00"
# test_start_time = "2019-06-01 00:00:00"
start_time = "2020-09-14 00:00:00"
end_time = "2021-01-18 16:00:00"
train_end_time = "2020-11-30 16:00:00"
test_start_time = "2020-12-01 00:00:00"
###################################
# train model
###################################
DATA_HANDLER_CONFIG0 = {
"start_time": start_time,
"end_time": end_time,
@@ -94,8 +75,6 @@ if __name__ == "__main__":
},
},
},
# You shoud record the data in specific sequence
# "record": ['SignalRecord', 'SigAnaRecord', 'PortAnaRecord'],
"dataset_backtest": {
"class": "DatasetH",
"module_path": "qlib.data.dataset",
@@ -115,26 +94,50 @@ if __name__ == "__main__":
},
},
}
##=============load the calendar for cache=============
# unnecessary, but may accelerate
Cal.calendar(freq="1min") # load the calendar for cache
Cal.get_calendar_day(freq="1min") # load the calendar for cache
##=============get data=============
def _init_qlib(self):
"""initialize qlib"""
# use yahoo_cn_1min data
QLIB_INIT_CONFIG = {**HIGH_FREQ_CONFIG, **self.SPEC_CONF}
provider_uri = QLIB_INIT_CONFIG.get("provider_uri")
if not exists_qlib_data(provider_uri):
print(f"Qlib data is not found in {provider_uri}")
sys.path.append(str(Path(__file__).resolve().parent.parent.parent.joinpath("scripts")))
from get_data import GetData
dataset = init_instance_by_config(task["dataset"])
xtrain, xtest = dataset.prepare(["train", "test"])
print(xtrain, xtest)
GetData().qlib_data(target_dir=provider_uri, interval="1min", region=REG_CN)
qlib.init(**QLIB_INIT_CONFIG)
dataset_backtest = init_instance_by_config(task["dataset_backtest"])
backtest_train, backtest_test = dataset_backtest.prepare(["train", "test"])
print(backtest_train, backtest_test)
def _prepare_calender_cache(self):
"""preload the calendar for cache"""
del xtrain, xtest
del backtest_train, backtest_test
# This code used the copy-on-write feature of Linux to avoid calculating the calendar multiple times in the subprocess
# This code may accelerate, but may be not useful on Windows and Mac Os
Cal.calendar(freq="1min")
get_calendar_day(freq="1min")
## example to show how to save the dataset and reload it, and how to use different data
if DROP_LOAD_DATASET:
def get_data(self):
"""use dataset to get highreq data"""
self._init_qlib()
self._prepare_calender_cache()
dataset = init_instance_by_config(self.task["dataset"])
xtrain, xtest = dataset.prepare(["train", "test"])
print(xtrain, xtest)
dataset_backtest = init_instance_by_config(self.task["dataset_backtest"])
backtest_train, backtest_test = dataset_backtest.prepare(["train", "test"])
print(backtest_train, backtest_test)
del xtrain, xtest
del backtest_train, backtest_test
def dump_and_load_dataset(self):
"""dump and load dataset state on disk"""
self._init_qlib()
self._prepare_calender_cache()
dataset = init_instance_by_config(self.task["dataset"])
dataset_backtest = init_instance_by_config(self.task["dataset_backtest"])
##=============dump dataset=============
dataset.to_pickle(path="dataset.pkl")
@@ -142,33 +145,18 @@ if __name__ == "__main__":
del dataset, dataset_backtest
##=============reload dataset=============
file_dataset = open("dataset.pkl", "rb")
dataset = pickle.load(file_dataset)
file_dataset.close()
with open("dataset.pkl", "rb") as file_dataset:
dataset = pickle.load(file_dataset)
file_dataset_backtest = open("dataset_backtest.pkl", "rb")
dataset_backtest = pickle.load(file_dataset_backtest)
file_dataset_backtest.close()
with open("dataset_backtest.pkl", "rb") as file_dataset_backtest:
dataset_backtest = pickle.load(file_dataset_backtest)
self._prepare_calender_cache()
##=============reload_dataset=============
dataset.init(init_type=DataHandlerLP.IT_LS)
dataset_backtest.init(init_type=DataHandlerLP.IT_LS)
dataset_backtest.init()
##=============reinit qlib=============
## Unless you want to modify the provider_uri and other configurations, reinit is unnecessary
qlib.init(
provider_uri=provider_uri,
custom_ops=[DayFirst, DayLast, FFillNan, Date, Select, IsNull],
redis_port=-1,
region=REG_CN,
auto_mount=False,
)
Cal.calendar(freq="1min") # load the calendar for cache
Cal.get_calendar_day(freq="1min") # load the calendar for cache
##=============test dataset=============
##=============get data=============
xtrain, xtest = dataset.prepare(["train", "test"])
backtest_train, backtest_test = dataset_backtest.prepare(["train", "test"])
@@ -176,3 +164,7 @@ if __name__ == "__main__":
print(backtest_train, backtest_test)
del xtrain, xtest
del backtest_train, backtest_test
if __name__ == "__main__":
fire.Fire(HighfreqWorkflow)

2
examples/workflow_by_code.py
View File

@@ -30,7 +30,7 @@ if __name__ == "__main__":
GetData().qlib_data(target_dir=provider_uri, region=REG_CN)
qlib.init(provider_uri=provider_uri, region=REG_CN, redis_port=233)
qlib.init(provider_uri=provider_uri, region=REG_CN)
market = "csi300"
benchmark = "SH000300"