# 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数据 )