带时间戳的日志输出，立即 flush

从 markets parquet 加载 token 映射 返回: token_id -> {market_id, condition_id, side, question}

追加写入一批数据

关闭所有 writers，确保数据写入磁盘

从事件中提取交易数据 只保留: - 两个交易所合约 (CTF_EXCHANGE, NEGRISK_CTF_EXCHANGE) 的 - OrderFilled 事件 Args: events: 解码后的事件列表 token_

保存进度和数据

清洗单个 trades DataFrame（用于增量处理） Args: trades_df: 原始 trades DataFrame Returns: 清洗后的 quant DataFrame

清洗单个 trades DataFrame 生成 users 数据（用于增量处理） Args: trades_df: 原始 trades DataFrame Returns: 清洗后的 users DataFrame

分批获取 返回: List[Dict]: 成功时返回记录列表（可能为空） None: RPC 请求失败时返回 None

获取市场列表

保存最后处理的区块到state.json

解析市场数据

处理交易数据批次 1. 过滤NaN价格和合约地址 2. 统一YES视角

处理用户数据批次 1. 过滤NaN价格和合约地址 2. 拆分maker/taker 3. 统一YES视角 4. 统一方向（SELL转BUY，token_amount取负）

发送请求

清理 DuckDB 临时文件

运行所有数据清洗

清洗交易数据（量化用）

清洗用户数据

增量获取新市场（高频运行，如每小时） 特性： - 增量获取新市场，不重复获取已有市场 - 支持安全退出（Ctrl+C） - 支持断点续传（--continue） - 市场数据较小，直接内存操作后一次性保存

获取链上数据（增量模式） 特性： - 使用 PyArrow 流式写入，不读取历史数据 - 支持安全退出（Ctrl+C） - 支持断点续传（--continue）

处理交易数据（带 market_id 关联和缺失 token 补全） 警告：此命令会一次性读取全部数据，仅适用于小数据集！ 对于大数据集（>1GB），请使用 process-historical 命令。

解码 OrderFilled 事件

批量解码

获取并处理一个区块范围

批量格式化

格式化 OrderFilled 事件为输出格式

获取区块时间戳

获取链上最新区块号

通过 token_id 获取市场

获取可用内存的一半作为限制

获取临时目录 - 集群用 SSD，本地用项目目录

保存当前处理到的区块号

解码 data 字段

解码 topic

解析 OrderFilled 事件

处理单个日志

使用最佳可用方法进行排序

更新 CSV 预览文件

批量获取时间戳

清洗交易数据 Args: input_path: 输入 trades.parquet 路径 output_path: 输出 quant.parquet 路径 batch_size: 批处理大小 test_rows:

清洗用户交易数据 Args: input_path: 输入 trades.parquet 路径 output_path: 输出 users.parquet 路径 batch_size: 批处理大小 test_rows

关闭所有 writer

合并所有 session 文件到主文件

分批处理历史数据（用于大文件，避免内存溢出） 使用方式： python3 run.py process-historical --batch-size 1000000 python3 run.py process-historical --continue

全量更新

更新未closed市场的状态（低频运行，如每周） 特性： - 只更新未closed的市场 - 支持安全退出（Ctrl+C） - 支持断点续传（--continue） - 市场数据较小，需要加载全部用于更新 注意：市场文件较小（~60MB），需要完整

将 batch 转换为目标 schema Args: batch: RecordBatch target_schema: 目标 schema Returns: 转换后的 RecordBatch

估算时间戳（减少 RPC 调用）

获取所有市场

获取指定区块范围的日志 返回: List[Dict]: 成功时返回记录列表（可能为空） None: RPC 请求失败时返回 None

批量获取缺失的 token 对应的市场

找出交易中没有映射的 token

获取上次处理的区块 优先级： 1. STATE_FILE 中的 last_block（最准确，记录了实际处理到的区块） 2. orderfilled.parquet 中的最大 block_number（备用） 3. 返回 0（首次运行）

获取区块范围内的 OrderFilled 日志 返回: List[Dict]: 成功时返回日志列表（可能为空） None: RPC 请求失败时返回 None

获取 RPC URL

迭代获取所有市场

加载上次处理到的区块号

合并两个 orderfilled 文件 Args: file1: 第一个文件（需要转换 schema） file2: 第二个文件（作为 schema 基准） output_file: 输出文件 auto_yes: 自

合并多个 parquet 文件 Args: input_files: 输入文件列表（按顺序） output_file: 输出文件路径 dry_run: 是否只显示信息不实际合并 auto_yes: 自动确认覆盖

合并所有session文件到主文件

合并当前 session 文件到主文件（仅当前 session）

读取失败区块列表

记录失败的区块范围

主循环：持续获取新区块

保存最新 N 条交易为 CSV 预览（保留所有字段）

设置日志，输出到控制台和文件

排序 quant.parquet 排序优先级: 1. event_id, 2. market_id, 3. timestamp

排序 users.parquet 排序优先级: 1. user, 2. timestamp

处理停止信号

添加新市场到映射

根据签名获取事件名

创建 token_id -> market 映射