#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ AU2512期货成交量在价格上的分布图 这个脚本专门用于生成当前成交量在价格上的分布图， 简单直观地展示不同价格水平下的成交量分布情况。 使用方法: python volume_price_distribution.py [data_file] 如果不指定文件，默认分析 data/au2512_20251013.parquet """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import warnings import os import sys import argparse warnings.filterwarnings('ignore') class VolumePriceDistributionAnalyzer: """成交量-价格分布分析器""" def __init__(self, data_file=None): """ 初始化分析器 Args: data_file (str): 数据文件路径，默认为 data/au2512_20251013.parquet """ self.data_file = data_file or "data/au2512_20251013.parquet" self.df = None self.setup_chinese_font() def setup_chinese_font(self): """设置中文字体支持""" try: # 直接使用已知可用的中文字体 chinese_fonts = ['Microsoft YaHei', 'SimHei', 'SimSun', 'KaiTi', 'FangSong', 'DengXian'] for font in chinese_fonts: try: plt.rcParams['font.sans-serif'] = [font] plt.rcParams['axes.unicode_minus'] = False plt.rcParams['font.size'] = 12 plt.rcParams['axes.titlesize'] = 16 plt.rcParams['axes.labelsize'] = 14 plt.rcParams['xtick.labelsize'] = 12 plt.rcParams['ytick.labelsize'] = 12 # 简单测试字体 fig, ax = plt.subplots(figsize=(1, 1)) ax.text(0.5, 0.5, '测试', fontsize=12) plt.close(fig) print(f"[*] 成功设置中文字体: {font}") return except: continue # 如果都失败，使用默认设置 print("[!] 无法设置中文字体，使用默认字体") plt.rcParams['font.sans-serif'] = ['DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False plt.rcParams['font.size'] = 12 except Exception as e: print(f"[!] 字体设置警告: {e}") plt.rcParams['font.sans-serif'] = ['DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False plt.rcParams['font.size'] = 12 def load_data(self): """加载数据""" try: if not os.path.exists(self.data_file): raise FileNotFoundError(f"数据文件不存在: {self.data_file}") print(f"正在加载数据: {self.data_file}") # 根据文件扩展名选择读取方式 if self.data_file.endswith('.parquet'): self.df = pd.read_parquet(self.data_file) elif self.data_file.endswith('.csv'): self.df = pd.read_csv(self.data_file) else: raise ValueError("不支持的文件格式，请使用 .parquet 或 .csv 文件") print(f"数据加载成功: {len(self.df):,} 条记录") return True except Exception as e: print(f"数据加载失败: {e}") return False def prepare_data(self): """准备分析数据""" print("\n=== 准备价格-成交量分布数据 ===") # 获取列名 price_col = None cumulative_volume_col = None # 自动检测关键列 for i, col in enumerate(self.df.columns): if '成交价' in col and price_col is None: price_col = col elif cumulative_volume_col is None and i == 3: # 累计成交量通常在第4列 cumulative_volume_col = col if price_col is None or cumulative_volume_col is None: raise ValueError("无法找到成交价或累计成交量列") print(f"使用价格列: {price_col}") print(f"使用累计成交量列: {cumulative_volume_col}") # 准备分析数据 analysis_data = self.df[[price_col, cumulative_volume_col]].copy() analysis_data.columns = ['价格', '累计成交量'] # 计算当前成交量 analysis_data['当前成交量'] = analysis_data['累计成交量'].diff().fillna(0) analysis_data.loc[analysis_data.index[0], '当前成交量'] = analysis_data.loc[analysis_data.index[0], '累计成交量'] # 移除零值和异常值 valid_data = analysis_data[ (analysis_data['当前成交量'] > 0) & (analysis_data['当前成交量'] < analysis_data['当前成交量'].quantile(0.999)) ].copy() print(f"原始数据点: {len(analysis_data):,}") print(f"有效数据点: {len(valid_data):,}") print(f"价格范围: {valid_data['价格'].min():.2f} - {valid_data['价格'].max():.2f}") print(f"成交量范围: {valid_data['当前成交量'].min():.2f} - {valid_data['当前成交量'].max():.2f} 手") print(f"价格间隔: 0.02 元/格") self.analysis_data = valid_data return True def create_volume_distribution_chart(self): """创建成交量在价格上的分布图""" print("\n=== 生成成交量价格分布图 ===") # 创建图表 plt.figure(figsize=(16, 10)) # 创建价格分组 prices = self.analysis_data['价格'] min_price = prices.min() max_price = prices.max() # 使用固定0.02价格间隔 interval_size = 0.02 price_bins = np.arange(np.floor(min_price * 50) / 50, # 向下取整到0.02的倍数 np.ceil(max_price * 50) / 50 + interval_size, # 向上取整 interval_size) bin_centers = (price_bins[:-1] + price_bins[1:]) / 2 # 计算每个价格分组的成交量 volume_by_price = [] for i in range(len(price_bins) - 1): mask = (self.analysis_data['价格'] >= price_bins[i]) & (self.analysis_data['价格'] < price_bins[i + 1]) volume_sum = self.analysis_data.loc[mask, '当前成交量'].sum() volume_by_price.append(volume_sum) volume_by_price = np.array(volume_by_price) # 绘制成交量分布柱状图 colors = plt.cm.RdYlBu_r(np.linspace(0, 1, len(bin_centers))) bars = plt.bar(bin_centers, volume_by_price, width=(price_bins[1] - price_bins[0]) * 0.8, color=colors, alpha=0.8, edgecolor='black', linewidth=0.5) # 设置标题和标签 plt.title('AU2512期货成交量在价格上的分布 (价格间隔: 0.02)', fontsize=20, fontweight='bold', pad=20) plt.xlabel('价格', fontsize=16, labelpad=10) plt.ylabel('累计成交量 (手)', fontsize=16, labelpad=10) plt.grid(True, alpha=0.3, axis='y') # 找出成交量最大的几个价格区间 top_indices = np.argsort(volume_by_price)[-5:] # 取前5个 # 标注成交量最大的价格区间 for i in top_indices: if volume_by_price[i] > 0: plt.annotate(f'{volume_by_price[i]:.0f}手', xy=(bin_centers[i], volume_by_price[i]), xytext=(0, 20), textcoords='offset points', ha='center', va='bottom', fontsize=11, fontweight='bold', color='red', bbox=dict(boxstyle='round,pad=0.3', facecolor='yellow', alpha=0.8), arrowprops=dict(arrowstyle='->', color='red', alpha=0.7)) # 添加统计信息 max_volume = volume_by_price.max() max_price_range = price_bins[np.argmax(volume_by_price)] max_price_range_end = price_bins[np.argmax(volume_by_price) + 1] total_volume = volume_by_price.sum() max_volume_percentage = (max_volume / total_volume) * 100 stats_text = f'''统计信息: • 价格区间: {min_price:.2f} - {max_price:.2f} • 总成交量: {total_volume:,.0f} 手 • 最大成交量: {max_volume:,.0f} 手 ({max_volume_percentage:.1f}%) • 最活跃价格区间: {max_price_range:.2f} - {max_price_range_end:.2f}''' plt.text(0.98, 0.98, stats_text, transform=plt.gca().transAxes, fontsize=12, ha='right', va='top', bbox=dict(boxstyle='round,pad=0.5', facecolor='lightgreen', alpha=0.9)) # 格式化y轴标签 plt.gca().yaxis.set_major_formatter(plt.FuncFormatter(lambda x, p: f'{x/1000:.0f}K')) # 调整布局 plt.tight_layout() # 保存图表 chart_file = 'au2512_volume_price_distribution.png' plt.savefig(chart_file, dpi=300, bbox_inches='tight', facecolor='white', edgecolor='none') plt.close() print(f"成交量价格分布图已保存: {chart_file}") return chart_file def print_summary(self): """打印分析摘要""" print("\n" + "="*60) print("成交量在价格上分布分析摘要") print("="*60) prices = self.analysis_data['价格'] volumes = self.analysis_data['当前成交量'] print(f"\n【数据概况】") print(f"数据文件: {self.data_file}") print(f"有效记录数: {len(self.analysis_data):,}") print(f"价格范围: {prices.min():.2f} - {prices.max():.2f}") print(f"成交量范围: {volumes.min():.2f} - {volumes.max():.2f} 手") print(f"平均成交量: {volumes.mean():.2f} 手") # 创建价格分组的详细数据 min_price = prices.min() max_price = prices.max() interval_size = 0.02 price_bins = np.arange(np.floor(min_price * 50) / 50, np.ceil(max_price * 50) / 50 + interval_size, interval_size) bin_centers = (price_bins[:-1] + price_bins[1:]) / 2 # 计算每个价格分组的成交量 volume_by_price = [] for i in range(len(price_bins) - 1): mask = (self.analysis_data['价格'] >= price_bins[i]) & (self.analysis_data['价格'] < price_bins[i + 1]) volume_sum = self.analysis_data.loc[mask, '当前成交量'].sum() volume_by_price.append(volume_sum) volume_by_price = np.array(volume_by_price) total_volume = volume_by_price.sum() # 找出成交量最大的价格区间 max_idx = np.argmax(volume_by_price) max_volume = volume_by_price[max_idx] max_price_start = price_bins[max_idx] max_price_end = price_bins[max_idx + 1] print(f"\n【最活跃价格区间】") print(f"价格区间: {max_price_start:.2f} - {max_price_end:.2f}") print(f"区间成交量: {max_volume:,.0f} 手") print(f"占总成交量比例: {(max_volume/total_volume*100):.1f}%") # 列出前20个成交最大的价格区间 print(f"\n【前20个成交最大的价格区间】") print(f"{'排名':>4} {'价格区间':>15} {'成交量':>12} {'占比':>8} {'累计占比':>10}") print("-" * 65) # 获取排序后的索引 sorted_indices = np.argsort(volume_by_price)[::-1] # 降序排列 cumulative_percentage = 0 for rank in range(min(20, len(sorted_indices))): idx = sorted_indices[rank] if volume_by_price[idx] > 0: # 只显示有成交量的区间 price_start = price_bins[idx] price_end = price_bins[idx + 1] volume = volume_by_price[idx] percentage = (volume / total_volume) * 100 cumulative_percentage += percentage print(f"{rank+1:>4} {price_start:>7.2f}-{price_end:<7.2f} {volume:>10,.0f} {percentage:>6.1f}% {cumulative_percentage:>8.1f}%") print(f"\n【生成的图表】") print(f" au2512_volume_price_distribution.png - 成交量价格分布图") print(f"\n" + "="*60) print("分析完成！") print("="*60) def run_analysis(self): """运行完整分析""" print("开始AU2512期货成交量价格分布分析...") # 加载数据 if not self.load_data(): return False # 准备数据 if not self.prepare_data(): return False # 创建图表 self.create_volume_distribution_chart() # 打印摘要 self.print_summary() return True def main(): """主函数""" parser = argparse.ArgumentParser(description='AU2512期货成交量价格分布分析工具') parser.add_argument('data_file', nargs='?', default='data/au2512_20251013.parquet', help='数据文件路径 (默认: data/au2512_20251013.parquet)') parser.add_argument('--output-dir', '-o', default='.', help='输出目录 (默认: 当前目录)') args = parser.parse_args() # 检查数据文件是否存在 if not os.path.exists(args.data_file): print(f"错误: 数据文件不存在 - {args.data_file}") print("\n使用方法:") print(" python volume_price_distribution.py [数据文件路径]") print(f" python volume_price_distribution.py # 使用默认文件: {args.data_file}") sys.exit(1) # 创建分析器并运行 analyzer = VolumePriceDistributionAnalyzer(args.data_file) success = analyzer.run_analysis() if success: print(f"\n分析完成！图表已保存到: {args.output_dir}") else: print(f"\n分析失败！") sys.exit(1) if __name__ == "__main__": main()