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

black formatting

Browse Source
This commit is contained in:
王雪
2021-01-20 20:29:59 +08:00
committed by you-n-g
parent 5ad1b4cc33
commit 784e73bceb
2 changed files with 39 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
qlib/data/data.py
View File

@@ -59,9 +59,7 @@ class CalendarProvider(abc.ABC):
list list
calendar list calendar list
""" """
raise NotImplementedError( raise NotImplementedError("Subclass of CalendarProvider must implement `calendar` method")
"Subclass of CalendarProvider must implement `calendar` method"
)
def locate_index(self, start_time, end_time, freq, future): def locate_index(self, start_time, end_time, freq, future):
"""Locate the start time index and end time index in a calendar under certain frequency. """Locate the start time index and end time index in a calendar under certain frequency.
@@ -183,9 +181,7 @@ class InstrumentProvider(abc.ABC):
return config return config
@abc.abstractmethod @abc.abstractmethod
def list_instruments( def list_instruments(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
self, instruments, start_time=None, end_time=None, freq="day", as_list=False
):
"""List the instruments based on a certain stockpool config. """List the instruments based on a certain stockpool config.
Parameters Parameters
@@ -204,13 +200,9 @@ class InstrumentProvider(abc.ABC):
dict or list dict or list
instruments list or dictionary with time spans instruments list or dictionary with time spans
""" """
raise NotImplementedError( raise NotImplementedError("Subclass of InstrumentProvider must implement `list_instruments` method")
"Subclass of InstrumentProvider must implement `list_instruments` method"
)
def _uri( def _uri(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
self, instruments, start_time=None, end_time=None, freq="day", as_list=False
):
return hash_args(instruments, start_time, end_time, freq, as_list) return hash_args(instruments, start_time, end_time, freq, as_list)
# instruments type # instruments type
@@ -241,9 +233,7 @@ class InstrumentProvider(abc.ABC):
) )
_df_list.append(_df.iloc[:, [0, -1]]) _df_list.append(_df.iloc[:, [0, -1]])
df = pd.concat(_df_list, sort=False).sort_values("save_inst") df = pd.concat(_df_list, sort=False).sort_values("save_inst")
df = df.drop_duplicates(subset=["save_inst"], keep="first").fillna( df = df.drop_duplicates(subset=["save_inst"], keep="first").fillna(axis=1, method="ffill")
axis=1, method="ffill"
)
_instruments_map = df.set_index("inst").iloc[:, 0].to_dict() _instruments_map = df.set_index("inst").iloc[:, 0].to_dict()
setattr(self, "_instruments_map", _instruments_map) setattr(self, "_instruments_map", _instruments_map)
return _instruments_map.get(instrument, instrument) return _instruments_map.get(instrument, instrument)
@@ -277,9 +267,7 @@ class FeatureProvider(abc.ABC):
pd.Series pd.Series
data of a certain feature data of a certain feature
""" """
raise NotImplementedError( raise NotImplementedError("Subclass of FeatureProvider must implement `feature` method")
"Subclass of FeatureProvider must implement `feature` method"
)
class ExpressionProvider(abc.ABC): class ExpressionProvider(abc.ABC):
@@ -300,14 +288,11 @@ class ExpressionProvider(abc.ABC):
self.expression_instance_cache[field] = expression self.expression_instance_cache[field] = expression
except NameError as e: except NameError as e:
get_module_logger("data").exception( get_module_logger("data").exception(
"ERROR: field [%s] contains invalid operator/variable [%s]" "ERROR: field [%s] contains invalid operator/variable [%s]" % (str(field), str(e).split()[1])
% (str(field), str(e).split()[1])
) )
raise raise
except SyntaxError: except SyntaxError:
get_module_logger("data").exception( get_module_logger("data").exception("ERROR: field [%s] contains invalid syntax" % str(field))
"ERROR: field [%s] contains invalid syntax" % str(field)
)
raise raise
return expression return expression
@@ -333,9 +318,7 @@ class ExpressionProvider(abc.ABC):
pd.Series pd.Series
data of a certain expression data of a certain expression
""" """
raise NotImplementedError( raise NotImplementedError("Subclass of ExpressionProvider must implement `Expression` method")
"Subclass of ExpressionProvider must implement `Expression` method"
)
class DatasetProvider(abc.ABC): class DatasetProvider(abc.ABC):
@@ -366,9 +349,7 @@ class DatasetProvider(abc.ABC):
pd.DataFrame pd.DataFrame
a pandas dataframe with <instrument, datetime> index. a pandas dataframe with <instrument, datetime> index.
""" """
raise NotImplementedError( raise NotImplementedError("Subclass of DatasetProvider must implement `Dataset` method")
"Subclass of DatasetProvider must implement `Dataset` method"
)
def _uri( def _uri(
self, self,
@@ -398,9 +379,7 @@ class DatasetProvider(abc.ABC):
whether to skip(0)/use(1)/replace(2) disk_cache. whether to skip(0)/use(1)/replace(2) disk_cache.
""" """
return DiskDatasetCache._uri( return DiskDatasetCache._uri(instruments, fields, start_time, end_time, freq, disk_cache)
instruments, fields, start_time, end_time, freq, disk_cache
)
@staticmethod @staticmethod
def get_instruments_d(instruments, freq): def get_instruments_d(instruments, freq):
@@ -412,9 +391,7 @@ class DatasetProvider(abc.ABC):
if isinstance(instruments, dict): if isinstance(instruments, dict):
if "market" in instruments: if "market" in instruments:
# dict of stockpool config # dict of stockpool config
instruments_d = Inst.list_instruments( instruments_d = Inst.list_instruments(instruments=instruments, freq=freq, as_list=False)
instruments=instruments, freq=freq, as_list=False
)
else: else:
# dict of instruments and timestamp # dict of instruments and timestamp
instruments_d = instruments instruments_d = instruments
@@ -504,9 +481,7 @@ class DatasetProvider(abc.ABC):
return data return data
@staticmethod @staticmethod
def expression_calculator( def expression_calculator(inst, start_time, end_time, freq, column_names, spans=None, g_config=None):
inst, start_time, end_time, freq, column_names, spans=None, g_config=None
):
""" """
Calculate the expressions for one instrument, return a df result. Calculate the expressions for one instrument, return a df result.
If the expression has been calculated before, load from cache. If the expression has been calculated before, load from cache.
@@ -571,9 +546,7 @@ class LocalCalendarProvider(CalendarProvider):
fname = self._uri_cal.format(freq + "_future") fname = self._uri_cal.format(freq + "_future")
# if future calendar not exists, return current calendar # if future calendar not exists, return current calendar
if not os.path.exists(fname): if not os.path.exists(fname):
get_module_logger("data").warning( get_module_logger("data").warning(f"{freq}_future.txt not exists, return current calendar!")
f"{freq}_future.txt not exists, return current calendar!"
)
fname = self._uri_cal.format(freq) fname = self._uri_cal.format(freq)
else: else:
fname = self._uri_cal.format(freq) fname = self._uri_cal.format(freq)
@@ -635,9 +608,7 @@ class LocalInstrumentProvider(InstrumentProvider):
_instruments.setdefault(row[0], []).append((row[1], row[2])) _instruments.setdefault(row[0], []).append((row[1], row[2]))
return _instruments return _instruments
def list_instruments( def list_instruments(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
self, instruments, start_time=None, end_time=None, freq="day", as_list=False
):
market = instruments["market"] market = instruments["market"]
if market in H["i"]: if market in H["i"]:
_instruments = H["i"][market] _instruments = H["i"][market]
@@ -658,20 +629,14 @@ class LocalInstrumentProvider(InstrumentProvider):
) )
for inst, spans in _instruments.items() for inst, spans in _instruments.items()
} }
_instruments_filtered = { _instruments_filtered = {key: value for key, value in _instruments_filtered.items() if value}
key: value for key, value in _instruments_filtered.items() if value
}
# filter # filter
filter_pipe = instruments["filter_pipe"] filter_pipe = instruments["filter_pipe"]
for filter_config in filter_pipe: for filter_config in filter_pipe:
from . import filter as F from . import filter as F
filter_t = getattr(F, filter_config["filter_type"]).from_config( filter_t = getattr(F, filter_config["filter_type"]).from_config(filter_config)
filter_config _instruments_filtered = filter_t(_instruments_filtered, start_time, end_time, freq)
)
_instruments_filtered = filter_t(
_instruments_filtered, start_time, end_time, freq
)
# as list # as list
if as_list: if as_list:
return list(_instruments_filtered) return list(_instruments_filtered)
@@ -698,9 +663,7 @@ class LocalFeatureProvider(FeatureProvider):
instrument = Inst.convert_instruments(instrument) instrument = Inst.convert_instruments(instrument)
uri_data = self._uri_data.format(instrument.lower(), field, freq) uri_data = self._uri_data.format(instrument.lower(), field, freq)
if not os.path.exists(uri_data): if not os.path.exists(uri_data):
get_module_logger("data").warning( get_module_logger("data").warning("WARN: data not found for %s.%s" % (instrument, field))
"WARN: data not found for %s.%s" % (instrument, field)
)
return pd.Series(dtype=np.float32) return pd.Series(dtype=np.float32)
# raise ValueError('uri_data not found: ' + uri_data) # raise ValueError('uri_data not found: ' + uri_data)
# load # load
@@ -721,13 +684,9 @@ class LocalExpressionProvider(ExpressionProvider):
expression = self.get_expression_instance(field) expression = self.get_expression_instance(field)
start_time = pd.Timestamp(start_time) start_time = pd.Timestamp(start_time)
end_time = pd.Timestamp(end_time) end_time = pd.Timestamp(end_time)
_, _, start_index, end_index = Cal.locate_index( _, _, start_index, end_index = Cal.locate_index(start_time, end_time, freq, future=False)
start_time, end_time, freq, future=False
)
lft_etd, rght_etd = expression.get_extended_window_size() lft_etd, rght_etd = expression.get_extended_window_size()
series = expression.load( series = expression.load(instrument, max(0, start_index - lft_etd), end_index + rght_etd, freq)
instrument, max(0, start_index - lft_etd), end_index + rght_etd, freq
)
# Ensure that each column type is consistent # Ensure that each column type is consistent
# FIXME: # FIXME:
# 1) The stock data is currently float. If there is other types of data, this part needs to be re-implemented. # 1) The stock data is currently float. If there is other types of data, this part needs to be re-implemented.
@@ -759,16 +718,12 @@ class LocalDatasetProvider(DatasetProvider):
start_time = cal[0] start_time = cal[0]
end_time = cal[-1] end_time = cal[-1]
data = self.dataset_processor( data = self.dataset_processor(instruments_d, column_names, start_time, end_time, freq)
instruments_d, column_names, start_time, end_time, freq
)
return data return data
@staticmethod @staticmethod
def multi_cache_walker( def multi_cache_walker(instruments, fields, start_time=None, end_time=None, freq="day"):
instruments, fields, start_time=None, end_time=None, freq="day"
):
""" """
This method is used to prepare the expression cache for the client. This method is used to prepare the expression cache for the client.
Then the client will load the data from expression cache by itself. Then the client will load the data from expression cache by itself.
@@ -836,9 +791,7 @@ class ClientCalendarProvider(CalendarProvider):
"future": future, "future": future,
}, },
msg_queue=self.queue, msg_queue=self.queue,
msg_proc_func=lambda response_content: [ msg_proc_func=lambda response_content: [pd.Timestamp(c) for c in response_content],
pd.Timestamp(c) for c in response_content
],
) )
result = self.queue.get(timeout=C["timeout"]) result = self.queue.get(timeout=C["timeout"])
return result return result
@@ -857,14 +810,11 @@ class ClientInstrumentProvider(InstrumentProvider):
def set_conn(self, conn): def set_conn(self, conn):
self.conn = conn self.conn = conn
def list_instruments( def list_instruments(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
self, instruments, start_time=None, end_time=None, freq="day", as_list=False
):
def inst_msg_proc_func(response_content): def inst_msg_proc_func(response_content):
if isinstance(response_content, dict): if isinstance(response_content, dict):
instrument = { instrument = {
i: [(pd.Timestamp(s), pd.Timestamp(e)) for s, e in t] i: [(pd.Timestamp(s), pd.Timestamp(e)) for s, e in t] for i, t in response_content.items()
for i, t in response_content.items()
} }
else: else:
instrument = response_content instrument = response_content
@@ -950,9 +900,7 @@ class ClientDatasetProvider(DatasetProvider):
start_time = cal[0] start_time = cal[0]
end_time = cal[-1] end_time = cal[-1]
data = self.dataset_processor( data = self.dataset_processor(instruments_d, column_names, start_time, end_time, freq)
instruments_d, column_names, start_time, end_time, freq
)
if return_uri: if return_uri:
return data, feature_uri return data, feature_uri
else: else:
@@ -984,12 +932,8 @@ class ClientDatasetProvider(DatasetProvider):
get_module_logger("data").debug("get result") get_module_logger("data").debug("get result")
try: try:
# pre-mound nfs, used for demo # pre-mound nfs, used for demo
mnt_feature_uri = os.path.join( mnt_feature_uri = os.path.join(C.get_data_path(), C.dataset_cache_dir_name, feature_uri)
C.get_data_path(), C.dataset_cache_dir_name, feature_uri df = DiskDatasetCache.read_data_from_cache(mnt_feature_uri, start_time, end_time, fields)
)
df = DiskDatasetCache.read_data_from_cache(
mnt_feature_uri, start_time, end_time, fields
)
get_module_logger("data").debug("finish slicing data") get_module_logger("data").debug("finish slicing data")
if return_uri: if return_uri:
return df, feature_uri return df, feature_uri
@@ -1007,9 +951,7 @@ class BaseProvider:
def calendar(self, start_time=None, end_time=None, freq="day", future=False): def calendar(self, start_time=None, end_time=None, freq="day", future=False):
return Cal.calendar(start_time, end_time, freq, future=future) return Cal.calendar(start_time, end_time, freq, future=future)
def instruments( def instruments(self, market="all", filter_pipe=None, start_time=None, end_time=None):
self, market="all", filter_pipe=None, start_time=None, end_time=None
):
if start_time is not None or end_time is not None: if start_time is not None or end_time is not None:
get_module_logger("Provider").warning( get_module_logger("Provider").warning(
"The instruments corresponds to a stock pool. " "The instruments corresponds to a stock pool. "
@@ -1017,9 +959,7 @@ class BaseProvider:
) )
return InstrumentProvider.instruments(market, filter_pipe) return InstrumentProvider.instruments(market, filter_pipe)
def list_instruments( def list_instruments(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
self, instruments, start_time=None, end_time=None, freq="day", as_list=False
):
return Inst.list_instruments(instruments, start_time, end_time, freq, as_list) return Inst.list_instruments(instruments, start_time, end_time, freq, as_list)
def features( def features(
@@ -1045,9 +985,7 @@ class BaseProvider:
if C.disable_disk_cache: if C.disable_disk_cache:
disk_cache = False disk_cache = False
try: try:
return DatasetD.dataset( return DatasetD.dataset(instruments, fields, start_time, end_time, freq, disk_cache)
instruments, fields, start_time, end_time, freq, disk_cache
)
except TypeError: except TypeError:
return DatasetD.dataset(instruments, fields, start_time, end_time, freq) return DatasetD.dataset(instruments, fields, start_time, end_time, freq)
@@ -1068,9 +1006,7 @@ class LocalProvider(BaseProvider):
elif type == "feature": elif type == "feature":
return DatasetD._uri(**kwargs) return DatasetD._uri(**kwargs)
def features_uri( def features_uri(self, instruments, fields, start_time, end_time, freq, disk_cache=1):
self, instruments, fields, start_time, end_time, freq, disk_cache=1
):
"""features_uri """features_uri
Return the uri of the generated cache of features/dataset Return the uri of the generated cache of features/dataset
@@ -1082,9 +1018,7 @@ class LocalProvider(BaseProvider):
:param end_time: :param end_time:
:param freq: :param freq:
""" """
return DatasetD._dataset_uri( return DatasetD._dataset_uri(instruments, fields, start_time, end_time, freq, disk_cache)
instruments, fields, start_time, end_time, freq, disk_cache
)
class ClientProvider(BaseProvider): class ClientProvider(BaseProvider):
@@ -1148,9 +1082,7 @@ def register_all_wrappers():
_calendar_provider = init_instance_by_config(C.calendar_provider, module) _calendar_provider = init_instance_by_config(C.calendar_provider, module)
if getattr(C, "calendar_cache", None) is not None: if getattr(C, "calendar_cache", None) is not None:
_calendar_provider = init_instance_by_config( _calendar_provider = init_instance_by_config(C.calendar_cache, module, provide=_calendar_provider)
C.calendar_cache, module, provide=_calendar_provider
)
register_wrapper(Cal, _calendar_provider, "qlib.data") register_wrapper(Cal, _calendar_provider, "qlib.data")
logger.debug(f"registering Cal {C.calendar_provider}-{C.calendar_cache}") logger.debug(f"registering Cal {C.calendar_provider}-{C.calendar_cache}")
@@ -1166,19 +1098,13 @@ def register_all_wrappers():
# This provider is unnecessary in client provider # This provider is unnecessary in client provider
_eprovider = init_instance_by_config(C.expression_provider, module) _eprovider = init_instance_by_config(C.expression_provider, module)
if getattr(C, "expression_cache", None) is not None: if getattr(C, "expression_cache", None) is not None:
_eprovider = init_instance_by_config( _eprovider = init_instance_by_config(C.expression_cache, module, provider=_eprovider)
C.expression_cache, module, provider=_eprovider
)
register_wrapper(ExpressionD, _eprovider, "qlib.data") register_wrapper(ExpressionD, _eprovider, "qlib.data")
logger.debug( logger.debug(f"registering ExpressioneD {C.expression_provider}-{C.expression_cache}")
f"registering ExpressioneD {C.expression_provider}-{C.expression_cache}"
)
_dprovider = init_instance_by_config(C.dataset_provider, module) _dprovider = init_instance_by_config(C.dataset_provider, module)
if getattr(C, "dataset_cache", None) is not None: if getattr(C, "dataset_cache", None) is not None:
_dprovider = init_instance_by_config( _dprovider = init_instance_by_config(C.dataset_cache, module, provider=_dprovider)
C.dataset_cache, module, provider=_dprovider
)
register_wrapper(DatasetD, _dprovider, "qlib.data") register_wrapper(DatasetD, _dprovider, "qlib.data")
logger.debug(f"registering DataseteD {C.dataset_provider}-{C.dataset_cache}") logger.debug(f"registering DataseteD {C.dataset_provider}-{C.dataset_cache}")

4
qlib/workflow/__init__.py
View File

@@ -38,9 +38,7 @@ class QlibRecorder:
try: try:
yield run yield run
except Exception as e: except Exception as e:
self.end_exp( self.end_exp(Recorder.STATUS_FA) # end the experiment if something went wrong
Recorder.STATUS_FA
) # end the experiment if something went wrong
raise e raise e
self.end_exp(Recorder.STATUS_FI) self.end_exp(Recorder.STATUS_FI)