摘要：从 Yahoo Finance 或 yfinance 换到另一套股票数据，最容易忽略的，是价格背后的计算口径。本文用苹果股票五个交易日的真实来源值，拆开显示精度、复权、交易日期和缺失记录四类问题，解释它们怎样影响看盘和回测核对。附一张换源检查表、两张图和可运行的离线对照代码，适合自己做研究、维护量化脚本，或与技术同事一起更换行情来源的读者。
原来的日线图已经看习惯了，选股脚本也跑了很久。准备换一套股票数据，最直观的检查通常是：找几天行情，看看收盘价是否相同。
如果连续五天都相同，能不能把新数据接上去，继续用原来的研究和回测？
还需要查清楚：两份价格有没有经过相同的调整、日期代表哪个交易日、记录有没有缺失，以及你的计算究竟用到了哪些字段。先确认比较的是同一种价格，再检查同一天的记录，比一看到差异就修改策略更有用。
下面这组实际整理的样本，正好说明这个问题。
一、五天收盘价相同，开盘价却差了半美分
取苹果股票 AAPL 在 2026年11月2日至6日 的五个交易日，分别查看 Yahoo Finance 官方历史页的展示值和 TickDB 返回的历史日线。
TickDB 是行情数据接口服务，本例使用它的股票历史 K 线作为候选新来源，便于保留接口原值，与已有来源逐日核对。这里关注的是换源时怎样检查数据，不做服务商排名。
Yahoo 一侧来自通过搜索工具读取的官方历史页可读快照；TickDB 一侧来自实际 REST 请求，选择日线和不复权参数。TickDB 于北京时间 2026年11月12日00:00 返回20根日线，以下选取其中五个已结束交易日。Yahoo 查询接口当时连接超时，所以本例没有把网页值称为 yfinance 导出结果，也不是两侧 API 同时取样。
金额单位均为美元。
| 交易日期 | Yahoo 页面开盘价 | TickDB 原值开盘价 | Yahoo 页面收盘价 | TickDB 原值收盘价 |
|---|---|---|---|---|
| 2026-11-02 | 342.67 | 342.67 | 338.32 | 338.32 |
| 2026-11-03 | 325.58 | 325.58 | 331.90 | 331.9 |
| 2026-11-04 | 341.57 | 341.565 | 349.51 | 349.51 |
| 2026-11-05 | 353.89 | 353.89 | 344.17 | 344.17 |
| 2026-11-06 | 338.73 | 338.73 | 342.37 | 342.37 |
五天收盘数值相同。331.90 和 331.9 在数值上也相等，只是书写位数不同。
11月4日的开盘价更值得看：网页显示 341.57，接口原值为 341.565，相差 0.005 美元，也就是半美分。
这时急着判断“哪家的数据错了”，会跳过最先应该解决的问题：你在比较页面显示，还是原始数值？
假设应用决定保留两位小数，并明确使用十进制的四舍五入规则，那么 341.565 可以显示成 341.57：
from decimal import Decimal, ROUND_HALF_UP
raw_price = Decimal("341.565")
display_price = raw_price.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
print(display_price) # 341.57
这是应用可以自行选定的显示方式。Yahoo 页面内部是否采用这一规则，不能从这一个数值反推。保存和计算时仍保留原值，不把两位小数的展示结果写回原始文件。
对只看报价卡片的读者，显示位数可能已经够用；如果你的规则用开盘价判断跳空，或者计算开盘到收盘的变化，就要核对计算实际读的是哪个值。是否影响信号，要看自己的条件和阈值，不能由半美分这个数字直接得出结论。
二、回测接不上，先别急着怀疑策略
显示问题通常比较容易解释。更隐蔽的问题是：两个字段都叫“收盘价”，背后却可能不是同一种价格序列。
1. 看盘价格与研究价格，服务的任务不同
看某一天的报价，你通常想知道当时成交或展示的价格；研究跨越拆股、分红的长期走势，还需要理解历史价格经过了什么调整。
拆股会改变每股价格和股数之间的关系。若直接拿处理方式不同的历史序列拼接，价格变化中就可能混入调整方式的影响。
最简单的收益率计算是：
当期变化率 = 当期价格 ÷ 前期价格 - 1
计算式很简单，输入却有前提：前后两个价格需要遵循适合这项计算的共同定义。未经核对地把一种调整方式的旧历史和另一种方式的新历史拼起来，算出的变化可能同时包含数据处理差异。
如果还要计算含分红的总回报，单看价格变化又不够，需要继续核对分红及再投资等计算约定。
2. 关闭自动调整，也不能直接理解为完全不复权
这是使用 yfinance 时容易踩到的一处细节。
按本篇核对的 yfinance 价格调整文档，auto_adjust=False 返回的 OHLC 仍是 Yahoo 已进行拆股调整的序列，另带包含分红调整的 Adj Close；auto_adjust=True 会对 OHLC 进行相应调整。
因此，旧脚本里的 False，不能只看名字就与另一套接口的“不复权”画等号。本例 TickDB 请求使用 adjust=none，需要按双方的实际定义核对。
图1：AI概念示意，解释不同价格调整设置。图中K线形状不代表实际行情或两源一致性结果。右上角统一标注“TickDB 实时行情 API”。
自己的研究跨越了拆股或分红日期，就把相关区间放进检查样本。这次五日样本没有提供事件前后的两源对照，不能代替这一步。
3. 保留旧结果，才能知道变化发生在哪里
更换来源前，先留下旧 CSV、原脚本和实际参数。记录调整设置、是否修复数据、是否删除缺失行、起止日期、库版本和导出时间。
随后固定一段研究区间与策略代码，分别核对数据、指标和信号。每次只处理一个已查明的问题。若数据与策略一起改，结果变化后很难追溯原因。
本篇只完成日线数据的对照，没有运行策略回测。实际收益、指标或信号是否发生变化，应在自己的研究环境中另做验证。
三、两份数据行数一样，为什么仍会漏日期？
“两边各有20行”只能说明记录数量相同，不能说明它们覆盖同一组交易日。
用一个明确标为合成测试数据的例子来看：
- 旧表有1月5日、6日、8日。
- 新表有1月5日、7日、8日。
- 测试预期清单要求1月5日至9日都有记录。
两边都是3行，但旧表独有6日，新表独有7日，9日两边都没有。只有同时检查日期和预期清单，这些差异才会出现。
图2：AI概念示意，使用虚构的 DEMO_A 软件测试数据。日期与数值用于解释程序逻辑，不是证券行情。
这对研究有几个直接影响：
| 数据问题 | 研究时容易忽略的影响 | 应先做什么 |
|---|---|---|
| 缺少某个日期 | “最近20根”可能对应不同的时间范围 | 列出实际交易日期，再决定是否补取 |
| 同一天出现多条日线 | 滚动计算可能重复使用同一日期 | 查明重复原因，不能随手删一条 |
| 日期标签不一致 | 同一天的数据可能错配到前后一天 | 核对交易所当地交易日与聚合规则 |
| 两侧共同缺失 | 两源互相对得上，却一起少了记录 | 引入经核对的预期清单 |
预期清单也要有依据。交易所日历可以帮助确认市场开市日期，个股是否停牌、数据是否应覆盖该标的，还需要继续核对。不要把“市场开市”直接等同于“每只股票都必须有一根完整日线”。
发现缺失后，先保留缺失状态，检查原始数据和处理代码。把空值补成零，或者不加说明地沿用前值，都会改变后续计算的输入。
四、真实案例的结果：先查口径，再谈是否沿用
将前面的五日真实输入交给配套对照程序，实际返回：
NOT_COMPARABLE: 5
ALIGN_METADATA_BEFORE_VALUES
HUMAN_REVIEW_REQUIRED
翻译成研究者能直接使用的话：
五个日期有对应记录，但价格可比条件还没有确认齐全，暂不进入逐列价格一致性判断。
原因包括：Yahoo 官方历史页的价格定义包含拆股调整，本例 TickDB 选择不复权；两侧交易阶段、日线日期聚合定义尚未共同确认，也没有还原某份旧 yfinance 脚本的真实设置。
这些条件来自来源资料和输入说明。程序按填写的条件检查，不会通过观察价格自行识别供应商的全部计算规则。
因此，五条“暂不可比”不是五条错误行情，也不是新来源不合格。接下来要核实定义、选择相同口径的序列，或完成有依据的转换；不能只把配置里的名称改成一样，让程序放行。
这次可以先确定一件事：页面显示统一小数位、原始数据保留原值；历史研究序列的替换继续等待口径确认。收盘价相同这个观察，保留下来，但不扩大成整套回测已经可以无缝沿用的结论。
五、换源前，把这张表过一遍
对维护个人研究工具或小型量化脚本的读者，先完成一只股票、一段固定区间的核对即可。遇到未确认项，就记录下来，不需要为凑“全部通过”而扩大下载范围。
| 检查项 | 要问清的问题 | 应留下的记录 |
|---|---|---|
| 证券身份 | 两个代码是否对应同一证券、同一市场 | 代码映射、币别与证券身份 |
| 价格含义 | OHLC、调整后收盘价分别如何处理拆股和分红 | 两侧实际设置与定义 |
| 日期覆盖 | 起止边界、交易阶段和交易日标签是否一致 | 日期清单、缺失及重复记录 |
| 数值差异 | 哪一列不同，是否只是显示位数 | 原值、差值与显示策略 |
| 研究依赖 | 自己的指标和信号用了哪些字段 | 固定代码版本的核对结果 |
| 使用范围 | 个人研究、团队共享和对外展示的条件是否不同 | 对应的数据使用条件 |
| 替换方式 | 哪些问题解决后才能换，发现问题如何恢复 | 原文件备份、停止条件与旧路径 |
对于团队，表里再加一列负责人。对于自己做研究的人，这张表就是一份换源笔记：以后解释某段结果为什么变了，有原始文件和设置可查。
六、愿意动手的读者，可以直接运行对照工具
工具使用 Python 3.10 或更新版本，只用标准库。它离线读取 CSV，不需要密钥，不下载行情，也不会自动替换你的研究数据。
将文末完整代码保存为 compare_migration.py，先运行：
python3 compare_migration.py --demo --out compare_runs
这会生成合成测试文件和三份结果：
row_comparison.csv：日期对应、单侧记录、重复与数值差异。comparison_report.json：可比条件和需要继续处理的问题。comparison_view.html：用浏览器打开查看，不必逐行读代码。
合成演示会得到：1条一致、1条仅旧表有、1条仅新表有、1条数值差异、1条共同缺失。它用于确认工具工作方式，与前面的真实五日案例分开。
换成自己的文件
两份输入均整理成单层表头：
symbol,trade_date,open,high,low,close
trade_date 使用核对过含义的 YYYY-MM-DD 交易日期。保留原始导出，另存整理后的副本。两侧各准备一个说明文件，分别命名为 old.meta.json、new.meta.json：
{
"source": "填写实际来源、版本与导出设置",
"interval": "1d",
"currency": "unknown",
"session": "unknown",
"adjustment_basis": "unknown",
"date_basis": "unknown",
"comparability_id": "unknown",
"semantics_reviewed": false,
"symbol_map": {
"AAPL": "AAPL.US"
}
}
这里的代码映射仅示例证券命名转换，使用时应替换为自己的真实映射。未知项照实保留；完成定义核对后，再填写已确认的值。comparability_id 指向双方同一份可比性确认记录，semantics_reviewed 也应在实际核对后才能改为 true。
如果已有经核对的预期记录清单，保存为 expected.csv：
symbol,trade_date
每行填写标准化后的证券代码及一个预期日期，再运行：
python3 compare_migration.py --old old.csv --new new.csv --old-meta old.meta.json --new-meta new.meta.json --expected expected.csv --out compare_runs
没有预期清单时，可以省略 --expected expected.csv；程序会将这项检查记为 NOT_CHECKED，不会替你确认两边共同缺失的日期。
它检查的是日线 OHLC。实时价格、成交量、分红总回报、策略绩效和成交执行，都需要对应的检查方式。即使输出样本内没有差异，也仍要核对自己的指标、信号和实际使用条件。
七、几个常见问题
不写代码，也能开始吗？
可以。先保存旧数据和设置，按第五节的表核对一只股票的一段历史。让技术同事运行工具时，重点看日期差异与尚未确认的定义，不必从192行代码开始读。
数据不同，应该先补数据还是改策略？
先定位是价格口径、日期覆盖、原值还是显示造成的差异。确认应有但缺少的记录后再补取，保留修正前后文件，用相同设置重跑。策略调整应有独立理由，不用改策略掩盖尚未解释的数据问题。
需要补查候选来源的接口定义，去哪里找？
本例使用的 TickDB 历史 K 线接口，其参数和字段可查阅[[TickDB API 文档 GitHub 仓库]]文末对照工具是本文配套代码，不声称已经收录在该仓库。
先完成一只股票的换源核对，再扩大到自己的研究范围。真正值得保留的成果，是一份能解释差异的记录，以及明确知道下一步该查什么。
资料与案例说明：方法资料核对于2026年11月11日，真实案例整理于11月12日。依据包括 yfinance 价格调整、下载与修复文档，Yahoo Finance 的 AAPL 官方历史页可读快照，TickDB 当次 REST 返回与交易日历。真实数据、合成测试和概念图已分别标明。AAPL 仅用于数据方法演示，不构成投资建议。
附录：完整离线对照代码
不需要运行工具的读者，可以将第五节检查表作为本次阅读的行动清单。以下代码供复现与进一步检查使用。
#!/usr/bin/env python3
"""Offline daily-OHLC comparison. Python 3.10+, standard library only.
--demo generates clearly labelled synthetic fixtures, never market observations.
No downloads, orders, automatic adjustment, fill-forward, or production cutover.
"""
import argparse
import csv
import hashlib
import html
import json
from collections import defaultdict
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
FIELDS = ("open", "high", "low", "close")
GATES = ("interval", "currency", "session", "adjustment_basis", "date_basis", "comparability_id")
def csv_write(path, rows, fields):
with path.open("w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fields)
writer.writeheader()
writer.writerows(rows)
def read_rows(path, mapping=None, expected=False):
groups, issues = defaultdict(list), []
with path.open(encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
required = {"symbol", "trade_date"} | (set() if expected else set(FIELDS))
if not required.issubset(reader.fieldnames or []):
raise ValueError(f"{path.name}: requires {sorted(required)}; flatten multi-level exports first")
for line, row in enumerate(reader, 2):
symbol = (row.get("symbol") or "").strip()
day = (row.get("trade_date") or "").strip()
try:
if not symbol or date.fromisoformat(day).isoformat() != day:
raise ValueError()
except ValueError:
issues.append({"line": line, "reason": "invalid_symbol_or_ISO_date"})
continue
symbol = (mapping or {}).get(symbol, symbol)
if not isinstance(symbol, str) or not symbol.strip():
issues.append({"line": line, "reason": "invalid_symbol_mapping"})
continue
groups[(symbol.strip(), day)].append(row)
return groups, issues
def prices(row):
try:
values = [Decimal(row[k]) for k in FIELDS]
if any(not n.is_finite() or n <= 0 for n in values):
return None
o, h, l, c = values
return values if h >= max(o, l, c) and l <= min(o, h, c) else None
except (KeyError, InvalidOperation, TypeError, ValueError):
return None
def compare(old_path, new_path, old_meta, new_meta, expected_path=None):
problems = []
for field in GATES:
a, b = old_meta.get(field), new_meta.get(field)
unknown = lambda v: not isinstance(v, str) or v.strip().lower() in ("", "unknown", "unconfirmed")
if unknown(a) or unknown(b) or a != b:
problems.append(f"{field}: unknown_or_different")
if old_meta.get("interval") != "1d" or new_meta.get("interval") != "1d":
problems.append("interval: this_example_accepts_daily_bars_only")
if old_meta.get("semantics_reviewed") is not True or new_meta.get("semantics_reviewed") is not True:
problems.append("semantics_reviewed: human_confirmation_required")
old, old_issues = read_rows(old_path, old_meta.get("symbol_map"))
new, new_issues = read_rows(new_path, new_meta.get("symbol_map"))
expected, expected_issues = read_rows(expected_path, expected=True) if expected_path else ({}, [])
rows = []
keys = sorted(set(old) | set(new) | set(expected))
for key in keys:
a, b = old.get(key, []), new.get(key, [])
changes = ""
if len(a) > 1 or len(b) > 1 or len(expected.get(key, [])) > 1:
status = "DUPLICATE_KEY"
elif not a and not b:
status = "MISSING_BOTH"
elif not a:
status = "NEW_ONLY"
elif not b:
status = "OLD_ONLY"
elif problems:
status = "NOT_COMPARABLE"
elif prices(a[0]) is None or prices(b[0]) is None:
status = "INVALID_OHLC"
else:
av, bv = prices(a[0]), prices(b[0])
changes = ";".join(f"{f}:{y-x}" for f, x, y in zip(FIELDS, av, bv) if x != y)
status = "VALUE_DIFFERENCE" if changes else "EXACT_MATCH"
rows.append({"symbol": key[0], "trade_date": key[1], "status": status,
"old_rows": len(a), "new_rows": len(b), "new_minus_old": changes,
"old_close": a[0].get("close", "") if len(a) == 1 else "",
"new_close": b[0].get("close", "") if len(b) == 1 else "",
"expected_key": key in expected if expected_path else "NOT_CHECKED"})
counts = {s: sum(r["status"] == s for r in rows) for s in sorted({r["status"] for r in rows})}
input_errors = {"old": old_issues, "new": new_issues, "expected": expected_issues}
if any(input_errors.values()) or counts.get("DUPLICATE_KEY"):
conclusion = "FIX_INPUT_KEYS"
elif problems:
conclusion = "ALIGN_METADATA_BEFORE_VALUES"
elif any(r["status"] != "EXACT_MATCH" for r in rows) or (expected_path and set(old) != set(expected)):
conclusion = "REVIEW_RECORD_DIFFERENCES"
elif not rows:
conclusion = "NO_RECORDS_TO_COMPARE"
else:
conclusion = "NO_DIFFERENCES_IN_THIS_SAMPLE"
return rows, {"comparability_issues": problems, "input_errors": input_errors,
"counts": counts, "conclusion": conclusion,
"expected_dates_check": "PROVIDED" if expected_path else "NOT_CHECKED",
"old_unique_keys": len(old), "new_unique_keys": len(new),
"old_source": old_meta.get("source"), "new_source": new_meta.get("source"),
"data_kind": "SYNTHETIC_NOT_MARKET_DATA" if any(
m.get("data_kind") == "SYNTHETIC_NOT_MARKET_DATA" for m in (old_meta, new_meta)) else "USER_DECLARED_INPUTS",
"deployment_signoff": "HUMAN_REVIEW_REQUIRED",
"note": "Exact decimal equality is a comparison result, not an accuracy or migration approval."}
def demo_files(folder):
"""All values are invented software-test data for DEMO_A, not a security."""
base = {"symbol": "DEMO_A", "open": "10", "high": "12", "low": "9", "close": "11"}
old = [dict(base, trade_date=f"2026-01-{d:02}") for d in (5, 6, 8)]
new = [dict(base, trade_date=f"2026-01-{d:02}") for d in (5, 7, 8)]
new[-1]["close"] = "11.05"
meta = dict(source="SYNTHETIC_DEMO", interval="1d", currency="USD",
session="regular", adjustment_basis="split_adjusted",
date_basis="exchange_local_trade_date", semantics_reviewed=True,
comparability_id="SYNTHETIC_TEST_BASIS_ONLY",
symbol_map={}, data_kind="SYNTHETIC_NOT_MARKET_DATA")
for name, data in (("old", old), ("new", new)):
csv_write(folder / f"{name}.csv", data, ["symbol", "trade_date", *FIELDS])
(folder / f"{name}.json").write_text(json.dumps(meta, indent=2))
expected = [{"symbol": "DEMO_A", "trade_date": f"2026-01-{d:02}"} for d in range(5, 10)]
csv_write(folder / "expected.csv", expected, ["symbol", "trade_date"])
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--demo", action="store_true")
for name in ("old", "new", "old-meta", "new-meta", "expected"):
parser.add_argument("--" + name, type=Path)
parser.add_argument("--out", type=Path, default=Path("comparison_runs"))
args = parser.parse_args()
folder = args.out / datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
folder.mkdir(parents=True, exist_ok=False)
if args.demo:
demo_files(folder)
args.old, args.new = folder / "old.csv", folder / "new.csv"
args.old_meta, args.new_meta = folder / "old.json", folder / "new.json"
args.expected = folder / "expected.csv"
if not all((args.old, args.new, args.old_meta, args.new_meta)):
parser.error("Use --demo or supply --old, --new, --old-meta and --new-meta")
old_meta = json.loads(args.old_meta.read_text(encoding="utf-8"))
new_meta = json.loads(args.new_meta.read_text(encoding="utf-8"))
rows, summary = compare(args.old, args.new, old_meta, new_meta, args.expected)
summary["mode"] = "SYNTHETIC_DEMO" if args.demo else "USER_SUPPLIED_FILES"
summary["created_at_utc"] = datetime.now(timezone.utc).isoformat()
summary["inputs"] = {n: str(getattr(args, n)) for n in ("old", "new", "old_meta", "new_meta", "expected")}
summary["declared_metadata"] = {"old": old_meta, "new": new_meta}
summary["input_sha256"] = {n: hashlib.sha256(getattr(args, n).read_bytes()).hexdigest()
for n in ("old", "new", "old_meta", "new_meta", "expected") if getattr(args, n)}
csv_write(folder / "row_comparison.csv", rows,
["symbol", "trade_date", "status", "old_rows", "new_rows", "old_close", "new_close", "new_minus_old", "expected_key"])
(folder / "comparison_report.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
columns = ("symbol", "trade_date", "old_close", "new_close", "status", "new_minus_old")
header = "".join(f"<th>{html.escape(c)}</th>" for c in columns)
body = "".join("<tr>" + "".join(f"<td>{html.escape(str(r[c]))}</td>" for c in columns) + "</tr>" for r in rows)
view = ("<!doctype html><html lang='en'><meta charset='utf-8'><meta name='viewport' content='width=device-width'>"
"<title>Migration comparison</title><style>body{max-width:1000px;margin:40px auto;padding:0 20px;"
"font:17px system-ui;background:#f8f4eb;color:#123349}table{border-collapse:collapse;width:100%}"
"th,td{padding:12px;border-bottom:1px solid #ccd3d6;text-align:left}pre{white-space:pre-wrap}"
".label{padding:14px;background:#fae2b9;border-radius:8px}</style>"
"<h1>迁移对齐结果 / Migration comparison</h1><p class='label'>"
+ html.escape(summary["data_kind"]) + " — demonstration or user-supplied input; not a production approval.</p>"
+ "<p>" + html.escape(summary["conclusion"]) + "</p><table><thead><tr>" + header
+ "</tr></thead><tbody>" + body + "</tbody></table><h2>Settings requiring review</h2><pre>"
+ html.escape(json.dumps(summary["comparability_issues"], ensure_ascii=False, indent=2))
+ "</pre><p>HUMAN_REVIEW_REQUIRED. Missing values stay missing. Differences require investigation.</p></html>")
(folder / "comparison_view.html").write_text(view, encoding="utf-8")
print(json.dumps({"output": str(folder), "mode": summary["mode"], "counts": summary["counts"],
"conclusion": summary["conclusion"], "deployment_signoff": summary["deployment_signoff"]}))
if __name__ == "__main__":
main()