因子表达式重构及quantstats替换：quantlab 1.3

今日工作：

1、因子表达式重构

2、整合ffn代码，去掉pyfolio， quantstats。

星球的合伙人计划，已有三名位同学入选，本周若超过10位，将组织第一次例会，探讨平台发展计划，答疑等。

最近做了一些调研，实盘对接做得好的几家，主要集中在期货量化。因为期货CTA本身就是合规，甚至是鼓励的。除了加密货币，最适合量化的领域是期货的CTA或者外汇的RA。

星球很多同学反馈，代码用起来有难度，其实是的，读别人的代码本身就有难度，何况叠加对量化、金融、人工智能的理解。

因此，后续我尽可能把易用性做出来，产品化、服务化。

系统代码已经发布至星球：

因子表达式重构的原因：

之前考虑到rank这个函数，需要截面排序，因此需要全体数据进行运算之后，才能rank，因此引入了group_by（‘date’），当然也引入了调试的复杂性。从帮助大家简单理解的角度，我决定去掉groupby。

Duckdbloader使用上没有变化：

在计算因子表达式时，单个dataframe循环计算：

def load(self, fields=None, names=None):
    dfs = self._load_dfs()
    # df = self._reset_index(df)

    if not fields or not names or len(fields) != len(names):
        return self._concat_dfs(dfs)
    else:
        # 截面rank需要
        dfs_expr = []
        for df in tqdm(dfs):
            for field, name in tqdm(zip(fields, names)):
                df[name] = calc_expr(df, field)
            dfs_expr.append(df)
        df = self._concat_dfs(dfs_expr)
        # todo 如果需要支持截面rank，逻辑加在这里
        return df

这里，表达式函数的实现就可以非常简单。

输入是pd.Series以及其它参数：

def shift(se: pd.Series, N):
    return se.shift(N)


def roc(se: pd.Series, N):
    return se / shift(se, N) - 1

rolling的函数实现也变得简单：

import numpy as np
import pandas as pd
from scipy.stats import percentileofscore


def rolling(se, N, func):
    ind = getattr(se.rolling(window=N), func)()
    return ind


def sum(se: pd.Series, N):
    return rolling(se, N, 'sum')


def max(se, N):
    return rolling(se, N, 'max')


def min(se, N):
    return rolling(se, N, 'min')


def std(se, N):
    return rolling(se, N, 'std')


def avg(se, N):
    se = rolling(se, N, 'mean')
    return se


def mean(se, N):
    return avg(se, N)


def idxmax(se, N):
    return rolling(se, N, 'argmax')


def idxmin(se, N):
    return rolling(se, N, 'argmin')


def ts_rank(se, N):
    return se.rolling(N).apply(lambda x: percentileofscore(x, x[-1]) / len(x))


def _zscore(se):
    return se - se.mean() / se.std()


def zscore(se: pd.Series, N):
    return se.rolling(window=N).apply(lambda x: _zscore(x))


def _slope(x):
    try:
        x = x / x.iloc[0]  # 这里做了一个“归一化”
        slope = np.polyfit(range(len(x)), x, 1)[0]
        return slope
    except:
        return -1


def slope(se, N):
    result = se.rolling(N, min_periods=2).apply(lambda x: slope(x))
    return result


def quantile(se, N, qscore):
    return se.rolling(N, min_periods=1).quantile(qscore)


def bias(se, N):
    return se.rolling(N).apply(lambda x: x / x.mean())


def _qcut(se: pd.Series, quantiles, N):
    if len(se) < 3:
        return
    return pd.qcut(se, quantiles, labels=range(0, N), duplicates='drop').astype('float')


def qcut(se: pd.Series, N):
    quantiles = [step / 100 for step in range(0, 100, int(100 / N))]
    if len(quantiles) <= N:
        quantiles.append(1)
    labels = se.rolling(N).apply(
        lambda sub_se: _qcut(sub_se, quantiles, N))
    return labels