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

Format with black

Browse Source
This commit is contained in:
Jactus
2020-10-29 13:22:49 +08:00
parent 490dbd908b
commit da9d1c8ac6
20 changed files with 290 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
qlib/data/dataset/__init__.py
View File

@@ -6,11 +6,12 @@ import pandas as pd
class Dataset(Serializable):
'''
"""
Preparing data for model training and inferencing.
'''
"""
def __init__(self, *args, **kwargs):
'''
"""
init is designed to finish following steps
- setup data
- The data related attributes' names should start with '_' so that it will not be saved on disk when serializing
@@ -18,7 +19,7 @@ class Dataset(Serializable):
- The name of essential state for preparing data should not start with '_' so that it could be serialized on disk when serializing.
The data could specify the info to caculate the essential data for preparation
'''
"""
self.setup_data(*args, **kwargs)
super().__init__()
@@ -51,14 +52,15 @@ class Dataset(Serializable):
class DatasetH(Dataset):
'''
"""
Dataset with Data(H)anler
User should try to put the data preprocessing functions into handler.
Only following data processing functions should be placed in Dataset
- The processing is related to specific model.
- The processing is related to data split
'''
"""
def __init__(self, handler: Union[dict, DataHandler], segments: list):
"""
Parameters
@@ -96,10 +98,9 @@ class DatasetH(Dataset):
self._handler = init_instance_by_config(handler, accept_types=DataHandler)
self._segments = segments
def prepare(self,
segments: Union[List[str], Tuple[str], str, slice],
col_set=DataHandler.CS_ALL,
**kwargs) -> Union[List[pd.DataFrame], pd.DataFrame]:
def prepare(
self, segments: Union[List[str], Tuple[str], str, slice], col_set=DataHandler.CS_ALL, **kwargs
) -> Union[List[pd.DataFrame], pd.DataFrame]:
"""
prepare the data for learning and inference
@@ -124,9 +125,7 @@ class DatasetH(Dataset):
[TODO:description]
"""
if isinstance(segments, (list, tuple)):
return [
self._handler.fetch(slice(*self._segments[seg]), col_set=col_set, **kwargs) for seg in segments
]
return [self._handler.fetch(slice(*self._segments[seg]), col_set=col_set, **kwargs) for seg in segments]
elif isinstance(segments, str):
return self._handler.fetch(slice(*self._segments[segments]), col_set=col_set, **kwargs)
else:

105
qlib/data/dataset/handler.py
View File

@@ -25,7 +25,7 @@ from . import loader as data_loader_module
# TODO: A more general handler interface which does not relies on internal pd.DataFrame is needed.
class DataHandler(Serializable):
'''
"""
The steps to using a handler
1. initialized data handler (call by `init`).
2. use the data
@@ -46,13 +46,21 @@ class DataHandler(Serializable):
SH600004 13.313329 11800983.0 13.313329 13.317701 0.183632 0.0042
SH600005 37.796539 12231662.0 38.258602 37.919757 0.970325 0.0289
'''
def __init__(self, instruments, start_time=None, end_time=None, data_loader: Tuple[dict, str, DataLoader]=None, init_data=True):
"""
def __init__(
self,
instruments,
start_time=None,
end_time=None,
data_loader: Tuple[dict, str, DataLoader] = None,
init_data=True,
):
# Set logger
self.logger = get_module_logger("DataHandler")
# Setup data loader
assert(data_loader is not None) # to make start_time end_time could have None default value
assert data_loader is not None # to make start_time end_time could have None default value
self.data_loader = init_instance_by_config(data_loader, data_loader_module, accept_types=DataLoader)
self.instruments = instruments
@@ -62,7 +70,7 @@ class DataHandler(Serializable):
self.init()
super().__init__()
def init(self, enable_cache: bool=True):
def init(self, enable_cache: bool = True):
"""
initialize the data.
In case of running intialization for multiple time, it will do nothing for the second time.
@@ -83,7 +91,9 @@ class DataHandler(Serializable):
self._data = self.data_loader.load(self.instruments, self.start_time, self.end_time)
# TODO: cache
def _fetch_df_by_index(self, df: pd.DataFrame, selector: Union[pd.Timestamp, slice, str, list], level: Union[str, int]) -> pd.DataFrame:
def _fetch_df_by_index(
self, df: pd.DataFrame, selector: Union[pd.Timestamp, slice, str, list], level: Union[str, int]
) -> pd.DataFrame:
"""
fetch data from `data` with `selector` and `level`
@@ -100,7 +110,7 @@ class DataHandler(Serializable):
idx_slc = idx_slc[1], idx_slc[0]
return df.loc(axis=0)[idx_slc]
CS_ALL = '__all'
CS_ALL = "__all"
def _fetch_df_by_col(self, df: pd.DataFrame, col_set: str) -> pd.DataFrame:
cln = len(df.columns.levels)
@@ -111,10 +121,12 @@ class DataHandler(Serializable):
else:
return df.loc(axis=1)[col_set]
def fetch(self,
selector: Union[pd.Timestamp, slice, str],
level: Union[str, int] = 'datetime',
col_set: Union[str, List[str]] = CS_ALL) -> pd.DataFrame:
def fetch(
self,
selector: Union[pd.Timestamp, slice, str],
level: Union[str, int] = "datetime",
col_set: Union[str, List[str]] = CS_ALL,
) -> pd.DataFrame:
"""
fetch data from underlying data source
@@ -157,32 +169,35 @@ class DataHandler(Serializable):
class DataHandlerLP(DataHandler):
'''
"""
DataHandler with **(L)earnable (P)rocessor**
'''
"""
# data key
DK_R = 'raw'
DK_I = 'infer'
DK_L = 'learn'
DK_R = "raw"
DK_I = "infer"
DK_L = "learn"
# process type
PTYPE_I = 'independent'
PTYPE_I = "independent"
# - _proc_infer_df will processed by infer_processors
# - _proc_learn_df will be processed by learn_processors
PTYPE_A = 'append'
PTYPE_A = "append"
# - _proc_infer_df will processed by infer_processors
# - _proc_learn_df will be processed by infer_processors + learn_processors
# - (e.g. _proc_infer_df processed by learn_processors )
def __init__(self,
instruments,
start_time=None,
end_time=None,
data_loader: Tuple[dict, str, DataLoader] = None,
infer_processors=[],
learn_processors=[],
process_type=PTYPE_A,
**kwargs):
def __init__(
self,
instruments,
start_time=None,
end_time=None,
data_loader: Tuple[dict, str, DataLoader] = None,
infer_processors=[],
learn_processors=[],
process_type=PTYPE_A,
**kwargs,
):
"""
Parameters
----------
@@ -217,10 +232,11 @@ class DataHandlerLP(DataHandler):
# Setup preprocessor
self.infer_processors = [] # for lint
self.learn_processors = [] # for lint
for pname in 'infer_processors', 'learn_processors':
for pname in "infer_processors", "learn_processors":
for proc in locals()[pname]:
getattr(self, pname).append(init_instance_by_config(proc, processor_module,
accept_types=(processor_module.Processor,)))
getattr(self, pname).append(
init_instance_by_config(proc, processor_module, accept_types=(processor_module.Processor,))
)
self.process_type = process_type
super().__init__(instruments, start_time, end_time, data_loader, **kwargs)
@@ -240,8 +256,7 @@ class DataHandlerLP(DataHandler):
"""
self.process_data(with_fit=True)
def process_data(self, with_fit: bool=False):
def process_data(self, with_fit: bool = False):
"""
process_data data. Fun `processor.fit` if necessary
@@ -281,11 +296,11 @@ class DataHandlerLP(DataHandler):
self._learn = _learn_df
# init type
IT_FIT_SEQ = 'fit_seq' # the input of `fit` will be the output of the previous processor
IT_FIT_IND = 'fit_ind' # the input of `fit` will be the original df
IT_LS = 'load_state' # The state of the object has been load by pickle
IT_FIT_SEQ = "fit_seq" # the input of `fit` will be the output of the previous processor
IT_FIT_IND = "fit_ind" # the input of `fit` will be the original df
IT_LS = "load_state" # The state of the object has been load by pickle
def init(self, init_type: str=IT_FIT_SEQ, enable_cache: bool=False):
def init(self, init_type: str = IT_FIT_SEQ, enable_cache: bool = False):
"""
Initialize the data of Qlib
@@ -314,15 +329,17 @@ class DataHandlerLP(DataHandler):
# TODO: Be able to cache handler data. Save the memory for data processing
def _get_df_by_key(self, data_key: str=DK_I) -> pd.DataFrame:
df = getattr(self, {self.DK_R: '_data', self.DK_I: "_infer", self.DK_L: "_learn"}[data_key])
def _get_df_by_key(self, data_key: str = DK_I) -> pd.DataFrame:
df = getattr(self, {self.DK_R: "_data", self.DK_I: "_infer", self.DK_L: "_learn"}[data_key])
return df
def fetch(self,
selector: Union[pd.Timestamp, slice, str],
level: Union[str, int] = 'datetime',
col_set=DataHandler.CS_ALL,
data_key: str = DK_I) -> pd.DataFrame:
def fetch(
self,
selector: Union[pd.Timestamp, slice, str],
level: Union[str, int] = "datetime",
col_set=DataHandler.CS_ALL,
data_key: str = DK_I,
) -> pd.DataFrame:
"""
fetch data from underlying data source
@@ -345,7 +362,7 @@ class DataHandlerLP(DataHandler):
df = self._fetch_df_by_index(df, selector, level)
return self._fetch_df_by_col(df, col_set)
def get_cols(self, col_set=DataHandler.CS_ALL, data_key: str=DK_I) -> list:
def get_cols(self, col_set=DataHandler.CS_ALL, data_key: str = DK_I) -> list:
"""
get the column names

57
qlib/data/dataset/loader.py
View File

@@ -8,44 +8,46 @@ from typing import Tuple
class DataLoader(ABC):
'''
"""
DataLoader is designed for loading raw data from original data source.
'''
"""
@abstractmethod
def load(self, instruments, start_time=None, end_time=None) -> pd.DataFrame:
"""
load the data as pd.DataFrame
load the data as pd.DataFrame
Parameters
----------
self : [TODO:type]
[TODO:description]
instruments : [TODO:type]
[TODO:description]
start_time : [TODO:type]
[TODO:description]
end_time : [TODO:type]
[TODO:description]
Parameters
----------
self : [TODO:type]
[TODO:description]
instruments : [TODO:type]
[TODO:description]
start_time : [TODO:type]
[TODO:description]
end_time : [TODO:type]
[TODO:description]
Returns
-------
pd.DataFrame:
data load from the under layer source
Returns
-------
pd.DataFrame:
data load from the under layer source
Example of the data:
The multi-index of the columns is optional.
feature label
$close $volume Ref($close, 1) Mean($close, 3) $high-$low LABEL0
datetime instrument
2010-01-04 SH600000 81.807068 17145150.0 83.737389 83.016739 2.741058 0.0032
SH600004 13.313329 11800983.0 13.313329 13.317701 0.183632 0.0042
SH600005 37.796539 12231662.0 38.258602 37.919757 0.970325 0.0289
Example of the data:
The multi-index of the columns is optional.
feature label
$close $volume Ref($close, 1) Mean($close, 3) $high-$low LABEL0
datetime instrument
2010-01-04 SH600000 81.807068 17145150.0 83.737389 83.016739 2.741058 0.0032
SH600004 13.313329 11800983.0 13.313329 13.317701 0.183632 0.0042
SH600005 37.796539 12231662.0 38.258602 37.919757 0.970325 0.0289
"""
pass
class QlibDataLoader(DataLoader):
'''Same as QlibDataLoader. The fields can be define by config'''
"""Same as QlibDataLoader. The fields can be define by config"""
def __init__(self, config: Tuple[list, tuple, dict], filter_pipe=None):
"""
Parameters
@@ -65,7 +67,7 @@ class QlibDataLoader(DataLoader):
Here is a few examples to describe the fields
TODO:
"""
self.is_group = isinstance(config, dict)
self.is_group = isinstance(config, dict)
if self.is_group:
self.fields = {grp: self._parse_fields_info(fields_info) for grp, fields_info in config.items()}
@@ -88,6 +90,7 @@ class QlibDataLoader(DataLoader):
df = D.features(D.instruments(instruments, filter_pipe=self.filter_pipe), exprs, start_time, end_time)
df.columns = names
return df
if self.is_group:
df = pd.concat({grp: _get_df(exprs, names) for grp, (exprs, names) in self.fields.items()}, axis=1)
else:

22
qlib/data/dataset/processor.py
View File

@@ -30,8 +30,7 @@ def get_group_columns(df: pd.DataFrame, group: str):
class Processor(Serializable):
def fit(self, df: pd.DataFrame=None):
def fit(self, df: pd.DataFrame = None):
"""
learn data processing parameters
@@ -40,7 +39,7 @@ class Processor(Serializable):
df : pd.DataFrame
When we fit and process data with processor one by one. The fit function reiles on the output of previous
processor, i.e. `df`.
"""
pass
@@ -81,16 +80,17 @@ class DropnaProcessor(Processor):
class DropnaLabel(DropnaProcessor):
def __init__(self, group='label'):
def __init__(self, group="label"):
super().__init__(group=group)
def is_for_infer(self) -> bool:
'''The samples are dropped according to label. So it is not usable for inference'''
"""The samples are dropped according to label. So it is not usable for inference"""
return False
class ProcessInf(Processor):
'''Process infinity '''
"""Process infinity """
def __call__(self, df):
def replace_inf(data):
def process_inf(df):
@@ -102,6 +102,7 @@ class ProcessInf(Processor):
data = data.groupby("datetime").apply(process_inf)
data.sort_index(inplace=True)
return data
return replace_inf(df)
@@ -126,6 +127,7 @@ class MinMaxNorm(Processor):
if not ignore[i]:
x[i] = (x[i] - min_val) / (max_val - min_val)
return x
df.loc(axis=1)[self.cols] = normalize(df[self.cols].values)
return df
@@ -151,17 +153,19 @@ class ZscoreNorm(Processor):
if not ignore[i]:
x[i] = (x[i] - mean_train) / std_train
return x
df.loc(axis=1)[self.cols] = normalize(df[self.cols].values)
return df
class CSZScoreNorm(Processor):
'''Cross Sectional ZScore Normalization'''
"""Cross Sectional ZScore Normalization"""
def __init__(self, fields_group=None):
self.fields_group = fields_group
def __call__(self, df):
# try not modify original dataframe
cols = get_group_columns(df,self.fields_group)
df[cols] = df[cols].groupby('datetime').apply(lambda df: (df - df.mean()).div(df.std()))
cols = get_group_columns(df, self.fields_group)
df[cols] = df[cols].groupby("datetime").apply(lambda df: (df - df.mean()).div(df.std()))
return df

3
qlib/data/dataset/utils.py
View File

@@ -24,9 +24,8 @@ def get_level_index(df: pd.DataFrame, level=Union[str, int]) -> int:
return df.index.names.index(level)
except (AttributeError, ValueError):
# NOTE: If level index is not given in the data, the default level index will be ('datetime', 'instrument')
return ('datetime', 'instrument').index(level)
return ("datetime", "instrument").index(level)
elif isinstance(level, int):
return level
else:
raise NotImplementedError(f"This type of input is not supported")