|
|
@@ -0,0 +1,426 @@
|
|
|
+# 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数据
|
|
|
+)
|
maxfeng