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

本文目录导读：

1. 主要功能模块：
2. 使用方法：
3. 输出结果：
4. 核心分析点：

我理解您想要一个综合的Python案例来演示"变向突破次数对比"（可能是股票/期货交易中的技术分析指标，或者是某种数据趋势分析），我来设计一个完整的案例，包含数据生成、变向突破检测、可视化对比等功能。

"""
变向突破次数对比分析系统
功能：
1. 生成模拟价格数据（正态分布/带趋势）
2. 检测价格变向突破（上破/下破关键位）
3. 对比不同策略或不同时间周期的突破次数
4. 可视化展示对比结果
"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from dataclasses import dataclass
from typing import List, Tuple, Dict
import warnings
warnings.filterwarnings('ignore')
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False    # 用来正常显示负号
# ============================ 1. 数据生成模块 ============================
class PriceDataGenerator:
    """生成模拟价格数据的类"""
    @staticmethod
    def generate_trend_data(n_points: int = 500, 
                           start_price: float = 100,
                           trend_strength: float = 0.05,
                           volatility: float = 2.0,
                           seed: int = 42) -> pd.DataFrame:
        """生成带趋势的价格数据"""
        np.random.seed(seed)
        # 生成时间索引
        dates = pd.date_range(start='2023-01-01', periods=n_points, freq='D')
        # 生成随机游走 + 趋势
        returns = np.random.normal(trend_strength, volatility, n_points)
        prices = start_price * np.exp(np.cumsum(returns))
        df = pd.DataFrame({
            'date': dates,
            'price': prices,
            'volume': np.random.randint(1000, 10000, n_points)
        })
        return df
    @staticmethod
    def generate_mean_reverting_data(n_points: int = 500,
                                    mean_price: float = 100,
                                    reversion_strength: float = 0.3,
                                    volatility: float = 5.0,
                                    seed: int = 42) -> pd.DataFrame:
        """生成均值回归的价格数据"""
        np.random.seed(seed)
        dates = pd.date_range(start='2023-01-01', periods=n_points, freq='D')
        prices = []
        current_price = mean_price
        for _ in range(n_points):
            # 均值回归：向均值靠拢 + 随机扰动
            reversion = reversion_strength * (mean_price - current_price)
            shock = np.random.normal(0, volatility)
            current_price += reversion + shock
            prices.append(current_price)
        df = pd.DataFrame({
            'date': dates,
            'price': prices,
            'volume': np.random.randint(1000, 10000, n_points)
        })
        return df
    @staticmethod
    def generate_regime_switching_data(n_points: int = 500,
                                      seed: int = 42) -> pd.DataFrame:
        """生成市场状态切换的数据（牛熊交替）"""
        np.random.seed(seed)
        dates = pd.date_range(start='2023-01-01', periods=n_points, freq='D')
        prices = []
        current_price = 100
        regime = 'bull'  # 初始状态
        for i in range(n_points):
            # 每100个时间点切换一次状态
            if i % 100 == 0:
                regime = 'bull' if np.random.random() > 0.5 else 'bear'
            if regime == 'bull':
                current_price *= (1 + np.random.normal(0.001, 0.008))
            else:
                current_price *= (1 + np.random.normal(-0.001, 0.008))
            prices.append(current_price)
        df = pd.DataFrame({
            'date': dates,
            'price': prices,
            'volume': np.random.randint(1000, 10000, n_points)
        })
        return df
# ============================ 2. 变向突破检测模块 ============================
@dataclass
class BreakoutConfig:
    """突破检测配置"""
    window_size: int = 20          # 窗口大小（用于计算支撑/压力位）
    threshold_percent: float = 0.5  # 突破阈值百分比
    min_hold_period: int = 3        # 最小持有期（避免频繁切换）
class BreakoutDetector:
    """检测价格变向突破的类"""
    def __init__(self, config: BreakoutConfig = None):
        self.config = config or BreakoutConfig()
    def detect_breakouts(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        检测所有变向突破点
        返回添加了突破标志的数据框
        """
        result_df = df.copy()
        result_df['upper_band'] = result_df['price'].rolling(
            window=self.config.window_size, center=True).max()
        result_df['lower_band'] = result_df['price'].rolling(
            window=self.config.window_size, center=True).min()
        # 计算突破阈值
        result_df['upper_threshold'] = result_df['upper_band'] * (1 + self.config.threshold_percent / 100)
        result_df['lower_threshold'] = result_df['lower_band'] * (1 - self.config.threshold_percent / 100)
        result_df['breakout_type'] = 'none'
        # 检测上破
        upper_breakout_mask = (result_df['price'] > result_df['upper_threshold']) & \
                              (result_df['price'].shift(1) <= result_df['upper_threshold'].shift(1))
        result_df.loc[upper_breakout_mask, 'breakout_type'] = 'upper'
        # 检测下破
        lower_breakout_mask = (result_df['price'] < result_df['lower_threshold']) & \
                              (result_df['price'].shift(1) >= result_df['lower_threshold'].shift(1))
        result_df.loc[lower_breakout_mask, 'breakout_type'] = 'lower'
        # 过滤短时间内重复信号
        result_df = self._filter_consolidated_signals(result_df)
        return result_df
    def _filter_consolidated_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """合并/过滤短时间内重复的突破信号"""
        breakouts = df[df['breakout_type'] != 'none'].copy()
        if len(breakouts) == 0:
            return df
        # 记录连续信号的间隔
        filtered_indices = []
        last_breakout_idx = -999
        for idx in breakouts.index:
            if idx - last_breakout_idx >= self.config.min_hold_period:
                filtered_indices.append(idx)
                last_breakout_idx = idx
        # 重置非突破信号
        df['breakout_type'] = 'none'
        df.loc[filtered_indices, 'breakout_type'] = breakouts.loc[filtered_indices, 'breakout_type']
        return df
    def count_breakouts(self, df: pd.DataFrame) -> Dict[str, int]:
        """统计各类突破次数"""
        breakout_counts = {
            'upper': len(df[df['breakout_type'] == 'upper']),
            'lower': len(df[df['breakout_type'] == 'lower']),
            'total': len(df[df['breakout_type'] != 'none'])
        }
        return breakout_counts
# ============================ 3. 对比分析模块 ============================
class BreakoutComparisonAnalyzer:
    """对比不同参数/策略下的突破情况"""
    def __init__(self):
        self.results = {}
    def compare_strategies(self, df: pd.DataFrame, 
                          configs: List[Tuple[str, BreakoutConfig]]) -> pd.DataFrame:
        """
        对比不同策略的突破表现
        返回包含各策略统计指标的DataFrame
        """
        comparison_results = []
        for strategy_name, config in configs:
            detector = BreakoutDetector(config)
            result_df = detector.detect_breakouts(df)
            counts = detector.count_breakouts(result_df)
            # 计算突破频率
            total_periods = len(df)
            counts['frequency'] = counts['total'] / total_periods
            counts['strategy'] = strategy_name
            counts['window_size'] = config.window_size
            counts['threshold'] = config.threshold_percent
            comparison_results.append(counts)
        return pd.DataFrame(comparison_results)
    def compare_time_periods(self, df: pd.DataFrame, 
                            n_periods: int = 4) -> Dict[str, pd.DataFrame]:
        """对比不同时间段的突破情况"""
        period_dfs = {}
        total_length = len(df)
        chunk_size = total_length // n_periods
        for i in range(n_periods):
            start_idx = i * chunk_size
            end_idx = (i + 1) * chunk_size if i < n_periods - 1 else total_length
            period_df = df.iloc[start_idx:end_idx].copy()
            period_dfs[f'Period_{i+1}'] = period_df
        return period_dfs
# ============================ 4. 可视化模块 ============================
class BreakoutVisualizer:
    """突破分析可视化"""
    @staticmethod
    def plot_price_with_breakouts(df: pd.DataFrame, title: str = "价格与突破点"):
        """绘制价格曲线并标注突破点"""
        fig, ax = plt.subplots(figsize=(14, 6))
        # 绘制价格
        ax.plot(df['date'], df['price'], 'b-', alpha=0.6, label='价格', linewidth=1.5)
        # 绘制上下轨
        if 'upper_band' in df.columns and 'lower_band' in df.columns:
            ax.fill_between(df['date'], df['lower_band'], df['upper_band'], 
                           alpha=0.2, color='gray', label='价格区间')
        # 标注突破点
        upper_breakouts = df[df['breakout_type'] == 'upper']
        lower_breakouts = df[df['breakout_type'] == 'lower']
        if not upper_breakouts.empty:
            ax.scatter(upper_breakouts['date'], upper_breakouts['price'], 
                      color='red', marker='^', s=100, zorder=5, label='上破')
        if not lower_breakouts.empty:
            ax.scatter(lower_breakouts['date'], lower_breakouts['price'], 
                      color='green', marker='v', s=100, zorder=5, label='下破')
        ax.set_title(title, fontsize=16, fontweight='bold')
        ax.set_xlabel('日期', fontsize=12)
        ax.set_ylabel('价格', fontsize=12)
        ax.legend(loc='best')
        ax.grid(True, alpha=0.3)
        plt.tight_layout()
        return fig
    @staticmethod
    def plot_comparison(comparison_df: pd.DataFrame, title: str = "策略对比"):
        """绘制不同策略的对比柱状图"""
        fig, ax = plt.subplots(figsize=(12, 6))
        # 设置位置
        strategies = comparison_df['strategy']
        x = np.arange(len(strategies))
        width = 0.25
        # 绘制各类突破次数
        bars1 = ax.bar(x - width, comparison_df['upper'], width, label='上破', color='red', alpha=0.7)
        bars2 = ax.bar(x, comparison_df['lower'], width, label='下破', color='green', alpha=0.7)
        bars3 = ax.bar(x + width, comparison_df['total'], width, label='总计', color='blue', alpha=0.7)
        ax.set_xlabel('策略', fontsize=12)
        ax.set_ylabel('突破次数', fontsize=12)
        ax.set_title(title, fontsize=16, fontweight='bold')
        ax.set_xticks(x)
        ax.set_xticklabels(strategies)
        ax.legend()
        # 添加数值标签
        for bars in [bars1, bars2, bars3]:
            for bar in bars:
                height = bar.get_height()
                ax.text(bar.get_x() + bar.get_width()/2., height,
                       f'{int(height)}', ha='center', va='bottom')
        ax.grid(True, alpha=0.3)
        plt.tight_layout()
        return fig
    @staticmethod
    def plot_period_comparison(period_results: Dict[str, pd.DataFrame], 
                              title: str = "时间段突破对比"):
        """绘制不同时间段的突破情况"""
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        for i, (period_name, df) in enumerate(period_results.items()):
            ax = axes[i//2, i%2]
            detector = BreakoutDetector()
            result_df = detector.detect_breakouts(df)
            ax.plot(df['date'], df['price'], 'b-', alpha=0.6, linewidth=1.5)
            # 标注突破
            upper = result_df[result_df['breakout_type'] == 'upper']
            lower = result_df[result_df['breakout_type'] == 'lower']
            ax.scatter(upper['date'], upper['price'], color='red', marker='^', s=60)
            ax.scatter(lower['date'], lower['price'], color='green', marker='v', s=60)
            counts = detector.count_breakouts(result_df)
            ax.set_title(f"{period_name}\n上破:{counts['upper']}, 下破:{counts['lower']}", fontsize=10)
            ax.grid(True, alpha=0.3)
        plt.suptitle(title, fontsize=16, fontweight='bold')
        plt.tight_layout()
        return fig
# ============================ 5. 主程序 ============================
def run_demo():
    """运行完整演示"""
    print("=" * 60)
    print("变向突破次数对比分析系统")
    print("=" * 60)
    # 1. 生成数据
    print("\n[1] 生成模拟价格数据...")
    data_gen = PriceDataGenerator()
    trend_data = data_gen.generate_trend_data(n_points=500, trend_strength=0.02, volatility=1.8)
    mean_reverting_data = data_gen.generate_mean_reverting_data(n_points=500, reversion_strength=0.2)
    regime_data = data_gen.generate_regime_switching_data(n_points=500)
    print("✓ 数据生成完成")
    # 2. 定义不同策略
    print("\n[2] 定义对比策略...")
    configs = [
        ("快速策略(窗口=10)", BreakoutConfig(window_size=10, threshold_percent=0.3, min_hold_period=2)),
        ("标准策略(窗口=20)", BreakoutConfig(window_size=20, threshold_percent=0.5, min_hold_period=3)),
        ("稳健策略(窗口=30)", BreakoutConfig(window_size=30, threshold_percent=0.8, min_hold_period=5)),
        ("激进策略(窗口=15)", BreakoutConfig(window_size=15, threshold_percent=0.2, min_hold_period=2)),
    ]
    # 3. 对比分析
    print("\n[3] 进行策略对比分析...")
    analyzer = BreakoutComparisonAnalyzer()
    # 对不同数据集进行分析
    datasets = {
        '趋势数据': trend_data,
        '均值回归数据': mean_reverting_data,
        '状态切换数据': regime_data
    }
    all_comparisons = {}
    for data_name, data in datasets.items():
        print(f"  分析{data_name}...")
        comparison = analyzer.compare_strategies(data, configs)
        all_comparisons[data_name] = comparison
        print(f"  {data_name} 突破统计:")
        print(comparison[['strategy', 'upper', 'lower', 'total']].to_string(index=False))
        print()
    # 4. 分析均值回归数据的时间段差异
    print("[4] 分析均值回归数据的时间段差异...")
    period_data = analyzer.compare_time_periods(mean_reverting_data, n_periods=4)
    print("\n各时间段突破详情:")
    detector = BreakoutDetector()
    for period, data in period_data.items():
        result = detector.detect_breakouts(data)
        counts = detector.count_breakouts(result)
        print(f"  {period}: 上破={counts['upper']}, 下破={counts['lower']}, 总计={counts['total']}")
    # 5. 可视化展示
    print("\n[5] 生成可视化图表...")
    visualizer = BreakoutVisualizer()
    # 5.1 趋势数据的价格与突破点
    fig1 = visualizer.plot_price_with_breakouts(trend_data, "趋势价格数据的变向突破")
    fig1.savefig('breakout_analysis_trend.png', dpi=150)
    print("✓ 趋势数据突破图已保存")
    # 5.2 策略对比柱状图
    fig2 = visualizer.plot_comparison(all_comparisons['均值回归数据'], "不同策略突破次数对比（均值回归数据）")
    fig2.savefig('strategy_comparison.png', dpi=150)
    print("✓ 策略对比图已保存")
    # 5.3 时间段对比
    fig3 = visualizer.plot_period_comparison(period_data, "均值回归数据不同时间段突破对比")
    fig3.savefig('period_comparison.png', dpi=150)
    print("✓ 时间段对比图已保存")
    # 6. 综合分析报告
    print("\n" + "=" * 60)
    print("综合分析报告")
    print("=" * 60)
    print("\n【各数据集最佳策略】")
    for data_name, comparison in all_comparisons.items():
        best_idx = comparison['total'].idxmax()
        best_strategy = comparison.loc[best_idx]
        print(f"  {data_name}: {best_strategy['strategy']} - 总突破{int(best_strategy['total'])}次")
    print("\n【各策略总体表现】")
    # 汇总所有数据集的统计
    summary = pd.concat(all_comparisons.values())
    strategy_summary = summary.groupby('strategy').agg({
        'upper': 'sum',
        'lower': 'sum',
        'total': 'sum'
    }).reindex([c[0] for c in configs])
    print(strategy_summary.to_string())
    # 找出方差最大的策略（波动性强）
    print("\n【策略稳定性分析】")
    for strategy in strategy_summary.index:
        strategy_data = summary[summary['strategy'] == strategy]
        std_value = strategy_data['total'].std()
        mean_value = strategy_data['total'].mean()
        cv = std_value / mean_value if mean_value != 0 else 0
        print(f"  {strategy}: 变异系数 = {cv:.2%}")
    print("\n" + "=" * 60)
    print("分析完成！所有图表已保存。")
    print("=" * 60)
if __name__ == "__main__":
    run_demo()

这个综合案例包含了以下核心功能：

主要功能模块：

1. 数据生成模块（PriceDataGenerator）：

   • 生成带趋势的数据（趋势行情）
   • 生成均值回归的数据（震荡行情）
   • 生成状态切换的数据（牛熊交替）

2. 突破检测模块（BreakoutDetector）：

   • 基于移动窗口的支撑/压力位计算
   • 上下突破信号检测
   • 信号过滤和合并

3. 对比分析模块（BreakoutComparisonAnalyzer）：

   • 多策略同步对比
   • 不同时间段对比
   • 频率和稳定性分析

4. 可视化模块（BreakoutVisualizer）：

   • 价格走势与突破标注图
   • 多策略柱状对比图
   • 时间段差异对比图

使用方法：

# 直接运行
python breakout_analysis.py

输出结果：

1. 各策略在不同市场环境下的突破次数统计表
2. 4张可视化图表（自动保存为PNG）
3. 综合分析报告（最佳策略、稳定性分析）

核心分析点：

• 策略参数对比：不同窗口大小和阈值对突破次数的影响
• 市场环境对比：趋势市 vs 震荡市 vs 牛熊交替
• 时间段对比：同一策略在不同时间段的稳定性
• 风险指标：变异系数（CV）反映策略稳定性

这个案例非常适合股票、期货、加密货币等金融数据分析场景,也可以扩展到其他领域的时间序列分析。