1. 本地量化数据库为什么需要一套数据质检系统做量化研究的人大多经历过这个阶段数据能查、代码能跑、回测有结果于是默认数据是可信的。直到某天发现某只股票 1992 年的最高价低于最低价或者复权后的收益率在某一天突然跳了 30%才意识到问题一直存在只是没人去看。本地量化数据库的数据质检系统本质上就是给数据库配一个体检流程在数据同步完成之后自动检查行情和复权因子是否完整、数值是否合理、内部是否自洽、有没有重复记录。它适合所有把股票、指数、ETF 数据落到本地做研究的人尤其是已经跑通了数据同步、但还没建立质量监控环节的开发者。我这次的做法是用 Codex 作为编码助手把质检规则、执行脚本和调度骨架搭起来同时用 TaoToken 的统一 Key 和 API 通道解决模型调用接入的问题。这样做的直接好处是质检脚本里需要模型判断的地方比如异常归类、规则解释、报告摘要不用再分别配置多家厂商的 Key一个 Base URL 加一个 Key 就能覆盖。整篇文章会给出可复制的config.toml配置骨架、质检规则清单、运行验证动作以及实际跑全量数据时踩到的报错和排查方式。你可以按步骤在本地复现一套能用的流程不需要一开始就追求覆盖所有市场。先说清楚边界第一版只聚焦股票、指数、ETF 的日线行情和复权因子期货和期权先记录问题、隔离处理。范围收敛之后规则设计和脚本实现才不会失控。2. TaoToken 统一 Key 接入与 config.toml 配置骨架在写质检脚本之前先把模型调用的通道固定下来。质检系统里模型主要用在三个地方一是把异常明细归类成可读的问题描述二是对复权收益率跳变做原因初判三是生成每日质检摘要。这些调用如果分散到不同厂商Key 管理和额度监控会很麻烦。TaoToken 在这里的角色是统一入口官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。你只需要在配置里写一个 Base URL 和一个 Key模型 ID 按需切换。2.1 先拿 Key再写配置进入控制台创建 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。创建之后复制出来后面写进config.toml。如果你还没决定用哪个模型可以先在模型对话页面试一下地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。Key 的管理页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 建议给质检系统单独建一个 Key方便按项目统计用量。2.2 config.toml 配置骨架下面这份配置可以直接复制路径放在项目根目录的config/config.toml。字段含义我写在注释里注意base_url不要带多余路径model按你实际可用的 ID 填。# config/config.toml # 本地量化数据库质检系统配置骨架 [llm] # TaoToken 统一 API 入口 base_url https://taotoken.net/api # 从控制台创建的 Key建议用环境变量注入 api_key ${TAOTOKEN_API_KEY} # 质检摘要和异常归类使用的模型 model gpt-5.4-mini # 单次请求超时秒 timeout 60 # 失败重试次数 max_retries 3 [llm.roles] # 设计、review、测试计划用强模型 planner gpt-5.5 # 机械编码和批量归类用轻量模型 executor gpt-5.4-mini [database] # 本地量化数据库路径 path ./data/zer0share.db # 首期接入的表 tables [stock_daily, index_daily, etf_daily, fund_adj] [quality] # 质检维度开关 enable_completeness true enable_accuracy true enable_consistency true enable_uniqueness true # 单条规则失败是否继续 continue_on_rule_error true # 小范围测试的起始日期 test_start_date 2024-01-01 test_end_date 2024-01-31 [quality.rules] # OHLC 关系检查 ohlc_relation true # 涨跌幅一致性检查 pct_change_consistency true # 成交额非负检查 amount_non_negative true # 复权收益率跳变检查 adj_return_jump true # 跳变阈值 adj_return_jump_threshold 0.5 [report] # 输出目录 output_dir ./reports # 同时输出总体概况和异常明细 summary true detail true配置写完之后用环境变量注入 Key避免明文写在文件里export TAOTOKEN_API_KEY你的Key如果你用的是 Claude Code 做执行端可以在项目里加一份.claude/settings.json把 Base URL 和 Key 的读取方式固定下来{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: ${TAOTOKEN_API_KEY} } }这样无论是 Codex CLI 还是 Claude Code读的都是同一套通道切换执行模型时不用改业务代码。需要说明的是模型 ID 和可用范围以控制台实际展示为准配置里的名称只是示例。3. 质检规则清单与可复制脚本骨架配置固定之后进入规则实现。质检规则按四个维度组织完整性、准确性、一致性、唯一性。第一版把一致性规则做扎实准确性先留接口。3.1 一致性规则清单日线行情的基础一致性规则如下这些规则单看每个字段都正常但组合到同一条记录里就可能出问题-- 一致性检查OHLC 关系与数值范围 SELECT * FROM stock_daily WHERE trade_date BETWEEN :start_date AND :end_date AND ( open 0 OR high 0 OR low 0 OR close 0 OR amount 0 OR volume 0 OR high MAX(open, close) OR low MIN(open, close) OR high low );涨跌幅一致性单独一条用pre_close反推-- 涨跌幅一致性close / pre_close - 1 与 pct_change 偏差 SELECT *, ABS((close / pre_close - 1) - pct_change) AS diff FROM stock_daily WHERE trade_date BETWEEN :start_date AND :end_date AND pre_close 0 AND ABS((close / pre_close - 1) - pct_change) 0.001;唯一性检查用分组计数-- 唯一性同一标的、交易日、频率只应有一条 SELECT ts_code, trade_date, freq, COUNT(*) AS cnt FROM stock_daily GROUP BY ts_code, trade_date, freq HAVING cnt 1;完整性检查需要交易日历和股票基础信息配合先算出理论上应该有哪些数据再和实际数据做差集。这部分我放在quality/completeness.py里核心逻辑是# quality/completeness.py import pandas as pd def check_missing_daily(calendar_df, stock_basic_df, daily_df, trade_date): 检查某交易日应有行情但缺失的标的 expected stock_basic_df[ (stock_basic_df[list_date] trade_date) (stock_basic_df[delist_date].isna() | (stock_basic_df[delist_date] trade_date)) ][ts_code] actual daily_df[daily_df[trade_date] trade_date][ts_code] missing set(expected) - set(actual) return sorted(missing)3.2 复权收益率跳变检查这是第一版里最有价值的一条规则。复权因子存在不代表数值正确只有把它应用到价格上、再检查收益率连续性才能发现隐藏问题。# quality/adj_check.py import pandas as pd def check_adj_return_jump(daily_df, adj_df, threshold0.5): 计算后复权收盘价检查连续收益率是否异常跳变 merged daily_df.merge(adj_df, on[ts_code, trade_date], howleft) merged merged.sort_values([ts_code, trade_date]) merged[adj_close] merged[close] * merged[adj_factor] merged[adj_return] merged.groupby(ts_code)[adj_close].pct_change() jumps merged[merged[adj_return].abs() threshold] return jumps[[ts_code, trade_date, adj_return, adj_factor]]阈值先设 0.5跑小范围数据时观察误报率再决定是否收紧。实测下来早期数据1994—1995、2006—2007跳变较多近年数据偶发需要人工核对。3.3 规则执行器骨架单条规则失败不能中断整批任务所以执行器要包一层异常捕获# quality/runner.py import logging from quality import consistency, uniqueness, completeness, adj_check logger logging.getLogger(__name__) RULES [ (ohlc_relation, consistency.check_ohlc_relation), (pct_change_consistency, consistency.check_pct_change), (uniqueness, uniqueness.check_duplicate), (adj_return_jump, adj_check.check_adj_return_jump), ] def run_all(conn, start_date, end_date, continue_on_errorTrue): results {} for name, func in RULES: try: results[name] func(conn, start_date, end_date) logger.info(rule %s done, %d issues, name, len(results[name])) except Exception as exc: logger.error(rule %s failed: %s, name, exc) results[name] {error: str(exc)} if not continue_on_error: raise return results到这里规则层和配置层就打通了。模型调用只在报告生成阶段介入把异常明细交给executor模型归类把整体结论交给planner模型复核。4. 运行验证从单月小数据到全量历史规则写完不要直接跑全市场全历史先用一个月数据验证命令、规则、输出和异常处理是否正常。4.1 小范围验证# 激活环境后运行单月质检 python -m quality.runner \ --config config/config.toml \ --start 2024-01-01 \ --end 2024-01-31 \ --tables stock_daily,fund_adj预期输出类似[INFO] rule ohlc_relation done, 0 issues [INFO] rule pct_change_consistency done, 2 issues [INFO] rule uniqueness done, 0 issues [INFO] rule adj_return_jump done, 1 issues [INFO] report written to ./reports/2024-01.json第一轮跑下来我这边发现股票复权因子缺了一天期货和期权数据里出现不少 NaN 和空值。按前面定的范围期货期权先跳过把股票、指数、ETF 的链路跑完整。4.2 全量历史运行小范围通过后把日期范围放开python -m quality.runner \ --config config/config.toml \ --start 1990-01-01 \ --end 2025-12-31 \ --tables stock_daily,fund_adj \ --output ./reports/full全量跑的时候Codex CLI 界面里只能看到 background terminal不容易判断进度。可以用/btw追问当前状态或者直接在脚本里加进度日志logger.info(progress: %s %d/%d, ts_code, idx, total)4.3 结果解读全量质检跑完后股票日线的异常集中在 1991 到 1993 年的早期行情包括 OHLC 关系异常、涨跌幅不一致和成交额缺失。这些数据日常研究基本不用处理方式是记录异常、明确影响范围不强行修复。ETF 数据里发现一批异常代码部分代码出现七位数字无法匹配复权因子。确认无效后删除记录不再为它们匹配复权数据。ETF 整体没有严重错误主要是少量amount0和部分交易日fund_adj覆盖率不足。复权收益率跳变检查发现了若干标的在复权后仍然跳变抽查确认问题来自复权因子本身。这提示后续可以基于日线里的pre_close自己推导单次调整关系作为独立校验。4.4 模型调用验证报告生成阶段会调用模型做异常归类验证一下通道是否正常curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer ${TAOTOKEN_API_KEY} \ -H Content-Type: application/json \ -d { model: gpt-5.4-mini, messages: [{role: user, content: 把以下异常归类OHLC关系异常、复权因子缺失}] }返回正常说明 Base URL、Key、Model ID 三件套都对上了。如果要在 Claude Code 里跑执行任务参考接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。5. 常见报错排查401、local proxy failed 与 reading choices跑质检脚本和模型调用时下面几类报错出现频率最高逐个说排查方式。5.1 401 UnauthorizedError: 401 Unauthorized - invalid api key原因通常是 Key 没注入或写错。检查顺序先确认环境变量存在echo $TAOTOKEN_API_KEY再确认config.toml里用的是${TAOTOKEN_API_KEY}而不是明文占位符。如果用的是 Claude Code检查.claude/settings.json里的ANTHROPIC_API_KEY是否指向同一个变量。Key 重新生成后记得同步更新旧 Key 失效会直接 401。5.2 local proxy failedError: local proxy failed - connection refused这类报错一般出现在本地网络配置或代理设置干扰了请求。排查方式是先确认base_url写的是https://taotoken.net/api没有多余路径或端口再检查系统环境变量里有没有残留的代理配置影响请求。把HTTP_PROXY、HTTPS_PROXY临时清掉再试unset HTTP_PROXY HTTPS_PROXY python -m quality.runner --config config/config.toml --start 2024-01-01 --end 2024-01-315.3 reading choices 报错KeyError: choices 或 reading choices failed这是响应结构不符合预期常见原因是模型 ID 写错或者请求体格式不对。先确认model字段是控制台里实际可用的 ID再检查请求体里messages是否是数组。如果返回体里是error字段而不是choices把完整响应打出来看resp requests.post(url, headersheaders, jsonpayload, timeout60) data resp.json() if choices not in data: logger.error(unexpected response: %s, data)5.4 OAuth 相关报错Error: OAuth token expired / invalid_grant如果你在 Claude Code 里用 OAuth 方式登录遇到这类报错先检查登录状态是否过期。用 API Key 方式接入时不会走 OAuth 流程配置里确保ANTHROPIC_API_KEY生效即可。Codex 侧如果用auth.json确认里面的字段和当前 Key 一致{ api_key: ${TAOTOKEN_API_KEY}, base_url: https://taotoken.net/api }5.5 规则执行中断sqlite3.OperationalError: no such table: fund_adj表名和config.toml里tables配置不一致。先确认数据库里实际有哪些表sqlite3 ./data/zer0share.db .tables再把tables改成实际存在的表名。如果某张表暂时没有数据把对应规则关掉不要让整批任务失败。6. 把质检接进日常同步流程第一版跑通之后最重要的动作不是继续加规则而是把质检固定成数据同步之后的常规环节。每天增量同步完成后自动运行核心规则出现异常时记录明细和严重级别但不阻塞其他规则继续执行。调度上可以用 cron 或系统定时任务把质检脚本挂到同步任务后面# 每天同步完成后运行增量质检 0 3 * * * cd /path/to/project python -m quality.runner --config config/config.toml --mode incremental ./logs/quality.log 21增量模式只检查最近几个交易日全量模式每周跑一次。报告输出到./reports异常明细按严重级别分文件方便后续查询和追踪。模型调用这块长期跑编码和 Agent 任务的话可以考虑 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。质检脚本里的模型调用集中在报告生成阶段用量不大但如果你同时跑策略研究和数据同步统一通道会省掉不少 Key 管理成本。后续我打算补两块一是基于pre_close自己推导单次调整关系作为复权因子的独立校验二是把准确性检查的跨数据源抽样比对接口填上。这两块都不急着一次做完先把现有的完整性、一致性、唯一性规则跑稳让异常清单持续产出比追求规则数量更有意义。数据质检不是上线前跑一次的验收而是同步之后的固定动作。跑起来之后你面对的就不再是数据到底有没有问题这个模糊问题而是一份可以查询、追踪和处理的异常清单。