同一策略在牛、熊、震荡市分别回测工具（教学版）

定位：去营销化、中立、可教学、可扩展

⚠️ 全文含免责声明与风险提示，不荐股、不承诺收益、不引导开户、无任何引流

一、实际应用场景描述

在智能证券投资课程中，策略在不同市场环境下的适应性（Regime Dependence）是区分"幸运"与"能力"的关键课题。

本程序适用于：

- 高校量化投资、策略开发课程实验

- 多市场环境策略压力测试教学

- 投资者认知训练：打破"牛市啥都灵"的迷思

- 量化竞赛中的策略鲁棒性验证

核心目标：

- 将历史行情标记为牛市、熊市、震荡市三种状态

- 用同一策略分别回测三种行情

- 量化行情对策略有效性的影响

- 用数据回答："牛市赚钱，到底是策略牛，还是行情牛？"

✅ 不做未来预测

✅ 不构成投资建议

✅ 仅作为历史回测教学示例

二、痛点引入（真实可感知）

痛点 表现

"牛市什么策略都赚" 2014–2015 年人人是股神

行情归因缺失 不知道赚的是 α 还是 β

策略盲目推广 牛市回测好就以为万能

熊市裸奔 没测过下跌市，一跌就崩

工具门槛高 专业多状态回测框架复杂

👉 需要一个轻量、本地、可解释、可复现的分行情回测工具

三、核心逻辑讲解（工程视角）

1️⃣ 数据模型设计

MarketRegimeSession

├── regime 行情状态（bull/bear/sideway）

├── index_name 指数名称

├── nav_list 净值 / 价格序列

└── regime_periods 各状态区间列表

2️⃣ 行情状态判定（教学用简化规则）

基于累计涨跌幅划分：

状态 判定条件

牛市 区间累计涨幅 > +20%

熊市 区间累计跌幅 < −20%

震荡市 介于两者之间

⚠️ 教学中强调：行情划分本身有主观性，不同标准结论可能不同。

3️⃣ 核心回测逻辑

策略（教学用）：均线交叉策略

规则 操作

短均线上穿长均线 买入（全仓）

短均线下穿长均线 卖出（清仓）

持币期间 收益 = 0%

4️⃣ 分行情回测流程

读取完整行情序列

按涨跌幅划分为牛 / 熊 / 震荡区间

对每个区间：

执行同一策略回测

记录收益、回撤、胜率

输出三行情对比报告

5️⃣ 关键指标对比

指标 牛市长啥样 熊市长啥样 震荡市长啥样

累计收益 应该高 应该低甚至负 中等

最大回撤 通常较小 暴露风险 频繁止损

交易次数 较少（趋势明确） 中等 很多（反复打脸）

胜率 可能虚高 暴露真实胜率 关键分水岭

四、Python 模块化代码（可直接运行）

📁 项目结构

regime_backtest_tool/

│

├── main.py

├── models.py

├── regime_classifier.py

├── backtester.py

├── comparator.py

├── reporter.py

├── storage.py

├── README.md

└── DISCLAIMER.md

✅ models.py（数据建模）

"""

models.py

多行情状态回测数据模型

"""

class MarketRegimeSession:

"""单段行情区间"""

def __init__(self, regime, index_name, nav_list, start_idx, end_idx):

"""

regime: "bull" / "bear" / "sideway"

nav_list: 完整净值序列

start_idx: 区间起始索引

end_idx: 区间结束索引

"""

self.regime = regime

self.index_name = index_name

self.nav_list = nav_list

self.start_idx = start_idx

self.end_idx = end_idx

✅ regime_classifier.py（行情状态分类器）

"""

regime_classifier.py

牛 / 熊 / 震荡市判定

"""

import numpy as np

def classify_regime(nav_list, window=20, bull_thresh=20.0, bear_thresh=-20.0):

"""

基于滚动累计涨跌幅判定行情状态

window: 滚动窗口大小

bull_thresh: 牛市阈值（%）

bear_thresh: 熊市阈值（%）

"""

regimes = []

for i in range(len(nav_list)):

if i < window:

regimes.append("warmup") # 预热期

continue

start_nav = nav_list[i - window]

end_nav = nav_list[i]

change_pct = (end_nav - start_nav) / start_nav * 100

if change_pct >= bull_thresh:

regimes.append("bull")

elif change_pct <= bear_thresh:

regimes.append("bear")

else:

regimes.append("sideway")

return regimes

def extract_regime_periods(nav_list, regimes):

"""

将连续的同种行情合并为区间

返回: [(regime, start_idx, end_idx), ...]

"""

periods = []

i = 0

n = len(regimes)

while i < n:

if regimes[i] == "warmup":

i += 1

continue

current_regime = regimes[i]

start = i

while i < n and regimes[i] == current_regime:

i += 1

end = i - 1

periods.append((current_regime, start, end))

return periods

✅ backtester.py（核心回测引擎）

"""

backtester.py

均线交叉策略回测引擎（教学版）

"""

import numpy as np

class MABacktester:

"""简单均线交叉策略"""

def __init__(self, short_window=5, long_window=20):

self.short_window = short_window

self.long_window = long_window

def run(self, nav_list, start_idx, end_idx):

"""

在指定区间内执行均线策略

"""

# 截取区间（需要额外 long_window 根用于计算均线）

actual_start = max(0, start_idx - self.long_window)

segment = nav_list[actual_start:end_idx + 1]

if len(segment) < self.long_window + 1:

return self._empty_result()

short_ma = self._moving_average(segment, self.short_window)

long_ma = self._moving_average(segment, self.long_window)

# 对齐：short_ma[i] 和 long_ma[i] 对应 segment 的第 i 个点

# 从第 long_window 根开始比较

offset = self.long_window - self.short_window

cash = 10000.0

shares = 0.0

in_position = False

trades = []

for i in range(max(offset, 0), len(segment)):

if i + self.short_window > len(segment):

break

short_val = short_ma[i] if i < len(short_ma) else None

long_val = long_ma[i] if i < len(long_ma) else None

if short_val is None or long_val is None:

continue

price = segment[i]

if not in_position and short_val > long_val:

# 买入

shares = cash / price

cash = 0.0

in_position = True

trades.append({"type": "buy", "price": price, "day": i})

elif in_position and short_val < long_val:

# 卖出

cash = shares * price

shares = 0.0

in_position = False

trades.append({"type": "sell", "price": price, "day": i})

# 期末清仓

if in_position and len(segment) > 0:

cash = shares * segment[-1]

shares = 0.0

final_value = cash

total_return = (final_value - 10000) / 10000 * 100

# 计算最大回撤

peak = 10000

max_dd = 0

portfolio_value = 10000

for i in range(max(offset, 0), len(segment)):

price = segment[i]

if in_position:

val = shares * price

else:

val = cash

if val > peak:

peak = val

dd = (peak - val) / peak * 100

if dd > max_dd:

max_dd = dd

# 胜率（有买有卖的交易对）

win_count = 0

trade_pairs = []

for t in trades:

if t["type"] == "buy":

trade_pairs.append(t)

elif t["type"] == "sell" and trade_pairs:

buy_t = trade_pairs.pop(0)

if t["price"] > buy_t["price"]:

win_count += 1

total_trades = len([t for t in trades if t["type"] == "sell"])

return {

"final_value": round(final_value, 2),

"total_return_pct": round(total_return, 2),

"max_drawdown_pct": round(max_dd, 2),

"total_trades": total_trades,

"winning_trades": win_count,

"win_rate_pct": round(win_count / total_trades * 100, 2) if total_trades > 0 else None,

"days": end_idx - start_idx + 1

}

def _empty_result(self):

return {

"final_value": 10000,

"total_return_pct": 0,

"max_drawdown_pct": 0,

"total_trades": 0,

"winning_trades": 0,

"win_rate_pct": None,

"days": 0

}

def _moving_average(self, data, window):

"""简单移动平均"""

if len(data) < window:

return []

return [np.mean(data[i-window:i]) for i in range(window, len(data)+1)]

✅ comparator.py（三行情对比）

"""

comparator.py

牛 / 熊 / 震荡市策略表现对比

"""

from backtester import MABacktester

def run_regime_comparison(nav_list, regime_periods, index_name):

"""

对同一策略在三种行情下分别回测

"""

backtester = MABacktester()

results = {}

for regime, start, end in regime_periods:

result = backtester.run(nav_list, start, end)

if regime not in results:

results[regime] = []

results[regime].append(result)

# 汇总每种行情

summary = {}

for regime, rlist in results.items():

avg_return = sum(r["total_return_pct"] for r in rlist) / len(rlist)

avg_dd = sum(r["max_drawdown_pct"] for r in rlist) / len(rlist)

total_trades = sum(r["total_trades"] for r in rlist)

win_rates = [r["win_rate_pct"] for r in rlist if r["win_rate_pct"] is not None]

avg_win_rate = sum(win_rates) / len(win_rates) if win_rates else None

summary[regime] = {

"periods": len(rlist),

"avg_return_pct": round(avg_return, 2),

"avg_max_drawdown_pct": round(avg_dd, 2),

"total_trades": total_trades,

"avg_win_rate_pct": round(avg_win_rate, 2) if avg_win_rate else None,

"details": rlist

}

return summary

✅ reporter.py（对比报告输出）

"""

reporter.py

牛 / 熊 / 震荡市策略表现对比报告

"""

def report(index_name, summary):

print("\n" + "=" * 65)

print(f"【同一策略 × 多行情回测对比报告】")

print(f"指数：{index_name}")

print("=" * 65)

regime_names = {

"bull": "🐂 牛市",

"bear": "🐻 熊市",

"sideway": "📊 震荡市"

}

for regime, data in summary.items():

name = regime_names.get(regime, regime)

print(f"\n{name}（共 {data['periods']} 个区间）")

print("-" * 65)

print(f" 平均收益：{data['avg_return_pct']}%")

print(f" 平均最大回撤：{data['avg_max_drawdown_pct']}%")

print(f" 总交易次数：{data['total_trades']}")

if data['avg_win_rate_pct'] is not None:

print(f" 平均胜率：{data['avg_win_rate_pct']}%")

else:

print(f" 平均胜率：N/A（无完整交易对）")

# 教学结论

print(f"\n{'=' * 65}")

print(f"\n💡 教学启示：")

print("-" * 65)

bull_ret = summary.get("bull", {}).get("avg_return_pct", 0) or 0

bear_ret = summary.get("bear", {}).get("avg_return_pct", 0) or 0

side_ret = summary.get("sideway", {}).get("avg_return_pct", 0) or 0

bull_dd = summary.get("bull", {}).get("avg_max_drawdown_pct", 0) or 0

bear_dd = summary.get("bear", {}).get("avg_max_drawdown_pct", 0) or 0

print(f"\n ① 牛市平均收益：{bull_ret}% | 熊市平均收益：{bear_ret}%")

print(f" → 差距 {round(bull_ret - bear_ret, 2)} 个百分点")

if bull_ret > 0 and bear_ret < 0:

print(f" ⚠️ 策略在牛市赚、熊市亏，说明：")

print(f" 「赚的是行情的钱，不是策略的能耐」")

elif bull_ret > 0 and bear_ret > 0:

print(f" ✅ 策略在牛熊市均盈利，具备一定鲁棒性")

elif bear_ret < -10:

print(f" ⚠️ 熊市平均亏损 {abs(bear_ret)}%，策略需改进或加风控")

print(f"\n ② 牛市回撤：{bull_dd}% | 熊市回撤：{bear_dd}%")

if bear_dd > bull_dd * 2:

print(f" ⚠️ 熊市回撤是牛市的 {round(bear_dd / max(bull_dd, 0.1), 1)} 倍")

print(f" → 同一策略在不同行情下风险暴露差异巨大")

# 震荡市分析

side_trades = summary.get("sideway", {}).get("total_trades", 0)

if side_trades > 10:

print(f"\n ③ 震荡市交易 {side_trades} 次")

print(f" ⚠️ 频繁交叉 → 反复止损 → 震荡市是均线策略的「照妖镜」")

print(f"\n 核心结论：")

print(f" 牛市什么策略都赚钱 ≠ 策略有效。")

print(f" 真正的策略能力 = 熊市和震荡市的表现。")

print("=" * 65)

✅ storage.py（本地存储）

"""

storage.py

JSON 本地存储

"""

import json

FILE_PATH = "regime_backtest_result.json"

def save_result(data):

with open(FILE_PATH, "w", encoding="utf-8") as f:

json.dump(data, f, ensure_ascii=False, indent=2)

✅ main.py（交互入口）

"""

main.py

牛 / 熊 / 震荡市策略回测对比工具

"""

from models import MarketRegimeSession

from regime_classifier import classify_regime, extract_regime_periods

from comparator import run_regime_comparison

from reporter import report

from storage import save_result

def main():

print("=== 同一策略 × 多行情回测对比工具（教学版）===")

print("量化「牛市什么策略都赚钱」是否成立\n")

index_name = input("指数名称（如 沪深300）：")

print(f"\n📌 请输入净值 / 价格序列（空格分隔）：")

navs = list(map(float, input().split()))

if len(navs) < 40:

print("⚠️ 数据量不足（至少需要 40 个数据点），无法可靠回测")

return

# 行情分类

regimes = classify_regime(navs, window=20)

regime_periods = extract_regime_periods(navs, regimes)

if not regime_periods:

print("⚠️ 未检测到有效行情区间")

return

# 统计各行情数量

bull_count = sum(1 for r, _, _ in regime_periods if r == "bull")

bear_count = sum(1 for r, _, _ in regime_periods if r == "bear")

side_count = sum(1 for r, _, _ in regime_periods if r == "sideway")

print(f"\n📊 行情分布：牛市 {bull_count} 段 | 熊市 {bear_count} 段 | 震荡 {side_count} 段")

# 执行回测

summary = run_regime_comparison(navs, regime_periods, index_name)

# 输出报告

report(index_name, summary)

# 保存结果

result_data = {

"index": index_name,

"total_data_points": len(navs),

"regime_distribution": {

"bull": bull_count,

"bear": bear_count,

"sideway": side_count

},

"summary": summary

}

save_result(result_data)

print("\n✅ 回测结果已保存")

if __name__ == "__main__":

main()

五、README 与使用说明

# 同一策略 × 多行情回测对比工具（教学版）

## 项目说明

将行情划分为牛 / 熊 / 震荡三种状态，用同一策略分别回测，量化行情对策略有效性的影响。

## 使用方式

bash

pip install numpy

python main.py

## 输入示例

指数名称：沪深300

净值序列：1.0 1.05 1.12 1.08 0.95 0.88 0.92 1.01 1.10 1.18 1.15 1.22 ...

## 策略说明

采用**均线交叉策略**（短均线上穿买入、下穿卖出）：

- 短均线：5 期

- 长均线：20 期

- 初始资金：10000 元（模拟）

## 适用范围

- 量化投资策略课程

- 策略鲁棒性教学

- 行情归因分析演示

## 注意事项

- 仅基于历史数据

- 不构成任何投资建议

- 使用前请阅读 DISCLAIMER.md

六、DISCLAIMER.md（免责声明与风险提示）

# 免责声明与风险提示

## 免责声明

本程序仅供**教学与科研用途**，用于演示策略在不同行情下的表现差异。

作者不提供任何证券交易建议，不推荐任何策略，不承诺任何收益。

## 风险提示

1. 行情状态判定具有主观性，不同阈值 / 方法会导致不同结论

2. 均线交叉策略仅为教学示例，不代表"最优策略"

3. 历史回测不代表未来表现，策略可能过拟合

4. 未考虑手续费、滑点、流动性等现实约束

5. 牛市赚钱 ≠ 策略有效，需结合熊市 / 震荡市综合判断

6. 策略鲁棒性需在多种市场环境下验证，单一行情回测不足为据

使用本工具产生的任何后果，作者概不负责。

七、核心知识点卡片（教学向）

分类 内容

Python 类、滑动窗口、列表切片、字典聚合

量化金融 行情状态分类、策略鲁棒性、行情归因

投资理念 打破"牛市策略万能"迷思

数据分析 分组统计、跨组对比、胜率分析

工程思想 模块化（分类器 / 回测器 / 对比器解耦）

风险教育 同一策略在不同行情下表现差异可能巨大

可扩展性 可接入真实行情 API、支持多策略并行对比

八、总结（工程师视角）

这是一个完全中立、去营销化、可教学的原型系统：

✅ 不鼓吹任何策略

✅ 不妖魔化任何行情

✅ 不伪装成策略甄选神器

它真正展示的是：

如何用 Python 把"牛市赚钱到底是运气还是实力"变成可量化、可对比、可反思的教学实验

核心教学价值：

传统观念 数据可能揭示的真相

"牛市什么策略都赚" 熊市可能亏得更多，赚的是 β 不是 α

"回测收益率 50% 就是好策略" 可能只在牛市有效，震荡市疯狂止损

"策略通过了回测" 没分行情测 = 没真正测过

"均线策略很稳" 震荡市频繁交叉 → 反复打脸

典型回测结果参考（教学示例）：

行情 均线策略平均收益 最大回撤

牛市 +15% ~ +35% 5% ~ 12%

熊市 −5% ~ −25% 15% ~ 40%

震荡市 −3% ~ +5% 8% ~ 18%

⚠️ 以上为教学示意，不代表任何真实策略表现。实际结果取决于参数、数据区间和手续费假设。

本文代码仅供学习与技术交流，不构成任何投资建议，股市有风险，入市需谨慎！

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！