大约 7 分钟
part13 实操实验：量化回测
一、实验目标
- 掌握基于历史行情的回测流程。
- 掌握“信号日”和“成交日”分离的写法。
- 掌握三种常见策略的比较方法。
二、实验数据与环境
仍然使用 个股行情.csv。本实验主要用到：
代码日期开盘价(元)最高价(元)最低价(元)收盘价(元)涨跌幅(%)
先复用 part12.md 中的统一准备代码，保证下面这些函数可用：
load_stock_data()get_stock()calc_fee()max_affordable_shares()
任务一：单股三策略回测
任务要求
以 600657.SH 为样本，完成以下内容：
- 高抛低吸策略回测。
- 均值回归策略回测。
- 趋势跟踪策略回测。
- 生成三种策略的对比表。
- 绘制策略对比图。
代码框架
需要补充TODO区域的核心买卖代码
# 导入绘图和数据处理库。
# 这里沿用前面章节已经熟悉的 pandas / matplotlib 写法。
import matplotlib.pyplot as plt
import pandas as pd
# =========================
# 1. 读取样本数据
# =========================
# 只保留回测最需要的列：
# - 日期：用于保证时间顺序正确
# - 开盘价：作为“今天成交价”
# - 收盘价：作为判断信号和计算期末资产的依据
# - 涨跌幅：作为高抛低吸策略的信号来源
stock = get_stock("600657.SH")[
["日期", "开盘价(元)", "收盘价(元)", "涨跌幅(%)"]
].dropna().copy()
# 重新编号，方便后面按位置逐行读取
stock = stock.reset_index(drop=True)
# =========================
# 2. 统一汇总函数
# =========================
# 把三种策略最后都要输出的指标统一封装起来。
# 这样后面比较时，字段名和计算口径都保持一致。
def build_summary(strategy_name, initial_cash, cash, shares, last_close, trade_df, total_fee):
# 期末总资产 = 现金 + 持股数量 × 最后一个交易日收盘价
final_asset = cash + shares * last_close
return {
"策略": strategy_name,
"初始资金": round(initial_cash, 2),
"期末现金": round(cash, 2),
"期末持股": int(shares),
"期末收盘价": round(last_close, 2),
"期末总资产": round(final_asset, 2),
"总收益率(%)": round((final_asset / initial_cash - 1) * 100, 2),
"交易次数": len(trade_df),
"总手续费": round(total_fee, 2),
}
# =========================
# 3. 高抛低吸策略
# =========================
def backtest_reversal(stock, initial_cash=100000.0, trade_size=100, buy_trigger=-5, sell_trigger=5):
# 初始状态：现金充足、没有持股、没有手续费、没有交易记录
cash = initial_cash
shares = 0
total_fee = 0.0
records = []
# 从第 2 天开始循环，因为第 1 天没有“昨天”的涨跌幅可以参考
for i in range(1, len(stock)):
# 昨天的涨跌幅作为今天的交易信号
signal_return = stock.loc[i - 1, "涨跌幅(%)"]
# 今天真正成交的日期和价格
trade_date = stock.loc[i, "日期"]
trade_price = stock.loc[i, "开盘价(元)"]
# TODO:
# 1. 昨天跌幅达到买入阈值时，尝试买入固定股数
# 2. 昨天涨幅达到卖出阈值时，尝试卖出固定股数
# 3. 每次交易都要计算手续费
# 4. 记录交易明细
# 买入时先判断现金是否足够覆盖“成交额 + 手续费”
# 卖出时还要确认当前持股是否足够卖出固定数量
# 交易记录建议保存：日期、方向、数量、价格、手续费、交易后现金、剩余持股
# 将交易记录整理成表格，便于打印和保存
trade_df = pd.DataFrame(records, columns=["日期", "操作", "数量(股)", "成交价", "手续费", "交易后现金", "剩余持股"])
# 汇总本策略的核心结果
summary = build_summary("高抛低吸", initial_cash, cash, shares, stock["收盘价(元)"].iloc[-1], trade_df, total_fee)
return trade_df, summary
# =========================
# 4. 均值回归策略
# =========================
def backtest_mean_reversion(stock, initial_cash=100000.0, train_ratio=0.6, min_train=60, band_ratio=1/3):
# 先划分训练集，再用训练集估计“正常价格水平”
split = max(int(len(stock) * train_ratio), min_train)
if split >= len(stock) - 1:
raise ValueError("样本太短，无法划分训练集和回测集。")
# 训练集只负责统计，不参与交易
train = stock.iloc[:split].copy()
mean_price = train["收盘价(元)"].mean()
std_price = train["收盘价(元)"].std()
# 用均值上下各一个标准差的一部分，构造买入线和卖出线
buy_line = mean_price - std_price * band_ratio
sell_line = mean_price + std_price * band_ratio
# 回测阶段重新初始化账户
cash = initial_cash
shares = 0
total_fee = 0.0
records = []
# 只从训练集结束后的下一天开始交易
for i in range(split + 1, len(stock)):
# 昨天的收盘价用于判断是否“偏离均值”
yesterday_close = stock.loc[i - 1, "收盘价(元)"]
trade_date = stock.loc[i, "日期"]
open_price = stock.loc[i, "开盘价(元)"]
# TODO:
# 1. 空仓且昨天收盘价低于买入线时，买入尽可能多的整手股票
# 2. 持仓且昨天收盘价高于卖出线时，全部卖出
# 3. 记录交易明细
# 这里的关键是：先用昨天判断方向，再用今天开盘价成交，
# 这样才符合“信号先出现，交易后执行”的回测逻辑。
# 输出交易明细和汇总结果
trade_df = pd.DataFrame(records, columns=["日期", "操作", "数量(股)", "成交价", "手续费", "交易后现金", "剩余持股"])
summary = build_summary("均值回归", initial_cash, cash, shares, stock["收盘价(元)"].iloc[-1], trade_df, total_fee)
# 把训练集统计出来的均值和阈值也放进结果里，便于解释策略
summary.update({"训练集均值": round(mean_price, 2), "买入线": round(buy_line, 2), "卖出线": round(sell_line, 2)})
return trade_df, summary
# =========================
# 5. 趋势跟踪策略
# =========================
def backtest_trend(stock, initial_cash=100000.0, breakout_window=10, exit_window=5):
# 复制一份数据，避免直接修改原表
df = stock.copy()
# 用 shift(1) 避免把“今天自己”算进参考窗口，防止未来函数错误
df["前10日最高收盘价"] = df["收盘价(元)"].rolling(breakout_window).max().shift(1)
df["前5日最低收盘价"] = df["收盘价(元)"].rolling(exit_window).min().shift(1)
# 回测账户初始化
cash = initial_cash
shares = 0
# 记录买入时的成交价和手续费，卖出时要用它计算单笔盈亏
entry_price = None
entry_fee = 0.0
total_fee = 0.0
win_count = 0
loss_count = 0
records = []
# 逐日扫描行情，寻找突破和破位信号
for i in range(1, len(df)):
yesterday = df.loc[i - 1]
today = df.loc[i]
# 前几天的数据不足时，滚动窗口还没形成，直接跳过
if pd.isna(yesterday["前10日最高收盘价"]) or pd.isna(yesterday["前5日最低收盘价"]):
continue
# TODO:
# 1. 空仓且昨天收盘价突破前10日高点时，今天开盘全仓买入
# 2. 持仓且昨天收盘价跌破前5日低点时，今天开盘全部卖出
# 3. 卖出后统计盈利笔数和亏损笔数
# 这个策略的核心是顺势而为：
# 突破说明可能进入强势阶段，跌破说明趋势可能转弱。
# 整理交易记录和汇总数据
trade_df = pd.DataFrame(records, columns=["日期", "操作", "数量(股)", "成交价", "手续费", "交易后现金", "剩余持股"])
summary = build_summary("趋势跟踪", initial_cash, cash, shares, df["收盘价(元)"].iloc[-1], trade_df, total_fee)
# 记录平仓后的胜负情况，便于观察趋势策略是否依赖少数大波段
summary.update({"已平仓盈利笔数": win_count, "已平仓亏损笔数": loss_count})
return trade_df, summary
# =========================
# 6. 运行比较
# =========================
# 依次运行三种策略，得到交易明细和汇总指标
reversal_trade, reversal_summary = backtest_reversal(stock)
mr_trade, mr_summary = backtest_mean_reversion(stock)
trend_trade, trend_summary = backtest_trend(stock)
# 把三种策略的汇总结果合并成一张对比表
compare = pd.DataFrame([reversal_summary, mr_summary, trend_summary]).set_index("策略")
# 只展示最关键的对比指标
print(compare[["期末总资产", "总收益率(%)", "交易次数", "总手续费", "期末持股"]])
# 左图比较收益率，右图比较交易次数
fig, axes = plt.subplots(1, 2, figsize=(12, 4))
axes[0].bar(compare.index, compare["总收益率(%)"], color=["#4C78A8", "#F58518", "#54A24B"])
axes[0].set_title("三种策略总收益率比较")
axes[0].set_ylabel("总收益率(%)")
axes[0].grid(axis="y", alpha=0.3)
axes[1].bar(compare.index, compare["交易次数"], color=["#4C78A8", "#F58518", "#54A24B"])
axes[1].set_title("三种策略交易次数比较")
axes[1].set_ylabel("交易次数")
axes[1].grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.show()
# 保存结果，方便后面写实验报告时直接引用
compare.to_csv("part13_lesson1_compare.csv", encoding="utf-8-sig")结果整理
请写出一段 120~180 字的结论，至少回答下面 3 个问题：
- 哪种策略的期末总资产更高。
- 哪种策略的交易更频繁。
- 哪种策略的手续费压力更大。
三、提交内容
- 任务一代码
- 策略对比表
- 实验结论