聚宽平台如何统计全市场 ETF 成交额集中度？（附 Python 源码）

在聚宽（JoinQuant）量化交易平台上，统计全市场 ETF 的成交额集中度可以通过以下几个步骤实现：

1. 获取所有已上市的 ETF 列表：使用 get_all_securities 获取所有基金，并筛选出 type 为 etf 的标的。
2. 过滤货币和债券 ETF：通过 get_security_info 查询基金的详细类型，过滤掉货币市场基金（money_market_fund 或 mmf）和债券基金（bond_fund）。
3. 获取当日成交额：使用 get_price 或 get_bars 获取这些 ETF 在指定交易日的成交额（money）。
4. 计算集中度：将成交额降序排列，计算累计成交额占比，并统计达到 50%、75%、90% 阈值所需的 ETF 数量。

Python 实现源码

您可以在聚宽的研究环境（Jupyter Notebook）中直接运行以下代码：

import pandas as pd
import numpy as np
from jqdata import *

def get_etf_concentration(target_date=None):
    # 1. 如果未指定日期，默认使用前一个交易日
    if target_date is None:
        target_date = get_trade_days(count=1)[0]
    else:
        target_date = pd.to_datetime(target_date).date()
    
    print(f"正在统计日期: {target_date} 的 ETF 成交额集中度...")
    
    # 2. 获取所有基金列表
    all_funds = get_all_securities(types=['fund'], date=target_date)
    
    # 3. 过滤出 ETF，并排除货币和债券 ETF
    valid_etfs = []
    for code in all_funds.index:
        info = get_security_info(code)
        # 过滤货币基金(mmf/money_market_fund)和债券基金(bond_fund)
        if info.type == 'etf' and '货币' not in info.display_name and '债' not in info.display_name:
            valid_etfs.append(code)
            
    print(f"过滤后有效的股票型/商品型 ETF 数量: {len(valid_etfs)}")
    
    # 4. 获取这些 ETF 在目标日期的成交额
    # 使用 get_price 获取单日数据
    price_df = get_price(valid_etfs, start_date=target_date, end_date=target_date, frequency='daily', fields=['money'])
    
    # 提取成交额并清洗数据
    etf_money = price_df['money'].iloc[0].dropna()
    etf_money = etf_money[etf_money > 0] # 过滤无成交量的ETF
    
    # 5. 按成交额降序排列
    etf_money_sorted = etf_money.sort_values(ascending=False)
    total_market_money = etf_money_sorted.sum()
    
    # 6. 计算累计占比
    cumulative_sum = etf_money_sorted.cumsum()
    cumulative_ratio = cumulative_sum / total_market_money
    
    # 7. 计算达到 50%, 75%, 90% 累计占比所需的 ETF 数量
    count_50 = np.searchsorted(cumulative_ratio, 0.50) + 1
    count_75 = np.searchsorted(cumulative_ratio, 0.75) + 1
    count_90 = np.searchsorted(cumulative_ratio, 0.90) + 1
    
    # 打印结果
    print("================ 统计结果 ================")
    print(f"当日有效 ETF 总成交额: {total_market_money / 1e8:.2f} 亿元")
    print(f"成交额前 5 名的 ETF 累计占比: {cumulative_ratio.iloc[4]*100:.2f}%")
    print(f"成交额前 10 名的 ETF 累计占比: {cumulative_ratio.iloc[9]*100:.2f}%")
    print(f"达到 50% 累计成交额所需的 ETF 数量: {count_50} 只 (占总数 {count_50/len(etf_money)*100:.2f}%)")
    print(f"达到 75% 累计成交额所需的 ETF 数量: {count_75} 只 (占总数 {count_75/len(etf_money)*100:.2f}%)")
    print(f"达到 90% 累计成交额所需的 ETF 数量: {count_90} 只 (占总数 {count_90/len(etf_money)*100:.2f}%)")
    
    # 返回前10名成交额明细
    top_10 = pd.DataFrame({
        '名称': [get_security_info(x).display_name for x in etf_money_sorted.index[:10]],
        '成交额(万)': etf_money_sorted.values[:10] / 10000,
        '占比': cumulative_ratio.values[:10]
    }, index=etf_money_sorted.index[:10])
    
    return top_10

# 运行统计
top_10_df = get_etf_concentration()
print("\n成交额前 10 名明细:")
print(top_10_df)

核心 API 说明

• get_all_securities(types=['fund'], date=date)：获取指定日期在市的所有基金列表。
• get_security_info(code)：获取标的详细信息，其 type 属性可以帮助我们识别 etf、bond_fund 等，同时通过 display_name 辅助过滤名字中含有“货币”或“债”的场内基金。
• get_price(security_list, ...)：批量获取指定 ETF 列表的成交额（money）数据。