AI量化社区，单元测试与重新引入pybroker

正在进行中的开发，包含但不限于：

1、引入pybroker引擎。

2、全流程单元测试，提升质量。

3、各种经典策略，前沿机器学习模型引入。

量化论坛建设中

AI量化是一个长期主义的事情，做持续有积累，有价值的事情。

上午把AI量化论坛简单搭建了一个，网站也有个样子了，还有优化中。

发现星友中，有好多初学者，也许是好奇，也许是奔着收益。

但无论如何，想学投资，以及想用量化对投资加持，至少这个选择是对的。

星球本身就是一个社群，为何还要一个论坛。

星球对于一些不要花钱，或者犹豫的同学，还是有一点门槛，论坛则是半开放的。正如，有了社群，我们一样有群。

群，最大的问题，就是有些问题要不停的重复回答，因此我想到了一个中间态，就是传统的bbs。大家可以去指定版块先检索答案，若没有解决，则可以再提问，指定版主来回答。

当然，星友会有私密的高级版块，还可以引入圈子等功能。

后面面向不同进度的同学，我们还是要准备课程的。

单元测试

量化是一个严肃的事情，如何确保程序质量，越小的开发团队，单元测试越发重要。

我们使用pytest单元测试框架。

从咱们很重要的信号选股算子为例：

这个算子，可以输出多条规则，比如“20日动量>0.08”，“收盘价突出上轨”之类的，at_least_count是至少满足多少条规则，比如3条中的两条。这在规则量化中非常实用。direction是信号的方向，是做多，空，还是平仓。

class SelectBySignal(Algo):
    def __init__(self, rules=[], at_least_count=1, direction='long'):
        # direction是信号的方向： long:多单， short空单， flat平仓。
        super(SelectBySignal, self).__init__()
        self.rules = rules

        if at_least_count > len(rules):
            at_least_count = len(rules)
        if at_least_count <= 0:
            at_least_count = 1
        self.at_least_count = at_least_count

        self.direction = direction

    def _check_if_matched(self, df_bar, rules, at_least_count):
        matched_items = []

        se_count = pd.Series(index=df_bar.index, data=0)
        for r in rules:
            se_count += df_bar.eval(r)

        matched_items = se_count[(se_count.values >= self.at_least_count)].index
        return matched_items

    def __call__(self, target):
        df_bar = target.df_bar
        matched = None
        if self.rules and len(self.rules):
            matched = self._check_if_matched(df_bar, self.rules, self.at_least_count)
        if len(matched) == 0:
            return True

        if self.direction == 'flat':
            target.temp['selected_flat'] = matched  # 卖出信号，如果允许short，那就是做空单，或者平仓
        elif self.direction == 'short':
            target.temp['selected_short'] = matched
        else:
            target.temp['selected'] = matched
        return True

借写单元测试，我重构了一版，使用df.eval，效率高，而且部分信号不需要预先计算了。

这里逻辑分支挺多的，有了单元测试，就不需要担心再次重构的问题了。

from engine.strategy import StrategyAlgo
from engine.algos import *
from unittest import mock


def test_select_by_signal():
    s = mock.MagicMock()

    df_bar = pd.DataFrame(
        data=[
            [3003, 0.08, 1.1],
            [16888, -0.03, 0.7],
        ],
        index=['000300.SH', 'SPX'],
        columns=['close', 'roc_20', 'rsrs_18']
    )
    s.df_bar = df_bar
    s.temp = {}

    algo = SelectBySignal(rules=['roc_20>0'])
    algo(s)
    assert list(s.temp['selected']) == ['000300.SH']

    s.temp = {}
    algo = SelectBySignal(rules=['roc_20<0','rsrs_18<0.8'], at_least_count=3)
    algo(s)
    assert list(s.temp['selected']) == ['SPX']

    s.temp = {}
    algo = SelectBySignal(rules=['roc_20<0', 'rsrs_18<0.8'], at_least_count=3, direction='flat')
    algo(s)
    assert list(s.temp['selected_flat']) == ['SPX']

    s.temp = {}
    algo = SelectBySignal(rules=['roc_20>-1', 'rsrs_18>1'], at_least_count=1, direction='short')
    algo(s)
    assert list(s.temp['selected_short']) == ['000300.SH', 'SPX']

    s.temp = {}
    algo = SelectBySignal(rules=['roc_20>-1', 'rsrs_18>2'], at_least_count=2, direction='short')
    algo(s)
    assert 'selected_short' not in s.temp.keys()

    s.temp = {}
    algo = SelectBySignal(rules=['roc_20<0', 'rsrs_18<0.8'], at_least_count=0, direction='flat')
    algo(s)
    assert list(s.temp['selected_flat']) == ['SPX']

几点体会，

1、python这样的脚本语言，有些分支错误会带到运行时，使用单元测试覆盖下很有必要。

2、写单元测试时，反而会帮我们补充很多边界条件（尤其是异常分支）。

3、代码重构后，可以轻松验证。

这就属于有积累，值得做的事情。

pybroker

之前其实写过系列文章：

基于pybroker的动量轮动+排序模型，年化11%（附代码）

pybroker代码框架梳理与WFA算法

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

pybroker：原生支持机器学习模型进行量化

pybroker：兼容传统规则和机器学习的高性能量化回测框架（附源码）

昨天我在星球里发起的讨论：

backtrader在提交多个订单的时候，是需要自己决定顺序的，比如先卖再买，先平仓再开仓。

要说错吧，也没错，但回测系统可以做得更好，因为这个肯定的。做轮动，再平衡，还要去自己算是买还卖，违背易用的原则。

因为我想再看看pybroker的原因。

pybroker代码看懂很容易，支持它自定义的DataSource也支持Dataframe。

咱们现有的多symbol的dataframe加一个‘date’列就可以使用了，现代技术栈还是比较方便，老框架喜欢封装成各种自己的类。

from pybroker import Strategy

from datafeed.dataloader import Duckdbloader
from config import DATA_DIR_CSVS

symbols = ['000300.SH', '000905.SZ']
loader = Duckdbloader(path=DATA_DIR_CSVS.joinpath('index').resolve(), symbols=symbols,
                      columns=['close', 'high', 'low', 'open', 'volume'],
                      start_date="20100101")

df = loader.load()
df['date'] = df.index
print(df)


def exec_fn(ctx):
    # Buy on a new 10 day high.
    if not ctx.long_pos():
        ctx.buy_shares = 100
        # Hold the position for 5 days.
        ctx.hold_bars = 5
        # Set a stop loss of 2%.
        ctx.stop_loss_pct = 2


strategy = Strategy(df, start_date='1/1/2022', end_date='7/1/2022')
strategy.add_execution(
    exec_fn, symbols)
# Run the backtest after 20 days have passed.
result = strategy.backtest(warmup=20)
print(result.metrics_df)

print(result.positions)

print(result.trades)

import matplotlib.pyplot as plt

chart = plt.subplot2grid((3, 2), (0, 0), rowspan=3, colspan=2)
chart.plot(result.portfolio.index, result.portfolio['market_value'])
plt.show()