大约 16 分钟
6 量化回测模拟
前言
本节重点
从这里开始，我们不只是看图、做统计，而是把想法写成明确的买卖规则。
但一定记住一句话：
回测不是为了证明自己一定能赚钱，而是为了提前发现规则的问题。
6.1 高抛低吸策略回测
前言
场景与知识
场景题目
如果昨天跌幅超过 5%，今天开盘买入 100 股；如果昨天涨幅超过 5%，今天开盘卖出 100 股。用 600339.SH 做一次最基础回测。
先抓关键词
- 这是一个“反转型”思路：跌多了买，涨多了卖。
- 交易信号来自昨天，实际成交发生在今天开盘。
- 一次只交易固定数量，更适合初学者理解。
金融知识补充
高抛低吸背后的假设是：
- 价格短期涨太多，可能回落。
- 价格短期跌太多，可能反弹。
这个想法在震荡市里可能有机会，但在单边趋势行情里容易失效：
- 如果一直跌，你可能会越买越套。
- 如果一直涨，你可能会过早卖掉强势仓位。
代码
# 读取一只股票的数据，并保留回测需要的列
stock = get_stock("600339.SH")[
["日期", "开盘价(元)", "收盘价(元)", "涨跌幅(%)"]
].dropna().copy()
# 添加这一行重置索引
stock = stock.reset_index(drop=True)
# 设定初始账户状态
initial_cash = 100000.0
cash = initial_cash
shares = 0
# 每次固定买卖 100 股
trade_size = 100
# 交易触发条件：
# 昨天跌幅 <= -5%，今天开盘买入
# 昨天涨幅 >= 5%，今天开盘卖出
buy_trigger = -5
sell_trigger = 5
# 统计手续费和交易记录
total_fee = 0.0
records = []
# 从第 2 行开始循环，因为每次交易都要参考“昨天”的涨跌幅
for i in range(1, len(stock)):
# 交易信号来自昨天
signal_return = stock.loc[i - 1, "涨跌幅(%)"]
# 真正成交发生在今天开盘
trade_date = stock.loc[i, "日期"]
trade_price = stock.loc[i, "开盘价(元)"]
if signal_return <= buy_trigger:
# 买入时，先算本次成交金额和手续费
turnover = trade_price * trade_size
fee = calc_fee(turnover)
# 现金足够时才允许买入
if cash >= turnover + fee:
cash -= turnover + fee
shares += trade_size
total_fee += fee
records.append(
[trade_date, "买入", trade_size, trade_price, fee, round(cash, 2), shares]
)
elif signal_return >= sell_trigger and shares >= trade_size:
# 卖出时，同样先算成交金额和手续费
turnover = trade_price * trade_size
fee = calc_fee(turnover)
cash += turnover - fee
shares -= trade_size
total_fee += fee
records.append(
[trade_date, "卖出", trade_size, trade_price, fee, round(cash, 2), shares]
)
# 把买卖记录整理成表格，便于查看
trade_df = pd.DataFrame(
records,
columns=["日期", "操作", "数量(股)", "成交价", "手续费", "交易后现金", "剩余持股"],
)
# 期末总资产 = 现金 + 持股市值
last_close = stock["收盘价(元)"].iloc[-1]
final_asset = cash + shares * last_close
# 汇总输出几个最重要的回测指标
summary = pd.Series(
{
"初始资金": round(initial_cash, 2),
"期末现金": round(cash, 2),
"期末持股": int(shares),
"期末收盘价": round(last_close, 2),
"期末总资产": round(final_asset, 2),
"总收益率": f"{(final_asset / initial_cash - 1):.2%}",
"交易次数": len(trade_df),
"总手续费": round(total_fee, 2),
}
)
# 如果没有任何交易，就打印提示；否则打印前几笔和后几笔记录
if trade_df.empty:
print("样本期内没有触发交易信号。")
else:
print(trade_df.head())
print(trade_df.tail())
trade_df.to_csv("trade_records.csv", index=False, encoding="utf-8-sig")
print("交易记录已保存到 trade_records.csv")
# 最后打印回测总览
print(summary)结果解读
日期 操作 数量(股) 成交价 手续费 交易后现金 剩余持股
0 2001-08-07 买入 100 5.9070 5 99404.30 100
1 2001-10-15 卖出 100 5.9326 5 99992.56 0
2 2001-11-08 买入 100 6.0875 5 99378.81 100
3 2002-01-15 买入 100 5.4071 5 98833.10 200
4 2002-01-24 卖出 100 5.7258 5 99400.68 100
日期 操作 数量(股) 成交价 手续费 交易后现金 剩余持股
388 2016-01-08 买入 100 6.28 5 96109.44 1700
389 2016-01-12 买入 100 5.64 5 95540.44 1800
390 2016-01-22 买入 100 5.43 5 94992.44 1900
391 2016-01-27 买入 100 5.03 5 94484.44 2000
392 2016-01-29 买入 100 4.80 5 93999.44 2100
交易记录已保存到 trade_records.csv
初始资金 100000.0
期末现金 93999.44
期末持股 2100
期末收盘价 5.38
期末总资产 105297.44
总收益率 5.30%
交易次数 393
总手续费 1965.0
dtype: object这段代码会输出两类信息：
- 交易记录：告诉你哪天买了、哪天卖了、当时价格是多少。
- 回测汇总：告诉你最后还剩多少现金、多少股票、总资产是多少。
最应该重点看的不是“赚没赚钱”这一项，而是下面 4 个问题：
交易次数多不多。
如果交易太频繁，真实手续费压力会更大。期末持股是不是很多。
如果最后仍持有大量股票，说明策略收益可能还没有真正落袋。总手续费高不高。
很多看似不错的策略，最后是被手续费拖垮的。期末总资产是否真的比初始资金高。
判断策略结果应该看总资产，而不是只看现金。
易错提醒
- 这里使用的是“昨天信号，今天开盘交易”，比“看到今天涨跌幅就立刻按今天价格成交”更合理。
- 这里只交易 100 股，是为了让逻辑更容易看懂，不代表最优仓位。
- 这仍然是教学级回测，没有考虑滑点、涨跌停不能成交、停牌等真实问题。
6.1 均值回归策略回测
前言
场景与知识
场景题目
假设股价偏离历史平均水平后会回到中间位置。我们用训练样本估计均值和波动，再对 600339.SH 做均值回归回测。
先抓关键词
训练样本先用来估计均值和标准差。买入线在均值下方。卖出线在均值上方。- 只有在训练结束后，才开始正式回测。
金融知识补充
均值回归的核心假设是：
- 价格不会一直无限偏离“正常水平”。
- 偏离太多后，有可能回到中间区域。
它更适合：
- 区间震荡明显的股票
- 没有长期单边大趋势的阶段
它不适合：
- 长时间单边上涨
- 长时间单边下跌
代码
# 读取 600339.SH，并保留开盘价和收盘价
stock = get_stock("600339.SH")[
["日期", "开盘价(元)", "收盘价(元)"]
].dropna().copy()
# 用前 60% 的样本做“训练集”，至少保证有 60 条数据
split = max(int(len(stock) * 0.6), 60)
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 / 3
sell_line = mean_price + std_price / 3
# 初始化账户
initial_cash = 100000.0
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, "开盘价(元)"]
if shares == 0 and yesterday_close <= buy_line:
# 如果昨天收盘价已经低于买入线，就尝试在今天开盘买入
qty = max_affordable_shares(cash, open_price)
if qty > 0:
turnover = qty * open_price
fee = calc_fee(turnover)
cash -= turnover + fee
shares = qty
total_fee += fee
records.append([trade_date, "买入", qty, open_price, fee, round(cash, 2), shares])
elif shares > 0 and yesterday_close >= sell_line:
# 如果昨天收盘价已经高于卖出线，就把持仓在今天开盘全部卖出
qty = shares
turnover = qty * open_price
fee = calc_fee(turnover)
cash += turnover - fee
shares = 0
total_fee += fee
records.append([trade_date, "卖出", qty, open_price, fee, round(cash, 2), shares])
# 整理交易记录
trade_df = pd.DataFrame(
records,
columns=["日期", "操作", "数量(股)", "成交价", "手续费", "交易后现金", "剩余持股"],
)
# 计算回测结束时的总资产
last_close = stock["收盘价(元)"].iloc[-1]
final_asset = cash + shares * last_close
# 打印训练得到的阈值和回测结果
print(f"训练集均值：{mean_price:.2f}")
print(f"买入线：{buy_line:.2f}")
print(f"卖出线：{sell_line:.2f}")
print(trade_df.head())
print(trade_df.tail())
print(
pd.Series(
{
"初始资金": round(initial_cash, 2),
"期末现金": round(cash, 2),
"期末持股": int(shares),
"期末总资产": round(final_asset, 2),
"总收益率": f"{(final_asset / initial_cash - 1):.2%}",
"交易次数": len(trade_df),
"总手续费": round(total_fee, 2),
}
)
)结果解读
训练集均值：5.87
买入线：5.12
卖出线：6.62
日期 操作 数量(股) 成交价 手续费 交易后现金 剩余持股
0 2012-07-17 买入 19700 5.07 9.988 111.01 19700
1 2014-12-05 卖出 19700 6.78 13.357 133663.66 0
2 2015-07-09 买入 29800 4.46 13.291 742.36 29800
3 2015-07-22 卖出 29800 6.60 19.668 197403.70 0
4 2016-01-27 买入 39100 5.03 19.667 710.03 39100
日期 操作 数量(股) 成交价 手续费 交易后现金 剩余持股
0 2012-07-17 买入 19700 5.07 9.988 111.01 19700
1 2014-12-05 卖出 19700 6.78 13.357 133663.66 0
2 2015-07-09 买入 29800 4.46 13.291 742.36 29800
3 2015-07-22 卖出 29800 6.60 19.668 197403.70 0
4 2016-01-27 买入 39100 5.03 19.667 710.03 39100
初始资金 100000.0
期末现金 710.03
期末持股 39100
期末总资产 211068.03
总收益率 111.07%
交易次数 5
总手续费 75.97
dtype: object结果解读
1️⃣ 策略逻辑回顾
买入条件 ：股价低于“买入线” 5.12 时，买入股票
卖出条件 ：股价高于“卖出线” 6.62 时，卖出股票
参考基准 ：训练集均值 5.87，用作策略的中心参考
交易单位 ：每次买卖固定数量（如示例里 19700 股、29800 股等）
手续费考虑：本章按万1费率计算，且最低收 5 元
说明策略本质是 “低买高卖” + 趋势跟踪，属于简单的量化策略。
2️⃣ 回测表现解读
交易次数 ：5 次，说明市场波动触发了少数明确机会
总收益率 ：111.07%，期末总资产 211068.03 元，翻倍效果明显
手续费消耗：75.97 元，占总资产比例更低，对策略影响很小
期末持仓 ：仍有 39100 股未卖出，期末现金为 710.03 元
说明最后一次买入后价格未达到卖出条件，策略还在等待合适的卖点
总结：策略成功捕捉了主要波段收益，但末期持仓风险依然存在。
6.2 趋势跟踪策略回测
前言
场景与知识
场景题目
如果昨天收盘价突破了前 10 日最高收盘价，今天开盘全仓买入；如果昨天收盘价跌破了前 5 日最低收盘价，今天开盘全部卖出。用 600339.SH 测一遍。
先抓关键词
- 这次的思路和均值回归相反，是“顺着强势方向走”。
- 买入条件是“创新高”。
- 卖出条件是“破位”。
- 趋势策略通常不追求高胜率，更看重少数大盈利交易。
金融知识补充
趋势跟踪的核心假设是：
- 强势可能继续强。
- 弱势可能继续弱。
因此它不去猜“会不会回调”，而是直接跟随已经形成的方向。
这类策略的典型特点常常是：
- 胜率不一定高
- 但一旦抓到大趋势，单次收益可能比较可观
代码
# 读取 600339.SH 的开盘价和收盘价
stock = get_stock("600339.SH")[
["日期", "开盘价(元)", "收盘价(元)"]
].dropna().copy()
# 计算“昨天之前 10 日的最高收盘价”和“昨天之前 5 日的最低收盘价”
# 这里一定要 shift(1)，避免把今天自己算进参考区间
stock["前10日最高收盘价"] = stock["收盘价(元)"].rolling(10).max().shift(1)
stock["前5日最低收盘价"] = stock["收盘价(元)"].rolling(5).min().shift(1)
# 初始化账户和统计变量
initial_cash = 100000.0
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(stock)):
# yesterday 用来判断信号，today 用来执行交易
yesterday = stock.loc[i - 1]
today = stock.loc[i]
# 如果前期滚动窗口还没算出来，就先跳过
if pd.isna(yesterday["前10日最高收盘价"]) or pd.isna(yesterday["前5日最低收盘价"]):
continue
if shares == 0 and yesterday["收盘价(元)"] > yesterday["前10日最高收盘价"]:
# 昨天收盘创新高，说明形成向上突破，
# 今天开盘尝试全仓买入
qty = max_affordable_shares(cash, today["开盘价(元)"])
if qty > 0:
turnover = qty * today["开盘价(元)"]
fee = calc_fee(turnover)
cash -= turnover + fee
shares = qty
entry_price = today["开盘价(元)"]
entry_fee = fee
total_fee += fee
records.append(
[today["日期"], "买入", qty, today["开盘价(元)"], fee, round(cash, 2), shares]
)
elif shares > 0 and yesterday["收盘价(元)"] < yesterday["前5日最低收盘价"]:
# 昨天收盘跌破前 5 日低点，说明趋势可能转弱，
# 今天开盘把持仓全部卖出
qty = shares
turnover = qty * today["开盘价(元)"]
fee = calc_fee(turnover)
cash += turnover - fee
total_fee += fee
# 计算这笔完整交易（从买入到卖出）的盈亏
trade_profit = (today["开盘价(元)"] - entry_price) * qty - entry_fee - fee
if trade_profit > 0:
win_count += 1
else:
loss_count += 1
# 卖出后账户回到空仓状态
shares = 0
entry_price = None
entry_fee = 0.0
records.append(
[today["日期"], "卖出", qty, today["开盘价(元)"], fee, round(cash, 2), shares]
)
# 整理交易记录
trade_df = pd.DataFrame(
records,
columns=["日期", "操作", "数量(股)", "成交价", "手续费", "交易后现金", "剩余持股"],
)
# 期末总资产 = 现金 + 剩余持股按最后收盘价估值
last_close = stock["收盘价(元)"].iloc[-1]
final_asset = cash + shares * last_close
# 打印交易记录和总体结果
print(trade_df.head())
print(trade_df.tail())
print(
pd.Series(
{
"初始资金": round(initial_cash, 2),
"期末现金": round(cash, 2),
"期末持股": int(shares),
"期末总资产": round(final_asset, 2),
"总收益率": f"{(final_asset / initial_cash - 1):.2%}",
"已平仓盈利笔数": win_count,
"已平仓亏损笔数": loss_count,
"交易次数": len(trade_df),
"总手续费": round(total_fee, 2),
}
)
)结果解读
核心策略：“昨天收盘创近10日新高 → 今天开盘买入；昨天收盘跌破近5日最低 → 今天开盘全仓卖出”
日期 操作 数量(股) 成交价 手续费 交易后现金 剩余持股
0 2001-02-23 买入 17500 5.6939 9.964 346.79 17500
1 2001-03-21 卖出 17500 5.8922 10.311 103449.97 0
2 2001-03-28 买入 16800 6.1133 10.270 736.26 16800
3 2001-04-24 卖出 16800 6.5207 10.955 110273.07 0
4 2001-05-16 买入 16200 6.7381 10.916 1104.93 16200
日期 操作 数量(股) 成交价 手续费 交易后现金 剩余持股
... 中间交易记录略 ...
初始资金 100000.0
期末现金 30038.93
期末持股 20800
期末总资产 141942.93
总收益率 41.94%
已平仓盈利笔数 60
已平仓亏损笔数 81
交易次数 283
总手续费 3291.11
dtype: object
这类策略最值得看的不是“胜率高不高”，而是：
- 有没有抓到少数关键的大趋势。
- 总资产最终有没有明显改善。
- 亏损笔数虽然多，但单次亏损是不是比较可控。
如果你看到：
已平仓盈利笔数不多，但期末总资产仍然不错，说明策略可能靠少数趋势段赚到了钱。交易次数很多、总手续费仍然不低，而且总资产改善有限，说明样本更像震荡市，趋势策略容易被来回打脸。
易错提醒
- 这里的买入和卖出规则用的是“昨天收盘信号，今天开盘成交”，逻辑要分清楚。
前10日最高收盘价和前5日最低收盘价都做了shift(1)，是为了防止把今天自己算进参考窗口里。- 趋势跟踪策略在震荡市场中可能明显不如均值回归。
7 多股票横向比较
前言
7.1 股票历史累计涨跌幅排名
前言
场景与知识
场景题目
统计 2015-01-01 之前，各股票的历史累计涨跌幅，并按从高到低排序。
先抓关键词
- 这是横向比较，不是单股票回测。
- 重点是“同一时间段、同一种计算方式”。
- 这里不再简单对日涨跌幅求和，而是用复利累计，更符合金融含义。
金融知识补充
为什么不能简单把每天的涨跌幅直接相加？
因为金融收益更接近“连乘”关系。
例如：
- 第一天涨
10% - 第二天下跌
10%
如果直接相加，好像等于 0%；
但真实净值是 1.1 × 0.9 = 0.99，其实亏了 1%。
所以这里要用：
单日净值 = 1 + 涨跌幅 / 100区间累计净值 = 每日净值连乘
代码
# 读入全市场数据，并删除涨跌幅缺失的记录
data = load_stock_data().dropna(subset=["涨跌幅(%)"]).copy()
# 设定研究截止日期：只看 2015-01-01 之前的数据
end_date = pd.Timestamp("2015-01-01")
# 先按时间筛选样本
sample = data[data["日期"] < end_date].copy()
# 把“百分比收益”转成“净值倍数”
# 例如涨 5% -> 1.05，跌 3% -> 0.97
sample["单日净值"] = 1 + sample["涨跌幅(%)"] / 100
# 理论上单日净值必须大于 0，否则连乘就没有正常金融含义
sample = sample[sample["单日净值"] > 0].copy()
# 按股票代码分组：
# - 交易天数：统计每只股票样本里有多少天
# - 区间累计净值：把每天净值连乘起来，得到更符合金融意义的累计结果
ranking = (
sample.groupby("代码")
.agg(交易天数=("单日净值", "size"), 区间累计净值=("单日净值", "prod"))
.sort_values("区间累计净值", ascending=False)
)
# 再把累计净值转回更直观的累计涨跌幅百分比
ranking["区间累计涨跌幅(%)"] = (ranking["区间累计净值"] - 1) * 100
# 只看排名前 10 的股票，并保留两位小数
print(ranking[["交易天数", "区间累计涨跌幅(%)"]].head(10).round(2))结果解读
交易天数 区间累计涨跌幅(%)
代码
600652.SH 5880 17668.68
600651.SH 5880 14698.96
600601.SH 5880 14458.17
600653.SH 5880 11062.55
600887.SH 4561 10967.00
600741.SH 4443 5664.71
600739.SH 4448 4884.71
600654.SH 5880 4711.80
600111.SH 4179 4091.83
600606.SH 5558 4064.29你会得到一个前 10 名的排名表，其中通常会看到：
交易天数：表示这只股票在样本期里有多少个交易日数据。区间累计涨跌幅(%)：表示在整个样本期里，按复利方式累计后的涨跌结果。
阅读这张表时要注意：
- 排名越靠前，不代表未来一定继续强，只能说明它在那个历史区间里更强。
- 如果某只股票交易天数太少，排名可能受样本长度影响。
- 做横向比较时，计算口径一定要统一，否则结果没有可比性。
易错提醒
- 一定要先把百分比转成净值，也就是
1 + r/100，金融回测、资金曲线、复利计算的标准做法'- 举例：
- 初始资金 1000 元
- r1 = 5% → v1 = 1 + 0.05 = 1.05
- r2 = -3% → v2 = 1 - 0.03 = 0.97
- 累积资金变化 = 初始资金 × v1 × v2= 1000 × 1.05 × 0.97 = 1018.5 元
- 举例：
- 如果你直接
groupby("代码").sum()，结果只是近似处理，不是严格累计收益。 - 时间筛选最好先转成真正的日期格式再比较，这样更稳妥。