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

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

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

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

import copy

from 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 seen
            stack[-1][1].append(string)
    print(string)
    return string

def 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 pset

def 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 creator

def 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 datetime
    dt1 = datetime(2021, 1, 1)
    LABEL_y = 'RETURN_OO_1'
    from itertools import count
    toolbox.register('map', map_exprs, gen=count(), label=LABEL_y, split_date=dt1)

    return toolbox

if __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 random


class RET_TYPE:
    # 是什么不重要
    # 只要addPrimitive中in_types, ret_type 与 PrimitiveSetTyped("MAIN", [], ret_type)中
    # 这三种type对应即可
    pass


# 改个名，因为从polars_ta中默认提取的annotation是Expr
# TODO 如果用使用其它库，这里可能要修改
Expr = RET_TYPE


def _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 pset


def 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 pset


def add_unary_ops(pset):
    from datafeed.expr_functions import unary_funcs
    for func in unary_funcs:
        pset.addPrimitive(dummy, [Expr], Expr, name=func)


def add_unary_rolling_ops(pset):
    from datafeed.expr_functions import unary_rolling_funcs
    for func in unary_rolling_funcs:
        pset.addPrimitive(dummy, [Expr, int], Expr, name=func)


def add_binary_ops(pset):
    from datafeed.expr_functions import binary_funcs
    for 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_funcs
    for 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 pset


def 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 pset


def dummy(*args):
    # 由于生成后的表达计算已经被map和evaluate接管，所以这里并没有用到，可随便定义
    print('dummy')
    return 1

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

可以计算出因子值：