因子挖掘之表达式引擎构建：二元滑动计算函数探讨。

01 因子挖掘之因子表达式

因子表达式无外乎四类：

一元直接计算：比如abs, log;

二元直接计算的：比如+，-，*，/比大小之类的;

一元rolling：比如ts_max, delta。

还有就是二元滑动：比如rolling_corr， rolling_std。

最后一种比较麻烦，Pandas原生不支持，或者说性能不好。

gip unary_ops = [Abs, Log]
binary_ops = [Add, Sub, Mul, Div, Greater, Less]
rolling_ops = [Ref, Mean, Sum, Std, Var, Max, Min, Med, Mad, Delta, WMA, EMA]
rolling_binary_ops = [Cov, Corr]

就以最后一种为例：

def rolling_binary(cls, day):
    def _calc(a, b):
        n = len(a)
        return np.array([f'{cls.__name__}({a[i]},{b[i]},{day})' for i in range(n)])
    return _calc

这里的计算函数，按最后一个维度展开，其实类似group_by_symbol。

因为数据是三维叠加的。

昨天在群里和星友们讨论的问题——qlib支持不支持截面，但有同学说DeepAlpha是支持的呀，应该它把所有数据搁到一起了。——不好意思，这种方式，仍然不支持截面，比如rank(a)

class RollingOperator(Operator):
    def __init__(self, operand: Union[Expression, float], delta_time: Union[int, DeltaTime]) -> None:
        self._operand = operand if isinstance(operand, Expression) else Constant(operand)
        if isinstance(delta_time, DeltaTime):
            delta_time = delta_time._delta_time
        self._delta_time = delta_time

    @classmethod
    def n_args(cls) -> int: return 2

    @classmethod
    def category_type(cls) -> Type['Operator']: return RollingOperator

    def evaluate(self, data: StockData, period: slice = slice(0, 1)) -> Tensor:
        start = period.start - self._delta_time + 1
        stop = period.stop
        # L: period length (requested time window length)
        # W: window length (dt for rolling)
        # S: stock count
        values = self._operand.evaluate(data, slice(start, stop))   # (L+W-1, S)
        values = values.unfold(0, self._delta_time, 1)              # (L, S, W)
        return self._apply(values)

用三维矩阵来存储数据，不直观还容易出错。

因此，咱们的因子表达式使用group by symbol和 group by symbol统一的范式，清晰简单。

WorldQuant101的3号因子，只要使用rank(open)和rank(volume)求相关系数，结果都是None。

我检查过，如果计算correlation(open,volume)是没有问题的，说明计算函数correlation没有问题——这里怀疑是rank后数值太相近？？。

# Alpha#3: (-1 * correlation(rank(open), rank(volume), 10))
se = calc_expr(df, expr='(-1 * correlation(rank(open), rank(volume), 10))')
df['alpha_03'] = se

另外一个点，

series_left.rolling(10).corr(right)可以实现两个序列的滑动相关计算，但这里不支持自定义函数，比如计算rsrs， numpy的内置函数是可以的（这里有认知的同学，欢迎留言探讨）。

@calc_by_symbol
def correlation(left: pd.Series, right: pd.Series, periods=20):
    #es: pd.Series = pd.Series(index=left.index)

  res = left.rolling(window=periods).corr(right)
    #left.rolling(window=periods).apply(func=func,right)
    res.loc[
        np.isclose(left.rolling(periods, min_periods=1).std(), 0, atol=2e-05)
        | np.isclose(right.rolling(periods, min_periods=1).std(), 0, atol=2e-05)
        ] = np.nan
    return res