一、别人的结果
2007年四季度附近✅
2008年秋季附近✅
2014-12✅
2018年初附近✅
2021-01附近
二、我的测试结果
开始聚合月度成交额（pandas groupby，不生成宽表）... 聚合完成，耗时：4.77 秒 月度成交额记录数：1024713 月份数量：187 股票数量：5561 ================================================== 计算成交额集中度... 集中度计算完成，耗时：28.00 秒 最近 5 个月集中度： month 2026-03 0.402632 2026-04 0.431910 2026-05 0.446701 2026-06 0.490294 2026-07 0.499735 Name: amount, dtype: float64 ================================================== 生成信号灯... 信号灯计算完成，耗时：0.03 秒 ================================================== 当前状态： 最新月份：2026-07 成交额集中度：49.97% 10年90%分位：48.74% 10年95%分位：51.30% z-score：1.40 信号灯：黄灯
三、代码
# -*- coding: utf-8 -*-
"""
成交额集中度指标复刻（pandas 优化版，不生成宽表）
数据来源：本地 Parquet 文件（全A股日线_2011-2015.parquet, 全A股日线_2016-2020.parquet, 全A股日线_2021-2026.parquet）
功能：计算月度成交额集中度，并生成信号灯
"""
import pandas as pd
import numpy as np
from datetime import datetime
import math
# ============================================================
# 1. 计算月度成交额集中度（输入为分组求和后的长表）
# ============================================================
def calc_concentration_from_long(monthly_amount_series):
"""
输入：monthly_amount_series
pandas Series，索引为 (month, stock)，值为该股票当月成交额合计
输出：pd.Series，index 为月份字符串（'YYYY-MM'），values 为集中度
"""
# 按 month 分组，对每组计算前5%股票的成交额占比
def apply_func(group):
total = group.sum()
if total == 0 or pd.isna(total):
return np.nan
n = len(group)
k = int(math.ceil(n * 0.05))
top_k_sum = group.nlargest(k).sum()
return top_k_sum / total
concentration = monthly_amount_series.groupby(level='month').apply(apply_func)
concentration.index.name = 'month'
return concentration.sort_index()
# ============================================================
# 2. 生成信号灯（保持原逻辑）
# ============================================================
def add_signal_light(concentration_series):
"""
给集中度序列添加滚动分位数、z-score 和信号灯
使用过去 120 个月（10 年）数据计算，滞后一个月，避免未来函数
"""
df = pd.DataFrame({'concentration': concentration_series})
df.index = pd.to_datetime(df.index)
df['roll_90'] = df['concentration'].rolling(120, min_periods=60).quantile(0.90).shift(1)
df['roll_95'] = df['concentration'].rolling(120, min_periods=60).quantile(0.95).shift(1)
df['roll_mean'] = df['concentration'].rolling(120, min_periods=60).mean().shift(1)
df['roll_std'] = df['concentration'].rolling(120, min_periods=60).std().shift(1)
df['zscore'] = (df['concentration'] - df['roll_mean']) / df['roll_std']
def get_light(row):
if pd.isna(row['concentration']) or pd.isna(row['roll_90']):
return '无数据'
if row['concentration'] >= 45.0 or row['zscore'] >= 2:
return '红灯'
elif row['concentration'] >= row['roll_95']:
return '橙灯'
elif row['concentration'] >= row['roll_90']:
return '黄灯'
else:
return '无'
df['light'] = df.apply(get_light, axis=1)
return df
# ============================================================
# 3. 主程序
# ============================================================
def main():
files = [
'全A股日线_2011-2015.parquet',
'全A股日线_2016-2020.parquet',
'全A股日线_2021-2026.parquet'
]
print("=" * 50)
print("开始读取 Parquet 文件...")
start_time = datetime.now()
print(f"当前时间：{start_time.strftime('%Y-%m-%d %H:%M:%S')}")
df_list = []
for f in files:
try:
file_start = datetime.now()
df = pd.read_parquet(f, columns=['stime', 'stock', 'amount'])
file_end = datetime.now()
print(f"已读取 {f}，形状：{df.shape}，耗时：{(file_end - file_start).total_seconds():.2f} 秒")
df_list.append(df)
except Exception as e:
print(f"读取 {f} 失败：{e}")
return
if not df_list:
print("没有读取到任何数据，程序终止。")
return
df_all = pd.concat(df_list, ignore_index=True)
read_end = datetime.now()
print(f"所有文件读取合并完成，耗时：{(read_end - start_time).total_seconds():.2f} 秒")
# 基本信息
print("\n" + "=" * 50)
print("数据基本信息：")
print(f"总行数：{len(df_all)}")
print(f"字段列表：{df_all.columns.tolist()}")
print(f"日期范围：{df_all['stime'].min()} 至 {df_all['stime'].max()}")
print(f"唯一股票代码数量：{df_all['stock'].nunique()}")
# 确保日期为 datetime 类型
df_all['stime'] = pd.to_datetime(df_all['stime'])
# 提取月份（格式：YYYY-MM）
df_all['month'] = df_all['stime'].dt.strftime('%Y-%m')
# 删除不再需要的 stime 列，减少内存占用
df_all.drop(columns=['stime'], inplace=True)
# ============================================================
# 聚合月度成交额：每只股票每月成交额合计（长表形式）
# ============================================================
print("\n" + "=" * 50)
print("开始聚合月度成交额（pandas groupby，不生成宽表）...")
agg_start = datetime.now()
monthly_amount_series = df_all.groupby(['month', 'stock'])['amount'].sum()
agg_end = datetime.now()
print(f"聚合完成，耗时：{(agg_end - agg_start).total_seconds():.2f} 秒")
print(f"月度成交额记录数：{len(monthly_amount_series)}")
print(f"月份数量：{monthly_amount_series.index.get_level_values('month').nunique()}")
print(f"股票数量：{monthly_amount_series.index.get_level_values('stock').nunique()}")
# ============================================================
# 计算集中度
# ============================================================
print("\n" + "=" * 50)
print("计算成交额集中度...")
conc_start = datetime.now()
concentration = calc_concentration_from_long(monthly_amount_series)
conc_end = datetime.now()
print(f"集中度计算完成，耗时：{(conc_end - conc_start).total_seconds():.2f} 秒")
print("最近 5 个月集中度：")
print(concentration.tail(5))
# ============================================================
# 生成信号灯
# ============================================================
print("\n" + "=" * 50)
print("生成信号灯...")
light_start = datetime.now()
concentration_df = add_signal_light(concentration)
light_end = datetime.now()
print(f"信号灯计算完成，耗时：{(light_end - light_start).total_seconds():.2f} 秒")
# ============================================================
# 当前状态
# ============================================================
latest = concentration_df.iloc[-1]
print("\n" + "=" * 50)
print("当前状态：")
print(f"最新月份：{concentration_df.index[-1].strftime('%Y-%m')}")
print(f"成交额集中度：{latest['concentration']:.2%}")
print(f"10年90%分位：{latest['roll_90']:.2%}")
print(f"10年95%分位：{latest['roll_95']:.2%}")
print(f"z-score：{latest['zscore']:.2f}")
print(f"信号灯：{latest['light']}")
# 保存结果
concentration_df.to_csv('成交额集中度_信号灯.csv')
print("\n结果已保存到 成交额集中度_信号灯.csv")
print("全部流程完成。")
if __name__ == '__main__':
main()
原文：https://zhuanlan.zhihu.com/p/2052689594123133662