Quantlab v3.9.2:策略集合

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

代码更新（更新策略集合）

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