原创内容第656篇，专注量化投资、个人成长与财富自由。

一层秋雨一层凉，上班途中雨大，北京正式入秋了。

世事大梦一场，人生几度秋凉。

世事一场大梦，人生几度秋凉？夜来风叶已鸣廊。看取眉头鬓上。酒贱常愁客少，月明多被云妨。中秋谁与共孤光。把盏凄然北望。

美联储降息，符合预期，美股暴涨。

一周一度，本周星球Quantlab更新重点如下：

1、《七天入门量化投资》专栏完成第6天的文章及代码。

2、年化收益61.9%，最大回撤-13.3%，卡玛比率达到4.6，ETF趋势策略优化——策略优化到61.9%，大家可重点参考思路。

3、全新的回测引擎——这一版本的重写，个人是比较满意的，基本都是向量化计算，可以达到极速回测之效果。

源码包、数据和策略集已经打包提交：

import bt
import numpy as np
import pandas as pd


class SelectTopK(bt.AlgoStack):
    def __init__(self, signal, K, dropN=0, sort_descending=True, all_or_none=False, filter_selected=True):
        super(SelectTopK, self).__init__(bt.algos.SetStat(signal),
                                         bt.algos.SelectN(K + dropN, sort_descending, all_or_none, filter_selected))
        self.dropN = dropN

    def __ceil__(self, target):
        super(SelectTopK, self).__ceil__()
        if self.dropN > 0:
            sel = target.temp["selected"]
            if self.dropN >= len(sel):
                target.temp['selected'] = []
            else:
                target.temp["selected"] = target.temp["selected"][self.dropN:]
            return True


from datafeed.dataloader import CSVDataloader
from datetime import datetime
from matplotlib import rcParams

rcParams['font.family'] = 'SimHei'


class Task:
    name: str
    desc: str
    symbols: list = []
    fields: list = []
    names: list = []
    start_date: str = '20100101'
    end_date: str = datetime.now().strftime('%Y%m%d')
    benchmark: str = '510300.SH'
    select: str = 'SelectAll'

    select_buy: str = None
    select_sell: str = None

    order_by_signal: str = None
    order_by_topK: int = 1
    order_by_dropN: int = 0
    order_by_DESC: bool = True  # 默认从大至小排序

    weight: str = 'WeighEqually'
    period: str = 'RunDaily'
    period_days: int = None


import importlib


class Engine:
    def _parse_rules(self, task: Task, df):

        def _rules(rules, b_or=False):
            signals = []
            if not rules:
                return None
            for r in rules:
                signals.append(eval(r))

            all = None
            for s in signals:
                if all is None:
                    all = s
                else:
                    if b_or:
                        all |= s
                    else:
                        all &= s
            return all

        # rules = ['roc_20>0.02']
        for factor in task.names:
            globals()[factor] = CSVDataloader.get_col_df(df, factor)
            # print(globals()[factor])

        all_buy = _rules(task.select_buy)
        all_sell = _rules(task.select_sell, b_or=True)  # 卖出 求或，满足一个即卖出
        return all_buy, all_sell

    def _get_algos(self, task: Task, df):

        bt_algos = importlib.import_module('bt.algos')

        if task.period == 'RunEveryNPeriods':
            algo_period = bt.algos.RunEveryNPeriods(n=task.period_days)
        else:
            algo_period = getattr(bt_algos, task.period)()

        algo_weight = getattr(bt_algos, task.weight)()

        algo_select_where = None
        # 信号规则
        if task.select_buy is not None:
            signal_buy, signal_sell = self._parse_rules(task, df)

            select_signal = np.where(signal_buy, 1, np.nan)
            select_signal = np.where(signal_sell, 0, select_signal)
            select_signal = pd.DataFrame(select_signal, index=signal_buy.index, columns=signal_buy.columns)
            select_signal.ffill(inplace=True)
            select_signal.fillna(0, inplace=True)
            algo_select_where = bt.algos.SelectWhere(signal=select_signal)

        # 排序因子
        algo_order_by = None
        if task.order_by_signal:
            signal_order_by = CSVDataloader.get_col_df(df, col=task.order_by_signal)
            algo_order_by = SelectTopK(signal=signal_order_by, K=task.order_by_topK, dropN=task.order_by_dropN,
                                       sort_descending=task.order_by_DESC)

        algos = []
        algos.append(algo_period)

        if algo_select_where:
            algos.append(algo_select_where)
        else:
            algos.append(bt.algos.SelectAll())

        if algo_order_by:
            algos.append(algo_order_by)

        if task.weight == 'WeighERC':
            algos.insert(0, bt.algos.RunAfterDays(days=126))
        algos.append(algo_weight)
        algos.append(bt.algos.Rebalance())

        return algos

    def run(self, task: Task):
        # 加载数据
        df = CSVDataloader.get_df(task.symbols, True, task.start_date, task.end_date)

        # 计算因子
        if len(task.symbols):
            df = CSVDataloader.calc_expr(df, fields=task.fields, names=task.names)

        s = bt.Strategy(task.name, self._get_algos(task, df))

        df_close = CSVDataloader.get_col_df(df, 'close')
        bkt = bt.Backtest(s, df_close, name='策略')

        bkts = [bkt]
        for bench in [task.benchmark]:
            data = CSVDataloader.get([bench])
            s = bt.Strategy(bench, [bt.algos.RunOnce(),
                                    bt.algos.SelectAll(),
                                    bt.algos.WeighEqually(),
                                    bt.algos.Rebalance()])
            stra = bt.Backtest(s, data, name='benchmark')
            bkts.append(stra)

        res = bt.run(*bkts)
        res.get_transactions()
        return res

代码下载：AI量化实验室 星球，已经运行三年多，1100+会员。quantlab代码交付至5.X版本，含几十个策略源代码，因子表达式引擎、遗传算法（Deap）因子挖掘引擎等，每周五迭代一次。

AI量化实验室——2024量化投资的星辰大海

02 五年财务自由退休

如果你发现自己没有特别擅长的事情，那就去读书，分享读书心得也可以。

（限时免费，感兴趣可入）五年之约，一起出发吧！

作者：AI量化实验室（专注量化投资、个人成长与财富自由）

 扩展  •  历史文章   

• 年化收益52%，最大回撤13%，卡玛比率3.77，ETF轮动系列大有可为（附策略代码和数据下载）。

• 年化12.6%，最大回撤才2.6的债券轮动策略，卡玛比4.79，稳稳的幸福

• 年化收益61.9%，最大回撤-13.3%，卡玛比率达到4.6，ETF趋势策略优化

• AI量化实验室——2024量化投资的星辰大海