Method _expression

qlib/data/cache.py:506–563

(self, instrument, field, start_time=None, end_time=None, freq="day")

        return hash_args(instrument, field, freq)

    def _expression(self, instrument, field, start_time=None, end_time=None, freq="day"):
        _cache_uri = self._uri(instrument=instrument, field=field, start_time=None, end_time=None, freq=freq)
        _instrument_dir = self.get_cache_dir(freq).joinpath(instrument.lower())
        cache_path = _instrument_dir.joinpath(_cache_uri)
        # get calendar
        from .data import Cal  # pylint: disable=C0415

        _calendar = Cal.calendar(freq=freq)

        _, _, start_index, end_index = Cal.locate_index(start_time, end_time, freq, future=False)

        if self.check_cache_exists(cache_path, suffix_list=[".meta"]):
            """
            In most cases, we do not need reader_lock.
            Because updating data is a small probability event compare to reading data.

            """
            # FIXME: Removing the reader lock may result in conflicts.
            # with CacheUtils.reader_lock(self.r, 'expression-%s' % _cache_uri):

            # modify expression cache meta file
            try:
                # FIXME: Multiple readers may result in error visit number
                if not self.remote:
                    CacheUtils.visit(cache_path)
                series = read_bin(cache_path, start_index, end_index)
                return series
            except Exception:
                series = None
                self.logger.error("reading %s file error : %s" % (cache_path, traceback.format_exc()))
            return series
        else:
            # normalize field
            field = remove_fields_space(field)
            # cache unavailable, generate the cache
            _instrument_dir.mkdir(parents=True, exist_ok=True)
            if not isinstance(eval(parse_field(field)), Feature):
                # When the expression is not a raw feature
                # generate expression cache if the feature is not a Feature
                # instance
                series = self.provider.expression(instrument, field, _calendar[0], _calendar[-1], freq)
                if not series.empty:
                    # This expression is empty, we don't generate any cache for it.
                    with CacheUtils.writer_lock(self.r, f"{str(C.dpm.get_data_uri(freq))}:expression-{_cache_uri}"):
                        self.gen_expression_cache(
                            expression_data=series,
                            cache_path=cache_path,
                            instrument=instrument,
                            field=field,
                            freq=freq,
                            last_update=str(_calendar[-1]),
                        )
                    return series.loc[start_index:end_index]
                else:
                    return series
            else:
                # If the expression is a raw feature(such as $close, $open)
                return self.provider.expression(instrument, field, start_time, end_time, freq)