量化策略绩效在mongo在存储与查询方案

我在想，那些量化平台里供人付费订阅的策略，不能说全部，而是说有一大部分，是这样运作的。

好比，每年年终，都可以评出一个批优质的主动型基金经理，这些基金，无一例外，成绩斐然，吸引大量投资进来。因为业绩实在太好看了，然后基金经理们事后归因，高瞻远瞩。

优秀基金经理，人家那履历，那经历，那背景，那投研能力，那团队，还算可以理解。但个人投资者，通过配置几个量化指标？

有几种情况是可能的，大类资产配置，趋势交易与轮动，跑赢大盘是没有问题的。

单一股票市场就是“赌风格”，对了，某一段时间的收益益会非常高。

比如我看到一个打榜的小市值：

长期跑输大盘，然后在某8月、9月，突然“押中”某支股票，拉升了收益率。看后面的表现不得而知，目前看在“横盘”。

关于量化，我们一直在说一个悖论：如果投资能赚大钱，为何还要做策略？自己闷头发财不好？

我自己的观点一直是：

投资理财是普通人复利之最佳杠杆。10%的长期年化很容易。我们做工具和策略，是提供你这方向的知识与概率。注意：仍然是概率。

我想做一些长期的事情。

与我之前不相信K线就可以投资股票一样，搞几个因子我同样不相信。这不符合常识与逻辑。

做一件事，最怕底层逻辑不通，瞎折腾。

一个好的组合策略，应当是熊市不大亏，牛市跟得上。

继续代码，把回测完成后，equities写回mongo备查：

@asset(description='可视化equities', group_name='task')
def visual_html():
    tasks = ['5a3907af-7852-11ee-aeb4-c858c0412ade']
    for id in tasks:
        items = mongo_utils.get_db()['tasks_equities'].find({"id": id}, {'_id': 0, '策略': 1, '基准': 1, 'date': 1})
        df = pd.DataFrame(list(items))

        df = df.iloc[-252:]
        html = visual_utils.build_linechart_html(df,'回测曲线', cols=['策略','基准'], width=400, height=260)
        mongo_utils.write_file_data('tasks_html',html, {'_id': id, 'encoding': 'utf8'})

        html = mongo_utils.read_file_data('tasks_html', id)
        print(html)

今天来一个比昨天还简单的策略，创业板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这样的框架。

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()})