Python量化：自动抓取期指持仓主力净空单，直接“画”进通达信里！（含中信净空单）

📌 前言

对于期货交易者来说，持仓排名数据是判断市场情绪和主力动向的重要指标。特别是：

• 前20大席位净空单占比：反映整体市场的多空力量对比
• 中信期货净空单占比：作为"期指一哥"，中信的持仓变化往往具有风向标意义

今天分享一个全自动化工具，只需运行一次脚本，就能自动完成从数据抓取到通达信注入的全流程！

🎯 核心功能

✅ 自动抓取：从中金所官网自动下载每日持仓排名XML数据

✅ 智能解析：自动识别主力合约，提取前20大席位数据

✅ 精准计算：自动计算两个核心指标

• 前20大净空单占比（JKDZB20）
• 中信期货净空单占比（ZXJKDZB）

✅ 直通通达信：直接写入通达信自定义数据文件，无需手动导入

✅ 批量回补：支持指定日期区间，批量回补历史数据

📊 两大核心指标解读

1️⃣ 前20大净空单占比 (JKDZB20)

计算公式：

JKDZB20 = (前20大卖单量 - 前20大买单量) / (前20大总持仓) × 100%

实战意义：

• 正值越大 → 前20大席位整体偏空，市场看跌情绪浓厚
• 负值越大 → 前20大席位整体偏多，市场看涨情绪强烈
• 快速上升 → 空头力量增强，警惕回调风险
• 快速下降 → 多头力量增强，可能迎来反弹

2️⃣ 中信期货净空单占比 (ZXJKDZB)

计算公式：

ZXJKDZB = (中信卖单量 - 中信买单量) / 前20大总持仓 × 100%

实战意义：

• 中信期货作为头部券商，其持仓变化具有领先指标作用
• 当中信大幅增仓空单时，往往预示着短期调整压力
• 当中信减仓空单或转多时，可能是行情反转信号

💻 代码架构解析

整个工具分为三大模块，结构清晰，易于维护：

📦 模块一：数据抓取与解析

pythondef get_xml_content(product, trade_date):"""从中金所官网下载XML数据"""    url = f"http://www.cffex.com.cn/sj/ccpm/{url_date}/{product}.xml"# 支持本地缓存，避免重复下载def parse_cffex_xml(content, product):"""解析XML，提取目标合约及持仓数据"""# 自动筛选前4个主力合约# 提取成交量、持买单量、持卖单量

亮点：

• 支持IF（沪深300）、IH（上证50）、IC（中证500）、IM（中证1000）四个品种
• 自动识别主力合约（取前4个近月合约）
• 本地缓存机制，避免重复下载

📦 模块二：清洗计算核心指标

pythondef calculate_metrics(raw_rows):"""计算两大核心指标"""# 1. 按席位聚合买卖单量# 2. 排序取前20大席位# 3. 计算净空单占比

核心逻辑：

python# 前20大净空单占比top20_buy = buy_df.head(20)["Volume"].sum()top20_sell = sell_df.head(20)["Volume"].sum()jkdzb20 = (top20_sell - top20_buy) / (top20_buy + top20_sell) * 100# 中信净空单占比citic_buy = buy_df[buy_df["Member"] == "中信期货(代客)"]["Volume"].sum()citic_sell = sell_df[sell_df["Member"] == "中信期货(代客)"]["Volume"].sum()zxjkdzb = (citic_sell - citic_buy) / top20_total * 100

📦 模块三：直通通达信写入

pythondef update_tdx_dat(file_path, date_int, value):"""更新通达信.dat自定义数据文件"""# 读取现有数据# 更新当天数值# 按日期排序重写文件def write_to_tdx(metrics, trade_date):"""批量写入通达信"""# signals_user_9: 前20大净空占比# signals_user_10: 中信净空占比

技术细节：

• 使用二进制格式 <If（小端序：4字节整数+4字节浮点数）
• 自动创建目录结构
• 增量更新，保留历史数据

⚙️ 配置说明

全局配置区

python# 1. 运行日期配置RUN_DATE = None        # 留空则默认今天START_DATE = 20260310  # 回补起始日期END_DATE = 20260410    # 回补结束日期# 2. 核心路径配置SAVE_DIR = r"D:"# XML临时下载目录TDX_BASE_DIR = r"D:\交易\通达信\T0002\signals"# 通达信signals目录

使用场景

场景1：每日自动更新

pythonRUN_DATE = None  # 自动使用今天日期START_DATE = NoneEND_DATE = None

场景2：批量回补历史数据

pythonRUN_DATE = NoneSTART_DATE = 20240101  # 从2024年1月1日开始END_DATE = 20241231    # 到2024年12月31日

📈 通达信图表展示

数据写入后，在通达信中可以通过以下方式查看：

1. 打开自定义数据管理器 → 新建技术指标公式

1. 建立通达信公式

   JKD:SIGNALS_USER(9,1) ; {前20大净空占比-IF}ZXJKD:SIGNALS_USER(10,1) ;{中信净空占比-IF}IF(CODELIKE('IF'),12,IF(CODELIKE('IH'),16,IF(CODELIKE('IC'),7,8))),DOTLINE,COLORGREEN;IF(CODELIKE('IF'),5,IF(CODELIKE('IH'),8,IF(CODELIKE('IC'),2,4))),DOTLINE,COLORRED;

2. 叠加到K线图或单独副图显示

推荐用法：

• 将净空单占比与指数走势叠加，观察背离信号
• 设置阈值警戒线（如±15%），超出时提示风险
• 结合成交量、资金流向等多维度分析

🔧 完整代码

python# -*- coding: utf-8 -*-import osimport reimport structimport xml.etree.ElementTree as ETfrom datetime import datetime, timedeltaimport pandas as pdimport requests# ==================== 【全局配置区】 ====================# 1. 运行日期配置 (留空则默认今天，也可填区间回补历史数据)RUN_DATE = None        START_DATE = 20260310      END_DATE = 20260410        # 2. 核心路径配置SAVE_DIR = r"D:"TDX_BASE_DIR = r"D:\交易\通达信交易版\T0002\signals"# ==================== 【系统常量区】 ====================PRODUCTS = ["IF", "IH", "IC", "IM"]CITIC_MEMBER_NAME = "中信期货(代客)"CONTRACT_PATTERN = re.compile(r"^([A-Z]{2})(\d{2})(\d{2})$")HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/91.0"}# 通达信写入任务配置TDX_TASKS = [    {'desc': '期权监控度(前20大净空占比)','dir': os.path.join(TDX_BASE_DIR, 'signals_user_9'),'items': [            {'metric': 'IF_JKDZB20', 'code': 'IFL8'},            {'metric': 'IH_JKDZB20', 'code': 'IHL8'},            {'metric': 'IC_JKDZB20', 'code': 'ICL8'},            {'metric': 'IM_JKDZB20', 'code': 'IML8'},        ],    },    {'desc': '期权最新监控度(中信净空占比)','dir': os.path.join(TDX_BASE_DIR, 'signals_user_10'),'items': [            {'metric': 'IF_ZXJKDZB', 'code': 'IFL8'},            {'metric': 'IH_ZXJKDZB', 'code': 'IHL8'},            {'metric': 'IC_ZXJKDZB', 'code': 'ICL8'},            {'metric': 'IM_ZXJKDZB', 'code': 'IML8'},        ],    },]# ==================== 【模块一：数据抓取与解析】 ====================def parse_config_date(value):return datetime.strptime(str(value), "%Y%m%d") if value else Nonedef iter_target_dates():    start_dt, end_dt = parse_config_date(START_DATE), parse_config_date(END_DATE)if start_dt and end_dt:        current = start_dtwhile current <= end_dt:            yield current            current += timedelta(days=1)return    yield parse_config_date(RUN_DATE) or datetime.now()def get_xml_content(product, trade_date):    date_str = trade_date.strftime("%Y%m%d")    url_date = trade_date.strftime("%Y%m/%d")    save_path = os.path.join(SAVE_DIR, date_str)    os.makedirs(save_path, exist_ok=True)    local_file = os.path.join(save_path, f"{product}.xml")if os.path.exists(local_file):        with open(local_file, "rb") as f: return f.read()    url = f"http://www.cffex.com.cn/sj/ccpm/{url_date}/{product}.xml"    try:        resp = requests.get(url, headers=HEADERS, timeout=10)if resp.status_code == 200:            with open(local_file, "wb") as f: f.write(resp.content)return resp.content    except Exception: passreturn Nonedef parse_cffex_xml(content, product):if not content: return []    try:        root = ET.fromstring(content)    except:return []    contract_ids = set()    rows = []    type_map = {"0": "成交量", "1": "持买单量", "2": "持卖单量"}for child in root.findall("data"):        inst_id = child.findtext("instrumentid", "").strip()        dtype = child.findtext("datatypeid", "").strip()if re.match(CONTRACT_PATTERN, inst_id) and inst_id.startswith(product):            contract_ids.add(inst_id)if dtype in type_map:                rows.append({"Product": product, "Instrument": inst_id,"Type": type_map[dtype], "Member": child.findtext("shortname", ""),"Volume": int(child.findtext("volume", "0")),"Change": int(child.findtext("varvolume", "0"))                })    targets = sorted(list(contract_ids), key=lambda x: x[2:])[:4]return [r for r in rows if r["Instrument"] in targets]# ==================== 【模块二：清洗计算核心指标】 ====================def calculate_metrics(raw_rows):if not raw_rows: return {}    df = pd.DataFrame(raw_rows)    metrics = {}for product in PRODUCTS:        df_prod = df[df["Product"] == product]if df_prod.empty: continue        df_agg = df_prod.groupby(["Member", "Type"])[["Volume", "Change"]].sum().reset_index()        buy_df = df_agg[df_agg["Type"] == "持买单量"].sort_values(by="Volume", ascending=False).reset_index()        sell_df = df_agg[df_agg["Type"] == "持卖单量"].sort_values(by="Volume", ascending=False).reset_index()        top20_buy = int(buy_df.head(20)["Volume"].sum())        top20_sell = int(sell_df.head(20)["Volume"].sum())        top20_total = top20_buy + top20_sell        jkdzb20 = ((top20_sell - top20_buy) / top20_total * 100) if top20_total else 0        metrics[f"{product}_JKDZB20"] = jkdzb20        citic_buy = int(buy_df[buy_df["Member"] == CITIC_MEMBER_NAME]["Volume"].sum()) if not buy_df[buy_df["Member"] == CITIC_MEMBER_NAME].empty else 0        citic_sell = int(sell_df[sell_df["Member"] == CITIC_MEMBER_NAME]["Volume"].sum()) if not sell_df[sell_df["Member"] == CITIC_MEMBER_NAME].empty else 0        zxjkdzb = ((citic_sell - citic_buy) / top20_total * 100) if top20_total else 0        metrics[f"{product}_ZXJKDZB"] = zxjkdzbreturn metrics# ==================== 【模块三：直通通达信写入】 ====================def update_tdx_dat(file_path, date_int, value):    records = {}    struct_fmt = struct.Struct('<If')if os.path.exists(file_path):        with open(file_path, 'rb') as f:while chunk := f.read(struct_fmt.size):if len(chunk) == struct_fmt.size:                    d_int, val = struct_fmt.unpack(chunk)                    records[d_int] = val    records[date_int] = float(value)    os.makedirs(os.path.dirname(file_path), exist_ok=True)    with open(file_path, 'wb') as f:for d, v in sorted(records.items()):            f.write(struct_fmt.pack(d, v))def write_to_tdx(metrics, trade_date):    date_int = int(trade_date.strftime("%Y%m%d"))for task in TDX_TASKS:for item in task['items']:            metric_val = metrics.get(item['metric'])if metric_val is not None:                prefix = "47"if'L8'in item['code'] else"0"                file_path = os.path.join(task['dir'], f"{prefix}_{item['code']}.dat")                update_tdx_dat(file_path, date_int, metric_val)print(f"✅ 通达信已更新: {item['code']} -> {metric_val:.2f}%")# ==================== 【主控流程】 ====================def main():print("🚀 启动期指持仓数据抓取与注入系统...")for trade_date in iter_target_dates():print(f"\n📅 正在处理日期: {trade_date.strftime('%Y-%m-%d')}")        all_raw_rows = []for product in PRODUCTS:            content = get_xml_content(product, trade_date)            rows = parse_cffex_xml(content, product)            all_raw_rows.extend(rows)if not all_raw_rows:print(f"⚠️ {trade_date.strftime('%Y-%m-%d')} 数据未发布或为非交易日，跳过。")continue        metrics = calculate_metrics(all_raw_rows)        write_to_tdx(metrics, trade_date)print("\n🎉 全部处理完成！请重启通达信查看图表更新。")if __name__ == "__main__":    main()

🎓 使用教程

第一步：安装依赖

bashpip install pandas requests

第二步：修改配置

根据你的实际情况修改以下配置：

pythonSAVE_DIR = r"D:"# XML缓存目录TDX_BASE_DIR = r"D:\交易\通达信\T0002\signals"# 你的通达信路径

第三步：运行脚本

bashpython cffex_to_tdx.py

第四步：重启通达信

运行完成后，重启通达信即可在图表中看到更新的数据！

📢 免责声明：本系统的代码仅供量化技术交流与编程学习，机构持仓数据不能作为绝对的交易依据（主力也有可能用期指做套期保值对冲）。市场有风险，投资需谨慎！