Quantlab服务器运行版本：更简化的策略模板

今天来一个比昨天还简单的策略，创业板ETF动量择时：年化21%，最大回撤18%。

创业板择时策略代码如下：

def gen_cy_picktime():
    proj = ProjConfig()
    proj.name = '创业板动量择时'
    proj.commission = 0.0001
    proj.slippage = 0.0001
    proj.symbols = ['159915.SZ']  # 证券池列表
    proj.benchmark = '159915.SZ'
    proj.start_date = '20100101'
    proj.data_folder = 'etfs'  # 这里指定data/数据目录

    fields = ['roc(close,20)']
    names = ['roc_20']
    proj.fields = fields
    proj.names = names

    # 这里是策略算子列表
    proj.algos.append(
        AlgoConfig(name=SelectBySignal().name, kwargs={'rules_buy': ['roc_20>0.08'], 'rules_sell': ['roc_20<-0.0']})

    )
    #proj.algos.append(
    #    AlgoConfig(name=SelectTopK().name, kwargs={'factor_name': 'roc_20'})
    #)
    proj.algos.append(AlgoConfig(name=WeightEqually().name))
    proj.algos.append(AlgoConfig(name=Rebalance().name))

    return proj

昨天的策略，再优化一下，年化20.3%，就两个标的：沪深300ETF和创业板ETF两个宽基ETF。

大家选择“大小盘轮动”策略即可：

代码在个位置：

之前的文章提及的DeepAlphaGen:

DeepAlphaGen:强化学习的因子组合挖掘：框架代码+数据下载

代码在如下目录：

吾日三省吾身

“凡事发生必有利于我”。

上周开始，眼晴不舒服，初步判断是干眼症。

想来，沉浸式写代码，做研究，用眼过度。

停下来，思考方向，人的精力真的很有限，专注做好能力圈的一点事情就好。

做有积累的事情，尽量自动化。一次时间投入，多次复用的事情。

如无必要，勿增实体。

投资应该是轻松，愉悦，甚至是略显无聊的事情，它不需要你特别努力，也不是你努力就有用的。关键是你要做对的事情，方向对就好。

后续的重心：

Dagster盘后更新数据，在线的， SAAS化的回测系统，不排除可转接对实盘。数据上会专注把ETF先搞扎实，当然技术栈，框架，因子挖掘，我相信都是通用的，大家愿意用在股票上，期货上，甚至加密货币上，都是可以的。

后端的框架应该还是Django，如果需要前后端分离，会使用django-ninja 一个类似fastapi的框架。前端暂时不想引入nodejs，太麻烦，还是喜欢bootstrap（jquery）的老方式。

我们那个年代的程序员，对于SEO是有执念的，也不知道有没有用。

SAAS版本AI量化回测系统，最大的挑战在于，任务可能需要异步运行，当然django也有celery这样的框架。

今天开始深度学习挖因子。

前面的文章开了个头：DeepAlphaGen:强化学习的因子组合挖掘

主流的公、私募量化，多因子模型是重中之重和热点方向。容量大，可以与前沿技术相结合。

东方证券的一张图：多因子选股体系

多因子选股体系主要包括 Alpha 模型、风险模型、交易成本模型和组合优化四个模块。Alpha 模型负责对股票收益或 Alpha 的预测，对组合收益的影响相对更大，是量化研究的重中之
重。传统的 Alpha 模型一般分为 Alpha 因子库构建和 Alpha 因子加权两个核心步骤。

其中：在 Alpha 因子构建中，可以引入的常见机器学习模型主要有两大类：遗传规划和神经网络。

先挖掘因子，再合成，忽略了因子之间的相互作用。

今天开始要代码实现的：一种新的因子组合挖掘框架，直接使用因子组合的表现来优化一个强化
学习因子生成器，最终生成的是一组公式因子集合，这些因子协同使用具有较高的选股效力。这
样做既能保留遗传规划算法公式化的优势，也能提升模型泛化能力，适应多种股票池，还能大幅
提升运算效率。

我对这篇论文感兴趣的核心在于，它具备一定程度上的通用性，集传统遗传规划的优点，可以显示生成表达式，结果了深度学习的泛化能力和端到端的能力。另外，原作者是提供代码的，不过它使用qlib的数据库，我进行了拆分，与咱们的开源项目，数据模块整合起来，让这个模块更加通用。

核心代码在Quantlab工程的如下位置：alphagen。

代码环境，需要：pytorch框架上的强化学习包：

stable_baselines3==2.0.0
sb3_contrib==2.0.0

核心的调用代码如下：

import json
import os
from datetime import datetime
from typing import Optional

import numpy as np
from sb3_contrib import MaskablePPO
from stable_baselines3.common.callbacks import BaseCallback

from alphagen.data.calculator import AlphaCalculator
from alphagen.models.alpha_pool import AlphaPool, AlphaPoolBase
from alphagen.rl.env.core import AlphaEnvCore
from alphagen.rl.env.wrapper import AlphaEnv
import torch

from alphagen.rl.policy import LSTMSharedNet
from alphagen.utils import reseed_everything
from alphagen.caculator.duckdb_caculator import DuckdbCalculator


def train(seed: int = 0, pool_capacity: int = 10, steps: int = 200_000, instruments: str = "csi300"):
    calculator_train = DuckdbCalculator(instrument=instruments,
                           start_time='2010-01-01',
                           end_time='2019-12-31')
    calculator_valid = DuckdbCalculator(instrument=instruments,
                           start_time='2020-01-01',
                           end_time='2020-12-31')
    calculator_test = DuckdbCalculator(instrument=instruments,
                          start_time='2021-01-01',
                          end_time='2022-12-31')

    pool = AlphaPool(
        capacity=pool_capacity,
        calculator=calculator_train,
        ic_lower_bound=None,
        l1_alpha=5e-3
    )

    reseed_everything(seed)

    device = torch.device('cuda:0')
    env = AlphaEnv(pool=pool, device=device, print_expr=True)

    name_prefix = f"new_{instruments}_{pool_capacity}_{seed}"
    timestamp = datetime.now().strftime('%Y%m%d%H%M%S')

    checkpoint_callback = CustomCallback(
        save_freq=10000,
        show_freq=10000,
        save_path='/path/for/checkpoints',
        valid_calculator=calculator_valid,
        test_calculator=calculator_test,
        name_prefix=name_prefix,
        timestamp=timestamp,
        verbose=1,
    )

    model = MaskablePPO(
        'MlpPolicy',
        env,
        policy_kwargs=dict(
            features_extractor_class=LSTMSharedNet,
            features_extractor_kwargs=dict(
                n_layers=2,
                d_model=128,
                dropout=0.1,
                device=device,
            ),
        ),
        gamma=1.,
        ent_coef=0.01,
        batch_size=128,
        tensorboard_log='/path/for/tb/log',
        device=device,
        verbose=1,
    )
    model.learn(
        total_timesteps=steps,
        callback=checkpoint_callback,
        tb_log_name=f'{name_prefix}_{timestamp}',
    )

其中DuckdbCaculator是我们实现的。

就是根据表达式计算因子，IC值等等。

使用了sb3的强化学习扩展包里的MaskablePPO算法。

01 金融日频数据入mongo

Quantlab的金融日线数据库一直是mongo。

我重写了Write_df函数：

重构前：

def write_df2(tb_name, df, db='datalake', drop_tb_if_exist=False):
    db = get_db()
    if drop_tb_if_exist:
        if tb_name in db.list_collection_names():
            db.drop_collection(tb_name)
    try:
        if len(df) == 0:
            print('长度为零')
        else:
            docs = df.to_dict(orient='records')
            get_db()[tb_name].insert_many(docs, ordered=False)
    except pymongo.errors.BulkWriteError as e:
        print(e)

重构后：

def write_df(tb_name, df, db='datalake', drop_tb_if_exist=False):
    db = get_db()
    if drop_tb_if_exist:
        if tb_name in db.list_collection_names():
            db.drop_collection(tb_name)
    try:
        if len(df) == 0:
            print('长度为零')
        else:
            docs = df.to_dict(orient='records')
            update_datas = []
            for doc in docs:
                update_datas.append(UpdateOne({'_id': doc['_id']}, {'$set': doc}, upsert=True))
            get_db()[tb_name].bulk_write(update_datas, ordered=False)

    except pymongo.errors.BulkWriteError as e:
        print(e)

这里的区别在于，如果_id字段存在，则跳过，还是更新。

重构后是更新。

akshare，我发现获取日线数据时，当前这一天，会获取盘中实时数据，如果写入后就不更新了，那么这个数据收盘后就是错的。

这样就完美解决了这个问题。

02 量化策略模板

本期重点实现：大类资产配置模板，轮动模板（含择时）

一个大类资产配置策略模板，只需要如下配置：（在algo的基础上，进行了封装，大家不必了解algo如何配置了）

@dataclass
class Task:
    name: str = ''
    desc: str = ''
    start_date: str = '20100101'
    end_date: str = None
    commission: int = 0.0001
    slippage: int = 0.0001
    benchmark: str = '000300.SH'
    symbols: list[str] = field(default_factory=list)
    template: str = '大类资产配置'  # 轮动，多策略组合

    # 权重
    weights: str = 'equally'  # or fixed
    weights_fixed: list[float] = field(default_factory=list)  # weights == fixed时有效

    # 调仓周期
    period: str = 'weekly'  # once, quarterly, monthly, yearly, days
    days: int = 5  # 如果period == days有效

根据策略模板可以自动化生成算子：

def _parse_period(self):
    module = importlib.import_module('engine.algos')
    if self.period == 'RunDays':
        algo_period = getattr(module, self.period)(self.days)
    else:
        if self.period in ['RunWeekly', 'RunOnce', 'RunMonthly', 'RunQuarterly', 'RunYearly']:
            algo_period = getattr(module, self.period)()
        else:
            algo_period = None
    return algo_period

def _parse_weights(self):
    module = importlib.import_module('engine.algos')
    if self.weights == 'WeightEqually':
        return WeightEqually()
    if self.weights == 'WeightFix':
        return WeightFix(self.weights_fixed)
    return None

def gen_algos(self):
    algos = []
    algo_period = self._parse_period()
    if algo_period:
        algos.append(algo_period)

    if self.template == '大类资产配置':
        algos.append(SelectAll())

    algo_weight = self._parse_weights()
    if algo_weight:
        algos.append(algo_weight)

    algos.append(Rebalance())
    return algos

这是一个可以在服务器上运行的版本，我去掉了多余的callback，提示之类的：

大家可以极简的配置一个策略：

from quantlab.engine.engine import Task

def hello_backtest():
    task = Task()
    task.name = '大类资产配置'
    task.symbols = ['510300.SH', '159915.SZ']
    task.template = '大类资产配置'
    task.weights = 'WeightEqually'
    task.period = 'RunMonthly'
    task.benchmark = '510300.SH'

还是那句话，投资并没有那么难，应该快乐而优雅。

我们做投资的目的，是为了更自由地去体验世界，去满足自己的好奇，而非投资本身。