Python量化分析：主力资金成本支撑位识别与筹码分布实战

import pandas as pd
import numpy as np
import qstock as qs
import matplotlib.pyplot as plt
from scipy.stats import gaussian_kde

def calculate_chip_distribution(df, target_date, price_range=100):
    """
    计算指定历史截止日期的筹码分布
    :param df: 包含'open', 'high', 'low', 'close', 'volume'列的DataFrame
    :param target_date: 目标截止日期（格式'YYYY-MM-DD'）
    :param price_range: 价格区间划分精度（单位：分，即0.01元）
    :return: 筹码分布数组（价格序列, 对应筹码占比序列）
    """
    # 筛选目标日期前的所有历史交易数据
    historical_data = df[df['date'] <= target_date].copy()
    # 定义价格最小变动单位
    price_step = 0.01  # 精确到分
    # 确定整体价格分析范围，预留边界
    min_price = historical_data['low'].min() - 1 * price_step
    max_price = historical_data['high'].max() + 1 * price_step
    price_bins = np.arange(min_price, max_price, price_step)

    # 初始化筹码分布数组
    chip_dist = np.zeros_like(price_bins, dtype=float)
    total_volume = historical_data['volume'].sum()

    # 遍历每一个交易日，分配当日筹码
    for _, row in historical_data.iterrows():
        o, h, l, c, v = row['open'], row['high'], row['low'], row['close'], row['volume']
        # 确定当日实际价格波动区间（简化处理为OHLC中的高低价）
        day_low = min(l, o)
        day_high = max(h, o)
        # 计算筹码在当日价格区间内的均匀分布密度
        if day_high == day_low:
            continue  # 避免除以零
        volume_per_price = v / (day_high - day_low)
        # 找到该日价格区间在总价格序列中的对应索引位置
        start_idx = np.searchsorted(price_bins, day_low, side='left')
        end_idx = np.searchsorted(price_bins, day_high, side='right')
        # 将当日筹码累加到全局分布中
        chip_dist[start_idx:end_idx] += volume_per_price * price_step

    # 将绝对量转换为占比，便于不同股票间的比较
    chip_dist /= total_volume
    return price_bins, chip_dist

def calculate_concentration(chip_prices, chip_dist):
    """
    计算筹码集中度指标（CR10, CR30）和基尼系数
    :param chip_prices: 价格数组
    :param chip_dist: 筹码分布占比数组
    :return: CR10, CR30, 基尼系数
    """
    # 将价格从高到低进行排序，对应筹码分布同步排序
    sorted_idx = np.argsort(-chip_prices)
    sorted_prices = chip_prices[sorted_idx]
    sorted_dist = chip_dist[sorted_idx]

    # 计算累计筹码分布
    cumulative_dist = np.cumsum(sorted_dist)
    total = cumulative_dist[-1]

    # 计算CR10：找到累计筹码占比达到10%的位置，并转换为价格区间占比
    cr10_idx = np.where(cumulative_dist >= 0.1)[0][0] + 1
    cr10 = cr10_idx / len(sorted_dist) * 100  # 数值越小，说明筹码越集中在高价区

    # 计算CR30
    cr30_idx = np.where(cumulative_dist >= 0.3)[0][0] + 1
    cr30 = cr30_idx / len(sorted_dist) * 100

    # 计算基尼系数（基于相对平均差公式）
    n = len(sorted_dist)
    gini = (np.sum(np.abs(np.subtract.outer(sorted_dist, sorted_dist)))) / (2 * n**2 * np.mean(sorted_dist))

    return cr10, cr30, gini

def analyze_chip_change(df, date1, date2):
    """
    分析两个特定日期之间的筹码分布变动情况
    :return: 统一价格区间数组, 筹码变动量数组（正值为流入，负值为流出）
    """
    # 分别计算两个日期的筹码分布
    p1, c1 = calculate_chip_distribution(df, date1)
    p2, c2 = calculate_chip_distribution(df, date2)

    # 合并两个日期的所有价格点，形成统一的分析区间
    all_prices = np.unique(np.concatenate([p1, p2]))
    # 通过线性插值，确保两个分布在同一组价格点上可比
    c1_interp = np.interp(all_prices, p1, c1, left=0, right=0)
    c2_interp = np.interp(all_prices, p2, c2, left=0, right=0)

    # 计算变动：后期分布减去前期分布
    change = c2_interp - c1_interp
    return all_prices, change

def calculate_volume_deviation(df, window=20):
    """
    计算成交量偏离度（当前成交量相对于近期平均成交量的波动幅度）
    :param window: 计算移动平均的窗口期，默认为20日
    :return: 添加了偏离度指标的DataFrame
    """
    df = df.copy()
    df['ma20_vol'] = df['volume'].rolling(window).mean()
    df['vol_deviation'] = (df['volume'] - df['ma20_vol']) / df['ma20_vol']
    return df

def analyze_price_volume(df, chip_prices, chip_dist):
    """
    结合筹码分布进行量价综合分析
    :return: 当前股价处的筹码密度, 量价背离信号序列
    """
    # 获取最新收盘价
    current_price = df[df['date'] == max(df['date'])]['close'].values[0]
    # 定位当前股价在筹码分布中的位置
    idx = np.searchsorted(chip_prices, current_price, side='right')
    current_chip = chip_dist[idx]

    # 构建一个简单的量价背离预警指标：成交量创阶段新高，但筹码集中度却在下降（CR10值上升）
    volume_high = df['volume'] == df['volume'].rolling(60).max()
    concentration_decrease = (df['cr10'].diff() > 0)  # CR10值增加意味着集中度下降
    volume_price_divergence = volume_high & concentration_decrease

    return current_chip, volume_price_divergence

# 获取历史数据
df = qs.get_data('中国平安', start='20230101')
df = df.reset_index()

# 定义分析的关键时间点
start_date = '2023-01-01'
end_date = qs.latest_trade_date()
key_dates = ['2023-06-30', '2023-12-29', '2024-06-28', '2024-12-31', end_date]

# 遍历关键日期，计算并存储筹码分布及指标
chip_data = {}
for date in key_dates:
    prices, dist = calculate_chip_distribution(df, date)
    chip_data[date] = (prices, dist)
    cr10, cr30, gini = calculate_concentration(prices, dist)
    # 将计算出的指标存回原DataFrame对应日期行
    df.loc[df['date'] == date, ['cr10', 'cr30', 'gini']] = [cr10, cr30, gini]

# 查看关键日期的指标
print(df[['date', 'close', 'cr10', 'cr30', 'gini']].dropna().round(2))