一、说明

老样子，又到了喜闻乐见的程序整理环节，我们今天会完成下列事项

• 资料拆分：以1:1比例分成训练集与验证集。
• 报表显示：包括年化报酬率、最大跌幅等日後分析与视觉化时需要的资料。
• 参数优化

训练器

主要作用为策略回测，策略参数优化，回测完毕後回传该股票的获利与风险报表。

def bt_trainer(stock_index, test_strategy, opt=False, opt_params=None):
    print(stock_index)

    train_df = load_stock(stock_index, start_year=2012, end_year=2015)
    if len(train_df) == 0:
        return pd.DataFrame()

    valid_df = load_stock(stock_index, start_year=2016, end_year=2019)
    if len(valid_df) == 0:
        return pd.DataFrame()

    train_holding_cash = (
        1000
        * train_df.groupby(pd.DatetimeIndex(train_df.index).to_period("M"))
        .nth(0)["Close"]
        .sum()
    )

    valid_holding_cash = (
        1000
        * valid_df.groupby(pd.DatetimeIndex(valid_df.index).to_period("M"))
        .nth(0)["Close"]
        .sum()
    )

    train_bt = Backtest(
        train_df,
        test_strategy,
        cash=train_holding_cash,
        commission=0.004,
        # exclusive_orders=True,
        trade_on_close=True,
    )

    valid_bt = Backtest(
        valid_df,
        test_strategy,
        cash=valid_holding_cash,
        commission=0.004,
        # exclusive_orders=True,
        trade_on_close=True,
    )

    if opt:
        train_result = train_bt.optimize(**opt_params)

        valid_result = valid_bt.run(**train_result["_strategy"].params)
    else:
        train_result = train_bt.run()
        valid_result = valid_bt.run()

    result = train_result.copy()
    result["stock_id"] = stock_index
    result["Strategy"] = test_strategy.__name__
    result["Params"] = result["_strategy"].params
    result["Train Return (Ann.) [%]"] = train_result["Return (Ann.) [%]"]
    result["Train Max. Drawdown [%]"] = train_result["Max. Drawdown [%]"]
    result["Train Sharpe Ratio"] = train_result["Sharpe Ratio"]
    result["Train SQN"] = train_result["SQN"]
    result["Valid Return (Ann.) [%]"] = valid_result["Return (Ann.) [%]"]
    result["Valid Max. Drawdown [%]"] = valid_result["Max. Drawdown [%]"]
    result["Valid Sharpe Ratio"] = valid_result["Sharpe Ratio"]
    result["Valid SQN"] = valid_result["SQN"]
    # result["Backtest Train"] = train_bt
    # result["Backtest Valid"] = valid_bt

    opt_str = "opt" if opt else "std"
    plot_dir = f"./data/Strategy/{test_strategy.__name__}/{opt_str}"
    Path(plot_dir).mkdir(parents=True, exist_ok=True)
    valid_bt.plot(
        filename=plot_dir + f"/{stock_index}.html",
        open_browser=False,
    )

    df = result.to_frame().transpose()

    return df

报表整理

def report_generate(frames, test_strategy, opt):
    final_df = pd.concat(frames).reset_index(drop=True)

    final_df = final_df.loc[
        final_df["Duration"] == final_df["Duration"].max()
    ].reset_index(drop=True)

    stock_name = pd.read_csv(
        "data/stock_id.csv", dtype={"stock_id": str, "stock_name": str}
    )

    profit_df = final_df.merge(stock_name, on="stock_id")

    report_df = profit_df[
        [
            "stock_id",
            "stock_name",
            "Train Return (Ann.) [%]",
            "Train Max. Drawdown [%]",
            "Train Sharpe Ratio",
            "Train SQN",
            "Valid Return (Ann.) [%]",
            "Valid Max. Drawdown [%]",
            "Valid Sharpe Ratio",
            "Valid SQN",
        ]
    ]

    opt_str = "opt" if opt else "std"

    report_dir = f"./data/Strategy/{test_strategy.__name__}/{opt_str}"
    Path(report_dir).mkdir(parents=True, exist_ok=True)

    report_df.to_csv(
        report_dir + "/Report.csv",
        index=False,
        header=True,
    )

    report_df = pd.read_csv(report_dir + "/Report.csv", dtype={"stock_id": str})

    return report_df

• 完整程序码

二、结语

回测的部分差不多结束了，明天会测试几个常见的几种策略。