我来为你设计一个综合Python案例,演示三种不同方式实现"变向突破次数对比"（比如在股票、交易数据分析中常见）。
场景：检测价格序列的变向突破次数
假设我们有一个股票价格序列,需要统计不同策略下价格突破前期高/低点的次数对比。
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from typing import List, Tuple, Optional
import warnings
warnings.filterwarnings('ignore')
class TrendBreakAnalyzer:
"""
趋势变向突破分析器
对比不同周期/策略下的突破次数
"""
def __init__(self, prices: List[float], dates: Optional[List[str]] = None):
"""
初始化价格数据
:param prices: 价格序列
:param dates: 日期序列（可选）
"""
self.prices = np.array(prices)
self.n = len(prices)
self.dates = dates if dates else [f'T-{self.n-i}' for i in range(self.n)]
# 计算基本统计量
self.high = np.maximum.accumulate(self.prices)
self.low = np.minimum.accumulate(self.prices)
def method1_simple_break(self, lookback: int = 5) -> Tuple[int, int]:
"""
方法1：简单N日突破
统计价格突破前N日高点/低点的次数
"""
up_breaks = 0
down_breaks = 0
for i in range(lookback, self.n):
# 前N日最高价
prev_high = np.max(self.prices[i-lookback:i])
prev_low = np.min(self.prices[i-lookback:i])
# 当前价格突破
if self.prices[i] > prev_high:
up_breaks += 1
elif self.prices[i] < prev_low:
down_breaks += 1
return up_breaks, down_breaks
def method2_donchian_channel(self, period: int = 20) -> Tuple[int, int]:
"""
方法2：唐奇安通道突破
使用上下轨进行突破统计
"""
if period > self.n:
return 0, 0
up_breaks = 0
down_breaks = 0
last_break = None # 记录上次突破方向
for i in range(period, self.n):
channel_high = np.max(self.prices[i-period:i])
channel_low = np.min(self.prices[i-period:i])
# 检测突破
if self.prices[i] > channel_high and last_break != 'up':
up_breaks += 1
last_break = 'up'
elif self.prices[i] < channel_low and last_break != 'down':
down_breaks += 1
last_break = 'down'
return up_breaks, down_breaks
def method3_zigzag_break(self, retrace_threshold: float = 0.05) -> Tuple[int, int]:
"""
方法3：之字形突破
基于价格回撤百分比识别转折点并统计突破
"""
if self.n < 3:
return 0, 0
# 识别转折点
turning_points = []
direction = None # 'up' 或 'down'
for i in range(1, self.n - 1):
# 局部高点
if self.prices[i] > self.prices[i-1] and self.prices[i] > self.prices[i+1]:
turning_points.append((i, 'high', self.prices[i]))
# 局部低点
elif self.prices[i] < self.prices[i-1] and self.prices[i] < self.prices[i+1]:
turning_points.append((i, 'low', self.prices[i]))
# 统计突破次数
up_breaks = 0
down_breaks = 0
last_trend = None
for idx, point_type, price in turning_points:
if point_type == 'high':
if last_trend == 'down' or last_trend is None:
# 检查是否突破前一个高点
if len(turning_points) > 1 and price > turning_points[-2][2]:
up_breaks += 1
last_trend = 'up'
elif point_type == 'low':
if last_trend == 'up' or last_trend is None:
if len(turning_points) > 1 and price < turning_points[-2][2]:
down_breaks += 1
last_trend = 'down'
return up_breaks, down_breaks
def compare_all_methods(self) -> pd.DataFrame:
"""
对比所有方法的突破次数
"""
results = {}
# 不同参数组合
params = {
'简单5日突破': {'method': 'simple', 'lookback': 5},
'简单10日突破': {'method': 'simple', 'lookback': 10},
'唐奇安20日': {'method': 'donchian', 'period': 20},
'唐奇安30日': {'method': 'donchian', 'period': 30},
'之字形5%': {'method': 'zigzag', 'threshold': 0.05},
'之字形10%': {'method': 'zigzag', 'threshold': 0.10}
}
for name, param in params.items():
if param['method'] == 'simple':
up, down = self.method1_simple_break(param['lookback'])
elif param['method'] == 'donchian':
up, down = self.method2_donchian_channel(param['period'])
else:
up, down = self.method3_zigzag_break(param['threshold'])
results[name] = {
'向上突破': up,
'向下突破': down,
'总突破': up + down,
'成功率': f"{up/(up+down)*100:.1f}%" if (up+down) > 0 else 'N/A'
}
df = pd.DataFrame(results).T
df.index.name = '策略'
return df
def visualize_breaks(self, method: str = 'simple', **kwargs):
"""
可视化突破点
"""
fig, axes = plt.subplots(2, 1, figsize=(15, 10))
# 价格和移动平均线
axes[0].plot(self.dates, self.prices, label='价格', color='blue', alpha=0.7)
if self.n > 20:
ma20 = pd.Series(self.prices).rolling(20).mean()
axes[0].plot(self.dates, ma20, label='MA20', color='orange', linestyle='--')
axes[0].set_title('价格走势与突破点')
axes[0].set_ylabel('价格')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# 标记突破点
break_points = {'up': [], 'down': []}
if method == 'simple':
lookback = kwargs.get('lookback', 5)
for i in range(lookback, self.n):
prev_high = np.max(self.prices[i-lookback:i])
prev_low = np.min(self.prices[i-lookback:i])
if self.prices[i] > prev_high:
break_points['up'].append(i)
axes[0].scatter(i, self.prices[i], color='green', s=100,
marker='^', zorder=5, alpha=0.8)
elif self.prices[i] < prev_low:
break_points['down'].append(i)
axes[0].scatter(i, self.prices[i], color='red', s=100,
marker='v', zorder=5, alpha=0.8)
# 第二个图：突破次数对比
methods_data = self.compare_all_methods()
x = range(len(methods_data))
width = 0.35
bars1 = axes[1].bar([i - width/2 for i in x], methods_data['向上突破'],
width, label='向上突破', color='green', alpha=0.7)
bars2 = axes[1].bar([i + width/2 for i in x], methods_data['向下突破'],
width, label='向下突破', color='red', alpha=0.7)
axes[1].set_xlabel('策略')
axes[1].set_ylabel('突破次数')
axes[1].set_title('各策略突破次数对比')
axes[1].set_xticks(x)
axes[1].set_xticklabels(methods_data.index, rotation=45, ha='right')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
# 在柱状图上添加数值标签
for bar in bars1:
height = bar.get_height()
axes[1].text(bar.get_x() + bar.get_width()/2., height,
f'{int(height)}', ha='center', va='bottom')
for bar in bars2:
height = bar.get_height()
axes[1].text(bar.get_x() + bar.get_width()/2., height,
f'{int(height)}', ha='center', va='bottom')
plt.tight_layout()
plt.show()
# 输出统计结果
print("\n" + "="*60)
print(f"变向突破次数统计对比 (基于{method}方法)")
print("="*60)
print(f"向上突破次数: {len(break_points['up'])}")
print(f"向下突破次数: {len(break_points['down'])}")
print(f"总突破次数: {len(break_points['up']) + len(break_points['down'])}")
print(f"突破频率: {(len(break_points['up']) + len(break_points['down'])) / self.n * 100:.2f}%")
# 示例运行
def generate_sample_data(n_days: int = 200) -> List[float]:
"""
生成模拟价格数据（含趋势和波动）
"""
np.random.seed(42)
# 基础价格
base_price = 100
# 生成趋势
trend = np.cumsum(np.random.randn(n_days) * 0.5)
trend = trend - trend[0] # 从0开始
# 添加季节性
seasonal = 5 * np.sin(np.linspace(0, 4*np.pi, n_days))
# 添加噪声
noise = np.random.randn(n_days) * 2
# 组合
prices = base_price + trend + seasonal + noise
# 确保价格为正
prices = np.maximum(prices, 10)
return prices.tolist()
# 主程序
if __name__ == "__main__":
# 生成模拟数据
prices = generate_sample_data(200)
dates = pd.date_range('2024-01-01', periods=200, freq='D').strftime('%Y-%m-%d').tolist()
# 创建分析器实例
analyzer = TrendBreakAnalyzer(prices, dates)
# 执行所有方法并对比
print("="*60)
print("趋势变向突破分析 - 综合对比")
print("="*60)
# 详细对比
result_df = analyzer.compare_all_methods()
print("\n各策略突破次数对比:")
print(result_df.to_string())
# 可视化
analyzer.visualize_breaks(method='simple', lookback=10)
# 额外分析：寻找最佳参数
print("\n" + "="*60)
print("参数优化分析")
print("="*60)
lookback_range = range(5, 31, 5)
results = []
for lookback in lookback_range:
up, down = analyzer.method1_simple_break(lookback)
total = up + down
results.append({
'周期': lookback,
'向上突破': up,
'向下突破': down,
'总突破': total,
'单日突破频率': f"{total/analyzer.n*100:.2f}%"
})
opt_df = pd.DataFrame(results)
print(opt_df.to_string(index=False))
print("\n分析结论:")
print(f"- 最优参数: 周期{opt_df.loc[opt_df['总突破'].idxmax(), '周期']}日")
print(f"- 最高突破频率: {opt_df['单日突破频率'].max()}")
print(f"- 平均突破次数: {opt_df['总突破'].mean():.1f}")
这个综合案例展示了：
三种不同突破策略
- 简单N日突破：最基础的方法
- 唐奇安通道：防止频繁交易
- 之字形突破：基于转折点识别
核心功能模块
- 数据生成与预处理
- 多种策略实现
- 性能对比分析
- 可视化展示
实际应用价值
- 股票/期货交易策略分析
- 技术指标回测
- 参数优化
运行结果会展示每种策略的突破次数对比、可视化图表以及参数优化建议，帮助你选择最适合当前市场环境的突破策略。
