尧图网页设计 
做比特币量化的人应该都绕不开 coinglass 上的资金费率、持仓量、多空比这些数据。我在早期偷过懒直接把从 coinglass 接口抓回来的原始资金流数据塞进策略回测结果被坑得很惨时间戳对不上、字段缺胳膊少腿、偶尔蹦出一个离谱的异常值直接把整个信号逻辑带偏。后来我抽了一个周末把“抓取”和“数据清洗”拆成独立模块专门处理 coinglass 原始资金流数据才算把这些烂摊子一次性收拾干净。这篇文章就是记录这个清洗模块的完整设计思路和实现细节包括抓层选型、pandas 清洗操作、模块化封装以及我踩过的典型坑。适合正在搭量化数据管道、或者需要处理加密市场数据的开发者参考。就算你的数据源不是 coinglass这套清洗逻辑基本也能直接搬过去用关键是里面那些莫名其妙的脏数据你大概率也会遇到。1. 资金流数据为什么需要专门清洗1.1 coinglass 原始资金流数据到底包含什么coinglass 上常见的资金流数据并不是单指“资金费率”一个字段。日常策略里会用到的大致有资金费率funding rate、合约持仓量open interest、多空持仓人数比、大户持仓量占比、成交加权价格、资金流入流出量等。这些数据来自各大合约交易所的公开接口coinglass 做的事情是把多所数据聚合、展示同时也会通过自己的后端接口把中间数据暴露出来。但恰恰因为它是聚合层原始接口返回的结构通常不是为本地计算设计的。我见过最典型的情况一次请求返回的 JSON 里嵌套了好几层同一个时间点在不同交易所分别有不同格式有的返回毫秒时间戳有的返回秒级字符串有的直接把数值做成带%后缀的文本。这些数据如果不做标准化直接读进 pandas后面每一步都能踩雷。我们需要认识到一点资金流数据的核心价值在于“多空力量的前瞻性”。资金费率反映杠杆方向的一致性持仓量变化反映资金进场离场强度。但这类数据的有效信号往往非常微弱如果原始数据里混着一个单位错乱的异常点就足以把整个因子序列污染掉。所以清洗不是可有可无的环节而是量化链路里最笨但最有必要的一道工序。1.2 原始数据的几种典型脏问题我把实际处理中遇到的脏数据问题分了六类每一类都是真实发生过的。第一类是时间戳格式不统一。同一个数据源里有的字段叫timestamp数值是毫秒有的字段叫timeStamp数值是秒还有的返回2025-01-01 08:00:00这种无时区字符串。pandas 的to_datetime能处理一部分但需要你先告诉它输入是什么格式否则它会用本地时区去猜猜错就差好几个小时。第二类是缺失值。某个交易所某一次结算没有上报数据对应字段直接给 null或者某一整段时间因为断网没抓到出现连续的 NaN 区块。缺失值不是简单地删掉就行资金费率这种有“延续性”的字段和成交量这种“累计性”字段处理方式完全不同。第三类是数值单位不统一。资金费率有时候是小数0.0001有时候是百分比0.01%有时候直接乘了 100 显示为0.01。持仓量有的返回张数、有的返回美元价值。单位错了后面算出来的资金流量完全不可用。第四类是异常尖峰。我遇到过一次某天的资金费率突然显示-0.25正常来说 BTC 资金费率极值一般也就 0.02 上下一看就是接口把某个拼接字段搞错了。还有持仓量瞬间归零又恢复的情况这种点如果不剔除会让“持仓量变化”这个因子出现虚假的暴涨暴跌。第五类是重复记录。同一根 K 线被接口重复返回或者你在增量抓取时和上一次任务的时间窗口重叠都会导致下游resample时出现索引冲突。如果不去重统计量会直接翻倍。第六类是时区问题。coinglass 展示层通常用 UTC但你的服务器和本地环境可能是东八区。pandas 默认会把不带时区信息的时间戳当作本地时间当你把两个来源的数据合并时就会出现一个小时级别的错位。1.3 清洗模块在量化链路中的位置完整的数据管道大概是这样的抓取 → 原始数据落盘 → 清洗 → 标准化存储 → 因子计算 → 策略信号。清洗模块就是中间这个承上启下的环节。它的目标不是把数据变得“更好看”而是让下游研究员面对一份类型正确、时间连续、值域合理、口径一致的表格数据。这个定位很重要。清洗模块不应该掺杂策略逻辑也不应该负责具体的因子计算它只负责把“原始状态”变成“可信状态”。这样同时带来一个核心优势策略改了不需要改清洗逻辑清洗逻辑改了只要保存好原始层随时可以重跑清洗流程不需要重新抓数据。我见过不少小型量化项目把所有处理塞在一个 Jupyter Notebook 里抓取、清洗、因子计算一路顺序执行。前期数据量小看不出来问题一旦数据量上升或者需要复盘就会发现自己根本说不清某天的数据到底是哪来的、中间发生了什么。把清洗模块独立出来就是给数据管道装一道隔离门错误不会向下游蔓延。2. 抓层设计把原始数据可靠地拿下来2.1 数据源接入方式选型先说结论优先使用官网提供的公开 API如果没有再考虑网页后端接口网页 HTML 解析是最后的选择能不用就不用。为什么这么排序公开 API 有正式文档、有鉴权机制、字段相对稳定是最好的数据来源。即使你需要 key通常也是免费申请的。网页后端接口也就是页面网络请求中常见的 JSON 接口也能用但它的参数和返回结构没有文档容易随着前端改版而失效。HTML 解析最不推荐页面结构一变整个解析逻辑就废了而且增加很多无意义的页面代码传输量抓取效率也很低。我自己的处理顺序是先去查数据源官网有没有 API 文档有就优先申请 key如果确实只有网页 JSON 接口就打开浏览器开发者工具观察一次正常请求的 Headers 和 Params按那个样子去模拟同时必须注意频率控制只获取公开数据遵守网站条款不给对方服务器制造压力。你可以把这件事想象成去食堂打饭窗口允许你打一份你非要从后厨把整锅端走那肯定会被拉黑。另外抓取模块里最容易被忽视的是请求头设置。很多接口会校验User-Agent少了一个字段就返回 403 或空数据。我一般会把浏览器中常见的 UA 字符串写进配置里而不是写死在代码中方便后期维护。2.2 抓取频率与增量更新策略资金费率通常是 8 小时结算一次但历史数据和更新节奏并不只和结算周期相关。在 coinglass 类数据源中历史持仓量、资金费率往往有逐小时甚至逐分钟的更新入口。如果你的策略只需要日频或 8 小时频不需要追求过高频率否则容易触发限流。我建议把抓取任务拆成两个全量历史任务一次性拉取从某日期到当前的所有历史数据写入原始层文件。增量定时任务每 5 分钟或每 1 小时拉取最新一段数据与本地最大时间戳比较只补缺失区间。增量更新有一个隐蔽的坑交易所可能对历史数据做修正。当你用“本地最大时间戳”作为增量起点时如果修正点在最大时间戳之前会漏掉被修正的数据。所以我会每隔一段时间跑一次针对近 7 天的全量覆盖而不是永远只做增量。简单说就是“增量为主定期补全”。抓取频率要根据数据源的更新时间来定。比如资金费率 8 小时结算一次你跑 1 分钟频率去抓大部分时候都在抓同一段数据除了增加被封几率没什么意义。我的经验是把频率控制在数据源更新时间的一半左右比如持仓量如果 5 分钟更新一次那我 2~3 分钟抓一次就够了。2.3 请求失败与限频处理程序化获取过程中最常碰到的状态码是 429请求过多、403被拒绝、503服务不可用和超时。遇到这些情况最忌讳的是无脑快速重试这只会让限流更严重。我采用指数退避策略第 1 次失败后等待 1 秒第 2 次失败等待 2 秒第 3 次等待 4 秒最多等待 64 秒总重试次数上限 5 次如果 5 次都失败就放弃本次抓取并记录日志告警。在每次请求之间加一个固定的小间隔比如 0.3~0.5 秒能大幅降低触发限流的概率。不要把自己伪装成“正规高频交易系统”更不要开几十个并发去抓除非你有官方许可。低频、守规矩、按接口文档来绝大多数公开数据源都不会限制你。如果被限制的是 IP 维度那就降低请求频率并检查是否超过了接口的每小时调用配额。不要试图搞什么绕过限制的手段该申请配额就申请配额或者换一个官方认可的访问入口这是最稳妥的路径。2.4 数据落盘与原始层保留抓下来的数据应当先落盘再做清洗不要抓一次洗一次直接扔掉。原始层raw data是数据管道里的“黑匣子备份”如果清洗逻辑后面发现有问题只要原始数据还在就可以重新处理。建议按日期分目录存储raw_data/ ├── 2025-01-01/ │ ├── BTCUSDT_funding.json │ ├── ETHUSDT_funding.json └── 2025-01-02/ ├── BTCUSDT_funding.json ├── ETHUSDT_funding.json小数据量用 JSON 或 CSV 都行数据量大推荐按日期存成 Parquet。Parquet 是列式存储压缩率比 CSV 高不少而且 pandas 读取时可以用columns参数只读需要的列节省内存。写入时要注意幂等性同一天的数据如果重复抓取应该用新文件覆盖旧文件而不是追加。追加会产生重复记录后面清洗的时候还要多做一步去重白白增加工作量。保存原始数据时不要做任何清洗保留接口返回的原始字段名和原始值这样才能保证“黑匣子”里存的是真正没被污染的数据。3. 清洗层实现用 pandas 把脏数据变成干净 DataFrame3.1 字段标准化与类型转换清洗层的第一个动作是统一字段名和数据类型。原始 JSON 中同一个业务概念可能有多种命名比如资金费率字段可能叫fundingRate、rate、funding_rate持仓量可能叫openInterest、open_interest_value、total_value。我会先做一个字段映射表把所有列统一成自己想要的英文命名规范。COLUMN_MAP { timestamp: ts, timeStamp: ts, fundingRate: funding_rate, rate: funding_rate, openInterest: open_interest, open_interest_value: open_interest, } def standardize(raw_df): df raw_df.rename(columnsCOLUMN_MAP) required_cols [ts, funding_rate, open_interest] missing_cols set(required_cols) - set(df.columns) if missing_cols: raise ValueError(f缺少字段: {missing_cols}) df[ts] pd.to_datetime(df[ts], utcTrue, errorscoerce) df[funding_rate] pd.to_numeric(df[funding_rate], errorscoerce) df[open_interest] pd.to_numeric(df[open_interest], errorscoerce) return df标准化之后所有字段的类型必须是明确的时间戳是datetime64[ns, UTC]数值是float64。这一步看起来简单但能省掉后面大量“类型不对导致聚合报错”的麻烦。pandas 里最常见的类型陷阱是列实际上是字符串df[open_interest].sum()会把数字拼接成字符串等发现问题时已经晚了。字段缺失检查也很重要。如果你的策略核心需要资金费率和持仓量而某一批次数据没有持仓量字段直接抛异常而不是静默继续。早失败早发现比带着缺字段的 DataFrame 跑完全流程要省事得多。3.2 缺失值处理策略缺失值处理没有万能公式必须按字段的业务含义区分。资金费率这种“定期结算、中间维持”的数据在两次结算之间其实不会改变。所以如果只是个别时间点缺失用前向填充ffill是合理的相当于把上一次结算费率维持到下一次。但如果连续缺失超过两次我倾向于不填充保留 NaN让下游因子计算跳过这段因为连续缺失往往说明数据源本身出了大问题强行填充只会制造假信号。持仓量则不同。持仓量是连续变化的市场状态不适合用前向填充搞出“平台期”。如果某个点缺失我会根据前后相邻点做线性插值因为持仓量短时间内的变化相对平滑插值结果可信度较高。但如果持仓量出现“缺失—恢复”的模式恢复点数值和缺失前差很多说明可能发生了数据中断或合约更换这种要直接标记。成交量类字段处理方式又不一样。成交量是区间累计值不能用前后插值因为某个时间段的成交量缺失就等于这个时间段没有成交应该补 0 或者干脆保留 NaN具体取决于下游是用它做流量估算还是总额统计。我通常在模块里把处理策略做成分离的规则比如funding_rate: 缺失时前向填充连续超过 2 个点填充失败则保留 NaNopen_interest: 缺失时线性插值volume: 缺失时补 0但打上is_estimated标志。3.3 异常值与价格极值过滤异常值过滤是清洗模块中最需要“贴近业务”的部分。千万不要直接丢给 z-score 一把梭因为金融时序数据本身就厚尾正常行情下的持仓量跳变也可能很剧烈简单套用统计阈值会误杀。我的做法分两层。第一层是绝对范围检查。以 BTC 资金费率为例正常结算值通常在±0.02之间极端行情偶尔到±0.05高于 0.05 基本可以认为是脏数据。所以你可以在配置里写一个funding_rate_range: [-0.05, 0.05]超过就置为 NaN。持仓量则检查负值和超出历史均值 N 倍的值。第二层是相邻点突变检查。如果一个值相对于前一个点变化超过 10 倍但后一个点又迅速回到前值附近这个点大概率是异常拼接或接口 bug。用 pandas 计算pct_change就能识别def filter_outliers(df): df df.copy() df.loc[df[funding_rate].abs() 0.05, funding_rate] None pct_prev df[funding_rate].pct_change().abs() pct_next df[funding_rate].pct_change(periods-1).abs() spike_mask (pct_prev 10) (pct_next 10) df.loc[spike_mask, funding_rate] None return df注意pct_change(periods-1)在末尾会产生 NaN实际使用时要先对边界做处理。上面例子只展示核心过滤逻辑生产代码里我会把spike_mask与df[funding_rate].notna()做交合避免把本来就缺失的行标记成异常。过滤后的异常点不要直接删除行更推荐把值置空然后交给缺失值处理流程去决定填充还是跳过。这样保留了“这里曾经发生过什么”的记录你可以在后续分析中统计异常率。3.4 时间序列对齐与重采样coinglass 聚合的数据来自多个交易所而不同交易所的结算时间并不完全对齐。有的在整点有的在 8 小时周期后的第 5 分钟。直接按原始时间序列做多交易所对比会遇到索引错位。我常用的对齐方法是按统一周期重采样。以资金费率为 8 小时频率为例df[ts] pd.to_datetime(df[ts], utcTrue) df df.set_index(ts).sort_index() resampled df.resample(8H).agg({ funding_rate: last, open_interest: last, volume: sum, }).dropna(howall)resample(8H)会把索引自动对齐到标准的 8 小时边界0点、8点、16点每个桶内按规则聚合。资金费率和持仓量取最后一个值因为到结算时才更新成交量则取总和因为成交是一个累计过程。重采样之后时间序列会变成严格等间隔的索引这对后续因子计算非常友好。但是要注意如果原始数据本身频率是 1 小时直接重采样到 8 小时会丢失中间波动不是所有因子都喜欢这种低频。所以我的模块会把重采样默认做成可配置项让下游自己选择需要的周期而不是一刀切。另外所有时间统一用 UTC。技术上把 pandas 时间索引设置为带时区的datetime64[ns, UTC]写入 Parquet 时尽量保留原始时间戳毫秒值或 ISO 8601 字符串。永远不要在存储层使用本地时间否则换台机器、换个时区数据就全乱了。3.5 多数据源冗余校验如果你的资金流数据可以从两个独立来源拿到同一时刻的数据那一定要做冗余校验。这是我发现数据源问题的最有效手段。做法很简单把两个源的数据按相同时间对齐计算差值超过阈值就告警。例如 BTC 资金费率在 Binance 官方接口和 coinglass 聚合接口得到的结算值应该一致。如果两个源在某个时间点的差值超过 0.001%绝对值则说明至少有一个源有问题。校验代码逻辑merged df_source_a.merge(df_source_b, onts, suffixes(_a, _b)) diff (merged[funding_rate_a] - merged[funding_rate_b]).abs() bad diff[diff 1e-5] if not bad.empty: print(f发现 {len(bad)} 个不一致时间点需要检查数据源)有一次我在做冗余校验时发现某个聚合数据源把资金费率的符号丢了所有负值都变成了正数。这个错误靠肉眼很难发现但和另一个源做差值对比时异常点非常明显。冗余校验不仅能保证数据质量也能让你对每个数据源的可靠性心里有底。4. 模块化封装与实战代码4.1 模块整体结构设计把抓取和清洗逻辑拆成模块文件核心目的只有一个让每一步可替换、可重跑、可维护。我建议的项目目录结构funding_data_module/ ├── config.yaml # 数据源地址、字段映射、清洗规则 ├── fetcher.py # 负责抓取原始数据 ├── cleaner.py # 负责清洗加工 ├── tasks.py # 定时任务与入口 ├── raw_data/ # 原始数据层 └── clean_data/ # 清洗结果层fetcher.py只负责和外部数据源交互拿到原始 JSON 后立刻落盘。cleaner.py负责读原始文件、做标准化和清洗、输出干净文件。tasks.py负责调度不掺入复杂逻辑。config.yaml则把字段映射、清洗阈值、请求参数都放到外部配置里避免改参数就要改代码。这样的好处是如果你的数据源从 coinglass 的网页接口换成了官方 API只需要改fetcher.py清洗逻辑完全不用动。如果某个字段的取值范围变化只需要改config.yaml代码逻辑依然保持不变。4.2 抓取函数实现示例下面是一个抓取历史资金费率数据的示例函数。重点展示分页、失败重试、限速这几件事。import time import requests API_KEY your_key_here BASE_URL https://api.example.com/v1 # 以实际接口文档为准 def fetch_history(symbolBTCUSDT, start_tsNone, end_tsNone): params { symbol: symbol, start_time: start_ts, end_time: end_ts, limit: 1000, } headers { User-Agent: data-research-script, XB-API-KEY: API_KEY, } rows [] cursor None while True: p {**params} if cursor: p[cursor] cursor for attempt in range(5): try: resp requests.get( f{BASE_URL}/funding-history, paramsp, headersheaders, timeout10 ) resp.raise_for_status() data resp.json() rows.extend(data.get(data, [])) cursor data.get(next_cursor) if not cursor: return rows break except requests.RequestException as e: wait 2 ** attempt print(f请求失败{wait}秒后重试: {e}) time.sleep(wait) else: raise RuntimeError(连续重试5次仍然失败) time.sleep(0.5)这个函数把分页游标、指数退避和固定间隔都封装在了一起。实际接口可能有不同的分页参数你只需要把cursor替换成接口文档里的分页字段名。重试逻辑的关键在于每次失败后的等待时间呈指数增加避免在服务端还没恢复时继续猛打。4.3 清洗函数实现示例抓取返回的是原始list[dict]清洗的第一步是转成 DataFrame然后依次执行标准化、去重、异常过滤、重采样。import pandas as pd def clean(raw_rows: list[dict]) - pd.DataFrame: df pd.DataFrame(raw_rows) # 1. 字段标准化与类型转换 df standardize(df) # 2. 去重避免重复抓取导致的数据膨胀 df df.drop_duplicates(subset[ts, funding_rate, open_interest]) # 3. 异常值过滤 df filter_outliers(df) # 4. 时间索引与重采样 df df.dropna(subset[ts]) df df.set_index(ts).sort_index() df df.resample(8H).agg({ funding_rate: last, open_interest: last, volume: sum, }).dropna(howall) return df.reset_index()这里有一个细节去重时不要把ts作为唯一键。如果同一个时间戳确实有两笔不同来源的合法数据删除后反而会丢信息。我一般会先按“时间戳 核心字段值”判断重复完全一致才删。清洗函数要保持“纯函数”特性输入原始 rows输出干净的 DataFrame不写数据库、不读文件。这样你可以直接把清洗函数用在单元测试里随便构造一份脏数据就能验证逻辑是否正确。4.4 调度与幂等性处理调度模块主要负责周期性地抓取、清洗、写入。我用过time.sleep循环也用过schedule库生产环境里更推荐APScheduler因为它支持任务持久化和失败重跑。下面是一个基于schedule的简单示例import schedule import time def job(): rows fetch_history( symbolBTCUSDT, start_tslast_clean_ts() - 24 * 3600 * 1000, end_tsnow_ms(), ) df clean(rows) save_clean_to_parquet(df, date_strlatest) schedule.every(30).minutes.do(job) while True: schedule.run_pending() time.sleep(1)幂等性需要特别强调。清洗结果写入时先按日期删除目标文件再写新文件。比如你想保存clean_data/2025-01-01.parquet会先检查文件是否存在存在就覆盖。为什么要这样因为一旦某次任务中途崩溃可能会留下半个文件的脏数据。用“整文件覆盖”策略每次写入都是完整结果重复执行多少次都不会产生脏状态。我还会在任务开头加一个简单的锁机制防止多个调度实例同时运行同一个任务。否则两个进程同时清洗同一天的数据最后写文件时互相覆盖结果不可预测。锁可以用文件锁fcntl或者 Redis 锁简单的场景用文件锁就够了。5. 常见问题与排查技巧实录5.1 抓取返回空数据或被拒绝这是被问得最多的一个问题。明明浏览器里能打开接口放到代码里就返回空列表或者直接 403。排查路径我一般按下面几步走先用浏览器的开发者工具看接口正常请求的完整 Headers 和 Params逐项和代码里的请求对比。最容易漏的是User-Agent、Referer、Origin这样的常规请求头。在命令行用curl原样跑一次确认接口本身是否还能访问。有些接口有缓存或地区限制浏览器里能通不代表服务器上能通。看响应状态码。429 表示请求太频繁403 表示权限或签名问题503 表示服务端不稳定。如果是 429优先降低请求频率在循环里加一个随机 sleep 区间0.3~1 秒而不是粗暴地连续请求。再补充一点如果接口需要签名时间戳必须和服务器保持一致。我遇到过本地时间比服务器快了几分钟导致签名始终不过的情况后来用 NTP 同步时间就解决了。下面是一个排查速查表症状可能原因快速处理返回空列表时间范围参数格式错误将时间戳转成毫秒整数再传返回 403缺少请求头或签名错误对照浏览器请求补齐 Headers返回 429频率过高被限流降低频率增加随机等待时间全部差 8 小时时区处理不一致统一使用 UTC落盘存毫秒资金费率差异巨大单位口径不一致在 config 中校准 scale内存不够DataFrame 全量加载按日分文件 float32 parquet5.2 时间戳时区错乱时间戳时区问题在 pandas 里表现得很隐蔽。你读取一个 CSV 里的2025-01-01 08:00:00pandas 默认把它当作本地时间可实际上它是 UTC 时间。当两个这样的序列合并时差的 8 个小时就出来了。我的方案是所有原始时间戳在进入清洗模块后立刻统一转成带时区的datetime64[ns, UTC]。具体写法是df[ts] pd.to_datetime(df[ts], utcTrue, errorscoerce)utcTrue会强制把无时区的时间解释为 UTC而不是本地时间。落盘时我会刻意保存成 Unix 毫秒整数或者带 UTC 偏移的 ISO 字符串避免再次读取时被环境时区干扰。还有一个特别容易忽略的坑当 DataFrame 的 index 是 UTC 时间保存到 Parquet 时它会被正确保留但保存成 CSV 后时区信息会丢失。所以如果你用 CSV 格式建议额外加一列ts_ms保存毫秒时间戳作为唯一准绳。5.3 资金费率与持仓量数据口径不一致coinglass 页面展示的资金费率常常是百分比比如 0.01%但接口返回可能是小数 0.0001。如果你直接对照页面值却发现清洗后的数据差了一个 100 倍那不是代码问题是单位口径问题。处理方式是为每个字段配置一个scale参数放在config.yaml中fields: funding_rate: source_unit: 0.0001 target_unit: 0.000001 scale: 100 open_interest: source_unit: contracts target_unit: usd scale: 25.7清洗时统一乘这个 scale把原始值转成你内部的标准单位。持仓量更需要小心因为有些交易所按张数计算一张 BTC 合约可能价值 100 USD 或 25.7 USD不同时间的换算值可能还会变。如果你要精确计算资金流量建议直接从原始源拿 USD 计价字段避免用张数手动换算。另外字段名也容易混淆。open_interest有时候代表名义持仓量美元有时候代表张数有时候代表币数量。我会在清洗后的 DataFrame 列名里直接加上单位比如open_interest_usd、funding_rate_pct这样下游使用的人一眼就能分清。5.4 pandas 处理大数据量时内存爆掉几千行数据小打小闹没问题但如果你抓了几年的分钟级数据DataFrame 轻松上千万行内存就会很紧张。我有四个实际建议第一个是按日期分文件处理不要一次性读取全部历史数据。每天一个 Parquet 文件清洗时循环处理然后写回各自的干净文件。这样单次内存占用只与一天的数据量有关全程跑下来不会爆。第二个是尽量减少精确小数位数。pandas 默认float64如果你只有 6 位小数精度其实够用可以转成float32内存直接减半。df[open_interest] df[open_interest].astype(float32)第三个是用category类型处理重复度高的字符串列。比如symbol、exchange这类列的取值只有几种转为category后内存占用会大幅降低。第四个是优先使用 Parquet 而不是 CSV。Parquet 是列式存储读取时可以只加载需要的列CSV 必须整表扫描IO 开销大很多。比如我只想读取ts和funding_rate两列用pd.read_parquet(path, columns[ts, funding_rate])就够了。内存优化的核心思路是不要把数据一次性塞进内存也不要让 pandas 默认用最占空间的类型。按日期分片 类型压缩 列式存储三管齐下绝大多数数据量问题都能解决。最后再说几句这个清洗模块从最初的糙版到现在稳定运行最大的变化不是代码变多而是我把“原始层”和“干净层”彻底分开了。早期图省事抓完就覆盖后来清洗逻辑出 bug想复查原始数据已经晚了只能重新抓浪费时间不说还容易漏数据。现在所有抓取的原始 JSON 都按日期完整保存清洗脚本想重跑多少遍都行这种可重跑性才是数据模块最值钱的地方。如果你刚起步做量化数据管道别着急写策略先把数据层的地板铺平。字段标准化、时间统一、异常过滤、幂等写入这些都是琐碎但回报极高的工程。等哪天下游因子突然变得特别离谱你回头看大概率是某个数据源悄悄改了格式而你因为保留了原始层和清洗日志半小时就能定位到问题。最后再分享一个小技巧日常调试时可以故意往清洗模块里塞一些“带毒数据”比如一个时间戳重复行、一个超范围资金费率、一个字符串数字混排的字段看看模块能不能正确报错或过滤。这种测试做得越多上线后就越稳。数据清洗这件事没有一劳永逸只有一次一次把它做扎实。 Coinglass资金流数据清洗模块设计与实现
做比特币量化的人应该都绕不开 coinglass 上的资金费率、持仓量、多空比这些数据。我在早期偷过懒直接把从 coinglass 接口抓回来的原始资金流数据塞进策略回测结果被坑得很惨时间戳对不上、字段缺胳膊少腿、偶尔蹦出一个离谱的异常值直接把整个信号逻辑带偏。后来我抽了一个周末把“抓取”和“数据清洗”拆成独立模块专门处理 coinglass 原始资金流数据才算把这些烂摊子一次性收拾干净。这篇文章就是记录这个清洗模块的完整设计思路和实现细节包括抓层选型、pandas 清洗操作、模块化封装以及我踩过的典型坑。适合正在搭量化数据管道、或者需要处理加密市场数据的开发者参考。就算你的数据源不是 coinglass这套清洗逻辑基本也能直接搬过去用关键是里面那些莫名其妙的脏数据你大概率也会遇到。1. 资金流数据为什么需要专门清洗1.1 coinglass 原始资金流数据到底包含什么coinglass 上常见的资金流数据并不是单指“资金费率”一个字段。日常策略里会用到的大致有资金费率funding rate、合约持仓量open interest、多空持仓人数比、大户持仓量占比、成交加权价格、资金流入流出量等。这些数据来自各大合约交易所的公开接口coinglass 做的事情是把多所数据聚合、展示同时也会通过自己的后端接口把中间数据暴露出来。但恰恰因为它是聚合层原始接口返回的结构通常不是为本地计算设计的。我见过最典型的情况一次请求返回的 JSON 里嵌套了好几层同一个时间点在不同交易所分别有不同格式有的返回毫秒时间戳有的返回秒级字符串有的直接把数值做成带%后缀的文本。这些数据如果不做标准化直接读进 pandas后面每一步都能踩雷。我们需要认识到一点资金流数据的核心价值在于“多空力量的前瞻性”。资金费率反映杠杆方向的一致性持仓量变化反映资金进场离场强度。但这类数据的有效信号往往非常微弱如果原始数据里混着一个单位错乱的异常点就足以把整个因子序列污染掉。所以清洗不是可有可无的环节而是量化链路里最笨但最有必要的一道工序。1.2 原始数据的几种典型脏问题我把实际处理中遇到的脏数据问题分了六类每一类都是真实发生过的。第一类是时间戳格式不统一。同一个数据源里有的字段叫timestamp数值是毫秒有的字段叫timeStamp数值是秒还有的返回2025-01-01 08:00:00这种无时区字符串。pandas 的to_datetime能处理一部分但需要你先告诉它输入是什么格式否则它会用本地时区去猜猜错就差好几个小时。第二类是缺失值。某个交易所某一次结算没有上报数据对应字段直接给 null或者某一整段时间因为断网没抓到出现连续的 NaN 区块。缺失值不是简单地删掉就行资金费率这种有“延续性”的字段和成交量这种“累计性”字段处理方式完全不同。第三类是数值单位不统一。资金费率有时候是小数0.0001有时候是百分比0.01%有时候直接乘了 100 显示为0.01。持仓量有的返回张数、有的返回美元价值。单位错了后面算出来的资金流量完全不可用。第四类是异常尖峰。我遇到过一次某天的资金费率突然显示-0.25正常来说 BTC 资金费率极值一般也就 0.02 上下一看就是接口把某个拼接字段搞错了。还有持仓量瞬间归零又恢复的情况这种点如果不剔除会让“持仓量变化”这个因子出现虚假的暴涨暴跌。第五类是重复记录。同一根 K 线被接口重复返回或者你在增量抓取时和上一次任务的时间窗口重叠都会导致下游resample时出现索引冲突。如果不去重统计量会直接翻倍。第六类是时区问题。coinglass 展示层通常用 UTC但你的服务器和本地环境可能是东八区。pandas 默认会把不带时区信息的时间戳当作本地时间当你把两个来源的数据合并时就会出现一个小时级别的错位。1.3 清洗模块在量化链路中的位置完整的数据管道大概是这样的抓取 → 原始数据落盘 → 清洗 → 标准化存储 → 因子计算 → 策略信号。清洗模块就是中间这个承上启下的环节。它的目标不是把数据变得“更好看”而是让下游研究员面对一份类型正确、时间连续、值域合理、口径一致的表格数据。这个定位很重要。清洗模块不应该掺杂策略逻辑也不应该负责具体的因子计算它只负责把“原始状态”变成“可信状态”。这样同时带来一个核心优势策略改了不需要改清洗逻辑清洗逻辑改了只要保存好原始层随时可以重跑清洗流程不需要重新抓数据。我见过不少小型量化项目把所有处理塞在一个 Jupyter Notebook 里抓取、清洗、因子计算一路顺序执行。前期数据量小看不出来问题一旦数据量上升或者需要复盘就会发现自己根本说不清某天的数据到底是哪来的、中间发生了什么。把清洗模块独立出来就是给数据管道装一道隔离门错误不会向下游蔓延。2. 抓层设计把原始数据可靠地拿下来2.1 数据源接入方式选型先说结论优先使用官网提供的公开 API如果没有再考虑网页后端接口网页 HTML 解析是最后的选择能不用就不用。为什么这么排序公开 API 有正式文档、有鉴权机制、字段相对稳定是最好的数据来源。即使你需要 key通常也是免费申请的。网页后端接口也就是页面网络请求中常见的 JSON 接口也能用但它的参数和返回结构没有文档容易随着前端改版而失效。HTML 解析最不推荐页面结构一变整个解析逻辑就废了而且增加很多无意义的页面代码传输量抓取效率也很低。我自己的处理顺序是先去查数据源官网有没有 API 文档有就优先申请 key如果确实只有网页 JSON 接口就打开浏览器开发者工具观察一次正常请求的 Headers 和 Params按那个样子去模拟同时必须注意频率控制只获取公开数据遵守网站条款不给对方服务器制造压力。你可以把这件事想象成去食堂打饭窗口允许你打一份你非要从后厨把整锅端走那肯定会被拉黑。另外抓取模块里最容易被忽视的是请求头设置。很多接口会校验User-Agent少了一个字段就返回 403 或空数据。我一般会把浏览器中常见的 UA 字符串写进配置里而不是写死在代码中方便后期维护。2.2 抓取频率与增量更新策略资金费率通常是 8 小时结算一次但历史数据和更新节奏并不只和结算周期相关。在 coinglass 类数据源中历史持仓量、资金费率往往有逐小时甚至逐分钟的更新入口。如果你的策略只需要日频或 8 小时频不需要追求过高频率否则容易触发限流。我建议把抓取任务拆成两个全量历史任务一次性拉取从某日期到当前的所有历史数据写入原始层文件。增量定时任务每 5 分钟或每 1 小时拉取最新一段数据与本地最大时间戳比较只补缺失区间。增量更新有一个隐蔽的坑交易所可能对历史数据做修正。当你用“本地最大时间戳”作为增量起点时如果修正点在最大时间戳之前会漏掉被修正的数据。所以我会每隔一段时间跑一次针对近 7 天的全量覆盖而不是永远只做增量。简单说就是“增量为主定期补全”。抓取频率要根据数据源的更新时间来定。比如资金费率 8 小时结算一次你跑 1 分钟频率去抓大部分时候都在抓同一段数据除了增加被封几率没什么意义。我的经验是把频率控制在数据源更新时间的一半左右比如持仓量如果 5 分钟更新一次那我 2~3 分钟抓一次就够了。2.3 请求失败与限频处理程序化获取过程中最常碰到的状态码是 429请求过多、403被拒绝、503服务不可用和超时。遇到这些情况最忌讳的是无脑快速重试这只会让限流更严重。我采用指数退避策略第 1 次失败后等待 1 秒第 2 次失败等待 2 秒第 3 次等待 4 秒最多等待 64 秒总重试次数上限 5 次如果 5 次都失败就放弃本次抓取并记录日志告警。在每次请求之间加一个固定的小间隔比如 0.3~0.5 秒能大幅降低触发限流的概率。不要把自己伪装成“正规高频交易系统”更不要开几十个并发去抓除非你有官方许可。低频、守规矩、按接口文档来绝大多数公开数据源都不会限制你。如果被限制的是 IP 维度那就降低请求频率并检查是否超过了接口的每小时调用配额。不要试图搞什么绕过限制的手段该申请配额就申请配额或者换一个官方认可的访问入口这是最稳妥的路径。2.4 数据落盘与原始层保留抓下来的数据应当先落盘再做清洗不要抓一次洗一次直接扔掉。原始层raw data是数据管道里的“黑匣子备份”如果清洗逻辑后面发现有问题只要原始数据还在就可以重新处理。建议按日期分目录存储raw_data/ ├── 2025-01-01/ │ ├── BTCUSDT_funding.json │ ├── ETHUSDT_funding.json └── 2025-01-02/ ├── BTCUSDT_funding.json ├── ETHUSDT_funding.json小数据量用 JSON 或 CSV 都行数据量大推荐按日期存成 Parquet。Parquet 是列式存储压缩率比 CSV 高不少而且 pandas 读取时可以用columns参数只读需要的列节省内存。写入时要注意幂等性同一天的数据如果重复抓取应该用新文件覆盖旧文件而不是追加。追加会产生重复记录后面清洗的时候还要多做一步去重白白增加工作量。保存原始数据时不要做任何清洗保留接口返回的原始字段名和原始值这样才能保证“黑匣子”里存的是真正没被污染的数据。3. 清洗层实现用 pandas 把脏数据变成干净 DataFrame3.1 字段标准化与类型转换清洗层的第一个动作是统一字段名和数据类型。原始 JSON 中同一个业务概念可能有多种命名比如资金费率字段可能叫fundingRate、rate、funding_rate持仓量可能叫openInterest、open_interest_value、total_value。我会先做一个字段映射表把所有列统一成自己想要的英文命名规范。COLUMN_MAP { timestamp: ts, timeStamp: ts, fundingRate: funding_rate, rate: funding_rate, openInterest: open_interest, open_interest_value: open_interest, } def standardize(raw_df): df raw_df.rename(columnsCOLUMN_MAP) required_cols [ts, funding_rate, open_interest] missing_cols set(required_cols) - set(df.columns) if missing_cols: raise ValueError(f缺少字段: {missing_cols}) df[ts] pd.to_datetime(df[ts], utcTrue, errorscoerce) df[funding_rate] pd.to_numeric(df[funding_rate], errorscoerce) df[open_interest] pd.to_numeric(df[open_interest], errorscoerce) return df标准化之后所有字段的类型必须是明确的时间戳是datetime64[ns, UTC]数值是float64。这一步看起来简单但能省掉后面大量“类型不对导致聚合报错”的麻烦。pandas 里最常见的类型陷阱是列实际上是字符串df[open_interest].sum()会把数字拼接成字符串等发现问题时已经晚了。字段缺失检查也很重要。如果你的策略核心需要资金费率和持仓量而某一批次数据没有持仓量字段直接抛异常而不是静默继续。早失败早发现比带着缺字段的 DataFrame 跑完全流程要省事得多。3.2 缺失值处理策略缺失值处理没有万能公式必须按字段的业务含义区分。资金费率这种“定期结算、中间维持”的数据在两次结算之间其实不会改变。所以如果只是个别时间点缺失用前向填充ffill是合理的相当于把上一次结算费率维持到下一次。但如果连续缺失超过两次我倾向于不填充保留 NaN让下游因子计算跳过这段因为连续缺失往往说明数据源本身出了大问题强行填充只会制造假信号。持仓量则不同。持仓量是连续变化的市场状态不适合用前向填充搞出“平台期”。如果某个点缺失我会根据前后相邻点做线性插值因为持仓量短时间内的变化相对平滑插值结果可信度较高。但如果持仓量出现“缺失—恢复”的模式恢复点数值和缺失前差很多说明可能发生了数据中断或合约更换这种要直接标记。成交量类字段处理方式又不一样。成交量是区间累计值不能用前后插值因为某个时间段的成交量缺失就等于这个时间段没有成交应该补 0 或者干脆保留 NaN具体取决于下游是用它做流量估算还是总额统计。我通常在模块里把处理策略做成分离的规则比如funding_rate: 缺失时前向填充连续超过 2 个点填充失败则保留 NaNopen_interest: 缺失时线性插值volume: 缺失时补 0但打上is_estimated标志。3.3 异常值与价格极值过滤异常值过滤是清洗模块中最需要“贴近业务”的部分。千万不要直接丢给 z-score 一把梭因为金融时序数据本身就厚尾正常行情下的持仓量跳变也可能很剧烈简单套用统计阈值会误杀。我的做法分两层。第一层是绝对范围检查。以 BTC 资金费率为例正常结算值通常在±0.02之间极端行情偶尔到±0.05高于 0.05 基本可以认为是脏数据。所以你可以在配置里写一个funding_rate_range: [-0.05, 0.05]超过就置为 NaN。持仓量则检查负值和超出历史均值 N 倍的值。第二层是相邻点突变检查。如果一个值相对于前一个点变化超过 10 倍但后一个点又迅速回到前值附近这个点大概率是异常拼接或接口 bug。用 pandas 计算pct_change就能识别def filter_outliers(df): df df.copy() df.loc[df[funding_rate].abs() 0.05, funding_rate] None pct_prev df[funding_rate].pct_change().abs() pct_next df[funding_rate].pct_change(periods-1).abs() spike_mask (pct_prev 10) (pct_next 10) df.loc[spike_mask, funding_rate] None return df注意pct_change(periods-1)在末尾会产生 NaN实际使用时要先对边界做处理。上面例子只展示核心过滤逻辑生产代码里我会把spike_mask与df[funding_rate].notna()做交合避免把本来就缺失的行标记成异常。过滤后的异常点不要直接删除行更推荐把值置空然后交给缺失值处理流程去决定填充还是跳过。这样保留了“这里曾经发生过什么”的记录你可以在后续分析中统计异常率。3.4 时间序列对齐与重采样coinglass 聚合的数据来自多个交易所而不同交易所的结算时间并不完全对齐。有的在整点有的在 8 小时周期后的第 5 分钟。直接按原始时间序列做多交易所对比会遇到索引错位。我常用的对齐方法是按统一周期重采样。以资金费率为 8 小时频率为例df[ts] pd.to_datetime(df[ts], utcTrue) df df.set_index(ts).sort_index() resampled df.resample(8H).agg({ funding_rate: last, open_interest: last, volume: sum, }).dropna(howall)resample(8H)会把索引自动对齐到标准的 8 小时边界0点、8点、16点每个桶内按规则聚合。资金费率和持仓量取最后一个值因为到结算时才更新成交量则取总和因为成交是一个累计过程。重采样之后时间序列会变成严格等间隔的索引这对后续因子计算非常友好。但是要注意如果原始数据本身频率是 1 小时直接重采样到 8 小时会丢失中间波动不是所有因子都喜欢这种低频。所以我的模块会把重采样默认做成可配置项让下游自己选择需要的周期而不是一刀切。另外所有时间统一用 UTC。技术上把 pandas 时间索引设置为带时区的datetime64[ns, UTC]写入 Parquet 时尽量保留原始时间戳毫秒值或 ISO 8601 字符串。永远不要在存储层使用本地时间否则换台机器、换个时区数据就全乱了。3.5 多数据源冗余校验如果你的资金流数据可以从两个独立来源拿到同一时刻的数据那一定要做冗余校验。这是我发现数据源问题的最有效手段。做法很简单把两个源的数据按相同时间对齐计算差值超过阈值就告警。例如 BTC 资金费率在 Binance 官方接口和 coinglass 聚合接口得到的结算值应该一致。如果两个源在某个时间点的差值超过 0.001%绝对值则说明至少有一个源有问题。校验代码逻辑merged df_source_a.merge(df_source_b, onts, suffixes(_a, _b)) diff (merged[funding_rate_a] - merged[funding_rate_b]).abs() bad diff[diff 1e-5] if not bad.empty: print(f发现 {len(bad)} 个不一致时间点需要检查数据源)有一次我在做冗余校验时发现某个聚合数据源把资金费率的符号丢了所有负值都变成了正数。这个错误靠肉眼很难发现但和另一个源做差值对比时异常点非常明显。冗余校验不仅能保证数据质量也能让你对每个数据源的可靠性心里有底。4. 模块化封装与实战代码4.1 模块整体结构设计把抓取和清洗逻辑拆成模块文件核心目的只有一个让每一步可替换、可重跑、可维护。我建议的项目目录结构funding_data_module/ ├── config.yaml # 数据源地址、字段映射、清洗规则 ├── fetcher.py # 负责抓取原始数据 ├── cleaner.py # 负责清洗加工 ├── tasks.py # 定时任务与入口 ├── raw_data/ # 原始数据层 └── clean_data/ # 清洗结果层fetcher.py只负责和外部数据源交互拿到原始 JSON 后立刻落盘。cleaner.py负责读原始文件、做标准化和清洗、输出干净文件。tasks.py负责调度不掺入复杂逻辑。config.yaml则把字段映射、清洗阈值、请求参数都放到外部配置里避免改参数就要改代码。这样的好处是如果你的数据源从 coinglass 的网页接口换成了官方 API只需要改fetcher.py清洗逻辑完全不用动。如果某个字段的取值范围变化只需要改config.yaml代码逻辑依然保持不变。4.2 抓取函数实现示例下面是一个抓取历史资金费率数据的示例函数。重点展示分页、失败重试、限速这几件事。import time import requests API_KEY your_key_here BASE_URL https://api.example.com/v1 # 以实际接口文档为准 def fetch_history(symbolBTCUSDT, start_tsNone, end_tsNone): params { symbol: symbol, start_time: start_ts, end_time: end_ts, limit: 1000, } headers { User-Agent: data-research-script, XB-API-KEY: API_KEY, } rows [] cursor None while True: p {**params} if cursor: p[cursor] cursor for attempt in range(5): try: resp requests.get( f{BASE_URL}/funding-history, paramsp, headersheaders, timeout10 ) resp.raise_for_status() data resp.json() rows.extend(data.get(data, [])) cursor data.get(next_cursor) if not cursor: return rows break except requests.RequestException as e: wait 2 ** attempt print(f请求失败{wait}秒后重试: {e}) time.sleep(wait) else: raise RuntimeError(连续重试5次仍然失败) time.sleep(0.5)这个函数把分页游标、指数退避和固定间隔都封装在了一起。实际接口可能有不同的分页参数你只需要把cursor替换成接口文档里的分页字段名。重试逻辑的关键在于每次失败后的等待时间呈指数增加避免在服务端还没恢复时继续猛打。4.3 清洗函数实现示例抓取返回的是原始list[dict]清洗的第一步是转成 DataFrame然后依次执行标准化、去重、异常过滤、重采样。import pandas as pd def clean(raw_rows: list[dict]) - pd.DataFrame: df pd.DataFrame(raw_rows) # 1. 字段标准化与类型转换 df standardize(df) # 2. 去重避免重复抓取导致的数据膨胀 df df.drop_duplicates(subset[ts, funding_rate, open_interest]) # 3. 异常值过滤 df filter_outliers(df) # 4. 时间索引与重采样 df df.dropna(subset[ts]) df df.set_index(ts).sort_index() df df.resample(8H).agg({ funding_rate: last, open_interest: last, volume: sum, }).dropna(howall) return df.reset_index()这里有一个细节去重时不要把ts作为唯一键。如果同一个时间戳确实有两笔不同来源的合法数据删除后反而会丢信息。我一般会先按“时间戳 核心字段值”判断重复完全一致才删。清洗函数要保持“纯函数”特性输入原始 rows输出干净的 DataFrame不写数据库、不读文件。这样你可以直接把清洗函数用在单元测试里随便构造一份脏数据就能验证逻辑是否正确。4.4 调度与幂等性处理调度模块主要负责周期性地抓取、清洗、写入。我用过time.sleep循环也用过schedule库生产环境里更推荐APScheduler因为它支持任务持久化和失败重跑。下面是一个基于schedule的简单示例import schedule import time def job(): rows fetch_history( symbolBTCUSDT, start_tslast_clean_ts() - 24 * 3600 * 1000, end_tsnow_ms(), ) df clean(rows) save_clean_to_parquet(df, date_strlatest) schedule.every(30).minutes.do(job) while True: schedule.run_pending() time.sleep(1)幂等性需要特别强调。清洗结果写入时先按日期删除目标文件再写新文件。比如你想保存clean_data/2025-01-01.parquet会先检查文件是否存在存在就覆盖。为什么要这样因为一旦某次任务中途崩溃可能会留下半个文件的脏数据。用“整文件覆盖”策略每次写入都是完整结果重复执行多少次都不会产生脏状态。我还会在任务开头加一个简单的锁机制防止多个调度实例同时运行同一个任务。否则两个进程同时清洗同一天的数据最后写文件时互相覆盖结果不可预测。锁可以用文件锁fcntl或者 Redis 锁简单的场景用文件锁就够了。5. 常见问题与排查技巧实录5.1 抓取返回空数据或被拒绝这是被问得最多的一个问题。明明浏览器里能打开接口放到代码里就返回空列表或者直接 403。排查路径我一般按下面几步走先用浏览器的开发者工具看接口正常请求的完整 Headers 和 Params逐项和代码里的请求对比。最容易漏的是User-Agent、Referer、Origin这样的常规请求头。在命令行用curl原样跑一次确认接口本身是否还能访问。有些接口有缓存或地区限制浏览器里能通不代表服务器上能通。看响应状态码。429 表示请求太频繁403 表示权限或签名问题503 表示服务端不稳定。如果是 429优先降低请求频率在循环里加一个随机 sleep 区间0.3~1 秒而不是粗暴地连续请求。再补充一点如果接口需要签名时间戳必须和服务器保持一致。我遇到过本地时间比服务器快了几分钟导致签名始终不过的情况后来用 NTP 同步时间就解决了。下面是一个排查速查表症状可能原因快速处理返回空列表时间范围参数格式错误将时间戳转成毫秒整数再传返回 403缺少请求头或签名错误对照浏览器请求补齐 Headers返回 429频率过高被限流降低频率增加随机等待时间全部差 8 小时时区处理不一致统一使用 UTC落盘存毫秒资金费率差异巨大单位口径不一致在 config 中校准 scale内存不够DataFrame 全量加载按日分文件 float32 parquet5.2 时间戳时区错乱时间戳时区问题在 pandas 里表现得很隐蔽。你读取一个 CSV 里的2025-01-01 08:00:00pandas 默认把它当作本地时间可实际上它是 UTC 时间。当两个这样的序列合并时差的 8 个小时就出来了。我的方案是所有原始时间戳在进入清洗模块后立刻统一转成带时区的datetime64[ns, UTC]。具体写法是df[ts] pd.to_datetime(df[ts], utcTrue, errorscoerce)utcTrue会强制把无时区的时间解释为 UTC而不是本地时间。落盘时我会刻意保存成 Unix 毫秒整数或者带 UTC 偏移的 ISO 字符串避免再次读取时被环境时区干扰。还有一个特别容易忽略的坑当 DataFrame 的 index 是 UTC 时间保存到 Parquet 时它会被正确保留但保存成 CSV 后时区信息会丢失。所以如果你用 CSV 格式建议额外加一列ts_ms保存毫秒时间戳作为唯一准绳。5.3 资金费率与持仓量数据口径不一致coinglass 页面展示的资金费率常常是百分比比如 0.01%但接口返回可能是小数 0.0001。如果你直接对照页面值却发现清洗后的数据差了一个 100 倍那不是代码问题是单位口径问题。处理方式是为每个字段配置一个scale参数放在config.yaml中fields: funding_rate: source_unit: 0.0001 target_unit: 0.000001 scale: 100 open_interest: source_unit: contracts target_unit: usd scale: 25.7清洗时统一乘这个 scale把原始值转成你内部的标准单位。持仓量更需要小心因为有些交易所按张数计算一张 BTC 合约可能价值 100 USD 或 25.7 USD不同时间的换算值可能还会变。如果你要精确计算资金流量建议直接从原始源拿 USD 计价字段避免用张数手动换算。另外字段名也容易混淆。open_interest有时候代表名义持仓量美元有时候代表张数有时候代表币数量。我会在清洗后的 DataFrame 列名里直接加上单位比如open_interest_usd、funding_rate_pct这样下游使用的人一眼就能分清。5.4 pandas 处理大数据量时内存爆掉几千行数据小打小闹没问题但如果你抓了几年的分钟级数据DataFrame 轻松上千万行内存就会很紧张。我有四个实际建议第一个是按日期分文件处理不要一次性读取全部历史数据。每天一个 Parquet 文件清洗时循环处理然后写回各自的干净文件。这样单次内存占用只与一天的数据量有关全程跑下来不会爆。第二个是尽量减少精确小数位数。pandas 默认float64如果你只有 6 位小数精度其实够用可以转成float32内存直接减半。df[open_interest] df[open_interest].astype(float32)第三个是用category类型处理重复度高的字符串列。比如symbol、exchange这类列的取值只有几种转为category后内存占用会大幅降低。第四个是优先使用 Parquet 而不是 CSV。Parquet 是列式存储读取时可以只加载需要的列CSV 必须整表扫描IO 开销大很多。比如我只想读取ts和funding_rate两列用pd.read_parquet(path, columns[ts, funding_rate])就够了。内存优化的核心思路是不要把数据一次性塞进内存也不要让 pandas 默认用最占空间的类型。按日期分片 类型压缩 列式存储三管齐下绝大多数数据量问题都能解决。最后再说几句这个清洗模块从最初的糙版到现在稳定运行最大的变化不是代码变多而是我把“原始层”和“干净层”彻底分开了。早期图省事抓完就覆盖后来清洗逻辑出 bug想复查原始数据已经晚了只能重新抓浪费时间不说还容易漏数据。现在所有抓取的原始 JSON 都按日期完整保存清洗脚本想重跑多少遍都行这种可重跑性才是数据模块最值钱的地方。如果你刚起步做量化数据管道别着急写策略先把数据层的地板铺平。字段标准化、时间统一、异常过滤、幂等写入这些都是琐碎但回报极高的工程。等哪天下游因子突然变得特别离谱你回头看大概率是某个数据源悄悄改了格式而你因为保留了原始层和清洗日志半小时就能定位到问题。最后再分享一个小技巧日常调试时可以故意往清洗模块里塞一些“带毒数据”比如一个时间戳重复行、一个超范围资金费率、一个字符串数字混排的字段看看模块能不能正确报错或过滤。这种测试做得越多上线后就越稳。数据清洗这件事没有一劳永逸只有一次一次把它做扎实。