★ FEATURED ARTICLE 2024年下半年我接了个让人头皮发麻的活从 CoinGlass 抓比特币永续合约的原始资金流数据再清洗成能让策略直接用的干净表。第一版我图省事抓回来就灌进数据库结果策略回测一跑全是洞某几天数据完全缺失某些时段费率突然多一个数量级更麻烦的是不同交易所的 BTC 合约代码在原始表里各写各的根本没法直接分组统计。后来我把这件事拆成抓取、存储、清洗、校验四层才把这条管线彻底理顺。这篇文章我把这些实操经验从头到尾写出来包括接口怎么选、请求节奏怎么控制、清洗规则怎么定、入库前要做哪些校验以及一组只有反复折腾过才会知道的坑。1. 项目拆解原始资金流数据为什么要单独洗很多人第一反应是CoinGlass 不是已经把数据聚合好了吗拿回来直接用不行吗我第一次也是这么想的直到打开返回的 JSON 才发现聚合平台只负责把各家交易所的数据拉到一起并不负责替你统一口径。1.1 先分清资金流和资金费率在动手之前必须先明确一件事所谓“原始资金流”不是 CEX 里那种买卖盘口的资金流入流出而是合约市场里的资金费率体系。永续合约没有到期日多空双方要想让盘口价格贴近现货价格交易所会定期让多头和空头互相支付资金费。多头多时多头付给空头空头多时空头付给多头。资金费率乘以仓位价值就是一次结算中的“资金流”。所以要算比特币的资金流最底层的数据至少包含两类资金费率funding rate每次结算时使用的费率数值。持仓量open interest用来乘以费率得到资金流金额。CoinGlass 的价值在于它把各家交易所的费率历史、持仓量、多空比等数据聚合到一个入口省去了自己一个个去对接交易所接口的麻烦。但麻烦也藏在这里每家交易所传上来的字段命名、单位、时间戳格式全都不一样聚合平台只是把这些差异原样暴露给了调用方。1.2 原始数据到底长什么样我通过公开数据接口拿到的一行典型记录大致是这个结构{ exchange: 某头部交易所, symbol: BTCUSDT, fundingTime: 1695926400000, fundingRate: 0.000123, avgPremiumIndex: -0.0000042, settlementInterval: 8 }看着挺规整但真实情况远没有这么简单。不同交易所的结算周期有的是 8 小时有的是 1 小时还有的是 4 小时资金费率有的给百分比有的给小数时间戳秒和毫秒混着来symbol 字段里有 BTCUSDT、BTC-USD、XBTUSD 各种写法。想在日线、周线级别聚合就必须先把这些混乱按统一规则收口。这里我把常见的“脏”情况整理成了一张表字段含义常见脏数据状态fundingTime资金费结算时间时区错乱、秒/毫秒混用fundingRate资金费率小数/百分比混用、负号丢失avgPremiumIndex标记价格与现货的偏离度null、精度不一致、部分交易所缺失symbol合约代码不同交易所前后缀不一致exchange交易所标识大小写不一致、别名混用settlementInterval结算周期缺失、单位不统一这张表基本就是我清洗模块的出发点了。1.3 把管线拆成四层吃过一次亏以后我再没把抓取和清洗写在一个大脚本里。现在的模块分四层抓取层只负责拉取数据不修数据、不分析数据。拿到原始 JSON 后原样落盘。存储层原始文件按日期归档保存。这一层看起来最没用却是后面所有清洗规则得以验证的关键。清洗层把散乱的原始数据变成统一 schema 的宽表主要做时间戳统一、字段映射、费率归一化、重复去除、异常标记。校验层跑完清洗后自动做条数对比、断点检测、极值检查通过后才允许写入正式数据库。这样分层的好处是显而易见的抓取挂了不影响清洗逻辑清洗规则要调整也不用重新抓数据。我见过不少人把抓取和清洗放在同一个函数里看起来省事一上线就后悔等你要重算一个月数据时没保留原始文件就只能重新抓一遍如果接口正好改版旧数据可能就再也对不上了。2. 抓取层接口选型、请求节奏与原始数据落盘抓取这层虽然叫“抓”但真正要想清楚的是接口选型和异常恢复机制。选错接口后面清洗规则再完善也白搭。2.1 为什么选 REST 而非 WebSocket做资金费率历史数据我推荐用 REST 接口拉分页历史。WebSocket 适合实时推送但它有个致命问题断线重连会留空档。你不可能 24 小时盯着一台机器保证它不掉线一旦中间的某个结算时间点没收到推送后面补数据时你会发现那段记录在本地永远缺一块。REST 接口按页码拉历史记录天然支持断点续传。今天拉完前 200 页明天继续从 201 页开始拉或者干脆每次都从开始日期重新拉一遍然后用清洗逻辑去重数据完整性更容易保证。具体接口选型上我建议优先使用返回列表型历史数据的公开接口而不是某一个币种的实时行情接口。实时接口只适合做盘中监控不适合做历史回测和离线清洗。2.2 请求节奏与限频处理CoinGlass 这类聚合平台不是给你一个人用的请求频率必须克制。我的经验值是单线程每秒最多 1 次请求每抓 100 页主动停 5 秒。这个节奏看起来保守实际上对数据清洗这种离线任务来说完全够用。如果被限频接口通常返回 429 状态码。我的重试策略是等待 60 秒后重试连续失败 5 次就放弃当前任务并发出告警而不是无限重试。无限重试会把“接口临时抖动”变成“封禁事故”。import time import requests def fetch_with_retry(session, url, params, max_retries5): for attempt in range(max_retries): resp session.get(url, paramsparams, timeout15) if resp.status_code 429: wait_time 60 * (attempt 1) time.sleep(wait_time) continue resp.raise_for_status() return resp.json() raise RuntimeError(重试次数耗尽任务已放弃)有一点必须说明我这里说的是控制自身请求频率不是让你用各种手段去突破平台的频率控制。做数据模块的人保持对数据源的尊重反而能少踩很多坑。2.3 原始数据一定要落盘我在模块里单独开了一层“原始存储”很多人不理解说“我最后不就是要一张干净表吗中间文件留它干嘛”。但我的建议是不管后面清洗规则怎么改原始数据都要原样留存。目录结构我是这样组织的raw/by_date/ 2024-06-15/ btcusdt_1.json btcusdt_2.json meta/ 2024-06-15_btcusdt.json每个 JSON 文件就是一次接口返回的原始内容meta 文件记录这次请求的时间、参数、成功条数。为什么要这么做因为清洗规则的调整是必然会发生的如果某天发现“费率的百分比转换逻辑写错了”需要重算历史数据原始文件还在就能第一时间重新清洗而不是再跑一遍抓取任务。3. 清洗规则设计把“能看的字段”变成“能算的数据”清洗层是整个模块的核心。我的设计原则很简单不追求在清洗时把数据“修”到完美而是把不确定性都留下标记计算的时候再决定用不用。3.1 时间戳统一秒和毫秒不能靠猜资金费率的结算时间部分交易所给秒级时间戳部分给毫秒级时间戳CoinGlass 返回的字段通常以毫秒为主但历史接口很容易混入老数据。我的做法是写一个自动判别函数大于 10^11 的按毫秒处理小于 10^11 的按秒处理统一先转成 UTC 时间再转成带时区的 pandas 时间戳import pandas as pd def normalize_time(ts): # 粗略判断毫秒/秒 if abs(ts) 10000000000: ts ts * 1000 return pd.to_datetime(ts, unitms, utcTrue)时区一律存 UTC展示层再转 UTC8。有人在清洗时就转成 UTC8我强烈不建议因为如果清洗表还要给其他时区的同事用就得反复横跳很容易乱。另外要特别注意“未来时间”问题。抓取时如果刚好有延迟任务在执行接口会把已经结算但尚未落到最终状态的记录也返回出来这些记录的时间戳可能刚好在当前时刻之后。我的规则是抓取时间点加上一个安全窗口超过这个窗口的时间戳一律打上is_future True标记不做删除由后续策略层决定是否丢弃。3.2 费率归一化正负号和小数百分比都得抠资金费率是最容易出问题的字段。同一个数值有的接口返回 0.0001有的返回 0.01%有的返回 1.01意味着百分数要除以 100。我的归一化规则如下如果某个数值的绝对值大于 1判定为百分比除以 100如果绝对值在 0 到 1 之间判定为小数费率保持不变绝对值过大、比如超过 50 的记录直接标记为异常不参与费率归一化。这个规则为什么放在“经验值”而不是“硬编码”因为市场极端行情下费率确实会出现 0.1 甚至 0.2 这种罕见值但出现 53 这种数字基本可以判断是单位换算错乱了。阈值要结合真实历史分布来定不要拍脑袋。还有负号的问题。资金费率本来就是可正可负的多头付空头时为正空头付多头时为负。有些接口为了节省字段会对负数做绝对值处理或者把符号放在另一个字段里。我在清洗时专门加了一步检查费率字段经过去重后负值比例如果低于 0.1%基本可以断定负号丢失需要找原始字段交叉验证。3.3 symbol 标准化不覆盖原值增加映射列不同交易所对同一标的的命名差异很大。BTC 永续在几家头部交易所的 symbol 分别可能是 BTCUSDT、BTC-USD、XBTUSD。对策略来说这些都是“比特币永续”但直接按 symbol 分组会把数据劈成好几份。我的处理方案是增加一个symbol_standard列用映射字典统一成btc_perp_usd这类内部代码原始 symbol 仍保留在symbol_raw字段中。有人喜欢直接把原值覆盖掉我提醒一句保留原值会让你在排查交易所特定问题时省很多事千万不要省这个字段。SYMBOL_MAP { BTCUSDT: btc_perp_usd, BTC-USD: btc_perp_usd, XBTUSD: btc_perp_usd, } def map_symbol(raw_symbol: str) - str: return SYMBOL_MAP.get(raw_symbol.upper(), raw_symbol.lower())3.4 缺失数据与异常值只标记不删除刚开始做清洗时我习惯把不满足条件的数据直接删掉后来发现这是最危险的处理方式。你删掉的到底是一条乱码还是一条市场极端波动下的真实记录清洗阶段根本区分不出来。我把清洗结果里的每一行都加上了质量标记字段quality_flag ok正常记录quality_flag missing_field有核心字段缺失quality_flag extreme费率数值超出合理范围quality_flag dup重复记录保留一条但标记其余重复项。断点检测也放在这一步。按交易所的结算周期计算8 小时制每天应该有 3 条记录1 小时制应该有 24 条。清洗完统计每个交易对每天的实际记录数和理论数对比差太多就先报警再决定要不要补抓。补抓的时候我建议重新拉当天全天数据而不是只插缺的那几条这样能用整体的连续性来验证。4. 抓取与清洗的完整实现一套可以直接改的模板前面讲了不少原则这节给出一个能跑通的最小实现。代码不一定覆盖所有极端场景但骨架是完整的。4.1 抓取端封装抓取端我封装成一个类核心方法是按日期区间拉历史资金费率。每次请求都带重试、带日志、带结果落盘。import json import logging import time from datetime import datetime, timezone import requests logger logging.getLogger(__name__) class CoinglassFundingFetcher: def __init__(self, sessionNone): self.session session or requests.Session() self.session.headers.update({User-Agent: DataPipeline/1.0}) def fetch_page(self, symbol: str, page: int, page_size: int 50): params { symbol: symbol, page: page, page_size: page_size, } # 这里的 URL 仅作示意实际以平台公开文档为准 url https://api.coinglass.com/api/funding-rate/history resp self.session.get(url, paramsparams, timeout15) if resp.status_code 429: time.sleep(60) return self.fetch_page(symbol, page, page_size) resp.raise_for_status() return resp.json() def fetch_all(self, symbol: str, save_path: str): page 1 collected [] while True: data self.fetch_page(symbol, pagepage) rows data.get(dataList) or [] if not rows: break collected.extend(rows) logger.info(symbol%s page%s rows%s, symbol, page, len(rows)) page 1 time.sleep(1) with open(save_path, w, encodingutf-8) as f: json.dump({fetched_at: datetime.now(timezone.utc).isoformat(), rows: collected}, f, ensure_asciiFalse) return collected日志一定要记全。我之前就因为没记日志某一次重试把同一个页面插了两遍最后靠“清洗阶段按时间戳去重”才发现。4.2 清洗端封装清洗函数接收原始 JSON 路径输出清洗后的 Parquet 或 CSV。整个过程就是“读入 - 归一化 - 去重 - 打标记 - 落盘”。import pandas as pd def clean_funding_file(raw_path: str, out_path: str) - pd.DataFrame: df pd.read_json(raw_path) raw_rows pd.json_normalize(df[rows]) # 1. 时间戳统一 raw_rows[funding_time] raw_rows[fundingTime].apply(normalize_time) # 2. 费率归一化 mask_pct raw_rows[fundingRate].abs() 1 raw_rows.loc[mask_pct, fundingRate] raw_rows.loc[mask_pct, fundingRate] / 100 # 3. symbol 标准化 raw_rows[symbol_standard] raw_rows[symbol].apply(map_symbol) # 4. 去重同一交易所、同一交易对、同一结算时间只保留一条 raw_rows raw_rows.drop_duplicates( subset[exchange, symbol_standard, funding_time], keeplast ) # 5. 标记缺失字段 raw_rows[quality_flag] ok raw_rows.loc[raw_rows[fundingRate].isna(), quality_flag] missing_field raw_rows.loc[raw_rows[fundingTime].isna(), quality_flag] missing_field raw_rows.to_parquet(out_path, indexFalse) return raw_rows这个清洗函数虽然短已经够应付日常使用。如果要生产化我建议再加上断点检测模块按“交易所 结算周期”分组计算每天的记录数和理论值比对。4.3 校验与入库数据没通过检查就别进正式表校验层设计成独立函数不跟清洗绑定。这样你在调试新规则时可以反复跑校验不用把清洗流程也一起带上。def validate_cleaned(df: pd.DataFrame) - list: problems [] # 1. 时间范围检查 min_time df[funding_time].min() max_time df[funding_time].max() if max_time pd.Timestamp.now(tzUTC): problems.append(存在未来时间戳) # 2. 每对每天条数与理论值对比 df[date] df[funding_time].dt.date count_table df.groupby([exchange, symbol_standard, date]).size() for (exchange, symbol, date), cnt in count_table.items(): expected 3 if exchange in [exchange_a] else 24 if cnt expected: problems.append(f{exchange}-{symbol}-{date} 记录数不足: {cnt}/{expected}) # 3. 极值检查 extreme_cnt (df[fundingRate].abs() 0.5).sum() if extreme_cnt 0: problems.append(f存在 {extreme_cnt} 条费率绝对值大于 0.5 的记录) return problems校验通过才允许写入正式表不通过就把问题列表发到消息队列等人处理。绝不自动跳过我在这里吃过的亏太多。5. 常见问题速查我能想到的坑都在这了这节我把实际运维中遇到的典型问题整理成一张速查表。以后你再遇到类似症状直接看表对应处理就行。现象疑似原因处理方式某天所有交易对都缺数据抓取进程挂了但没有重启策略补跑当日全量抓取校验重算单个交易对某小时缺一条接口分页时漏数据重新拉当日数据按去重规则插入费率全是正数负号在源头被处理掉交叉检查原字段补上符号后再清洗数值突然大十倍百分比/小数未归一化检查归一化阈值重新清洗时间戳出现未来时间结算任务延迟或接口缓存打 is_future 标记等待次日重算同一时间戳出现两条重试机制导致重复插入按交易所交易对时间戳去重5.1 问题一重复记录到底该留哪条这个问题最隐蔽。表面上看重复记录一模一样留哪条都一样但实际不是。如果两条记录分别是两次不同时间请求到的第二次请求拿到的可能包含修正后的数据尤其是费率在极端行情下被回溯修正过。所以我建议去重时保留fetched_at最新的一条而不是最早的一条。我在代码里用keeplast就是这个原因。5.2 问题二资金费率百分制和小数制混用这个坑我栽得特别深。有一次清洗完之后做月度统计发现某天数据比其他月份大百倍查了半天才发现那天的数据源里有一批记录是百分数格式归一化规则没覆盖。后来的解决办法很笨但很稳每次清洗完都画一张费率分布图肉眼扫一眼有无“脱群点”。分布图远比一堆统计指标直观任何异常数量级都会在图上暴露出明显的断档。5.3 问题三断点排查当某天的记录数不足理论值时我一开始会去重跑单天的抓取发现还是缺。后来才明白单天补抓往往会再次命中同一个分页错误正确做法是重算整个时间区间然后靠全量去重把旧数据覆盖掉。规矩不复杂补数据永远补区间不补单点。单点数据没有上下文校验很容易把错误数据当成正确数据写进去。6. 后续扩展与个人建议这个清洗模块最让我心服口服的一点是它把“不可控的原始数据”变成了“可控的中间层”。现在策略团队无论要日频还是周频的资金流数据都从同一张干净表里取值口径一致结果可复现。如果你要从这里往深走我可以给三条个人建议。第一清洗规则一定要带版本号。规则改动是必然发生的把版本号写进输出文件名或表名例如funding_rate_clean_v2.parquet这样规则出问题时可以快速定位“哪些数据是用旧版本规则算的”。第二抓取时的元信息要保留。我每次抓取都会记录fetched_at、请求参数、返回条数。后续出问题对账时这些元信息能帮你快速判断是“抓少了”还是“洗错了”。第三自动化任务尽量错峰。我的定时任务放在每个整点后 15 分钟跑避开各家交易所集中结算的高峰时间看起来只是一个小小的调度设置却能大幅减少遇到接口抖动的概率。最后再分享一个小技巧清洗结果落库之前先抽样 100 条人工看一眼原始 JSON 和清洗表的对照。别嫌麻烦这 100 条记录省下来的排查时间远远超过你花掉的这几分钟。数据清洗这种活耐心比技术更重要。
阅读完成 · 觉得有帮助？
