Quantlab 4.1：基于Deap遗传算法多股票因子挖掘

原创文章第549篇，专注“AI量化投资、世界运行的规律、个人成长与财富自由"。

遗传算法本身并不复杂，但gplearn的实现，把问题复杂化了，尤其在因子挖掘这个场景。

使用deap进行因子挖掘的代码在如下位置：

import copyfrom deap_patch import *  # noqa
from deap import base, creator, gp
from deap import tools
from datafeed.mining.deap_alpha.add_ops import *def convert_inverse_prim(prim, args):"""Convert inverse prims according to:[Dd]iv(a,b) -> Mul[a, 1/b][Ss]ub(a,b) -> Add[a, -b]We achieve this by overwriting the corresponding format method of the sub and div prim."""prim = copy.copy(prim)converter = {'Add': lambda *args_: "{}+{}".format(*args_),'Mul': lambda *args_: "{}*{}".format(*args_),'fsub': lambda *args_: "{}-{})".format(*args_),'fdiv': lambda *args_: "{}*Pow({}, -1))".format(*args_),'fmul': lambda *args_: "{}*{}".format(*args_),'fadd': lambda *args_: "{}+{}".format(*args_),'fmax': lambda *args_: "max_({},{})".format(*args_),'fmin': lambda *args_: "min_({},{})".format(*args_),'isub': lambda *args_: "{}-{}".format(*args_),'idiv': lambda *args_: "{}/{}".format(*args_),'imul': lambda *args_: "{}*{}".format(*args_),'iadd': lambda *args_: "{}+{}".format(*args_),'imax': lambda *args_: "max_({},{})".format(*args_),'imin': lambda *args_: "min_({},{})".format(*args_),}#print(prim.name)#if prim.name == "imul":#print('命中：',prim.name)#print(prim)prim_formatter = converter.get(prim.name, prim.format)return prim_formatter(*args)def stringify_for_sympy(f):"""Return the expression in a human readable string."""string = ""stack = []for node in f:stack.append((node, []))while len(stack[-1][1]) == stack[-1][0].arity:prim, args = stack.pop()string = convert_inverse_prim(prim, args)if len(stack) == 0:break  # If stack is empty, all nodes should have been seenstack[-1][1].append(string)print(string)return stringdef map_exprs(evaluate, invalid_ind, gen, label, split_date):#print(invalid_ind)sources = [f'GP_{i:04d}={stringify_for_sympy(expr)}' for i, expr in enumerate(invalid_ind)]print(sources)def init_pset():pset = gp.PrimitiveSetTyped("MAIN", [], RET_TYPE)pset = add_constants(pset)pset = add_operators(pset)pset = add_factors(pset)return psetdef init_creator():# 可支持多目标优化# TODO 必须元组，1表示找最大值,-1表示找最小值FITNESS_WEIGHTS = (1.0, 1.0)creator.create("FitnessMulti", base.Fitness, weights=FITNESS_WEIGHTS)creator.create("Individual", gp.PrimitiveTree, fitness=creator.FitnessMulti)return creatordef init_toolbox(creator):toolbox = base.Toolbox()pset = init_pset()toolbox.register("expr", gp.genHalfAndHalf, pset=pset, min_=2, max_=5)toolbox.register("individual", tools.initIterate, creator.Individual, toolbox.expr)toolbox.register("population", tools.initRepeat, list, toolbox.individual)toolbox.register("evaluate", print)  # 不单独做评估了，在map中一并做了from datetime import datetimedt1 = datetime(2021, 1, 1)LABEL_y = 'RETURN_OO_1'from itertools import counttoolbox.register('map', map_exprs, gen=count(), label=LABEL_y, split_date=dt1)return toolboxif __name__ == '__main__':random.seed(9527)creator = init_creator()toolbox = init_toolbox(creator)pop = toolbox.population(n=100)fitnesses = toolbox.map(toolbox.evaluate, pop)

重点看add_ops添加咱们自己的因子表达式引擎支持的函数：

# TODO: 请在此文件中添加算子和因子
# TODO: 由于部分算子计算过慢，这里临时屏蔽了
import randomclass RET_TYPE:# 是什么不重要# 只要addPrimitive中in_types, ret_type 与 PrimitiveSetTyped("MAIN", [], ret_type)中# 这三种type对应即可pass# 改个名，因为从polars_ta中默认提取的annotation是Expr
# TODO 如果用使用其它库，这里可能要修改
Expr = RET_TYPEdef _random_int_():return random.choice([1, 3, 5, 10, 20, 40, 60])def add_constants(pset):"""添加常量"""# !!! 名字一定不能与其它名字重，上次与int一样，结果其它地方报错 [<class 'deap.gp.random_int'>]pset.addEphemeralConstant('_random_int_', _random_int_, int)return psetdef add_operators_base(pset):"""基础算子"""# 无法给一个算子定义多种类型，只好定义多个不同名算子，之后通过helper.py中的convert_inverse_prim修正pset.addPrimitive(dummy, [Expr, Expr], Expr, name='fadd')pset.addPrimitive(dummy, [Expr, Expr], Expr, name='fsub')pset.addPrimitive(dummy, [Expr, Expr], Expr, name='fmul')pset.addPrimitive(dummy, [Expr, Expr], Expr, name='fdiv')pset.addPrimitive(dummy, [Expr, Expr], Expr, name='fmax')pset.addPrimitive(dummy, [Expr, Expr], Expr, name='fmin')pset.addPrimitive(dummy, [Expr, int], Expr, name='iadd')pset.addPrimitive(dummy, [Expr, int], Expr, name='isub')pset.addPrimitive(dummy, [Expr, int], Expr, name='imul')pset.addPrimitive(dummy, [Expr, int], Expr, name='idiv')return psetdef add_unary_ops(pset):from datafeed.expr_functions import unary_funcsfor func in unary_funcs:pset.addPrimitive(dummy, [Expr], Expr, name=func)def add_unary_rolling_ops(pset):from datafeed.expr_functions import unary_rolling_funcsfor func in unary_rolling_funcs:pset.addPrimitive(dummy, [Expr, int], Expr, name=func)def add_binary_ops(pset):from datafeed.expr_functions import binary_funcsfor func in binary_funcs:pset.addPrimitive(dummy, [Expr, Expr], Expr, name=func)def add_binary_rolling_ops(pset):from datafeed.expr_functions import binary_roilling_funcsfor func in binary_roilling_funcs:pset.addPrimitive(dummy, [Expr, Expr, int], Expr, name=func)def add_operators(pset):"""添加算子"""pset = add_operators_base(pset)add_unary_ops(pset)add_binary_ops(pset)add_unary_rolling_ops(pset)add_binary_rolling_ops(pset)return psetdef add_factors(pset):pset.addTerminal(1, Expr, name='OPEN')pset.addTerminal(1, Expr, name='HIGH')pset.addTerminal(1, Expr, name='LOW')pset.addTerminal(1, Expr, name='CLOSE')pset.addTerminal(1, Expr, name='VOLUME')# pset.addTerminal(1, Expr, name='AMOUNT')return psetdef dummy(*args):# 由于生成后的表达计算已经被map和evaluate接管，所以这里并没有用到，可随便定义print('dummy')return 1

生成的因子，直接转化为咱们因子表达式可以兼容的表达式：

可以计算出因子值：