因子表达式，积木式策略开发与pybroker框架整合（附源码）

原创文章第229篇，专注“个人成长与财富自由、世界运作的逻辑与投资"。

我是自己做做投资，写一些工具辅助投资思考，投研。

未来一定是一个AI的时代，所有行业都将会被重塑造，包括金融。

投资理念刷新

长线投资部分，固收+（会考虑主动型基金），以及优质指数型基金。

何谓“优质“，长期来看就看指数的ROE。

省心的角度，就从下面的指数里选，基本不太会错。在相对低位的时候”定投“。

今天白酒和消费红利设置定投，因为百分位在40%左右。

中证煤炭与中证医疗抽空要研究一下基本面。

因子表达式

尽管短期内不太可能，投资决策100%交给机器，但机器可以自动化，智能化的帮助我们做不少事情。投资应该是75%的科学加25%的艺术。艺术的部分就是咱们人类的范畴，但用于那个75%至关重要。

我们准备好了数据，下一步就是：因子，特别需要因子表达式，因为如果我们想到一个策略，要去计算这个因子，费时费力不说，关键是容易出错。如果要计算上百个因子，那就太费事了。好在ta-lib这样的库比较成熟了。但仍然不方便。

比如RSRS这个指标，就是求两个序列的斜率，这个talib是不提供的。

希望使用这样的表达式——slope(high,low,18,600)

调用talib的函数使用——ta_RSI(close,20)。

其实实现起来也不难，就是python的eval，外加正则表达式即可。

而且支持调talib的所有函数：

比如ATR：

像布林带这种，返回多个结果，可以指定具体为哪一条，不指定，默认取第0个。

核心代码实现比较简单：在inds.py下，打包发布在星球里，请大家前往下载：

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


def ta(fn, se, *args, **kwargs):
    result = getattr(talib, fn)(se, *args)
    if type(result) is tuple:
        if 'get_result' in kwargs.keys():
            if kwargs['get_result'] >= len(result):
                print('参数超过个数，返回第一个')
                return result[0]
            else:
                return result[kwargs['get_result']]
        return result[0]
    return result


def expr_transform(expr):
    # close/shift(close,5) -1
    for col in ['open', 'high', 'low', 'close', 'volume']:
        expr = expr.replace(col, 'bar_data.{}'.format(col))
    return expr


def indicator_fn(bar_data, expr):
    expr = expr_transform(expr)
    se = eval(expr)
    return se
    # se = eval('ta("RSI",bar_data.close,20)')  # ta_RSI(close,20)
    return se


def to_indicator(name, expr):
    return pybroker.indicator(name, indicator_fn, expr=expr)

”积木式“开发

一个策略，无论是规则型，还是机器模型，无外乎几个步骤：

择时：比如每月、每周调仓这种再平衡策略。（默认是每天运行）

1、选股（select_all，select_by_rule，select_by_model，这里可能会先对某个指标进行排序，然后选择top几个）

2、分配权重

3、调仓。

使用pybroker的api，修改了原来的算子，逻辑类似，使用了pyb.param全局传参，原先我们使用context。另外就是pyb会内置一个ctxs。

from pybroker import ExecContext
import pybroker as pyb

from loguru import logger
from quant import utils

class RunOnce:
    def __init__(self):
        self.done = False

    def __call__(self, *args, **kwargs):
        done = self.done
        pyb.param('is_done', done)
        self.done = True

class SelectAll:
    def __init__(self, universe):
        self.universe = universe

    def __call__(self, *args, **kwargs):
        if pyb.param('is_done') is True:
            return
        pyb.param('selected', self.universe)


class WeightEqually:
    def __init__(self):
        pass

    def __call__(self, ctxs: dict[str, ExecContext], *args, **kwargs):
        if pyb.param('is_done') is True:
            return
        # 这里就是全部当前要持仓的（含已经持仓了的）
        selected = pyb.param('selected')
        if selected is None:
            logger.error('selected参数不存在，返回')
            return
        pyb.param('selected', None)

        N = len(selected)

        if N > 0:
            weight = 1 / N
            for symbol, ctx in ctxs.items():
                if symbol not in selected:
                    utils.order_tanget_percent(ctx, 0)

            for symbol, ctx in ctxs.items():
                if symbol in selected:
                    utils.order_tanget_percent(ctx,weight)

        else:
            logger.error('selected为空，全部清仓。')
            for symbol, ctx in ctxs.items():
                utils.order_tanget_percent(ctx,0.0)

这些算子未来是可以复用的。

这样我们写一个策略就非常容易了。

symbols = ['000300.SH', 'SPX']
e = Engine(universe=symbols, config=config)
e.set_algos([
    RunOnce(),
    SelectAll(symbols),
    WeightEqually()
])

result = e.run()
e.show_result(result)

代码已发布至星球，包括程序主体与测试的notebook。

下周继续，大家周末快乐，同时要开始做单因子IC、IR分析。