综合Python案例，变向突破次数对比？

wen python案例 2026-07-27 147
我来为你设计一个综合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}")
这个综合案例展示了：