本文用真实回测数据证明——当市场进入极度狂热状态时，趋势跟踪量化策略的"追涨"信号恰好出现在动量衰竭的尾部，成为最后接盘的那批资金。我们不仅讲逻辑，更用代码和数据验证
利用换手率建立拥挤度指标这个是常见的拥挤的指标，还要其他的比如资金的拥挤，情绪的拥挤等，常见的是换手率
# 简易狂热预警公式
def frenzy_warning(df, window=20, z_threshold=2.0):
"""
判断当前是否处于"狂热陷阱"区域
"""
# 计算换手率Z-score
turnover_ma = df['turnover'].rolling(window).mean()
turnover_std = df['turnover'].rolling(window).std()
z_score = (df['turnover'] - turnover_ma) / turnover_std
# 趋势状态
is_uptrend = (df['close'] > df['close'].rolling(20).mean()) & \
(df['close'].rolling(20).mean() > df['close'].rolling(60).mean())
# 狂热预警：高拥挤 + 仍在上涨趋势中
# 注意：这里不是建议做空，而是建议"不开新仓"或"收紧止损"
return z_score > z_threshold, z_score
2026年4月3日，AI算力板块一片欢腾，7连阳、累计涨幅23%，换手率飙升至历史均值的3.2倍。各大股票群里"算力永不眠"的口号此起彼伏。然而4天后，板块急转直下，4个交易日暴跌14%。龙虎榜数据显示，量化席位在这波下跌中的净卖出占比高达51%。在上涨的最后一天，大量趋势跟踪策略触发了买入信号。它们成了最后的接盘者。这个案例让我开始反思一个被忽视的问题：当"羊群效应"（市场情绪极度狂热）撞上"冰冷代码"（量化策略的机械执行），追涨杀跌不仅不会赚钱，反而成了量化策略的最大陷阱。
羊群效应：金融市场中的"羊群效应"，指的是投资者盲目跟随大众行为，忽视自己的独立判断。当某个板块连续大涨时，"再不进场就来不及了"的焦虑感会驱使大量资金涌入。但问题在于：羊群涌入的时候，往往已经是行情的中后段。
量化陷阱：这里的"陷阱"不是指量化策略本身有bug，而是指：在市场极度狂热时，趋势跟踪策略的"追涨"入场信号，恰好出现在动量衰竭的尾部，导致策略在最高位附近建仓，随后被连锁止损击穿。换句话说，量化策略的机械执行，在狂热期变成了羊群效应的"算法化表达"——它比散户更纪律、更快速，但也更不挑时机
常见的原因是：趋势跟踪策略的核心逻辑是"确认趋势后介入"。当20日均线突破信号出现时，股价往往已经上涨了15%-20%。在狂热行情中，这个信号出现的时点，可能已经是行情的第7根或第8根阳线——离顶部往往只有一步之遥。
A股市场中，大量量化产品使用相似的因子库（动量、波动率、资金流向）。当同一个突破信号被数百个模型同时捕捉，它们会在同一价格区间执行买入，短期内推高股价。
有量化策略都有严格的止损纪律（如-5%或-7%止损）。当股价从高位回落触发止损时，大量策略在同一价位卖出，形成 "多杀多"的局面——这正是AI板块那波行情的真实写照
Z-score（标准化值）(当前换手率 - 过去60日均值) / 过去60日标准差，根据常见的经验大概的参考
下面开始验证，主力利用QMt的数据，因为QMt本身没有换手率这个指标需要自己计算，换手率等于成交量比流通市值，需要注意，QMT给的数据，成交量单位是手，流通市值是股，100股等于1手，可转债10股一手
计算的代码参考
from xg_qmt_trader import xg_qmt_trader
api=xg_qmt_trader(
account='',
account_type='STOCK',
url='127.0.0.1',
port='8888')
api.connect()
stock='.SH'
start_time='20260101'
end_time='20260722'
df=api.get_market_data_ex(
fields=[],
stock_code=[stock],
period='1d',
start_time='20260101',
end_time='20260722',
count=-1,
dividend_type='follow',
fill_data=True,
subscribe=True
)
df=df[stock]
df=api.data_to_pandas(df)
df=df[['time','close', 'high', 'low', 'open','amount','volume']]
df['time']=df['time'].apply(lambda x: str(api.conv_time(x))[:8])
inst=api.get_instrument_detail(stockcode=stock)
FloatVolume=inst['FloatVolume']
df['turnover']=(df['volume']*100/FloatVolume)
print(df)
下面利用芯片ETF来验证这个思路猜想，看偏移度，拥挤度，先读取历史行情数据
数据标准化，计算换手率的分位数，一般超90%比较危险
def calculate_indicators(df):
"""
计算趋势策略所需的技术指标
"""
df = df.copy()
# 均线
df['ma20'] = df['close'].rolling(20).mean()
df['ma60'] = df['close'].rolling(60).mean()
# ===== 修复：使用更稳健的方式计算换手率分位数 =====
def calc_rank(series):
"""计算当前值在窗口内的分位数（0-1）"""
min_val = series.min()
max_val = series.max()
if max_val > min_val:
return (series.iloc[-1] - min_val) / (max_val - min_val)
else:
return 0.5
df['turnover_rank'] = df['turnover'].rolling(60).apply(
calc_rank, raw=False
)
# 换手率Z-score
df['turnover_mean'] = df['turnover'].rolling(60).mean()
df['turnover_std'] = df['turnover'].rolling(60).std()
df['turnover_zscore'] = (df['turnover'] - df['turnover_mean']) / df['turnover_std']
# 收益计算
df['ret_5d'] = df['close'].pct_change(5) * 100
df['ret_10d'] = df['close'].pct_change(10) * 100
df['ret_20d'] = df['close'].pct_change(20) * 100
# 每日收益率
df['ret_daily'] = df['close'].pct_change() * 100
return df
定义市场过热的交易规则
def generate_signals(df):
"""
生成趋势策略信号
策略规则：收盘价 > MA20 且 MA20 > MA60 时买入，跌破MA20卖出
"""
df = df.copy()
# 趋势突破信号（买入）
df['trend_buy'] = (df['close'] > df['ma20']) & (df['ma20'] > df['ma60'])
# 趋势卖出信号
df['trend_sell'] = (df['close'] < df['ma20'])
# 信号变化：检测新信号出现
df['buy_signal'] = df['trend_buy'] & (~df['trend_buy'].shift(1).fillna(False))
df['sell_signal'] = df['trend_sell'] & (~df['trend_sell'].shift(1).fillna(False))
# 判断是否处于"狂热"状态（换手率分位数>0.9 或 Z-score>2）
df['is_frenzy'] = (df['turnover_rank'] > 0.9) | (df['turnover_zscore'] > 2.0)
# 狂热陷阱信号：在狂热状态下触发的买入信号
df['trap_signal'] = df['buy_signal'] & df['is_frenzy']
return df
执行的思路建议
这个是题材的，如果需要检验全市场一般可以考虑沪深300ETF，A500等可以体现市场全局的ETF来检验市场的拥挤度，我本来是利用上证指数的，但是指数数据出来是流通股是0，也合理，市场分析拥挤的大概情况
不懂的问我就可以，加我备注入群可以加入量化研究群
思路代码参考
# ============================================================
# 量化策略回测：趋势策略在狂热市场中的有效性验证
# 标题：当"羊群效应"撞上"冰冷代码"
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import warnings
from xg_qmt_trader import xg_qmt_trader
warnings.filterwarnings('ignore')
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
print("=" * 60)
print("量化策略回测：趋势策略在狂热市场中的有效性验证")
print("=" * 60)
#账户
account=''
account_type='STOCK'
url='127.0.0.1'
port='8888'
api=xg_qmt_trader(
account=account,
account_type=account_type,
url=url,
port=port)
#链接QMt数据
api.connect()
# ============================================================
# 第一步：获取数据（以人工智能ETF为例）
# ============================================================
def get_stock_data(symbol=".SH", start_date="20260101", end_date="20260722"):
"""
获取股票/ETF数据
symbol: 300ETF
"""
try:
stock=symbol
df=api.get_market_data_ex(
fields=[],
stock_code=[stock],
period='1d',
start_time=start_date,
end_time=end_date,
count=-1,
dividend_type='follow',
fill_data=True,
subscribe=True
)
df=df[stock]
df=api.data_to_pandas(df)
df=df[['time','close', 'high', 'low', 'open','amount','volume']]
df['time']=df['time'].apply(lambda x: str(api.conv_time(x))[:8])
df['date']=df['time'].apply(lambda x: pd.to_datetime(x))
inst=api.get_instrument_detail(stockcode=stock)
FloatVolume=inst['FloatVolume']
df['turnover']=(df['volume']*100/FloatVolume)
return df
except Exception as e:
print(f"⚠️ 数据获取失败：{e}")
print(" 使用模拟数据代替...")
return generate_mock_data(start_date, end_date)
def generate_mock_data(start_date, end_date):
"""
当真实数据获取失败时，生成模拟数据用于演示
"""
dates = pd.date_range(start=start_date, end=end_date, freq='D')
dates = dates[dates.weekday < 5] # 只保留工作日
n = len(dates)
np.random.seed(42)
# 模拟价格：包含趋势和波动
trend = np.linspace(0, 0.8, n) + np.random.normal(0, 0.02, n).cumsum() * 0.5
price = 1.0 * np.exp(trend)
# 生成OHLC数据
close = price
high = close * (1 + np.abs(np.random.normal(0, 0.01, n)))
low = close * (1 - np.abs(np.random.normal(0, 0.01, n)))
open_price = close * (1 + np.random.normal(0, 0.005, n))
# 成交量：在狂热期放大
volume_base = 1000000
frenzy_periods = (dates >= pd.Timestamp('2025-03-20')) & (dates <= pd.Timestamp('2025-04-15'))
volume = volume_base * (1 + 2 * np.random.random(n))
volume[frenzy_periods] = volume[frenzy_periods] * 3
# 换手率：同样在狂热期放大
turnover = 2 + 4 * np.random.random(n)
turnover[frenzy_periods] = turnover[frenzy_periods] * 2.5
df = pd.DataFrame({
'date': dates,
'open': open_price,
'high': high,
'low': low,
'close': close,
'volume': volume,
'turnover': turnover,
'amount': close * volume / 10000
})
print(f"✓ 使用模拟数据：{len(df)}个交易日")
return df
# ============================================================
# 第二步：计算技术指标（修复版）
# ============================================================
def calculate_indicators(df):
"""
计算趋势策略所需的技术指标
"""
df = df.copy()
# 均线
df['ma20'] = df['close'].rolling(20).mean()
df['ma60'] = df['close'].rolling(60).mean()
# ===== 修复：使用更稳健的方式计算换手率分位数 =====
def calc_rank(series):
"""计算当前值在窗口内的分位数（0-1）"""
min_val = series.min()
max_val = series.max()
if max_val > min_val:
return (series.iloc[-1] - min_val) / (max_val - min_val)
else:
return 0.5
df['turnover_rank'] = df['turnover'].rolling(60).apply(
calc_rank, raw=False
)
# 换手率Z-score
df['turnover_mean'] = df['turnover'].rolling(60).mean()
df['turnover_std'] = df['turnover'].rolling(60).std()
df['turnover_zscore'] = (df['turnover'] - df['turnover_mean']) / df['turnover_std']
# 收益计算
df['ret_5d'] = df['close'].pct_change(5) * 100
df['ret_10d'] = df['close'].pct_change(10) * 100
df['ret_20d'] = df['close'].pct_change(20) * 100
# 每日收益率
df['ret_daily'] = df['close'].pct_change() * 100
return df
# ============================================================
# 第三步：定义策略信号
# ============================================================
def generate_signals(df):
"""
生成趋势策略信号
策略规则：收盘价 > MA20 且 MA20 > MA60 时买入，跌破MA20卖出
"""
df = df.copy()
# 趋势突破信号（买入）
df['trend_buy'] = (df['close'] > df['ma20']) & (df['ma20'] > df['ma60'])
# 趋势卖出信号
df['trend_sell'] = (df['close'] < df['ma20'])
# 信号变化：检测新信号出现
df['buy_signal'] = df['trend_buy'] & (~df['trend_buy'].shift(1).fillna(False))
df['sell_signal'] = df['trend_sell'] & (~df['trend_sell'].shift(1).fillna(False))
# 判断是否处于"狂热"状态（换手率分位数>0.9 或 Z-score>2）
df['is_frenzy'] = (df['turnover_rank'] > 0.9) | (df['turnover_zscore'] > 2.0)
# 狂热陷阱信号：在狂热状态下触发的买入信号
df['trap_signal'] = df['buy_signal'] & df['is_frenzy']
return df
# ============================================================
# 第四步：策略回测
# ============================================================
def backtest_strategy(df):
"""
回测趋势策略，并区分正常市场和狂热市场
"""
df = df.copy()
df = df.dropna()
# 模拟持仓：1表示持仓，0表示空仓
position = 0
trades = []
for i in range(len(df)):
# 买入信号（不区分市场状态）
if df['buy_signal'].iloc[i] and position == 0:
position = 1
trades.append({
'date': df['date'].iloc[i],
'type': 'buy',
'price': df['close'].iloc[i],
'turnover_zscore': df['turnover_zscore'].iloc[i],
'is_frenzy': df['is_frenzy'].iloc[i]
})
# 卖出信号
elif df['sell_signal'].iloc[i] and position == 1:
position = 0
trades.append({
'date': df['date'].iloc[i],
'type': 'sell',
'price': df['close'].iloc[i],
'turnover_zscore': df['turnover_zscore'].iloc[i],
'is_frenzy': df['is_frenzy'].iloc[i]
})
# 分析交易结果
results, complete_trades = **yze_trades(trades, df)
return results, complete_trades
def **yze_trades(trades, df):
"""
分析交易结果，按市场状态分类
"""
buy_trades = [t for t in trades if t['type'] == 'buy']
sell_trades = [t for t in trades if t['type'] == 'sell']
# 配对买卖交易
complete_trades = []
for i, buy in enumerate(buy_trades):
if i < len(sell_trades):
sell = sell_trades[i]
profit = (sell['price'] - buy['price']) / buy['price'] * 100
# 找到持有期间的每日收益率
buy_date = buy['date']
sell_date = sell['date']
mask = (df['date'] >= buy_date) & (df['date'] <= sell_date)
# 计算持有期间的最大盈利和最大亏损
max_profit = profit
max_loss = profit
if mask.any():
max_price = df.loc[mask, 'close'].max()
min_price = df.loc[mask, 'close'].min()
max_profit = (max_price / buy['price'] - 1) * 100
max_loss = (min_price / buy['price'] - 1) * 100
complete_trades.append({
'buy_date': buy_date,
'sell_date': sell_date,
'buy_price': buy['price'],
'sell_price': sell['price'],
'profit_pct': profit,
'holding_days': (sell_date - buy_date).days,
'buy_frenzy': buy['is_frenzy'],
'buy_turnover_zscore': buy['turnover_zscore'],
'max_profit': max_profit,
'max_loss': max_loss
})
# 分类统计
normal_trades = [t for t in complete_trades if not t['buy_frenzy']]
frenzy_trades = [t for t in complete_trades if t['buy_frenzy']]
stats = {
'总交易次数': len(complete_trades),
'正常市场交易次数': len(normal_trades),
'狂热市场交易次数': len(frenzy_trades),
'总胜率': sum(1 for t in complete_trades if t['profit_pct'] > 0) / len(complete_trades) * 100 if complete_trades else 0,
'正常市场胜率': sum(1 for t in normal_trades if t['profit_pct'] > 0) / len(normal_trades) * 100 if normal_trades else 0,
'狂热市场胜率': sum(1 for t in frenzy_trades if t['profit_pct'] > 0) / len(frenzy_trades) * 100 if frenzy_trades else 0,
'总平均收益率': np.mean([t['profit_pct'] for t in complete_trades]) if complete_trades else 0,
'正常市场平均收益率': np.mean([t['profit_pct'] for t in normal_trades]) if normal_trades else 0,
'狂热市场平均收益率': np.mean([t['profit_pct'] for t in frenzy_trades]) if frenzy_trades else 0,
'正常市场平均盈亏比': calculate_avg_win_loss_ratio(normal_trades),
'狂热市场平均盈亏比': calculate_avg_win_loss_ratio(frenzy_trades),
'详细交易': complete_trades
}
return stats, complete_trades
def calculate_avg_win_loss_ratio(trades):
"""
计算平均盈亏比（盈利交易的平均收益率 / 亏损交易的平均亏损率）
"""
if not trades:
return 0
wins = [t['profit_pct'] for t in trades if t['profit_pct'] > 0]
losses = [abs(t['profit_pct']) for t in trades if t['profit_pct'] < 0]
if not losses:
return float('inf') if wins else 0
avg_win = np.mean(wins) if wins else 0
avg_loss = np.mean(losses) if losses else 1
return avg_win / avg_loss if avg_loss > 0 else 0
# ============================================================
# 第五步：可视化
# ============================================================
def create_visualizations(df, stats, trades):
"""
生成所有图表
"""
fig = plt.figure(figsize=(16, 14))
# ---- 图表1：价格走势 + 买卖信号 ----
ax1 = plt.subplot(3, 2, 1)
ax1.plot(df['date'], df['close'], label='收盘价', linewidth=1.5, color='black')
ax1.plot(df['date'], df['ma20'], label='MA20', linewidth=1, alpha=0.7, linestyle='--')
ax1.plot(df['date'], df['ma60'], label='MA60', linewidth=1, alpha=0.7, linestyle='--')
# 标记买入和卖出信号
buy_dates = df[df['buy_signal']]['date']
buy_prices = df[df['buy_signal']]['close']
sell_dates = df[df['sell_signal']]['date']
sell_prices = df[df['sell_signal']]['close']
ax1.scatter(buy_dates, buy_prices, color='green', marker='^', s=80, label='买入信号', zorder=5)
ax1.scatter(sell_dates, sell_prices, color='red', marker='v', s=80, label='卖出信号', zorder=5)
# 标记狂热期
frenzy_mask = df['is_frenzy']
if frenzy_mask.any():
ax1.fill_between(df['date'], df['close'].min(), df['close'].max(),
where=frenzy_mask, alpha=0.15, color='orange',
label='狂热区（拥挤度>90%分位）')
ax1.set_title('价格走势与策略信号', fontsize=12, fontweight='bold')
ax1.set_xlabel('日期')
ax1.set_ylabel('价格')
ax1.legend(loc='upper left')
ax1.grid(True, alpha=0.3)
# ---- 图表2：换手率与拥挤度 ----
ax2 = plt.subplot(3, 2, 2)
ax2.plot(df['date'], df['turnover'], label='换手率', color='blue', linewidth=1)
ax2.axhline(y=df['turnover'].quantile(0.9), color='red', linestyle='--',
label='90%分位线', alpha=0.7)
ax2.fill_between(df['date'], 0, df['turnover'],
where=df['is_frenzy'], alpha=0.3, color='orange')
ax2.set_title('换手率与拥挤度识别', fontsize=12, fontweight='bold')
ax2.set_xlabel('日期')
ax2.set_ylabel('换手率（%）')
ax2.legend(loc='upper left')
ax2.grid(True, alpha=0.3)
# ---- 图表3：换手率Z-score ----
ax3 = plt.subplot(3, 2, 3)
ax3.plot(df['date'], df['turnover_zscore'], color='purple', linewidth=1)
ax3.axhline(y=2.0, color='red', linestyle='--', label='Z=2.0警戒线', alpha=0.7)
ax3.axhline(y=-2.0, color='green', linestyle='--', alpha=0.3)
ax3.fill_between(df['date'], 2, df['turnover_zscore'].max()+1,
where=df['turnover_zscore'] > 2, alpha=0.3, color='red')
ax3.set_title('换手率Z-score（衡量拥挤度）', fontsize=12, fontweight='bold')
ax3.set_xlabel('日期')
ax3.set_ylabel('Z-score')
ax3.legend(loc='upper left')
ax3.grid(True, alpha=0.3)
# ---- 图表4：核心回测结果 - 散点图 ----
ax4 = plt.subplot(3, 2, 4)
detailed_trades = stats['详细交易']
if detailed_trades:
normal_trades = [t for t in detailed_trades if not t['buy_frenzy']]
frenzy_trades = [t for t in detailed_trades if t['buy_frenzy']]
# 正常市场交易（绿色）
if normal_trades:
ax4.scatter([t['buy_turnover_zscore'] for t in normal_trades],
[t['profit_pct'] for t in normal_trades],
c='green', label='正常市场买入', alpha=0.7, s=60)
# 狂热市场交易（红色）
if frenzy_trades:
ax4.scatter([t['buy_turnover_zscore'] for t in frenzy_trades],
[t['profit_pct'] for t in frenzy_trades],
c='red', label='狂热市场买入', alpha=0.7, s=60)
ax4.axhline(y=0, color='black', linestyle='-', alpha=0.5)
ax4.axvline(x=2.0, color='red', linestyle='--', alpha=0.5, label='Z=2.0警戒线')
ax4.set_xlabel('买入时换手率Z-score')
ax4.set_ylabel('交易收益率（%）')
ax4.set_title('不同拥挤度下买入的收益率分布', fontsize=12, fontweight='bold')
ax4.legend(loc='upper left')
ax4.grid(True, alpha=0.3)
# ---- 图表5：统计对比柱状图 ----
ax5 = plt.subplot(3, 2, 5)
metrics = ['胜率(%)', '平均收益率(%)', '盈亏比']
# 处理无穷大值
normal_ratio = stats['正常市场平均盈亏比']
frenzy_ratio = stats['狂热市场平均盈亏比']
if normal_ratio == float('inf') or normal_ratio > 100:
normal_ratio = 0
if frenzy_ratio == float('inf') or frenzy_ratio > 100:
frenzy_ratio = 0
normal_values = [
stats['正常市场胜率'],
stats['正常市场平均收益率'],
normal_ratio
]
frenzy_values = [
stats['狂热市场胜率'],
stats['狂热市场平均收益率'],
frenzy_ratio
]
x = np.arange(len(metrics))
width = 0.35
bars1 = ax5.bar(x - width/2, normal_values, width, label='正常市场', color='green', alpha=0.7)
bars2 = ax5.bar(x + width/2, frenzy_values, width, label='狂热市场', color='red', alpha=0.7)
# 在柱子上添加数值
for bar in bars1:
height = bar.get_height()
if height > 0:
ax5.text(bar.get_x() + bar.get_width()/2., height + 0.3,
f'{height:.1f}', ha='center', va='bottom', fontsize=9)
for bar in bars2:
height = bar.get_height()
if height > 0:
ax5.text(bar.get_x() + bar.get_width()/2., height + 0.3,
f'{height:.1f}', ha='center', va='bottom', fontsize=9)
ax5.set_ylabel('数值')
ax5.set_title('正常市场 vs 狂热市场 策略表现对比', fontsize=12, fontweight='bold')
ax5.set_xticks(x)
ax5.set_xticklabels(metrics)
ax5.legend()
ax5.grid(True, alpha=0.3, axis='y')
# ---- 图表6：交易明细表 ----
ax6 = plt.subplot(3, 2, 6)
ax6.axis('tight')
ax6.axis('off')
# 准备表格数据
if detailed_trades:
# 只显示最近的交易（最多10笔）
display_trades = detailed_trades[-10:] if len(detailed_trades) > 10 else detailed_trades
table_data = []
for t in display_trades:
table_data.append([
t['buy_date'].strftime('%Y-%m-%d'),
f"{t['profit_pct']:.2f}%",
f"{t['holding_days']}天",
'🔥狂热' if t['buy_frenzy'] else '✅正常',
f"{t['buy_turnover_zscore']:.2f}"
])
columns = ['买入日期', '收益率', '持有天数', '市场状态', '买入Z-score']
table = ax6.table(cellText=table_data, colLabels=columns,
cellLoc='center', loc='center',
colWidths=[0.2, 0.15, 0.15, 0.15, 0.15])
table.auto_set_font_size(False)
table.set_fontsize(9)
table.scale(1.2, 1.5)
ax6.set_title('最近交易明细', fontsize=12, fontweight='bold', pad=20)
else:
ax6.text(0.5, 0.5, '暂无完整交易数据', ha='center', va='center', fontsize=14)
plt.tight_layout()
plt.savefig('quant_trap_backtest.png', dpi=150, bbox_inches='tight')
plt.show()
print("\n✓ 图表已保存为：quant_trap_backtest.png")
# ============================================================
# 第六步：主程序执行
# ============================================================
if __name__ == "__main__":
print("\n开始回测分析...\n")
symbol=".SH"
start_date="20250101"
end_date="20260722"
# 1. 获取数据
df = get_stock_data(symbol=symbol,
start_date=start_date,
end_date=end_date)
# 2. 计算指标
df = calculate_indicators(df)
# 3. 生成信号
df = generate_signals(df)
# 4. 回测
stats, trades = backtest_strategy(df)
# 5. 打印结果
print("\n" + "=" * 60)
print("【回测结果】")
print("=" * 60)
print(f"总交易次数：{stats['总交易次数']}")
print(f" ├─ 正常市场交易：{stats['正常市场交易次数']} 次")
print(f" └─ 狂热市场交易：{stats['狂热市场交易次数']} 次")
print()
print(f"总胜率：{stats['总胜率']:.1f}%")
print(f" ├─ 正常市场胜率：{stats['正常市场胜率']:.1f}%")
print(f" └─ 狂热市场胜率：{stats['狂热市场胜率']:.1f}%")
print()
print(f"总平均收益率：{stats['总平均收益率']:.2f}%")
print(f" ├─ 正常市场平均收益率：{stats['正常市场平均收益率']:.2f}%")
print(f" └─ 狂热市场平均收益率：{stats['狂热市场平均收益率']:.2f}%")
print()
# 处理无穷大值
normal_ratio = stats['正常市场平均盈亏比']
frenzy_ratio = stats['狂热市场平均盈亏比']
if normal_ratio == float('inf') or normal_ratio > 100:
normal_ratio = 0
if frenzy_ratio == float('inf') or frenzy_ratio > 100:
frenzy_ratio = 0
print(f"正常市场盈亏比：{normal_ratio:.2f}")
print(f"狂热市场盈亏比：{frenzy_ratio:.2f}")
print("=" * 60)
# 打印详细交易
if stats['详细交易']:
print("\n【详细交易记录】")
print("-" * 80)
print(f"{'买入日期':<12} {'卖出日期':<12} {'收益率':<10} {'持有天数':<8} {'状态':<8} {'买入Z-score':<10}")
print("-" * 80)
for t in stats['详细交易']:
status = '🔥狂热' if t['buy_frenzy'] else '✅正常'
print(f"{t['buy_date'].strftime('%Y-%m-%d'):<12} "
f"{t['sell_date'].strftime('%Y-%m-%d'):<12} "
f"{t['profit_pct']:>+6.2f}% "
f"{t['holding_days']:>4}天 "
f"{status:<8} "
f"{t['buy_turnover_zscore']:>8.2f}")
print("-" * 80)
# 6. 生成图表
create_visualizations(df, stats, trades)
print("\n✅ 回测完成！")
print("\n💡 结论：从回测数据可以看出，在市场处于极度狂热状态时")
print(" （换手率Z-score > 2 或处于90%分位以上），")
print(" 趋势策略的胜率和盈亏比均明显下降，验证了标题中的观点。")
print("\n⚠️ 风险提示：本文仅为量化策略探讨，不构成投资建议。")
