@@ -0,0 +1,392 @@
""" 五大期货交易所前20会员成交持仓排名数据获取工具。
交易所对照:
- CZCE 郑商所 → get_rank_table_czce (玻璃FG、纯碱SA、甲醇MA等)
- SHFE 上期所 → get_shfe_rank_table (铜CU、铝AL、螺纹RB、原油SC等)
- DCE 大商所 → get_dce_rank_table (铁矿石I、豆粕M、棕榈P等)
- CFFEX 中金所 → get_cffex_rank_table (IF、IC、IH、T等)
- GFEX 广期所 → futures_gfex_position_rank (工业硅SI、碳酸锂LC等)
交易所公布规则:
- CZCE: 同时公布品种汇总 + 各合约明细
- SHFE/CFFEX: 只公布合约明细，品种汇总由函数自行加总
- DCE: 只公布品种排名
- GFEX: 公布合约明细
"""
import akshare as ak
import pandas as pd
from datetime import datetime , timedelta
EXCHANGE_CODE = {
" CZCE " : " CZCE " ,
" SHFE " : " SHFE " ,
" DCE " : " DCE " ,
" CFFEX " : " CFFEX " ,
" GFEX " : " GFEX " ,
}
# 常见品种 -> 交易所映射
VARIETY_EXCHANGE = {
# 郑商所 CZCE
" FG " : " CZCE " , " SA " : " CZCE " , " MA " : " CZCE " , " TA " : " CZCE " ,
" RM " : " CZCE " , " OI " : " CZCE " , " ZC " : " CZCE " , " SF " : " CZCE " ,
" SM " : " CZCE " , " UR " : " CZCE " , " PF " : " CZCE " , " PX " : " CZCE " ,
" SH " : " CZCE " , " PR " : " CZCE " , " PL " : " CZCE " , " AP " : " CZCE " ,
" CJ " : " CZCE " , " CY " : " CZCE " , " PK " : " CZCE " , " CF " : " CZCE " ,
" SR " : " CZCE " ,
# 上期所 SHFE
" CU " : " SHFE " , " AL " : " SHFE " , " ZN " : " SHFE " , " PB " : " SHFE " ,
" NI " : " SHFE " , " SN " : " SHFE " , " AU " : " SHFE " , " AG " : " SHFE " ,
" RB " : " SHFE " , " HC " : " SHFE " , " WR " : " SHFE " , " FU " : " SHFE " ,
" BU " : " SHFE " , " RU " : " SHFE " , " SC " : " SHFE " , " NR " : " SHFE " ,
" SP " : " SHFE " , " SS " : " SHFE " , " LU " : " SHFE " , " BC " : " SHFE " ,
" AO " : " SHFE " , " BR " : " SHFE " , " EC " : " SHFE " ,
# 大商所 DCE
" C " : " DCE " , " CS " : " DCE " , " A " : " DCE " , " B " : " DCE " ,
" M " : " DCE " , " Y " : " DCE " , " P " : " DCE " , " FB " : " DCE " ,
" BB " : " DCE " , " JD " : " DCE " , " L " : " DCE " , " V " : " DCE " ,
" PP " : " DCE " , " J " : " DCE " , " JM " : " DCE " , " I " : " DCE " ,
" EG " : " DCE " , " RR " : " DCE " , " EB " : " DCE " , " PG " : " DCE " ,
" LH " : " DCE " , " LG " : " DCE " , " BZ " : " DCE " ,
# 中金所 CFFEX
" IF " : " CFFEX " , " IC " : " CFFEX " , " IH " : " CFFEX " , " IM " : " CFFEX " ,
" T " : " CFFEX " , " TF " : " CFFEX " , " TS " : " CFFEX " , " TL " : " CFFEX " ,
# 广期所 GFEX
" SI " : " GFEX " , " LC " : " GFEX " , " PS " : " GFEX " ,
}
# 各交易所的品种代码列表（用于 get_rank_sum）
ALL_VARS_CZCE = [ " WH " , " PM " , " CF " , " SR " , " TA " , " OI " , " RI " , " MA " , " FG " , " RS " , " RM " , " ZC " ,
" JR " , " LR " , " SF " , " SM " , " WT " , " TC " , " GN " , " RO " , " ER " , " SRX " , " SRY " ,
" WSX " , " WSY " , " CY " , " AP " , " UR " , " CJ " , " SA " , " PK " , " PF " , " PX " , " SH " , " PR " , " PL " ]
ALL_VARS_DCE = [ " C " , " CS " , " A " , " B " , " M " , " Y " , " P " , " FB " , " BB " , " JD " , " L " , " V " , " PP " ,
" J " , " JM " , " I " , " EG " , " RR " , " EB " , " PG " , " LH " , " LG " , " BZ " ]
ALL_VARS_SHFE = [ " CU " , " AL " , " ZN " , " PB " , " NI " , " SN " , " AU " , " AG " , " RB " , " WR " , " HC " ,
" FU " , " BU " , " RU " , " SC " , " NR " , " SP " , " SS " , " LU " , " BC " , " AO " , " BR " , " EC " ]
ALL_VARS_CFFEX = [ " IF " , " IC " , " IM " , " IH " , " T " , " TF " , " TS " , " TL " ]
ALL_VARS_GFEX = [ " SI " , " LC " , " PS " ]
# 没有统一字段名映射时用的 to_int 辅助
def _to_int ( x ) :
if isinstance ( x , str ) :
return int ( x . replace ( " , " , " " ) )
return int ( x )
# ---------------------------------------------------------------------------
# 底层接口，按交易所获取当日前20排名
# ---------------------------------------------------------------------------
def fetch_top20 ( date : str = None , exchange : str = None , variety : str = None ) - > pd . DataFrame :
""" 获取指定交易日该品种的前20会员成交持仓排名。
主力合约检测调用 :func:`main_contract`，本函数自动定位。
Parameters
----------
date : str or None
交易日，格式 YYYY-MM-DD 或 YYYYMMDD。None 表示最新。
exchange : str or None
交易所代码（CZCE/SHFE/DCE/CFFEX/GFEX）。None 时从 variety 推算。
variety : str
品种代码（如 " FG " , " RB " , " I " , " IF " ）。如果为 None 则返回全品种。
Returns
-------
pd.DataFrame
前20会员排名数据。
列：rank, party_name, vol, vol_chg,
long_party_name, long_open_interest, long_open_interest_chg,
short_party_name, short_open_interest, short_open_interest_chg,
symbol, exchange
"""
if exchange is None and variety is not None :
exchange = VARIETY_EXCHANGE . get ( variety . upper ( ) )
if exchange is None :
raise ValueError ( f " 未知品种 { variety } ，请指定 exchange 参数 " )
if exchange == " CZCE " :
return _fetch_czce ( date , variety )
elif exchange == " SHFE " :
return _fetch_shfe ( date , variety )
elif exchange == " DCE " :
return _fetch_dce ( date , variety )
elif exchange == " CFFEX " :
return _fetch_cffex ( date , variety )
elif exchange == " GFEX " :
return _fetch_gfex ( date , variety )
else :
raise ValueError ( f " 不支持的交易所: { exchange } " )
def _fetch_czce ( date = None , variety = None ) :
rank = ak . get_rank_table_czce ( date = date )
if variety :
variety = variety . upper ( )
if variety in rank :
# 品种级别汇总
return _normalize_czce ( rank [ variety ] , variety , " CZCE " )
# 合约级别: 找到该品种的所有合约
contract_keys = [ k for k in rank if k . startswith ( variety ) and k != variety ]
if not contract_keys :
raise ValueError ( f " 未找到品种 { variety } 的数据 " )
# 返回汇总版
return _normalize_czce ( rank . get ( variety , pd . DataFrame ( ) ) , variety , " CZCE " )
else :
# 全部品种 - 返回汇总
return rank # dict, 调用方自己处理
def _fetch_shfe ( date = None , variety = None ) :
vars_list = [ variety . upper ( ) ] if variety else None
rank = ak . get_shfe_rank_table ( date = date , vars_list = vars_list )
# SHFE 返回 dict: {合约代码: DataFrame}
rows = [ ]
for symbol , df in rank . items ( ) :
for _ , r in df . iterrows ( ) :
rows . append ( {
" rank " : int ( r [ " 名次 " ] ) ,
" party_name " : r [ " 会员简称 " ] ,
" vol " : r [ " 成交量 " ] ,
" vol_chg " : r [ " 成交量增减 " ] ,
" long_party_name " : r [ " 持买单量-会员简称 " ] ,
" long_open_interest " : r [ " 持买单量-合计 " ] ,
" long_open_interest_chg " : r [ " 持买单量-增减 " ] ,
" short_party_name " : r [ " 持卖单量-会员简称 " ] ,
" short_open_interest " : r [ " 持卖单量-合计 " ] ,
" short_open_interest_chg " : r [ " 持卖单量-增减 " ] ,
" symbol " : symbol ,
} )
df = pd . DataFrame ( rows )
if variety :
df = df [ df [ " symbol " ] . str . startswith ( variety . upper ( ) ) ]
return df
def _fetch_dce ( date = None , variety = None ) :
vars_list = [ variety . upper ( ) ] if variety else None
rank = ak . get_dce_rank_table ( date = date , vars_list = vars_list )
rows = [ ]
for symbol , df in rank . items ( ) :
for _ , r in df . iterrows ( ) :
rows . append ( {
" rank " : int ( r [ " 名次 " ] ) ,
" party_name " : r [ " 会员简称 " ] ,
" vol " : r [ " 成交量 " ] ,
" vol_chg " : r [ " 成交量增减 " ] ,
" long_party_name " : r [ " 持买单-会员简称 " ] ,
" long_open_interest " : r [ " 持买单 " ] ,
" long_open_interest_chg " : r [ " 持买单增减 " ] ,
" short_party_name " : r [ " 持卖单-会员简称 " ] ,
" short_open_interest " : r [ " 持卖单 " ] ,
" short_open_interest_chg " : r [ " 持卖单增减 " ] ,
" symbol " : symbol ,
} )
return pd . DataFrame ( rows )
def _fetch_cffex ( date = None , variety = None ) :
vars_list = [ variety . upper ( ) ] if variety else None
rank = ak . get_cffex_rank_table ( date = date , vars_list = vars_list )
rows = [ ]
for symbol , df in rank . items ( ) :
for _ , r in df . iterrows ( ) :
rows . append ( {
" rank " : int ( r [ " 名次 " ] ) ,
" party_name " : r [ " 会员简称 " ] ,
" vol " : r [ " 成交量 " ] ,
" vol_chg " : r [ " 成交量增减 " ] ,
" long_party_name " : r [ " 持买单-会员简称 " ] ,
" long_open_interest " : r [ " 持买单-合计 " ] ,
" long_open_interest_chg " : r [ " 持买单-增减 " ] ,
" short_party_name " : r [ " 持卖单-会员简称 " ] ,
" short_open_interest " : r [ " 持卖单-合计 " ] ,
" short_open_interest_chg " : r [ " 持卖单-增减 " ] ,
" symbol " : symbol ,
} )
return pd . DataFrame ( rows )
def _fetch_gfex ( date = None , variety = None ) :
vars_list = [ variety . upper ( ) ] if variety else None
rank = ak . futures_gfex_position_rank ( date = date , vars_list = vars_list )
rows = [ ]
if isinstance ( rank , dict ) :
for symbol , df in rank . items ( ) :
for _ , r in df . iterrows ( ) :
rows . append ( {
" rank " : int ( r [ " 名次 " ] ) ,
" party_name " : r [ " 会员简称 " ] ,
" vol " : r [ " 成交量 " ] ,
" vol_chg " : r [ " 成交量增减 " ] ,
" long_party_name " : r [ " 持买单-会员简称 " ] ,
" long_open_interest " : r [ " 持买单-合计 " ] ,
" long_open_interest_chg " : r [ " 持买单-增减 " ] ,
" short_party_name " : r [ " 持卖单-会员简称 " ] ,
" short_open_interest " : r [ " 持卖单-合计 " ] ,
" short_open_interest_chg " : r [ " 持卖单-增减 " ] ,
" symbol " : symbol ,
} )
return pd . DataFrame ( rows )
def _normalize_czce ( df , variety , exchange ) :
""" 标准化 CZCE 品种汇总 DataFrame 字段名 """
df = df . copy ( )
df [ " symbol " ] = df . get ( " symbol " , variety )
df [ " exchange " ] = exchange
return df
# ---------------------------------------------------------------------------
# 主力合约检测
# ---------------------------------------------------------------------------
def main_contract ( variety : str , date : str = None ) - > str :
""" 检测该品种在指定交易日的主力合约代码。
逻辑：成交量/持仓量最大的合约即为主力。
"""
variety = variety . upper ( )
exchange = VARIETY_EXCHANGE . get ( variety )
if exchange is None :
raise ValueError ( f " 未知品种 { variety } " )
rank = ak . get_rank_table_czce ( date = date ) if exchange == " CZCE " else { }
if exchange == " CZCE " :
# CZCE: 合约级别的 key 是 品种+年月 (FG609), 品种级是 FG
contract_keys = [ k for k in rank if k . startswith ( variety ) and k != variety ]
if not contract_keys :
raise ValueError ( f " 未找到品种 { variety } 的合约数据 " )
vols = { }
for ck in contract_keys :
df = rank [ ck ]
vol_sum = df [ " vol " ] . apply ( _to_int ) . sum ( )
vols [ ck ] = vol_sum
return max ( vols , key = vols . get )
else :
# 其他交易所：通过成交额最大的标的判断
raise NotImplementedError ( f " { exchange } 主力合约检测待实现 " )
# ---------------------------------------------------------------------------
# 品种汇总对比（多交易所多品种）
# ---------------------------------------------------------------------------
def rank_summary ( date : str = None , varieties : list = None ) - > pd . DataFrame :
""" 多品种前N会员成交量/持仓汇总对比（仅支持 CZCE 品种）。
使用 get_rank_table_czce 的品种级别汇总数据，
返回各品种前5/10/15/20成交量和多空持仓总和。
Parameters
----------
date : str
交易日。
varieties : list
品种代码列表。默认 FG, SA, MA, CF, TA, UR, PF, SM。
Returns
-------
pd.DataFrame
每行一个品种，含品种级前N成交量和多空持仓总和。
"""
if varieties is None :
varieties = [ " FG " , " SA " , " MA " , " CF " , " TA " , " UR " , " PF " , " SM " ]
rank = ak . get_rank_table_czce ( date = date )
rows = [ ]
for var in varieties :
var = var . upper ( )
if var not in rank :
continue
df = rank [ var ]
df [ " vol_int " ] = df [ " vol " ] . apply ( _to_int )
df [ " long_int " ] = df [ " long_open_interest " ] . apply ( _to_int )
df [ " short_int " ] = df [ " short_open_interest " ] . apply ( _to_int )
total_vol = df [ " vol_int " ] . sum ( )
vol_5 = df . head ( 5 ) [ " vol_int " ] . sum ( )
vol_10 = df . head ( 10 ) [ " vol_int " ] . sum ( )
long_sum = df [ " long_int " ] . sum ( )
short_sum = df [ " short_int " ] . sum ( )
rows . append ( {
" var " : var ,
" vol_total " : total_vol ,
" vol_top5 " : vol_5 ,
" vol_top10 " : vol_10 ,
" vol_top20 " : df [ " vol_int " ] . sum ( ) ,
" long_sum " : long_sum ,
" short_sum " : short_sum ,
" net " : long_sum - short_sum ,
} )
return pd . DataFrame ( rows ) . sort_values ( " vol_total " , ascending = False )
# ---------------------------------------------------------------------------
# 主力合约前20成交量
# ---------------------------------------------------------------------------
def main_top20 ( variety : str , date : str = None ) - > pd . DataFrame :
""" 获取品种主力合约的前20会员成交量排名。
Parameters
----------
variety : str
品种代码，如 " FG " 。
date : str
交易日。None 表示最新。
Returns
-------
pd.DataFrame
主力合约前20会员成交量排名 + 多空持仓排名。
"""
exchange = VARIETY_EXCHANGE . get ( variety . upper ( ) )
if exchange != " CZCE " :
raise NotImplementedError ( f " 目前仅支持 CZCE 品种的主力合约前20查询 " )
rank = ak . get_rank_table_czce ( date = date )
contract_keys = [ k for k in rank if k . startswith ( variety . upper ( ) ) and k != variety . upper ( ) ]
vols = { ck : rank [ ck ] [ " vol " ] . apply ( _to_int ) . sum ( ) for ck in contract_keys }
main_code = max ( vols , key = vols . get )
df = rank [ main_code ] . copy ( )
df [ " symbol " ] = main_code
# 数值列处理（CZCE 返回字符串含逗号）
for col in [ " vol " , " vol_chg " , " long_open_interest " , " long_open_interest_chg " ,
" short_open_interest " , " short_open_interest_chg " ] :
df [ col ] = df [ col ] . apply ( _to_int )
return df . sort_values ( " rank " )
# ---------------------------------------------------------------------------
# 输出格式化
# ---------------------------------------------------------------------------
def print_top20 ( df : pd . DataFrame , title : str = " " ) :
""" 打印前20成交量排名 + 多空前10。 """
if title :
print ( f " \n { ' = ' * 60 } " )
print ( f " { title } " )
print ( f " { ' = ' * 60 } " )
print ( f " \n 【成交量前20】 " )
print ( f " { ' 排名 ' : >4 } { ' 会员 ' : <20 } { ' 成交量 ' : >10 } { ' 变化 ' : >10 } " )
print ( " - " * 50 )
for _ , r in df . iterrows ( ) :
print ( f " { int ( r [ ' rank ' ] ) : >4 } { str ( r [ ' vol_party_name ' ] ) : <20 } "
f " { int ( r [ ' vol ' ] ) : >10, } { int ( r [ ' vol_chg ' ] ) : >10, } " )
print ( f " \n 【持多单前10】 " )
print ( f " { ' 排名 ' : >4 } { ' 会员 ' : <20 } { ' 多单 ' : >10 } { ' 变化 ' : >10 } " )
print ( " - " * 50 )
for _ , r in df . head ( 10 ) . iterrows ( ) :
print ( f " { int ( r [ ' rank ' ] ) : >4 } { str ( r [ ' long_party_name ' ] ) : <20 } "
f " { int ( r [ ' long_open_interest ' ] ) : >10, } { int ( r [ ' long_open_interest_chg ' ] ) : >10, } " )
print ( f " \n 【持空单前10】 " )
print ( f " { ' 排名 ' : >4 } { ' 会员 ' : <20 } { ' 空单 ' : >10 } { ' 变化 ' : >10 } " )
print ( " - " * 50 )
for _ , r in df . head ( 10 ) . iterrows ( ) :
print ( f " { int ( r [ ' rank ' ] ) : >4 } { str ( r [ ' short_party_name ' ] ) : <20 } "
f " { int ( r [ ' short_open_interest ' ] ) : >10, } { int ( r [ ' short_open_interest_chg ' ] ) : >10, } " )
fish