因子挖掘”流水线“的新架构（代码+数据）更新

星球下一步的计划：针对ETF，股票，期货建立因子挖掘流水线。

今天的两件事情：

1、补充gplearn函数集，向world quant 101对齐，同时争取与咱们因子表达式复用。

2、发布3.1版本。

使用我们准备好的脚本，下载好螺纹钢的主连合编数据，后续咱们也会准备分钟级数据等，但对于咱们的因子挖掘没有本质的影响：

symbol = 'RB0'
df = ak.futures_main_sina(symbol="V0", start_date="19900101", end_date=datetime.now().strftime('%Y%m%d'))

df.rename(columns={'日期': 'date', '开盘价': 'open', '最高价': 'high', '最低价': 'low', '收盘价': 'close', '成交量': 'volume',
                   '持仓量': 'open_interest', '动态结算价': 'vwap'}, inplace=True)
print(df['date'])
df['date'] = df['date'].apply(lambda x: str(x).replace('-', ''))
from config import DATA_DIR

df.to_csv(DATA_DIR.joinpath('futures').joinpath(symbol + '.csv'), index=None)
print(df)

导入gpquant的代码，实现在适合咱们因子挖掘的23个基础函数（四则运算，正余弦等）和23个时间序列函数（偏度，峰度，相关性以及常见的技术指标）：

function_map = {
    "square": square1,
    "sqrt": sqrt1,
    "cube": cube1,
    "cbrt": cbrt1,
    "sign": sign1,
    "neg": neg1,
    "inv": inv1,
    "abs": abs1,
    "sin": sin1,
    "cos": cos1,
    "tan": tan1,
    "log": log1,
    "sig": sig1,
    "add": add2,
    "sub": sub2,
    "mul": mul2,
    "div": div2,
    "max": max2,
    "min": min2,
    "mean": mean2,
    "clear by cond": clear_by_cond3,
    "if then else": if_then_else3,
    "if cond then else": if_cond_then_else4,
    "ts delay": ts_delay2,
    "ts delta": ts_delta2,
    "ts pct change": ts_pct_change2,
    "ts mean return": ts_mean_return2,
    "ts max": ts_max2,
    "ts min": ts_min2,
    "ts sum": ts_sum2,
    "ts product": ts_product2,
    "ts mean": ts_mean2,
    "ts std": ts_std2,
    "ts median": ts_median2,
    "ts midpoint": ts_midpoint2,
    "ts skew": ts_skew2,
    "ts kurt": ts_kurt2,
    "ts inverse cv": ts_inverse_cv2,
    "ts cov": ts_cov3,
    "ts corr": ts_corr3,
    "ts autocorr": ts_autocorr3,
    "ts maxmin": ts_maxmin2,
    "ts zscore": ts_zscore2,
    "ts regression beta": ts_regression_beta3,
    "ts linear slope": ts_linear_slope2,
    "ts linear intercept": ts_linear_intercept2,
    "ts argmax": ts_argmax2,
    "ts argmin": ts_argmin2,
    "ts argmaxmin": ts_argmaxmin2,
    "ts rank": ts_rank2,
    "ts ema": ts_ema2,
    "ts dema": ts_dema2,
    "ts kama": ts_kama4,
    "ts AROONOSC": ts_AROONOSC3,
    "ts WR": ts_WR4,
    "ts CCI": ts_CCI4,
    "ts ATR": ts_ATR4,
    "ts NATR": ts_NATR4,
    "ts ADX": ts_ADX4,
    "ts MFI": ts_MFI5,
}

另外，gplearn要用于因子挖掘，还有一个重要的扩展，就是fitness适应度（也就是我们的优化目标，可以是年化收益，夏普比等，大家下载代码后，也可以自行扩展）：

fitness_map = {
    "annual return": ann_return,
    "sharpe ratio": sharpe_ratio,
    "mean absolute error": mean_absolute_error,
    "mean square error": mean_square_error,
    "direction accuracy": direction_accuracy,
}

gplearn的代码：

import numpy as np
import pandas as pd
from gplearn import fitness
# 数据集
from gplearn.genetic import SymbolicRegressor

train_data = pd.read_csv('../data/IC_train.csv', index_col=0, parse_dates=[0])
test_data = pd.read_csv('../data/IC_test.csv', index_col=0, parse_dates=[0])
feature_names = list(train_data.columns)
train_data.loc[:, 'y'] = np.log(train_data['Close'].shift(-4) / train_data['Close'])
train_data.dropna(inplace=True)

from examples.backtest import BackTester

class SymbolicTestor(BackTester):  # 回测的设定
    def init(self):
        self.params = {'factor': pd.Series}

    @BackTester.process_strategy
    def run_(self, *args, **kwargs) -> dict[str: int]:
        factor = np.array(self.params['factor'])
        long_cond = factor > 0
        short_cond = factor < 0
        self.backtest_env['signal'] = np.where(long_cond, 1, np.where(short_cond, -1, np.nan))
        self.construct_position_(keep_raw=True, max_holding_period=1200, take_profit=None, stop_loss=None)

# 回测环境（适应度函数）
comm = [0 / 10000, 0 / 10000]  # 买卖费率
bt = SymbolicTestor(train_data, transact_base='PreClose', commissions=(comm[0], comm[1]))  # 加载数据，根据Close成交,comm是买-卖

def score_func_basic(y, y_pred, sample_weight):  # 因子评价指标
    try:
        _ = bt.run_(factor=y_pred)
        factor_ret = _['annualized_mean']/_['max_drawdown'] if _['max_drawdown'] != 0 else 0 # 可以把max_drawdown换成annualized_std
    except:
        factor_ret = 0
    return factor_ret

def my_gplearn(function_set, my_fitness, pop_num=100, gen_num=3, tour_num=10, random_state = 42, feature_names=None):
    # pop_num, gen_num, tour_num的几个可选值：500, 5, 50; 1000, 3, 20; 1000, 15, 100
    metric = fitness.make_fitness(function=my_fitness, # function(y, y_pred, sample_weight) that returns a floating point number.
                        greater_is_better=True,  # 上述y是输入的目标y向量，y_pred是genetic program中的预测值，sample_weight是样本权重向量
                        wrap=False)  # 不保存，运行的更快 # gplearn.fitness.make_fitness(function, greater_is_better, wrap=True)
    return SymbolicRegressor(population_size=pop_num,  # 每一代公式群体中的公式数量 500，100
                              generations=gen_num,  # 公式进化的世代数量 10，3
                              metric=metric,  # 适应度指标，这里是前述定义的通过 大于0做多，小于0做空的 累积净值/最大回撤 的评判函数
                              tournament_size=tour_num,  # 在每一代公式中选中tournament的规模，对适应度最高的公式进行变异或繁殖 50
                              function_set=function_set,
                              const_range=(-1.0, 1.0),  # 公式中包含的常数范围
                              parsimony_coefficient='auto',
                              # 对较大树的惩罚,默认0.001，auto则用c = Cov(l,f)/Var( l), where Cov(l,f) is the covariance between program size l and program fitness f in the population, and Var(l) is the variance of program sizes.
                              stopping_criteria=100.0,  # 是对metric的限制（此处为收益/回撤）
                              init_depth=(2, 3),  # 公式树的初始化深度，树深度最小2层，最大6层
                              init_method='half and half',  # 树的形状，grow生分枝整的不对称，full长出浓密
                              p_crossover=0.8,  # 交叉变异概率 0.8
                              p_subtree_mutation=0.05,  # 子树变异概率
                              p_hoist_mutation=0.05,  # hoist变异概率 0.15
                              p_point_mutation=0.05,  # 点变异概率
                              p_point_replace=0.05,  # 点变异中每个节点进行变异进化的概率
                              max_samples=1.0,  # The fraction of samples to draw from X to evaluate each program on.
                              feature_names=feature_names, warm_start=False, low_memory=False,
                              n_jobs=1,
                              verbose=1,
                              random_state=random_state)

# 函数集
function_set=['add', 'sub', 'mul', 'div', 'sqrt', 'log',  # 用于构建和进化公式使用的函数集
                    'abs', 'neg', 'inv', 'sin', 'cos', 'tan', 'max', 'min',
                    # 'if', 'gtpn', 'andpn', 'orpn', 'ltpn', 'gtp', 'andp', 'orp', 'ltp', 'gtn', 'andn', 'orn', 'ltn', 'delayy', 'delta', 'signedpower', 'decayl', 'stdd', 'rankk'
                    ] # 最后一行是自己的函数，目前不用自己函数效果更好

my_cmodel_gp = my_gplearn(function_set, score_func_basic, random_state=0,
                          feature_names=feature_names)  # 可以通过换random_state来生成不同因子
my_cmodel_gp.fit(train_data.loc[:, :'rank_num'].values, train_data.loc[:, 'y'].values)
print(my_cmodel_gp)

# 策略结果
factor = my_cmodel_gp.predict(test_data.values)
bt_test = SymbolicTestor(test_data, transact_base='PreClose', commissions=(comm[0], comm[1]))  # 加载数据，根据Close成交,comm是买-卖
bt_test.run_(factor=factor)
md = bt_test.summary()
print(md.out_stats)
print(bt.fees_factor)
md.plot_(comm=comm, show_bool=True)

分钟线，如果不加费率，那么效果还是非常好的：年化81%，最大回撤10%。

代码已经发布，大家前往星球下载：【星球优惠券】AI量化实验室——量化投资的星辰大海

直接运行gpquant_demo.py即可。

吾日三省吾身

这几天，除了思考AI量化实验室2024年的方向，额外思考2024年的规划。

回溯了2021-2022年，基本就是定了一个数字，一个收益率。

可以说是美好的愿景，但目标无法转化为系统，意义是有限的。

Z计划=投资计划，500万，10%的长期年化收益。10%是投资能力的代表，使用基金做大类资产配置就可以“轻松”做到。这个需要拉长周期来看，而且是被动收入，建立好投资体系，等待美好自然发生即可。

更重要的是本金的积累。

对于多数普通人而言，如何建立B计划，发展另一条收入管道非常关键，你的收入结构如何优化，如何赚到更多工资以外、理财以外的收入（一开始不必刻意强调被动收入）。

因子挖掘的流水线，一定是量化投资的未来。

分解下来，

一是数据，最常用的量价数据，无论是高频还是日频，对于我们处理没什么影响。

二是挖掘流程。gplearn或者深度强化学习，这也是通过框架。

三、扩展函数集，这个是可以积累的，比如ts_rank，甚至各种技术指标。

四、因子合成：加权，线性，或者树模型，深度学习。

五、回测，止损，投资体系。

六、实盘对接

后续我们会围绕这些主题，持续完善咱们的quantlab。按星球惯例，预计明天统计更新迭代后的代码。【星球优惠券】AI量化实验室——量化投资的星辰大海

从上面的路线图来，回测系统只占其中一小部分，新手一般认为这基本就是全部。其实不是。

市场上开源的量化回测系统很多，而且你自己实现一个简单的也超级容易，星球里有好几个版本。另外，回测到实盘，或者说，所谓“无缝切换”，其实不是什么大问题，broker，获取数据的方式，本身就不一样，只是少一点代码罢了，不必为了兼容实盘而兼容。实盘系统本身比回测更简单。

我一直在考虑回测系统，希望功能要够，足够简单，这样我们才能专注在策略和因子上，而不是与框架“做斗争”。

回测系统按运行方式有三类：向量化，时间驱动，事件驱动。

一般现在市面上主流是时间驱动，就是按bar来循环。真正的事件驱动pyalgotrade显得有点多余。这个好处是与实盘的运行机制一样，vnpy早看也是，后来改成时间驱动，因为调试起来更加容易。

按如上规划：回测系统够用即可，重点策略，核心是因子。

数据自动更新之类的，也不在回测系统范围内，自行加载CSV即可。

下载期货主连合约数据：

import akshare as ak
from datetime import datetime
import pandas as pd
'''
0  V0 dce    PVC连续
1  P0 dce    棕榈油连续
2  B0 dce    豆二连续
3  M0 dce    豆粕连续
4  I0 dce    铁矿石连续
5  JD0    dce    鸡蛋连续
6  L0 dce    塑料连续
7  PP0    dce    聚丙烯连续
8  FB0    dce    纤维板连续
9  BB0    dce    胶合板连续
10 Y0 dce    豆油连续
11 C0 dce    玉米连续
12 A0 dce    豆一连续
13 J0 dce    焦炭连续
14 JM0    dce    焦煤连续
15 CS0    dce    淀粉连续
16 EG0    dce    乙二醇连续
17 RR0    dce    粳米连续
18 EB0    dce    苯乙烯连续
19 LH0    dce    生猪连续
20 TA0    czce   PTA连续
21 OI0    czce   菜油连续
22 RS0    czce   菜籽连续
23 RM0    czce   菜粕连续
24 ZC0    czce   动力煤连续
25 WH0    czce   强麦连续
26 JR0    czce   粳稻连续
27 SR0    czce   白糖连续
28 CF0    czce   棉花连续
29 RI0    czce   早籼稻连续
30 MA0    czce   甲醇连续
31 FG0    czce   玻璃连续
32 LR0    czce   晚籼稻连续
33 SF0    czce   硅铁连续
34 SM0    czce   锰硅连续
35 CY0    czce   棉纱连续
36 AP0    czce   苹果连续
37 CJ0    czce   红枣连续
38 UR0    czce   尿素连续
39 SA0    czce   纯碱连续
40 PF0    czce   短纤连续
41 PK0    czce   花生连续
42 FU0    shfe   燃料油连续
43 SC0    ine    上海原油连续
44 AL0    shfe   铝连续
45 RU0    shfe   天然橡胶连续
46 ZN0    shfe   沪锌连续
47 CU0    shfe   铜连续
48 AU0    shfe   黄金连续
49 RB0    shfe   螺纹钢连续
50 WR0    shfe   线材连续
51 PB0    shfe   铅连续
52 AG0    shfe   白银连续
53 BU0    shfe   沥青连续
54 HC0    shfe   热轧卷板连续
55 SN0    shfe   锡连续
56 NI0    shfe   镍连续
57 SP0    shfe   纸浆连续
58 NR0    ine    20号胶连续
59 SS0    shfe   不锈钢连续
60 LU0    ine    低硫燃料油连续
61 BC0    ine    国际铜连续
62 IF0    cffex  沪深300指数期货连续
63 TF0    cffex  5年期国债期货连续
64 IH0    cffex  上证50指数期货连续
65 IC0    cffex  中证500指数期货连续
66 TS0    cffex  2年期国债期货连续
'''
symbol = 'V0'
df = ak.futures_main_sina(symbol="V0", start_date="19900101", end_date=datetime.now().strftime('%Y%m%d'))
print(df)
df.rename(columns={'日期':'date','开盘价':'open', '最高价':'high', '最低价':'low','收盘价':'close', '成交量':'volume','持仓量':'open_interest','动态结算价':'vwap'}, inplace=True)

from config import DATA_DIR
df.to_csv(DATA_DIR.joinpath(symbol+'.csv'),index=None)

数据格式已经处理好了。

新增CSVLoader:

from config import DATA_DIR_CACHE_H5, DATA_DIR
import os


class CSVDataloader(Dataloader):
    def __init__(self, path:WindowsPath, symbols, start_date='20100101', end_date=datetime.now().strftime('%Y%m%d')):
        super(CSVDataloader, self).__init__(path, symbols, start_date, end_date)

    def _load_dfs(self):
        dfs = []
        csvs = os.listdir(self.path.resolve())
        for csv in csvs:
            df = pd.read_csv(self.path.joinpath(csv).resolve(), index_col=None)
            df.set_index('date', inplace=True)

        dfs.append(df)
        return dfs