表达式设计

基本思路

解析公式: 可以将每个公式看作一个字符串，然后用 Python 的字符串处理功能提取其中的变量、和运算符。其中，带问号的运算符因 python 不支持，需要进行特殊处理，使用 ast 替换为自定义的 where 函数实现
映射函数: 将公式中的函数名，例如 rank, ts_argmax, stddev 等，映射到实现的 python 函数
执行计算: 根据公式的逻辑和运算符，依次调用相应的函数，并传入对应的变量或中间结果，最终得到计算结果

解析代码

解析公式的代码如下，parse_and_calculate_alpha 输入参数为公式字符串和 pandas 股票数据：

class RewriteNode(ast.NodeTransformer):
    # 重写 AST 节点，用于转换三元表达式
    def visit_IfExp(self, node):
        # 交换 node 的 body 和 test
        node.body, node.test = node.test, node.body
        # 将 AST 节点转换为字符串
        test = ast.unparse(node.test)
        body = ast.unparse(node.body)
        orelse = ast.unparse(node.orelse)
        # 构建新的 where 函数调用节点
        newnode = ast.parse(f"where({test}, {body}, {orelse})").body[0]
        return newnode

def ternary_trans(formula):
    # 替换公式中的三元运算符为 if-else 形式
    formula = formula.replace('?', ' if ')
    formula = formula.replace(':', ' else ')
    # 解析修改后的公式为 AST 树
    tree = ast.parse(formula)
    # 遍历 AST 树，应用 RewriteNode 转换
    for node in ast.walk(tree):
        RewriteNode().visit(node)
    # 将转换后的 AST 树转换回字符串
    return ast.unparse(tree)

def where(c: pd.Series, t: pd.Series, f: pd.Series) -> pd.Series:
    return pd.Series(np.where(c, t, f))

def parse_and_calculate_alpha(formula, data):
    # 暂时不支持的函数，返回空数据
    todolist=['indneutralize','cap','filter','self','banchmarkindex']
    for todo in todolist:
        if todo in formula:
            return pd.Series(index=data.index)

    # 替换运算公式为python运算符
    formula=formula.replace("||"," | ")
    formula=formula.replace("&&"," & ")
    formula=formula.replace("^"," ** ")

    # 将 $xxx 替换为 data['xxx']
    for col in data.columns:
        formula = formula.replace(f'${col}', f"data['{col}']")

    print(f"变量替换后的公式: {formula}")

    # 使用 ast 处理三元表达式
    if '?' in formula:
        formula = ternary_trans(formula)
    print(f"三元表达式替换后的公式: {formula}")

    # 解析公式字符串为 AST 树
    tree = ast.parse(formula, mode='eval')
    print(f"ast 解析后的结果: {ast.dump(tree)}")  # 打印 ast 解析后的结果
    # 创建 ast.Expression 对象
    expression = ast.Expression(body=tree.body)

    # 将转换后的 AST 树编译为可执行的代码对象
    code = compile(expression, filename='<alpha_formula>', mode='eval') 

    # 将 FUNCTION_MAP 中的所有函数添加到全局环境
    global_env = {'__builtins__': {'True': True, 'False': False}}
    for func_name, func in FUNCTION_MAP.items():
        global_env[func_name] = func

    # 将 data 变量添加到局部环境
    local_env = {'data': data,'where':where}
    # 执行代码并返回结果
    result = eval(code, global_env, {**local_env})

    return pd.Series(result, index=data.index)