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

关于投资，要到大道至简，很轻松愉悦。

要么精细，高频、深度学习。

我不太建议，尤其是新同学，所谓的自动化的主动策略，然后看似自动化的实盘。

交易本身就是要么赚，要么亏。（50%）。你觉得自己时而行，进而不行，长期看，大概率不行。

大类资产，风险平价是入门首选。

要么整高科技，高频，深度学习。

我这段时间做的是前者，我认为对大家更有帮助。

01 量化策略在服务器的存储

量化策略的表达，是一个字典或json，当然它包含嵌套的数据结构。如果要通过关系数据库如mysql去存储。至少得设计好几个关联表。

但我们其实就是存储一个数据结构，nosql就非常合适。

from quantlab.engine.engine import Task
import uuid
from dataclasses import asdict
from quanttask.utils import mongo_utils

def task_portfolio():
    task = Task()
    task._id = '5a3907af-7852-11ee-aeb4-c858c0412ade'
    task.name = '大类资产配置'
    task.desc = '沪深300指数与创业板指数，等权的配置'

    task.symbols = ['510300.SH', '159915.SZ']
    task.template = '大类资产配置'
    task.weights = 'WeightEqually'
    task.period = 'RunMonthly'
    task.benchmark = '510300.SH'
    print(asdict(task))
    return task


if __name__ == '__main__':
    import pandas as pd
    tasks = [task_portfolio()]
    df = pd.DataFrame(tasks)

    mongo_utils.write_df('tasks', df )

同样，回测结果直接保存回mongo供调用：

02 策略在网页上的呈现

mongo也是支持跨表联合查询的：

def strategies(request):
    nav_flag = 888
    #items = list(mongo_utils.get_db()['tasks_performance'].find({}))

    items = mongo_utils.get_db()['tasks_performance'].aggregate(
        [
            {
                '$lookup':
                    {
                        "from": "tasks",  # 需要联合查询的另一张表B
                        "localField": "_id",  # 表A的字段
                        "foreignField": "_id",  # 表B的字段
                        "as": "task_docs"  # 根据A、B联合生成的新字段名
                    },
            },
            {
                '$project':  # 联合查询后需要显示哪些字段，1：显示
                    {
                        'task_docs.name': 1,
                        'task_docs.desc': 1,
                        'task_docs.author': 1,
                        'CAGR':1,
                        '最大回撤':1,
                        'benchmark':1,
                        '_id': 0,
                    },
            },
            {
                '$match':  # 根据哪些条件进行查询
                    {

                    }
            }
        ]
    )

    items =  list(items)
    print(items)
    tasks = []
    for item in items:
        task = item.copy()
        task['bechmark_CARG'] = task['benchmark'][0]['CAGR']
        task['author'] = task['task_docs'][0]['author']
        task['name'] = task['task_docs'][0]['name']
        task['desc'] = task['task_docs'][0]['desc']
        tasks.append(task)

    items = tasks
    return render(request, 'quant/mall.html', {**locals()})

最终呈现如下的样子如下：

后续可以继续美化和优化。

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

每天给大家写AI量化代码的星球，目前有一个自研的AI量化框架，算子化，积木式。整合多因子，因子挖掘，深度学习，强化学习等。

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

前面的文章开了个头：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'