☰
我用 Codex 给本地量化数据库搭了一套数据质检系统:TaoToken 统一 Key 接入与 config.toml 配置骨架
2026/10/3 6:16:39 网站建设 项目流程

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api 。你只需要在配置里写一个 Base URL 和一个 Key,模型 ID 按需切换。

2.1 先拿 Key,再写配置

进入控制台创建 API Key,地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。创建之后复制出来,后面写进config.toml。如果你还没决定用哪个模型,可以先在模型对话页面试一下,地址是 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。

Key 的管理页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ,建议给质检系统单独建一个 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, threshold=0.5): """计算后复权收盘价,检查连续收益率是否异常跳变""" merged = daily_df.merge(adj_df, on=["ts_code", "trade_date"], how="left") 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_error=True): 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 整体没有严重错误,主要是少量amount=0和部分交易日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_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。

5. 常见报错排查:401、local proxy failed 与 reading choices

跑质检脚本和模型调用时,下面几类报错出现频率最高,逐个说排查方式。

5.1 401 Unauthorized

Error: 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 failed

Error: 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-31

5.3 reading choices 报错

KeyError: 'choices' 或 reading 'choices' failed

这是响应结构不符合预期,常见原因是模型 ID 写错,或者请求体格式不对。先确认model字段是控制台里实际可用的 ID,再检查请求体里messages是否是数组。如果返回体里是error字段而不是choices,把完整响应打出来看:

resp = requests.post(url, headers=headers, json=payload, timeout=60) 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 2>&1

增量模式只检查最近几个交易日,全量模式每周跑一次。报告输出到./reports,异常明细按严重级别分文件,方便后续查询和追踪。

模型调用这块,长期跑编码和 Agent 任务的话,可以考虑 Coding Plan,地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。质检脚本里的模型调用集中在报告生成阶段,用量不大,但如果你同时跑策略研究和数据同步,统一通道会省掉不少 Key 管理成本。

后续我打算补两块:一是基于pre_close自己推导单次调整关系,作为复权因子的独立校验;二是把准确性检查的跨数据源抽样比对接口填上。这两块都不急着一次做完,先把现有的完整性、一致性、唯一性规则跑稳,让异常清单持续产出,比追求规则数量更有意义。

数据质检不是上线前跑一次的验收,而是同步之后的固定动作。跑起来之后,你面对的就不再是"数据到底有没有问题"这个模糊问题,而是一份可以查询、追踪和处理的异常清单。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询