huice/volume_price_distribution.py

#!/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()