Quantlab3.9.1:新增风险收益分析（附源码和数据）

咱们星球围绕一个核心：开发有效的可实盘的策略。

task的定义在：

engine/task.py中：需要的配置项很少：

name = "创业板动量择时"
desc = "创业板动量择时：在动量大的时候买入，动量小的时候卖出"
symbols = ['159915.SZ']
algo_period = "RunDaily"
algo_period_days = 20 #在RunPeriod时有效
rules_buy = ['roc_20>N1']
at_least_buy = 1
rules_sell = ['roc_20<N2']
at_least_sell = 1
order_by = ""
topK = 1
dropN = 0
b_ascending = 0
algo_weight = "WeightEqually"
algo_weight_fix = []
feature_names = ['roc_20']
features = ['roc(close,20)']
start_date = "20100101"
end_date = ""
commission = 0.0001
slippage = 0.0001
init_cash = 1000000
benchmark = "510300.SH"

回测结果如下：

创成长与红利低波动的智能beta策略（年化29.3%，最大回撤24%）：

全球大类资产轮动，年化26.1%：

再来创业板择时：

import importlib
from dataclasses import dataclass, field, asdict

from datafeed.dataloader import CSVDataloader
from engine.algos import *
import json

@dataclass
class Task:
    name: str = '策略名称'
    desc: str = '策略描述'

    # 标的池
    symbols: list[str] = field(default_factory=list)

    algo_period: str = 'RunDaily'
    algo_period_days: int = 20  # 仅当RunDays时需要

    # 规则选标的：当rules_buy或rules_sell至少有一条规则时，添加SelectBySignal算子， 在SelectAll之后
    rules_buy: list[str] = field(default_factory=list)  # roc(close,20)>0.08
    at_least_buy: int = 1
    rules_sell: list[str] = field(default_factory=list)  # roc(close,20)<0
    at_least_sell: int = 1

    # 排序算子：当order_by不为空时，在选股之后，添加SelectTopK算子
    order_by: str = ''  # 比如roc(close,20)，或者 roc(close,20)+ slope(close,20)
    topK: int = 1
    dropN: int = 0
    b_ascending: bool = 0  # 默认降序, 0=False

    # 仓位分配，默认等权
    algo_weight: str = 'WeightEqually'
    algo_weight_fix: list = field(default_factory=list)  # 当WeightFix时需要

    feature_names: list = field(default_factory=list)  # roc(close,20)
    features: list = field(default_factory=list)  # roc_20

    # 回测时用户可以改变的，不变就使用默认值，字符串为空表示不设置
    start_date: str = '20100101'
    end_date: str = ''
    commission: float = 0.0001
    slippage: float = 0.0001
    init_cash: int = 100 * 10000
    benchmark: str = '510300.SH'

    def __str__(self):
        return self.name

    def load_datas(self):
        logger.info('开始加载数据...')  #
        loader = CSVDataloader(DATA_DIR.joinpath('universe'), self.symbols, start_date=self.start_date,
                               end_date=self.end_date)
        df = loader.load(fields=self.features, names=self.feature_names)
        df['date'] = df.index
        df.dropna(inplace=True)
        return df

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

    def _parse_weights(self):
        module = importlib.import_module('engine.algos')
        if self.algo_weight == 'WeightEqually':
            return WeightEqually()
        if self.algo_weight == 'WeightFix':
            if len(self.symbols) != len(self.algo_weight_fix):
                logger.error('固定权重 != symbols数 ')
                return
            return WeightFix(self.algo_weight_fix)
        if self.algo_weight == 'WeightERC':
            return WeightERC()
        return None

    def get_algos(self):
        algos = []
        algos.append(self._parse_period())
        algos.append(SelectAll())
        if len(self.rules_buy) or len(self.rules_sell):
            algos.append(SelectBySignal(rules_buy=self.rules_buy,
                                        buy_at_least_count=self.at_least_buy,
                                        rules_sell=self.rules_sell,
                                        sell_at_least_count=self.at_least_sell
                                        ))

        if len(self.order_by):
            algos.append(SelectTopK(factor_name=self.order_by, K=self.topK, drop_top_n=self.dropN,
                                    b_ascending=self.b_ascending))

        algos.append(self._parse_weights())
        algos.append(Rebalance())
        return algos

    def to_toml(self, name):
        import toml
        from config import DATA_DIR
        with open(DATA_DIR.joinpath('tasks').joinpath(name + '.json'), "w", encoding='UTF-8') as f:
            toml.dump(asdict(self),f)

    def to_json(self, name):
        import json
        from config import DATA_DIR
        with open(DATA_DIR.joinpath('tasks').joinpath(name + '.json'), "w", encoding='UTF-8') as f:
            json.dump(asdict(self), f, ensure_ascii=False)


def task_from_json(name):
    with open(DATA_DIR.joinpath('tasks').joinpath(name), "r", encoding='UTF-8') as f:
        json_data = json.load(f)
        return Task(**json_data)


@dataclass
class TaskAssetsAllocation(Task):

    def get_algos(self):
        return [
            self._parse_period(),
            SelectAll(),
            self._parse_weights(),
            Rebalance()
        ]


@dataclass
class TaskRolling(Task):  # 轮动策略模板
    def get_algos(self):
        return [
            RunAlways(),
            SelectBySignal(rules_buy=self.rules_buy,
                           buy_at_least_count=self.at_least_buy,
                           rules_sell=self.rules_sell,
                           sell_at_least_count=self.at_least_sell
                           ),
            SelectTopK(factor_name=self.order_by, K=self.topK, drop_top_n=self.dropN,
                       b_ascending=self.b_ascending),
            self._parse_weights(),
            Rebalance()
        ]


@dataclass
class TaskRolling_Model(Task):  # 轮动策略模板
    def get_algos(self):
        return [
            RunAlways(),
            SelectBySignal(rules_buy=self.rules_buy,
                           buy_at_least_count=self.at_least_buy,
                           rules_sell=self.rules_sell,
                           sell_at_least_count=self.at_least_sell
                           ),
            SelectByModel(model_name=self.model_name, feature_names=self.feature_names),
            SelectTopK(factor_name=self.order_by, K=self.topK, drop_top_n=self.dropN,
                       b_ascending=self.b_ascending),
            self._parse_weights(),
            Rebalance()
        ]


@dataclass
class TaskPickTime(Task):  # 择时策略模板
    def get_algos(self):
        return [
            RunAlways(),
            SelectBySignal(rules_buy=self.rules_buy,
                           buy_at_least_count=self.at_least_buy,
                           rules_sell=self.rules_sell,
                           sell_at_least_count=self.at_least_sell
                           ),
            WeightEqually(),
            Rebalance()
        ]

吾日三省吾身

专注，一针顶破天。

有所为有所不为，有所必为。

能力圈与行动圈。

量化投资的本质还是投资。

投资没有圣杯，它是一个无限游戏。

投机就像赌场，为什么让很多人着迷，因为真的有人会赢，而且赢得很多。

塔勒布在《随机漫步的傻瓜》里讲的，地球70亿人，真就有那么几个幸运儿，他纯粹就是运气好。

但最怕这样的人出书立说，把他的经历当成经验让我们信以为真。

股神并不炒股，股神的公司80%的资产全资控股的子公司，合纵连横。可以为股神是经营了一个大集团公司，可以打通多个领域，多条产业链，相互扶持。——这个神话不可以复制。

普通人也确有人可以以交易为生，但冷暖自知。

个人曾经有一段不短的产品经理的职业经历，我把平台当产品来经营的——并不刻意所谓知识付费。

产品最重要的事：为谁解决什么问题？

几类用户：

1、使用者：不会代码，查学理财，甚至会一点量化投资，可以订阅别人的策略。————获得策略，理解策略，创建策略，分享策略。

2、策略或组合主理人。通过分享策略参数赚钱。

01 量化新手第一步

除了数据之外，量化新手第一步应该做什么呢？——时间序列的风险、收益分析。

无论是什么投资标的，无外乎就是时间序列，你按日频也好，分钟也罢都是如此。在时间序列之上，传统的技术分析就是用这些价量数据的统计关键，形态分析构建无数的规则和因子。

无论是股票，指数，ETF，可转债还是期货，都可以搁到这里进行分析，上图，我对沪深300，中证500以及创业板指数，做了历史风险收益分析，以及互相的相关系数计算。

可以看到，小市值风险高、长期预期收益也高，与我们的认知一致。

import pandas as pd
import streamlit as st
from config import DATA_DIR

from datafeed.data_builder import get_basic_list
from engine.performance import PerformanceUtils

@st.cache_data
def load_basic_list():
    return get_basic_list()


def func_basic(x):
    item = load_basic_list()[x]
    return '{}|{}'.format(item['name'], item['symbol'])


def build_page():
    st.write('时间序列风险，收益分析')
    symbols = st.multiselect(label='请选择投资标的', default=['000300.SH'], options=list(load_basic_list().keys()),
                             format_func=lambda x: func_basic(x))
    cols = []
    for s in symbols:
        df = pd.read_csv(DATA_DIR.joinpath('quotes').joinpath('{}.csv'.format(s)))
        df['date'] = df['date'].apply(lambda x:str(x))
        df.index = df['date']
        se = df['close']
        se.name = s
        cols.append(se)
    print(cols)
    all = pd.concat(cols, axis=1)

    all.sort_values(by='date', ascending=True,inplace=True)
    all.dropna(inplace=True)
    equity = (all.pct_change()+1).cumprod()
    print(equity)

    st.line_chart(equity)

    df_ratio,df_corr = PerformanceUtils().calc_equity(df_equity=equity)

    col1, col2 = st.columns(2)
    with col1:
        st.write(df_ratio)
    with col2:
        st.write(df_corr)