add period ops class

microsoft · you-n-g · Mar 10, 2022 · Mar 9, 2021 · Mar 9, 2021 · Mar 10, 2021
commit 3e92169f859a06b26683be24178c42dfb21efaae
diff --git a/qlib/config.py b/qlib/config.py
@@ -189,6 +189,7 @@ def set_conf_from_C(self, config_c):
         "region": REG_CN,
         ## Custom Operator
         "custom_ops": [],
+        "custom_period_ops": [],
     },
 }
 
@@ -297,11 +298,13 @@ def set(self, default_conf="client", **kwargs):
     def register(self):
         from .utils import init_instance_by_config
         from .data.ops import register_all_ops
+        from .data.ops_period import register_all_period_ops
         from .data.data import register_all_wrappers
         from .workflow import R, QlibRecorder
         from .workflow.utils import experiment_exit_handler
 
         register_all_ops(self)
+        register_all_period_ops(self)
         register_all_wrappers(self)
         # set up QlibRecorder
         exp_manager = init_instance_by_config(self["exp_manager"])

diff --git a/qlib/data/base.py b/qlib/data/base.py
@@ -9,6 +9,7 @@
 import pandas as pd
 
 
+
 class Expression(abc.ABC):
     """Expression base class"""
 
@@ -224,3 +225,208 @@ class ExpressionOps(Expression):
     """
 
     pass
+
+
+class PExpression(abc.ABC):
+    """PExpression base class"""
+
+    def __str__(self):
+        return type(self).__name__
+
+    def __repr__(self):
+        return str(self)
+
+    def __gt__(self, other):
+        from .ops import PGt
+
+        return PGt(self, other)
+
+    def __ge__(self, other):
+        from .ops import PGe
+
+        return PGe(self, other)
+
+    def __lt__(self, other):
+        from .ops import PLt
+
+        return PLt(self, other)
+
+    def __le__(self, other):
+        from .ops import PLe
+
+        return PLe(self, other)
+
+    def __eq__(self, other):
+        from .ops import PEq
+
+        return PEq(self, other)
+
+    def __ne__(self, other):
+        from .ops import PNe
+
+        return PNe(self, other)
+
+    def __add__(self, other):
+        from .ops import PAdd
+
+        return PAdd(self, other)
+
+    def __radd__(self, other):
+        from .ops import PAdd
+
+        return PAdd(other, self)
+
+    def __sub__(self, other):
+        from .ops import PSub
+
+        return PSub(self, other)
+
+    def __rsub__(self, other):
+        from .ops import PSub
+
+        return PSub(other, self)
+
+    def __mul__(self, other):
+        from .ops import PMul
+
+        return PMul(self, other)
+
+    def __rmul__(self, other):
+        from .ops import PMul
+
+        return PMul(self, other)
+
+    def __div__(self, other):
+        from .ops import PDiv
+
+        return PDiv(self, other)
+
+    def __rdiv__(self, other):
+        from .ops import PDiv
+
+        return PDiv(other, self)
+
+    def __truediv__(self, other):
+        from .ops import PDiv
+
+        return PDiv(self, other)
+
+    def __rtruediv__(self, other):
+        from .ops import PDiv
+
+        return PDiv(other, self)
+
+    def __pow__(self, other):
+        from .ops import PPower
+
+        return PPower(self, other)
+
+    def __and__(self, other):
+        from .ops import PAnd
+
+        return PAnd(self, other)
+
+    def __rand__(self, other):
+        from .ops import PAnd
+
+        return PAnd(other, self)
+
+    def __or__(self, other):
+        from .ops import POr
+
+        return POr(self, other)
+
+    def __ror__(self, other):
+        from .ops import POr
+
+        return POr(other, self)
+
+
+    @abc.abstractmethod
+    def load_period_data(self, instrument, start_offset, end_offset, cur_index):
+        raise NotImplementedError("This function must be implemented in your newly defined feature")
+
+    @abc.abstractmethod
+    def get_period_offset(self, cur_index):
+        raise NotImplementedError("This function must be implemented in your newly defined feature")
+
+    def load(self, instrument, start_index, end_index, freq):
+        from .cache import H
+
+        # cache
+        args = str(self), instrument, start_index, end_index, freq
+        if args in H["f"]:
+            return H["f"][args]
+        if start_index is None or end_index is None or start_index > end_index:
+            raise ValueError("Invalid index range: {} {}".format(start_index, end_index))
+
+        resample_series = pd.Series(index=pd.RangeIndex(start_index, end_index + 1), dtype='float32', name=str(self))
+        for cur_index in range(start_index, end_index + 1):
+            start_offset, end_offset = self.get_period_offset(cur_index)
+            resample_data[cur_index] = self.load_period_data(instrument, start_offset, end_offset, cur_index).iloc[-1]
+
+        H["f"][args] = resample_series
+        return resample_data
+
+    def get_longest_back_rolling(self):
+        return 0
+
+    def get_extended_window_size(self):
+        return 0, 0
+
+
+class PFeature(PExpression):
+
+    def __init__(self, name=None):
+        if name:
+            self._name = name.lower()
+        else:
+            self._name = type(self).__name__.lower()
+
+    def __str__(self):
+        return "$" + self._name
+
+    def load_period_data(self, instrument, start_offset, end_offset, cur_index):
+        ### Zhou Code
+        return pd.Series([1,2,3])
+
+    def get_period_offset(self, cur_index):
+        return 0
+
+
+class PExpressionOps(PExpression):
+    """Operator Expression
+
+    This kind of feature will use operator for feature
+    construction on the fly.
+    """
+
+    pass
+
+class OpsWrapper:
+    """Ops Wrapper"""
+
+    def __init__(self):
+        self._ops = {}
+
+    def reset(self):
+        self._ops = {}
+
+    def register(self, ops_list):
+        for operator in ops_list:
+            if not issubclass(operator, ExpressionOps) and not issubclass(operator, PExpressionOps):
+                raise TypeError("operator must be subclass of ExpressionOps or PExpressionOps, not {}".format(operator))
+
+            if operator.__name__ in self._ops:
+                get_module_logger(self.__class__.__name__).warning(
+                    "The custom operator [{}] will override the qlib default definition".format(operator.__name__)
+                )
+            self._ops[operator.__name__] = operator
+
+    def __getattr__(self, key):
+        if key not in self._ops:
+            raise AttributeError("The operator [{0}] is not registered".format(key))
+        return self._ops[key]
+
+
+Operators = OpsWrapper()
diff --git a/qlib/data/cache.py b/qlib/data/cache.py
@@ -31,9 +31,7 @@
 )
 
 from ..log import get_module_logger
-from .base import Feature
-
-from .ops import Operators
+from .base import Feature, PFeature, Operators
 
 
 class QlibCacheException(RuntimeError):

diff --git a/qlib/data/data.py b/qlib/data/data.py
@@ -19,10 +19,9 @@
 
 from .cache import H
 from ..config import C
-from .ops import Operators
 from ..log import get_module_logger
 from ..utils import parse_field, read_bin, hash_args, normalize_cache_fields, code_to_fname
-from .base import Feature
+from .base import Feature, PFeature, Operators
 from .cache import DiskDatasetCache, DiskExpressionCache
 from ..utils import Wrapper, init_instance_by_config, register_wrapper, get_module_by_module_path
 
@@ -480,7 +479,6 @@ def expression_calculator(inst, start_time, end_time, freq, column_names, spans=
         _calendar = Cal.calendar(freq=freq)
         data.index = _calendar[data.index.values.astype(int)]
         data.index.names = ["datetime"]
-
         if spans is None:
             return data
         else:

diff --git a/qlib/data/ops.py b/qlib/data/ops.py
@@ -74,7 +74,6 @@ class NpElemOperator(ElemOperator):
     """
 
     def __init__(self, feature, func):
-        self.feature = feature
         self.func = func
         super(NpElemOperator, self).__init__(feature)
 
@@ -289,8 +288,6 @@ class NpPairOperator(PairOperator):
     """
 
     def __init__(self, feature_left, feature_right, func):
-        self.feature_left = feature_left
-        self.feature_right = feature_right
         self.func = func
         super(NpPairOperator, self).__init__(feature_left, feature_right)
 
@@ -1489,39 +1486,12 @@ def __init__(self, feature_left, feature_right, N):
 ]
 
 
-class OpsWrapper:
-    """Ops Wrapper"""
-
-    def __init__(self):
-        self._ops = {}
-
-    def reset(self):
-        self._ops = {}
-
-    def register(self, ops_list):
-        for operator in ops_list:
-            if not issubclass(operator, ExpressionOps):
-                raise TypeError("operator must be subclass of ExpressionOps, not {}".format(operator))
-
-            if operator.__name__ in self._ops:
-                get_module_logger(self.__class__.__name__).warning(
-                    "The custom operator [{}] will override the qlib default definition".format(operator.__name__)
-                )
-            self._ops[operator.__name__] = operator
-
-    def __getattr__(self, key):
-        if key not in self._ops:
-            raise AttributeError("The operator [{0}] is not registered".format(key))
-        return self._ops[key]
-
-
-Operators = OpsWrapper()
-
-
 def register_all_ops(C):
     """register all operator"""
     logger = get_module_logger("ops")
 
+    from .base import Operators
+
     Operators.reset()
     Operators.register(OpsList)