JUKUAN/Lib/Options/etf_monthly_ma_analysis.py

# ETF月线级别均线分析工具
# 功能：
# 1. 获取50ETF、300ETF、创业板ETF的价格数据
# 2. 计算月线级别的5、10、20日均线
# 3. 计算当月收盘价与各均线的距离
# 4. 生成可视化图表和数据导出

import jqdata
from jqdata import *
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import datetime
from typing import Dict, List, Tuple

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

class ETFMonthlyMAAnalysis:
    """ETF月线级别均线分析器"""
    
    def __init__(self, start_date='2024-01-01', end_date='2024-12-31'):
        """
        初始化分析器
        :param start_date: 开始日期
        :param end_date: 结束日期
        """
        self.start_date = start_date
        self.end_date = end_date
        
        # ETF配置 - 基于现有策略的符号定义
        self.etf_symbols = {
            '50ETF': '510050.XSHG',      # 上证50ETF
            '300ETF': '510300.XSHG',     # 沪深300ETF (用户提到的500ETF应该是指这个)
            '创业板ETF': '159915.XSHE'    # 创业板ETF
        }
        
        # 移动平均线配置
        self.ma_periods = [5, 10, 20]  # 5日、10日、20日均线
        
        # 存储数据
        self.price_data = {}
        self.monthly_data = {}
        self.analysis_results = {}
        
        print(f"初始化ETF月线分析器")
        print(f"分析时间范围: {start_date} 到 {end_date}")
        print(f"分析ETF: {list(self.etf_symbols.keys())}")
        print(f"均线周期: {self.ma_periods}日")
    
    def fetch_etf_prices(self) -> Dict[str, pd.DataFrame]:
        """
        获取所有ETF的价格数据
        返回: {etf_name: price_dataframe}
        """
        print(f"\n开始获取ETF价格数据...")
        
        for etf_name, symbol in self.etf_symbols.items():
            try:
                print(f"  获取{etf_name}({symbol})价格数据...")
                
                # 使用现有策略的get_price方法获取数据
                price_data = get_price(
                    symbol, 
                    start_date=self.start_date, 
                    end_date=self.end_date, 
                    fields=['close', 'open', 'high', 'low', 'volume'],
                    frequency='daily'
                )
                
                if price_data.empty:
                    print(f"    警告: {etf_name}无价格数据")
                    continue
                
                # 确保索引为日期格式
                if not isinstance(price_data.index, pd.DatetimeIndex):
                    price_data.index = pd.to_datetime(price_data.index)
                
                self.price_data[etf_name] = price_data
                
                print(f"    成功获取{etf_name}数据: {len(price_data)}条记录")
                print(f"    数据范围: {price_data.index[0].strftime('%Y-%m-%d')} 到 {price_data.index[-1].strftime('%Y-%m-%d')}")
                print(f"    价格范围: {price_data['close'].min():.3f} - {price_data['close'].max():.3f}")
                
            except Exception as e:
                print(f"    错误: 获取{etf_name}价格数据失败 - {e}")
                continue
        
        print(f"\nETF价格数据获取完成，成功获取{len(self.price_data)}个ETF的数据")
        return self.price_data
    
    def calculate_monthly_ma(self) -> Dict[str, pd.DataFrame]:
        """
        计算月线级别的移动平均线
        返回: {etf_name: monthly_data_with_ma}
        """
        print(f"\n开始计算月线级别的移动平均线...")
        
        for etf_name, price_df in self.price_data.items():
            try:
                print(f"  处理{etf_name}的月线数据...")
                
                # 转换为月线数据 - 取每月最后一个交易日的数据
                monthly_df = price_df.resample('M').agg({
                    'open': 'first',    # 月初开盘价
                    'high': 'max',      # 月内最高价
                    'low': 'min',       # 月内最低价
                    'close': 'last',    # 月末收盘价
                    'volume': 'sum'     # 月成交量
                })
                
                # 计算移动平均线
                for period in self.ma_periods:
                    ma_column = f'MA{period}'
                    monthly_df[ma_column] = monthly_df['close'].rolling(window=period).mean()
                    print(f"    计算{period}日均线完成")
                
                # 计算当月收盘价与各均线的距离
                for period in self.ma_periods:
                    ma_column = f'MA{period}'
                    distance_column = f'Distance_MA{period}'
                    monthly_df[distance_column] = monthly_df['close'] - monthly_df[ma_column]
                    
                    # 计算距离的百分比
                    pct_distance_column = f'Distance_MA{period}_Pct'
                    monthly_df[pct_distance_column] = (monthly_df['close'] - monthly_df[ma_column]) / monthly_df[ma_column] * 100
                
                # 添加月份标识
                monthly_df['年月'] = monthly_df.index.strftime('%Y-%m')
                
                self.monthly_data[etf_name] = monthly_df
                
                print(f"    {etf_name}月线数据计算完成: {len(monthly_df)}个月")
                
                # 显示最近几个月的数据示例
                if len(monthly_df) > 0:
                    latest_data = monthly_df.tail(3)
                    print(f"    最近数据示例:")
                    for idx, row in latest_data.iterrows():
                        month_str = idx.strftime('%Y-%m')
                        close_price = row['close']
                        ma5_dist = row.get('Distance_MA5', np.nan)
                        ma10_dist = row.get('Distance_MA10', np.nan)
                        ma20_dist = row.get('Distance_MA20', np.nan)
                        print(f"      {month_str}: 收盘{close_price:.3f}, 距MA5={ma5_dist:.3f}, 距MA10={ma10_dist:.3f}, 距MA20={ma20_dist:.3f}")
                
            except Exception as e:
                print(f"    错误: 处理{etf_name}月线数据失败 - {e}")
                continue
        
        print(f"\n月线移动平均线计算完成，处理了{len(self.monthly_data)}个ETF")
        return self.monthly_data
    
    def analyze_distance_statistics(self) -> Dict[str, Dict]:
        """
        分析距离统计信息
        返回: {etf_name: statistics}
        """
        print(f"\n开始分析距离统计信息...")
        
        for etf_name, monthly_df in self.monthly_data.items():
            try:
                print(f"  分析{etf_name}的距离统计...")
                
                stats = {}
                
                # 对每个均线周期进行统计
                for period in self.ma_periods:
                    distance_col = f'Distance_MA{period}'
                    pct_distance_col = f'Distance_MA{period}_Pct'
                    
                    if distance_col in monthly_df.columns:
                        distance_data = monthly_df[distance_col].dropna()
                        pct_distance_data = monthly_df[pct_distance_col].dropna()
                        
                        if len(distance_data) > 0:
                            stats[f'MA{period}'] = {
                                '绝对距离_均值': distance_data.mean(),
                                '绝对距离_标准差': distance_data.std(),
                                '绝对距离_最大值': distance_data.max(),
                                '绝对距离_最小值': distance_data.min(),
                                '百分比距离_均值': pct_distance_data.mean(),
                                '百分比距离_标准差': pct_distance_data.std(),
                                '百分比距离_最大值': pct_distance_data.max(),
                                '百分比距离_最小值': pct_distance_data.min(),
                                '正向距离次数': (distance_data > 0).sum(),
                                '负向距离次数': (distance_data < 0).sum(),
                                '数据点总数': len(distance_data)
                            }
                            
                            print(f"    MA{period}统计: 均值={stats[f'MA{period}']['绝对距离_均值']:.3f}, "
                                  f"正向{stats[f'MA{period}']['正向距离次数']}次, "
                                  f"负向{stats[f'MA{period}']['负向距离次数']}次")
                
                self.analysis_results[etf_name] = stats
                
            except Exception as e:
                print(f"    错误: 分析{etf_name}距离统计失败 - {e}")
                continue
        
        print(f"\n距离统计分析完成")
        return self.analysis_results
    
    def plot_distance_chart(self, save_path: str = None):
        """
        绘制距离图表
        :param save_path: 保存路径，如果为None则直接显示
        """
        print(f"\n开始绘制距离图表...")
        
        # 设置图表布局：每个ETF一个子图
        n_etfs = len(self.monthly_data)
        if n_etfs == 0:
            print("没有数据可以绘制")
            return
        
        fig, axes = plt.subplots(n_etfs, 1, figsize=(14, 6 * n_etfs))
        if n_etfs == 1:
            axes = [axes]  # 确保axes是列表
        
        # 颜色配置
        colors = ['red', 'blue', 'green']
        ma_labels = [f'距MA{period}日线' for period in self.ma_periods]
        
        for idx, (etf_name, monthly_df) in enumerate(self.monthly_data.items()):
            ax = axes[idx]
            
            # 准备数据
            months = monthly_df.index
            month_labels = [dt.strftime('%Y-%m') for dt in months]
            
            # 绘制距离线
            for i, period in enumerate(self.ma_periods):
                distance_col = f'Distance_MA{period}'
                if distance_col in monthly_df.columns:
                    distance_data = monthly_df[distance_col]
                    ax.plot(months, distance_data, 
                           label=ma_labels[i], 
                           color=colors[i], 
                           linewidth=2, 
                           marker='o', 
                           markersize=4)
            
            # 添加零线
            ax.axhline(y=0, color='black', linestyle='--', alpha=0.5, linewidth=1)
            
            # 设置标题和标签
            ax.set_title(f'{etf_name} 月收盘价与移动平均线距离', fontsize=14, fontweight='bold')
            ax.set_xlabel('月份', fontsize=12)
            ax.set_ylabel('距离 (元)', fontsize=12)
            ax.legend(fontsize=10)
            ax.grid(True, alpha=0.3)
            
            # 设置x轴标签
            if len(months) > 12:
                # 如果月份太多，只显示部分标签
                step = max(1, len(months) // 12)
                ax.set_xticks(months[::step])
                ax.set_xticklabels([dt.strftime('%Y-%m') for dt in months[::step]], rotation=45)
            else:
                ax.set_xticklabels(month_labels, rotation=45)
            
            # 添加统计信息文本
            if etf_name in self.analysis_results:
                stats_text = []
                for period in self.ma_periods:
                    ma_key = f'MA{period}'
                    if ma_key in self.analysis_results[etf_name]:
                        mean_dist = self.analysis_results[etf_name][ma_key]['绝对距离_均值']
                        positive_count = self.analysis_results[etf_name][ma_key]['正向距离次数']
                        negative_count = self.analysis_results[etf_name][ma_key]['负向距离次数']
                        stats_text.append(f'MA{period}: 均值{mean_dist:.3f}, 正{positive_count}负{negative_count}')
                
                if stats_text:
                    ax.text(0.02, 0.98, '\n'.join(stats_text), 
                           transform=ax.transAxes, 
                           fontsize=9, 
                           verticalalignment='top',
                           bbox=dict(boxstyle='round,pad=0.3', facecolor='lightgray', alpha=0.8))
        
        plt.tight_layout()
        
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
            print(f"图表已保存到: {save_path}")
        else:
            plt.show()
        
        print("距离图表绘制完成")
    
    def export_results_to_csv(self, output_dir: str = "./"):
        """
        导出结果到CSV文件
        :param output_dir: 输出目录
        """
        print(f"\n开始导出结果到CSV文件...")
        
        for etf_name, monthly_df in self.monthly_data.items():
            try:
                # 准备导出数据
                export_df = monthly_df.copy()
                
                # 重新排列列的顺序，把重要信息放在前面
                column_order = ['年月', 'close']
                
                # 添加均线列
                for period in self.ma_periods:
                    column_order.append(f'MA{period}')
                
                # 添加距离列
                for period in self.ma_periods:
                    column_order.append(f'Distance_MA{period}')
                    column_order.append(f'Distance_MA{period}_Pct')
                
                # 添加其他列
                other_columns = [col for col in export_df.columns if col not in column_order]
                column_order.extend(other_columns)
                
                # 重新排序并导出
                export_df = export_df.reindex(columns=column_order)
                
                # 构建文件名
                filename = f"{output_dir}/ETF_月线分析_{etf_name}_{self.start_date}_{self.end_date}.csv"
                export_df.to_csv(filename, encoding='utf-8-sig', index=True)
                
                print(f"  {etf_name}数据已导出到: {filename}")
                
            except Exception as e:
                print(f"  错误: 导出{etf_name}数据失败 - {e}")
        
        print("CSV导出完成")
    
    def run_complete_analysis(self, save_chart: bool = True, export_csv: bool = True):
        """
        运行完整的分析流程
        :param save_chart: 是否保存图表
        :param export_csv: 是否导出CSV
        """
        print("="*60)
        print("开始ETF月线级别均线距离分析")
        print("="*60)
        
        try:
            # 1. 获取ETF价格数据
            print(f"\n步骤1: 获取ETF价格数据")
            self.fetch_etf_prices()
            
            if not self.price_data:
                print("错误: 没有成功获取任何ETF价格数据")
                return
            
            # 2. 计算月线均线
            print(f"\n步骤2: 计算月线级别移动平均线")
            self.calculate_monthly_ma()
            
            if not self.monthly_data:
                print("错误: 没有成功计算任何月线数据")
                return
            
            # 3. 分析距离统计
            print(f"\n步骤3: 分析距离统计信息")
            self.analyze_distance_statistics()
            
            # 4. 绘制图表
            print(f"\n步骤4: 绘制距离图表")
            if save_chart:
                chart_filename = f"ETF月线距离分析_{self.start_date}_{self.end_date}.png"
                self.plot_distance_chart(save_path=chart_filename)
            else:
                self.plot_distance_chart()
            
            # 5. 导出CSV
            if export_csv:
                print(f"\n步骤5: 导出分析结果")
                self.export_results_to_csv()
            
            # 6. 输出总结
            print(f"\n" + "="*60)
            print("分析总结")
            print("="*60)
            
            for etf_name, stats in self.analysis_results.items():
                print(f"\n{etf_name}分析结果:")
                for ma_period, stat_data in stats.items():
                    print(f"  {ma_period}:")
                    print(f"    平均距离: {stat_data['绝对距离_均值']:.3f} 元")
                    print(f"    平均距离百分比: {stat_data['百分比距离_均值']:.2f}%")
                    print(f"    正向/负向比例: {stat_data['正向距离次数']}/{stat_data['负向距离次数']}")
            
            print(f"\n分析完成! 总共处理了{len(self.monthly_data)}个ETF，{sum(len(df) for df in self.monthly_data.values())}个月度数据点")
            
        except Exception as e:
            print(f"分析过程中发生错误: {e}")
            import traceback
            traceback.print_exc()

# =============================================================================
# 直接运行分析 - 可以修改下面的参数来自定义分析
# =============================================================================

print("="*60)
print("ETF月线级别均线距离分析工具")
print("="*60)
print("功能：分析ETF价格与5、10、20日均线的距离关系")
print("ETF标的：50ETF、300ETF、创业板ETF")
print()

# 分析参数配置（可根据需要修改）
START_DATE = '2022-01-01'  # 开始日期
END_DATE = '2024-12-31'    # 结束日期

print(f"分析时间范围: {START_DATE} 到 {END_DATE}")
print()

# 创建分析器实例
analyzer = ETFMonthlyMAAnalysis(
    start_date=START_DATE, 
    end_date=END_DATE
)

# 运行完整分析
analyzer.run_complete_analysis(
    save_chart=False,  # 保存图表到文件
    export_csv=False   # 导出CSV数据
)