三步获得一个年化22%的动量策略，附python代码

import importlibfrom dataclasses import dataclass, asdictfrom typing import List, Dict
import backtrader as btimport numpy as npimport pandas as pd

from backtrader_strategy import StrategyTemplatefrom backtrader_algos import *


from matplotlib import rcParamsfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom collections import defaultdict
rcParams['font.family'] = 'SimHei'

@dataclassclass Task:    name: str = '策略'    symbols: List[str] = field(default_factory=list)
    start_date: str = '20100101'    end_date: str = datetime.now().strftime('%Y%m%d')
    benchmark: str = '510300.SH'    select: str = 'SelectAll'
    select_buy: List[str] = field(default_factory=list)    buy_at_least_count: int = 0    select_sell: List[str] = field(default_factory=list)    sell_at_least_count: int = 1
    order_by_signal: str = ''    order_by_topK: int = 1    order_by_dropN: int = 0    order_by_DESC: bool = True  # 默认从大至小排序
    weight: str = 'WeightEqually'    weight_fixed: Dict[str, int] = field(default_factory=dict)


    
    period: str = 'RunDaily'    period_days: int = None

@dataclassclass StrategyConfig:    name: str = '策略'    desc: str = '策略描述'    config_json: Dict[str, int] = field(default_factory=dict)    author: str = ''

class AlgoStrategy(StrategyTemplate):    #params = ('$P1',25)
    def __init__(self, algo_list):        super(AlgoStrategy, self).__init__()        self.algos = algo_list
    def prenext(self):        #return        self.next()
    def next(self):        #print(f"next - 当前日期: {self.datetime.date(0)}")        self.temp = {}
        for algo in self.algos:            if algo(self) is False:  # 如果algo返回False,直接不运行                return


from datafeed.csv_dataloader import CsvDataLoaderfrom datafeed.factor_expr import FactorExprclass DataFeed:    def __init__(self, task: Task):        dfs = CsvDataLoader().read_dfs(symbols=task.symbols,start_date=task.start_date, end_date=task.end_date)
        fields = list(set(task.select_buy + task.select_sell))        if task.order_by_signal:            fields += [task.order_by_signal]        names = fields        df_all = FactorExpr().calc_formulas(dfs,fields)        self.df_all = df_all

    def get_factor_df(self, col):        df_factor = self.df_all.pivot_table(values=col, index=self.df_all.index, columns='symbol')        if col == 'close':            df_factor = df_factor.ffill()        return df_factor

class Engine:    def __init__(self, path='quotes'):        self.path = path        self._init_engine()

    def _parse_rules(self, task: Task):
        def _rules(rules, at_least):            if not rules or len(rules) == 0:                return None
            all = None            for r in rules:                if r == '':                    continue
                df_r = self.datafeed.get_factor_df(r)                if df_r is not None:                    df_r = df_r.replace({True: 1, False: 0})                    df_r = df_r.astype('Int64')
                    #print(df_r)                    #df_r = df_r.astype(int)                #else:                    #print(r)                if all is None:                    all = df_r                else:                    all += df_r            return all >= at_least
        buy_at_least_count = task.buy_at_least_count        if


    
 buy_at_least_count <= 0:            buy_at_least_count = len(task.select_buy)
        all_buy = _rules(task.select_buy, at_least=buy_at_least_count)        all_sell = _rules(task.select_sell, task.sell_at_least_count)  # 卖出 求或，满足一个即卖出
        if all_sell is not None:            all_sell = all_sell.fillna(True)        if all_buy is not None:            all_buy = all_buy.fillna(False)        return all_buy, all_sell
    def _get_algos(self, task: Task):
        bt_algos = importlib.import_module('backtrader_algos')
        if task.period == 'RunEveryNPeriods':            algo_period = bt.algos.RunEveryNPeriods(n=task.period_days, run_on_last_date=True)        else:            algo_period = getattr(bt_algos,task.period)()
        algo_select_where = None        # 信号规则        signal_buy, signal_sell = self._parse_rules(task)        if signal_buy is not None or signal_sell is not None:  # 至少一个不为None            df_close = self.datafeed.get_factor_df('close')            if signal_buy is None:                select_signal = np.ones(df_close.shape)  # 注意这里是全1，没有select_buy就是全选                select_signal = pd.DataFrame(select_signal, columns=df_close.columns, index=df_close.index)            else:                select_signal = np.where(signal_buy, 1, np.nan)  # 有select_buy的话，就是买入，否则选置Nan表示 hold状态不变            if signal_sell is not None:                select_signal = np.where(signal_sell, 0, select_signal)  # select_sell置为0，就是清仓或不选            select_signal = pd.DataFrame(select_signal, index=df_close.index, columns=df_close.columns)            select_signal.ffill(inplace=True)  # 这一句非常关键，ffill就是前向填充，保持持仓状态不变。即不符合buy，也不符合sell，保持不变。            select_signal.fillna(0, inplace=True)            algo_select_where = SelectWhere(signal=select_signal)
        # 排序因子        algo_order_by = None        if task.order_by_signal:            signal_order_by = self.datafeed.get_factor_df(col=task.order_by_signal)            algo_order_by = SelectTopK(signal=signal_order_by, K=task.order_by_topK, drop_top_n=task.order_by_dropN,                                       b_ascending=task.order_by_DESC==False)
        algos = []        algos.append(algo_period)
        if algo_select_where:            algos.append(algo_select_where)        else:            algos.append(SelectAll())
        if algo_order_by:            algos.append(algo_order_by)

        algo_weight = WeightEqually()        if task.weight == 'WeightFix':            algo_weight =  WeightFix(weights_dict=task.weight_fixed)
        algos.append(algo_weight)
        force_update=False        if task.weight == 'WeightFix':            force_update = True        algos.append(ReBalance(force_update))        return algos

    def _init_engine(self):        cerebro = bt.Cerebro()        cerebro.broker.setcash(1000000.0)  # 设置初始资金        # cerebro.broker.setcommission(0.0005)        # 添加PyFolio分析器        cerebro.addanalyzer(bt.analyzers.PyFolio, _name='_PyFolio')
        cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')        cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')        cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')        cerebro.broker.set_coc(True)  # 设置Cheat-On-Close，确保在收盘时执行订单        self.cerebro = cerebro
    def _prepare_run(self, symbols, start_date, end_date, commissions=0.0):
        dfs = CsvDataLoader().read_dfs(symbols)        self.cerebro.broker.setcommission(commissions)


    
        for s, data in dfs.items():            data['openinterest'] = 0            data.set_index('date', inplace=True)            data.index = pd.to_datetime(data.index)            data.sort_index(ascending=True, inplace=True)            data = data[                (data.index >= pd.to_datetime(start_date)) &                (data.index <= pd.to_datetime(end_date))                ]
            data = bt.feeds.PandasData(                dataname=data,                fromdate=pd.to_datetime(start_date),                todate=pd.to_datetime(end_date),                timeframe=bt.TimeFrame.Days,                name=s,            )            self.cerebro.adddata(data)

    def run_strategy(self, strategy, symbols,start_date='20101001', end_date=datetime.now().strftime('%Y%m%d'),*args,**kwargs):        self._prepare_run(symbols,start_date, end_date)        self.cerebro.addstrategy(strategy,*args,**kwargs)        self.results = self.cerebro.run()        portfolio_stats = self.results[0].analyzers.getbyname('_PyFolio')        returns, positions, transactions, _ = portfolio_stats.get_pf_items()        returns.index = returns.index.tz_convert(None)
        #equity = (1 + returns).cumprod()        self.perf = (1 + returns).cumprod().calc_stats()        # equity.plot()        # import matplotlib.pyplot as plt        # plt.show()

    def run(self, task: Task, commissions=0.0):        self._prepare_run(task.symbols,task.start_date,task.end_date, commissions)        self.datafeed = DataFeed(task)
        self.cerebro.addstrategy(AlgoStrategy, algo_list=self._get_algos(task))        self.results = self.cerebro.run()
        portfolio_stats = self.results[0].analyzers.getbyname('_PyFolio')        returns, positions, transactions, _ = portfolio_stats.get_pf_items()        returns.index = returns.index.tz_convert(None)        # import empyrical        # ret = empyrical.annual_return(returns)        # print('年化收益',ret)
        returns.name = '策略'

        equity = pd.DataFrame((1 + returns).cumprod())        import ffn        #print('长度：',len(equity))        #equity.calc_stats().display()        datas = [equity]        for bench in [task.benchmark]:            df = CsvDataLoader().read_df([bench],start_date=task.start_date, end_date=task.end_date)            df.set_index('date',inplace=True)            df.index = pd.to_datetime(df.index)            data = df.pivot_table(values='close', index=df.index, columns='symbol')            data.columns = ['benchmark']            datas.append(data)

        all_returns = pd.concat(datas, axis=1).pct_change()        all_returns.dropna(inplace=True)
        self.perf = (1 + all_returns).cumprod().calc_stats()        return self.results
    def opt(self, strategy,symbols,start_date='20101001', end_date=datetime.now().strftime('%Y%m%d'),*args,**kwargs):        self._prepare_run(symbols, start_date, end_date)        self.cerebro.optstrategy(            strategy,            period=[5,10,15,20,25,30]        )
        # 打印结果        def get_my_analyzer(result):            analyzer = {}            # 返回参数            analyzer['period'] = result.params.period            #analyzer['period2'] = result.params.lower
            pyfolio = result.analyzers.getbyname('_PyFolio')            returns, positions, transactions, gross_lev = pyfolio.get_pf_items()            import empyrical as em            analyzer['年化收益率'] = em.annual_return(returns)            return analyzer


    

        self.results = self.cerebro.run(stdstats=False)        ret = []        for i in self.results:            print(i)            ret.append(get_my_analyzer(i[0]))
        df = pd.DataFrame(ret)        print(df)


    def stats(self):
        print(self.perf.display())
    def plot(self):        self.perf.plot()        import matplotlib.pyplot as plt        # 设置字体为 SimHei（黑体）        plt.rcParams['font.sans-serif'] = ['SimHei']        # 解决坐标轴负号显示问题        plt.rcParams['axes.unicode_minus'] = False        plt.show()

import requests, json
if __name__ == '__main__':    t = Task()    t.name = '全球大类资产-修正斜率轮动'    etfs = [        '510300.SH',  # 沪深300ETF        '159915.SZ',  # 创业板        '518880.SH',  # 黄金ETF        '513100.SH',  # 纳指ETF        '159985.SZ',  # 豆柏ETF        '511880.SH',  # 银华日利ETF    ]
    t.symbols = etfs
    #t.select_sell = ["roc(close,21)>0.17"]    t.order_by_signal =  "slope(close,25)"
    e = Engine()    e.run(t)    e.stats()    e.plot()