摘要：黄仁勋离职后，跟踪美股AI产业链的投资者，该继续研究英伟达，还是把目光转向AMD、博通和台积电？本文结合TickDB MCP取得的行情与业务分部数据，以及公司原始披露，沿着客户采购、产品替代和订单变化逐项核查，整理一张可以交给AI继续追踪的候选研究表，帮助读者判断下一步该查谁、查什么。 如果你一直在跟踪英伟达，看到黄仁勋离职的消息，接下来最关心的可能是：原来的投资判断还成立吗？如果客户开始考虑其他方案，又该先研究哪家公司？ 2026年11月12日，英伟达公告确认黄仁勋辞任，现任首席财务官Colette Kress出任临时CEO。当天美股常规交易收盘，英伟达下跌4.34%，AMD上涨5.86%，博通上涨5.54%。 看到这个分化，很容易产生一个念头：是不是该把研究重点从英伟达转向AMD和博通？ 用AI选股工具继续追问之前，值得先把问题拆开。AMD有加速计算产品，博通涉及定制AI芯片和网络，台积电则处在制造环节。它们与英伟达的关系不同，后续需要核查的事情也不同。 研究名单可以先调整优先级。至于某家公司是否因此增加收入，还要沿着客户的采购决定继续往下查。 先看同一天的行情 再决定把时间花在哪里 先把四家公司放到同一个完整交易日里。下面的数据取自TickDB，使用2026年11月12日美股常规交易收盘，对照11月11日收盘计算。 标的 11月11日收盘 11月12日收盘 当日涨跌幅 英伟达 NVDA 173.73 166.19 −4.34% AMD 552.90 585.30 ＋5.86% 博通 AVGO 345.22 364.36 ＋5.54% 台积电 TSM 413.35 405.49 −1.90% 半导体ETF SOXX 155.40 154.99 −0.26% 价格单位为美元；TSM为台积电美股ADR。涨跌幅计算方法为“当日收盘÷前一交易日收盘−1”。数据于11月13日采集，并核对了日线、收盘行情和前收价格。 这张表首先告诉我们，同属AI产业链，股票当天的表现并不一致。连制造环节的台积电也下跌了，简单用“英伟达承压，其他公司都受益”来理解，解释不了这些差别。 AMD和博通的上涨值得继续查。SOXX则提供一个同日的行业参照，帮助我们看到个股与板块之间的差异。至于各家公司为什么涨跌，还需要结合当时的其他消息和经营信息。 对投资者来说，这一步得到的是研究线索：哪些公司值得多花时间，哪些原来的判断需要重新检查。 AMD已经上涨 接下来还该查什么 AMD上涨5.86%，足以让跟踪这条产业链的人多看一眼。要判断这次消息会怎样影响它的生意，还得回到客户。 客户为什么原来选择英伟达？如果准备增加另一套方案，需要迁移哪些软件、重新验证哪些工作负载？AMD能不能按需要的时间交付？这些问题有了进展，才轮到新增订单，以及订单何时进入收入。 先看已经披露的数字。 AMD在11月3日公布的2026年第三季度业绩中，数据中心收入为85.46亿美元，约占总收入的63.89%，包括EPYC处理器和Instinct GPU；该季度数据中心收入同比增长97%。 这个比例把服务器CPU等业务也算在里面。研究AI加速器机会时，还要继续拆产品和客户，不能拿整个数据中心部门的收入来估算GPU替代空间。 再把时间放回去：这份业绩早于11月12日的离职公告。 它说明AMD原有的数据中心业务在增长；要研究这次事件带来的变化，需要继续寻找公告之后的采购调整、新增部署和订单信息。本轮核对的材料里，还没有找到客户因这次离职而转单给AMD的直接披露。 所以，可以先把AMD放到重点研究栏，旁边写清下一项任务：跟踪哪些客户扩大部署、何时交付、是否上调采购或收入预期。日后出现新公告，再判断它补上了哪一条证据。 已有增长和新增变化要分开记。否则，很容易把一家公司本来就在做成的生意，全部算到刚发生的新闻头上。 博通和台积电 不能放进同一个受益逻辑 看到博通也上涨，下一步要先弄清它卖的是什么。 博通在9月2日公布的2026财年第三季度业绩中，披露AI半导体收入约167亿美元，涉及定制AI加速器和网络业务。这里值得继续研究的是：客户在增加哪类计算方案，配套网络需求如何变化，以及哪些项目可能形成后续收入。 这与“某个客户减少英伟达GPU采购，把同样数量的订单交给博通”之间，还隔着产品路线和客户项目的核查。定制芯片和网络业务也要分别看，不能把它们合成一项笼统的替代机会。 收入分类同样容易造成误读。该季度博通半导体解决方案收入约208.39亿美元，占公司总收入约70.42%；官方披露的AI半导体收入约占总收入56.44%，这个比例根据已取整的167亿美元估算。 如果把70.42%直接写成“AI收入占比”，就把其他半导体业务也算进去了。选择不同的分子，读者对这家公司与AI需求的关联程度，会得到不同印象。 再看台积电，研究问题又变了。 本轮接口中的台积电业务分部列为100%晶圆制造。这个100%描述的是报告中的业务分类，不能读成全部收入都来自AI，更不能读成全部来自英伟达。 英伟达年报列出了台积电等制造合作方；AMD季度报告披露了对台积电先进工艺晶圆的依赖；博通季度报告也披露了台积电在其外包晶圆生产中的重要地位。 这意味着，客户即使增加另一家芯片设计公司的产品，制造环节仍可能落在同一家代工厂。台积电后续的收入影响，要继续看相关客户的总需求、产品组合以及产能和交付情况。 换了研究对象仍要检查共同依赖。把英伟达换成AMD或博通，并没有自动回答制造环节的问题；这也是为什么一张只有公司名字的AI概念股名单，往往不够用。 回到英伟达 原来的判断究竟要改哪一项 研究其他公司时，英伟达本身也值得留在表里。 如果你原来关注的是客户持续采购，那么接下来查的是客户是否改变采购计划；如果关注的是新产品交付，就查产品路线、交付节奏和客户部署有没有变化。管理层更替带来的不确定性，最终要落到这些具体事项上。 英伟达11月12日公告中，董事会表示预期公司运营、客户承诺和产品路线继续推进，并保留11月17日的财报安排。这是管理层对后续工作的表态，后续披露和执行情况仍需跟踪。 这也让下一次研究有了明确问题：新的财报、客户信息或产品进展，有没有改变自己原先使用的收入和利润假设？ 如果只有股价变了，就先更新价格记录；如果客户需求或交付预期也变了，再调整经营判断，随后重新比较估值。把这几步分开，日后回看时，才知道自己为什么修改了看法。 让AI交一张能够继续查的表 到这里，可以把四家公司整理成下面这张候选研究表。它记录的是研究方向，各行不代表买卖顺序。 公司 已有研究依据 下一步重点查什么 什么信息值得更新判断 英伟达 事件主体，已有客户与产品业务 客户采购、产品交付、后续业绩披露 原有需求或交付假设出现实质变化 AMD 有加速计算产品，数据中心业务已增长 新增部署、软件适配、供货与订单 客户采购规模或收入预期有新披露 博通 有定制AI加速器与网络业务 分开核查客户项目、定制芯片和网络需求 具体项目或订单变化，而非笼统行业关联 台积电 多家公司的共同制造环节 总体需求、产品组合、产能与交付 相关客户的总量或产品结构发生变化 实际使用时，在每一行旁边再加上来源、披露日期和最近检查日期。新材料进来，先更新对应事项，不必每次把整张名单推倒重来。 TickDB是面向开发者、量化研究和AI应用的市场数据服务。本轮通过MCP取得了四家公司的行情、K线、业务分部和估值快照，把价格变化与业务结构放进同一份研究记录；公司原始披露则用来核对产品、客户关系和实际财政期间。对于准备把这套方法接入自己的研究助手或应用的团队，可以查看TickDB的AI工具与MCP接入说明。 估值快照在这里先作为后续研究输入。利润预期和财务期间还没对齐时，不急着按一个市盈率数字给四家公司排便宜贵贱。 下面这段任务可以直接交给能够访问相应数据与原始资料的AI助手： 请围绕2026年11月12日英伟达管理层变更事件，建立NVDA、AMD、AVGO、TSM四家公司的候选研究表。 先按实际产品、客户或制造关系说明每家公司为什么值得研究。再使用同一完整美股常规交易日的行情，记录日期、收盘和前收，并加入SOXX作为行业参照。 请把事件前已披露的业务进展，与事件后新增的采购、部署和订单信息分开。每项注明原始来源、披露日期与实际财政期间；数据中心收入、半导体收入和AI收入分别记录。 对尚未找到新增订单依据的候选，保留下一项核查任务。最后告诉我：每家公司还缺哪条重要信息，出现什么新披露时值得更新判断。无法取得的数据请标为待补，不要按一天的涨跌生成买入排名。 一条新闻过去之后，这张表仍然可以继续使用。下一次客户宣布新部署、公司调整指引或产品交付发生变化，把新材料补进去，就能看清：最初只是市场价格有了反应，后来有没有真正走到业务和收入。 资料与数据说明 NVIDIA：《NVIDIA Announces CEO Transition》，2026年11月12日；相关Form 8-K。 AMD：《Third Quarter 2026 Financial Results》，2026年11月3日，财政季度截至2026年9月26日；相关Form 10-Q。 Broadcom：《Third Quarter Fiscal Year 2026 Financial Results》，2026年9月2日，财政季度截至2026年8月2日；相关Form 10-Q。 NVIDIA：截至2026年1月25日财年的Form 10-K，制造与竞争相关章节。 TickDB MCP：2026年11月13日采集的证券身份、交易日历与时段、行情、日线、业务分部及估值快照。文中价格比较使用11月12日完整美股常规交易日，未将当日波动归为单一事件的影响。 亲测最好用的AI编写量化策略工具，可以让 AI 直接生成同花顺Supermind上可运行的策略代码，代码质量远高于直接使用 DeepSeek、GPT等平台。 😃Supermind AI工具入口：https://easyquant.ai/ 大家可以直接向AI描述策略，然后一键生成可运行的完整策略代码，也可以把它当做一个API 查询工具。 迷茫市场中的破局之道 在当前复杂多变的投资环境中，不少散户投资者正面临着前所未有的挑战。面对震荡反复的市场行情，许多人陷入了不知如何下手的困境，迫切渴望能够有一套成熟、清晰的全球宏观逻辑作为操作的指引。 “最近很多朋友留言问说，现在的市场太难了，不知道怎么操作。能不能参考一下老谈的逻辑和方向？” 面对大家的困惑，破局的关键就在于学会解码全球宏观联动逻辑。只有看清全球资金流动、美联储货币政策、通胀走势以及地缘局势对不同资产的影响，才能在迷茫的市场中找准方向，建立起属于自己的操作节奏与定力。 核心干货：必须掌握的全球宏观联动逻辑口诀 为了方便大家记忆与实操，我们将复杂的全球宏观经济联动规律精炼为一套通俗易懂的口诀。这套口诀精准揭示了美联储加息/降息、美元走势、通胀预期、地缘局势及企业盈利对各类资产（黄金、资源、科技、银行、能源等）的深远影响： 宏观/外部触发因素与口诀 对市场/资产板块的具体影响与表现 美加息股市弱 美联储加息周期下，全球资金收紧、折现率抬升，股市普遍承压走弱 美降息股市强 美联储开启降息，市场流动性持续释放，驱动估值修复与股市走强 美元强黄金弱 美元走势强劲时，资金回流美元计价资产，对无息资产黄金形成压制 地缘乱资源涨 地缘局势动荡混乱，引发避险需求与供给中断担忧，推高大宗资源价格 通胀降科技涨 通货膨胀回落降温，加息预期缓和，利好高弹性科技板块估值抬升 通胀涨银行涨 通货膨胀上升阶段，利率预期与息差走高，带动银行板块盈利与表现 盈利增美股强 企业盈利实现持续增长，强劲的基本面支撑美股市场保持强势 美元贬科技强 美元贬值走弱时，全球资金风险偏好提升，科技板块表现尤为强劲 战事停能源调 地缘冲突平息或战事停止，溢价逐渐消除，能源板块面临回落调整 核心资产联动规律总结 **●**美联储政策与股市：加息压制市场估值与流动性，降息则为股市注入强劲的上涨动力。 **●美元与黄金/**科技：美元走强直接压制黄金表现；美元贬值则大幅提升资金风险偏好，对科技板块构成直接利好。 **●**通胀与行业轮动：通胀下降推动估值修复、驱动科技股上涨；通胀上升则提升利差预期，利好银行股表现。 **●**地缘局势与资源能源：地缘动荡推高资源类资产价格，而战事趋缓或停止则意味着地缘溢价挤出，能源板块随之回调。 **●**盈利基本面：企业盈利的持续增长，是支撑美股等权益市场保持长牛强势的底层核心动力。 九月展望与心智建设：从“否极泰来”到账户逆风翻盘 面对投资道路上的风浪与挫折，除了掌握硬核的宏观逻辑，建设强大的投资心智同样不可或缺。 “祝愿我们九月份账户都能逆风翻盘，匹太来（否极泰来）！” 无论当前市场多么艰难，只要坚定信心、做足功课，就一定能熬过黑暗。请大家把**“小红心”点好，并在评论区留下“匹太来”这四个字——这既是对“否极泰来”的美好期许，也是对自身投资定力的激励。把基本功做扎实，九月份我们完全有机会实现账户的逆风翻盘**！ 实战落地：如何通过日常心法与复盘提升投资能力 将宏观逻辑转化为实际的投资收益，需要日复一日的知行合一。哪怕你的投资基础较为薄弱，也可以通过以下三个精细化步骤一步步提升实战能力： ●每日必刷“谈门心法”：每天坚持查看置顶第一条视频，系统吸收老谈最新研究的投资方向与宏观逻辑。“谈门心法”通俗易懂，哪怕只有小学文化也能轻松看懂，帮你快速找准每日的操作方向。 ●****每周必看直播复盘（付盘）：每个周末准时进入直播间参与深度复盘，跟随老谈系统梳理盘面细节与结构演变，将宏观逻辑与实际走势进行对照验证。 **●**打卡验证实力与逻辑：点亮小红心，在评论区打出“匹太来”，持续关注老谈几天。通过实际盘面的走势来验证老谈的研究逻辑与硬核实力，切实感受自身投资分析能力的飞跃。 投资路漫漫，并肩作战同行 投资从来不是一件一蹴而就的事情，它需要宏观的大局观、扎实的实战心法以及不屈不挠的心态。看清全球宏观局势，一步一个脚印地做好每日与每周的复盘，逆风翻盘终将成为现实。 平时在 SuperMind 上开发量化策略时，不知道大家有没有遇到过这样的情况： 明明已经有了比较清晰的选股思路或交易逻辑，但真正写成可运行的策略代码，还是需要花不少时间。尤其是涉及平台 API、数据获取、调仓逻辑和仓位管理时，经常需要反复修改和调试。 虽然现在 DeepSeek、ChatGPT 等通用 AI 工具也能辅助编程，但生成的代码有时会出现平台接口不匹配、函数调用错误等问题，实际运行前还需要不少调整。 最近在推广一款专门面向量化策略开发的 AI 工具——AI 量化代码助手，分享给有类似需求的朋友。 目前支持 SuperMind、QMT、PTrade、聚宽等国内主流量化平台，主要功能包括： AI 生成策略代码：直接用自然语言描述选股条件、交易信号和调仓规则，辅助生成适用于 SuperMind 的策略代码。 修改已有策略：可以在原有代码基础上调整买卖条件、选股逻辑、仓位管理和止损机制，不需要每次都从头编写。 辅助排查代码问题：帮助分析报错信息、检查代码逻辑，减少重复调试的时间。 多平台策略开发：可以针对不同量化平台的 API 和代码规范生成对应代码，方便使用多个平台的开发者。 例如，直接输入： “帮我编写一个适用于 SuperMind 的多因子选股策略，根据市盈率、ROE 和动量指标筛选股票，每周调仓一次，并设置单只股票最大持仓比例。” AI 就可以根据这些要求辅助生成对应的策略代码，后续也能继续通过自然语言修改交易逻辑。 相比通用 AI 编程工具，这类专门面向国内量化平台的助手，在平台 API 适配和策略代码开发方面可能更有针对性。 当然，AI 生成的策略仍然需要通过实际运行和历史回测验证，尤其要注意未来函数、交易成本以及策略过拟合等问题。工具主要是帮助提高开发效率，并不保证策略能够盈利。 工具地址：https://qmtai.cn/?le 也想借这个帖子和社区里的朋友交流一下： 大家平时使用 AI 编写 SuperMind 策略时，最常遇到的是 API 兼容问题，还是策略逻辑修改和代码调试方面的问题？ 欢迎交流使用经验，也希望能收集一些实际需求和建议，进一步完善工具对 SuperMind 的支持。 量价关系与进出场时机 在股票交易的实战中，研判进出场时机是决定最终盈亏的关键所在，而“量是价的先行指标”，成交量与价格的动态互动正是资金动向最真实的体现。许多投资者常常陷入“买在回调高位、卖在起涨前夜”的困境，其根源就在于未能读懂量价语言。 本文基于视频“牢记四种量价口诀，理解进出场时机”的核心精华，结合资深实战经验，为您系统拆解四大黄金量价口诀。通过剖析量价形态背后的筹码博弈逻辑，并辅以位置判别、次日确认与止损策略，帮助您建立严密的交易体系，精准把握每一次买卖买点。 进场信号：两大“十字星”形态解析 “十字星”形态代表多空双方在特定位置的力量暂收平衡。然而，脱离成交量与研判位置的十字星毫无意义。结合成交量的放大与缩减，才能将其转化为明确的进场指导。 1.缩量见十字：上涨马上来 口诀解析：当股价经过一轮调整或在上升途中的洗盘阶段，收出成交量明显萎缩的十字星（“缩量见十字”），表明市场的抛盘力量已经消耗殆尽，筹码锁定良好，多空双方在极低交投下达成微妙平衡。这往往是上涨马上来的前兆，预示着企稳反转在即。 实战要点与进场策略： ●**位置上下文：该形态最有效的发生位置是在上升趋势中的回调支撑位**（如20日均线、前期密集成交区）或长期下跌后的底部止跌区域。若发生在长期暴涨后的高位，则不适用。 **●**买入确认：建议不宜在十字星当日盲目左侧建仓，可等待次日收出一根放量阳线突破十字星高点时进行右侧确认进场。 **●**风控设置：将止损位严格设定在该缩量十字星的最低价下方。 2.放量见十字：进场正当时 口诀解析：股价在关键位置收出十字星的同时，成交量大幅放大（“放量见十字”），这代表多空双方发生了剧烈的筹码交换与换手。主力资金在此位置强力接盘、吸筹或洗盘，多方力量蓄势待发，是明确的买入信号，意味着进场正当时。 实战要点与进场策略： ●**位置上下文：重点关注重要颈线位、前高突破口或双底右脚**位置的放量十字星，这通常是主力吃清阻力筹码的信号。需要警惕的是，若股价已连续大涨数倍，高位放量巨量十字星则可能为主力出货，需严格区分。 **●**买入确认：若出现在关键底部或突破位，可以在收盘前尾盘布局，或等待次日开盘走出向上阳线确认后跟进。 **●**风控设置：将止损位设定在该放量十字星的下影线低点，一旦跌破说明主力接盘失败，需及时止损。 回调研判：两种“回调量能”趋势分析 股价在回调过程中的量能变化，是区分“主力出货（危险形态）”与“洗盘蓄势（机会形态）”的试金石。 1. 回调见梯量：往后还有底 回调见梯量，往后还有底 口诀解析：“梯量”是指成交量在股价回调过程中呈现如阶梯状逐日放大的形态（即下探时量能一日高于一日）。这种形态说明随着股价下跌，市场恐慌盘与套牢盘持续涌出，主力资金不仅没有护盘，甚至可能在带头抛售。探底过程远未结束，预示着往后还有底。 实战要点与风控策略： **●**量能判定：连续3天以上的下跌阴线伴随着成交量阶梯式放大，表明抛压具有持续性。 **●**操作禁忌：此时切忌盲目“抄底”或“加仓摊薄成本”。面对“梯量下跌”，唯一正确的策略是保持观望，甚至离场规避风险，直至量能出现明显的萎缩企稳信号。 2. 回调出地量：上升量 回调出地量，上升量 口诀解析：股价在回调过程中，成交量持续萎缩，最终出现极度低迷的“地量”（“回调出地量”）。“地量”代表市场卖盘彻底枯竭，洗盘达到极致。此时一旦伴随量能温和放大（“上升量”），代表主力资金重新发力拉升，后市看涨信号强烈。 实战要点与进场策略： **●**地量定义：实战中，“地量”通常指成交量萎缩至5日或10日均量线的1/2以下，甚至创出阶段性新低量。 ●**买入确认：地量出现后，需密切跟踪盘面。当随后出现一根放量阳线（即****“上升量”****突破降雨趋势）**时，为最佳右侧买点。 **●**风控设置：止损位可设在地量形成时的阶段性股价最低点。 四种量价口诀速查表 为了方便投资者在日常盯盘与交易决策中快速查阅与复盘，现将四大核心量价口诀及实战要点整理为速查表： 量价口诀形态 市场含义 / 操作提示 实战关键点 缩量见十字 上涨马上来 关键支撑位缩量止跌，等待次日阳线确认进场，止损设十字星低点 放量见十字 进场正当时 底部或突破位资金剧烈换手，主力吸筹接盘，后市看涨 回调见梯量 往后还有底 下跌中成交量呈阶梯状放大，抛压持续，切勿盲目抄底，保持观望 回调出地量 上升量（后市看涨） 卖盘耗尽后出极低量，随后伴随放量阳线（上升量）启动，右侧重仓跟进 熟记量价口诀是技术分析的基本功，但股市交易绝非机械照搬。在实际操作中，投资者应始终将量价口诀与股价所处的趋势位置、关键支撑阻力位以及严密的止损纪律相结合，灵活研判，方能在股市中实现稳健获利。 摘要：从 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() 摘要：从 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() 摘要：用AI复盘自选股，难点常常在于下一步查谁、查什么。面向有量化或技术基础的A股投资者及投研团队，本文通过TickDB MCP调用股票行情API，以2026年11月11日寒武纪、海光信息和中科曙光的真实数据，比较当日涨跌、近20个交易日表现与成交变化，整理一张带后续研究问题的观察表，并提供可复制提示词与重算代码。 11月11日收盘，寒武纪涨7.40%，海光信息涨3.34%，中科曙光涨0.84%。如果它们都在你的自选股里，今晚只够认真研究一家公司，你会先点开谁？ 先看谁，要看你今晚想弄清什么。这次使用的[TickDB]是一项面向量化研究和AI应用的市场数据服务。我们通过它的MCP入口，让AI取得三只股票的行情快照和历史K线，再把当天涨幅放回近期走势里比较。 结果很有意思：当天涨得最多的寒武纪，近20个交易日累计下跌3.12%；当天只涨0.84%的中科曙光，累计上涨14.82%。 研究当天的明显反弹，与复盘近一个月的累计表现，需要追问的问题不同。把研究目标告诉AI，观察表里的比较和问题，才能围绕这项任务展开。 一、先把当天涨幅放回近期背景 我们先选定寒武纪、海光信息和中科曙光，再查询数据。这是三只科技股组成的自选股观察样本。 把当天与近期表现放在一起，差异就清楚了： 图1：同一组股票，换一个比较窗口，看到的相对表现顺序随之改变。 数据说明：样本交易日为2026年11月11日，快照时间为北京时间15:00:00至15:00:02。近期涨跌按前复权收盘价计算，从10月14日收盘至11月11日收盘，共20个交易间隔，使用21个收盘价。 只看当天，寒武纪排第一；看这20个交易日，中科曙光排第一。 两种排序都有用途。当天涨幅帮助你定位这一天的变化，近期涨跌帮助你了解一段时间的累计表现。研究顺序要结合你准备验证的问题来安排。 二、同一只股票，为什么今天大涨，近20天却仍在下跌？ 先把寒武纪的几个价格摆出来。按这次取回的前复权日线，10月14日收盘价是929元，11月10日是838元，11月11日回到900元。 从838元涨到900元，当天涨幅约7.40%；拿900元与929元相比，仍低约3.12%。当天的涨幅很醒目，过去这段时间的回落也确实存在。两组数字各有自己的比较起点。 中科曙光则呈现另一种情况：10月14日至11月11日，前复权收盘价从69.03元升至79.26元，累计上涨14.82%；单看11月11日，它从前一天的78.60元涨到79.26元，涨幅只有0.84%。在这组三只股票中，它当天涨幅最小，20日累计涨幅却最高。 这里值得带走的，是给AI布置任务时要补充的一个条件：你准备比较哪段时间。 “帮我看看今天谁涨得多”，会把注意力带向当天变化；“比较这几只股票近20个交易日的表现”，会让你看到另一组差异。 研究窗口有了，下一步才容易具体：查哪段时间的公告，核对哪些经营变化，下一个交易日继续观察什么。 三、再加一层成交背景 价格之后，再看当天的交易活跃度。 这次用一个容易复算的比较：把当日成交量，除以这只股票此前20个交易日的平均成交量。平均值中不包含当日。 计算结果是： 寒武纪约1.09倍，比自身此前均量略高。 海光信息约0.80倍，低于自身此前均量。 中科曙光约0.60倍，也低于自身此前均量。 三只股票当天都上涨，但成交相对各自历史水平的变化不同。 对寒武纪，后续可以继续观察价格变化与成交能否配合；对中科曙光，可以把近期累计上涨与当天相对较低的成交放在一起，记录下一次复盘要关注的变化。 这里比较的是每只股票相对自己的历史。若直接把不同股票的绝对成交量摆在一起，股本规模等差异也会混进来。这种“先和自己比”的办法，更适合本篇要完成的自选股复盘。 表中的成交比值按上述20日均量计算，与行情软件中的盘中量比口径不同。 四、让观察表最后一列写下研究问题 到这一步，AI已经有了当天表现、近期背景和成交变化。接下来要把它们接到具体工作上。 如果你今晚想研究当天明显反弹的股票，可以先追查寒武纪这次上涨前后有哪些新披露，再记录之后的价格与成交表现。 如果你正在复盘近一个月的相对表现，中科曙光值得单独展开：这段时间披露了哪些经营信息，哪些公司事项需要继续核对？ 例如，中科曙光2026年三季报在解释业绩变化时，提到了高端计算机等IT设备销量增加、配套软件销售增加。这些信息可以帮助你继续查产品销售和业务结构。它们是具体的研究材料，不能仅凭发布时间就认定为某一天上涨的原因。 寒武纪、海光信息也各有自己的经营披露。三份三季报中的前三季度营业收入均同比增长，而这组三只股票的20日价格表现方向并不相同。研究需要继续落到各家公司，核对增长来自哪里、后续披露有没有改变原先的判断。 最终，观察表可以整理成这样： 股票 已观察到的情况 下一步查什么 寒武纪 当天涨幅最大，20日累计表现仍为负；成交略高于此前均量 上涨前后有哪些新披露？后续价格与成交怎样变化？ 海光信息 当天上涨，20日累计表现为负；成交低于此前均量 近期经营披露中，有哪些变化值得重新核对？ 中科曙光 20日累计涨幅居样本第一；当天成交低于此前均量 这段时间有哪些经营信息与公司事项？下一次复盘重点观察什么？ 这张表把已经看到的情况与后续工作连在一起。你可以按自己的研究目标调整先后，而不必停留在一段泛泛的行情点评里。 五、把这段任务交给AI，再换成自己的名单 在已连接上述MCP入口的AI客户端中，可以复制下面的任务，把股票名单换成自己持续跟踪的对象。 请帮我做一份A股自选股收盘复盘表。 观察名单： 688256.SH 寒武纪 688041.SH 海光信息 603019.SH 中科曙光 研究目标： 比较当天变化与近期累计表现，安排下一步查资料的顺序。 请按下面的方法处理： 1. 核对股票名称与代码，查询当前行情快照并注明数据时间。 2. 以最近一个完整交易日为复盘截止日，取得每只股票至少21根 完整交易日的前复权日线，并核对比较日期是否一致。 3. 用完整日线计算当日涨跌，以及最近20个交易间隔的涨跌； 用截止日成交量除以此前20个交易日的平均成交量，均值排除当日。 4. 输出股票、截止日、当日涨跌、20日涨跌、成交比值， 再为每只股票写一条具体的后续研究问题。 5. 写清比较起点与计算方法。数据不足或日期不齐时标明缺口。 盘中快照与完整日线分开使用，不把尚未收盘的数据当作全天结果。 6. 把已观察到的事实和仍要查的原因分开；若引用公告， 注明公告名称与日期，不补写没有来源的消息或目标价。 本次实际使用的核心工具是get_ticker和get_kline，日线参数为1d、forward。三只股票各取得25根日线，覆盖的日期与本次交易日历逐日一致；快照价格、昨收和成交量也与对应日线核对。 行情数据由工具取得，比较和表格整理由AI及计算程序完成。把研究目标、名单、时间范围和输出要求一次说清楚，后续追问就能沿着同一项任务继续。 六、用这段代码重算文中的数字 下面是从本次日线提取的最小计算输入，包含窗口起点收盘价、昨收、收盘价、当日成交量和此前20日平均成交量。 代码使用Python标准库，复制后即可运行，重算本篇示例；它不发起新的行情请求。换股票或换日期时，先取得同口径数据，再替换输入。 from decimal import Decimal as D # 名称、代码、20个交易间隔前收盘、昨收、收盘、当日量、此前20日均量 # 价格为本次取得的前复权日线口径；成交量沿用接口单位。 samples = [ ('寒武纪', '688256.SH', '929', '838', '900', '116347', '107017.75'), ('海光信息', '688041.SH', '201.46', '190', '196.35', '164721', '206017'), ('中科曙光', '603019.SH', '69.03', '78.60', '79.26', '169962', '284811.75'), ] print('股票,当日涨跌,20日涨跌,成交量/此前20日均量') for name, symbol, start, previous, close, volume, mean_volume in samples: start, previous, close, volume, mean_volume = map( D, (start, previous, close, volume, mean_volume) ) if min(start, previous, mean_volume) <= 0: raise ValueError(f'{symbol}: 计算基准必须大于0') day_change = (close / previous - 1) * 100 period_change = (close / start - 1) * 100 volume_ratio = volume / mean_volume print(f'{name},{day_change:+.2f}%,{period_change:+.2f}%,{volume_ratio:.2f}倍') 运行结果： 股票,当日涨跌,20日涨跌,成交量/此前20日均量 寒武纪,+7.40%,-3.12%,1.09倍 海光信息,+3.34%,-2.54%,0.80倍 中科曙光,+0.84%,+14.82%,0.60倍 其中，20个交易间隔的涨跌，需要用截止日与20个间隔前的收盘价比较，也就是至少21个收盘价。计算此前20日均量时，要把截止日排除。把这两个基准固定下来，换一份名单也能沿用同样的方法。 七、下次复盘，从一条明确的问题开始 下一次打开自选股，不妨先写下今晚要解决的问题：研究当天的明显变化，还是回顾近期累计表现？ 再让AI取回相应数据，把比较结果放在一起，并在观察表最后一列写下接下来要查的材料。 寒武纪当天涨7.40%，中科曙光近20个交易日涨14.82%，都值得被准确记录。它们会把研究引向不同的问题。问题明确了，你才知道有限的时间应该先花在哪里。 参考资料与数据来源： 中科寒武纪科技股份有限公司《2026年第三季度报告》，巨潮资讯网，存档日期2026年10月24日。 海光信息技术股份有限公司《2026年第三季度报告》，巨潮资讯网，存档日期2026年10月20日。 曙光信息产业股份有限公司《2026年第三季度报告》，巨潮资讯网，存档日期2026年10月28日。 行情快照、K线查询官方文档，核对日期2026年11月11日。 本文MCP实际返回与Python计算记录，样本交易日2026年11月11日。 文中股票用于展示数据研究方法，不构成投资建议。 **摘要：做行情看板、量化回测或盯盘工具时，最麻烦的往往不是业务逻辑，而是行情数据源。本文基于 DeepTick 的 REST 与 WebSocket 接口，完整演示港股、美股、A 股实时报价、历史 K 线与逐笔成交的接入方式，附 Python 可运行代码与字段说明。 一、背景 金融类应用开发中，行情数据通常需要覆盖三类能力： 快照报价：当前最新价、开高低、成交量额，用于看板、列表页； 历史 K 线：用于画图、技术指标计算与回测； 逐笔成交（Tick）：用于高频盯盘、成交明细展示。 传统做法是 HTTP 轮询，简单但延迟高、请求量大。更合理的组合是：REST 拉快照和历史数据 + WebSocket 推实时数据。本文以 DeepTick 为例，把这条链路完整走一遍。 二、接入准备 2.1 认证方式 DeepTick 的 REST 接口通过请求头 deeptoken 认证，WebSocket 在建立连接时通过 additional_headers 传入同一个 token。 python headers = { "deeptoken": "your_api_key" } 2.2 标的代码（symbol）规则 DeepTick 统一使用 代码.市场 格式，跨市场规则如下： 市场 后缀 示例 说明 港股 .HK 700.HK、9988.HK 5 位数字代码 美股 .US AAPL.US、TSLA.US 1–4 位字母 A 股沪市 .SH 601398.SH 60 开头 A 股深市 .SZ 300750.SZ、000001.SZ 00 / 30 开头 加密货币 .CR BTCUSDT.CR 加密对 注意：REST 实时报价一次只能查 1 个标的；K 线接口单次最多 20 个标的；WebSocket 可在同一订阅中混合多市场。 三、REST 接口实战 3.1 实时报价：GET /api/v1/quote 拉取单个标的的当前最新价与当日统计，不含买卖盘口，逐笔成交请走 WebSocket tick。 请求参数 参数 类型 必填 说明 symbols string 是 一个标的，如AAPL.US；传多个会返回错误码8005 Python 示例 python import requests url = "https://api.deeptick.cc/api/v1/quote" headers = {"deeptoken": "your_api_key"} params = {"symbols": "AAPL.US"} try: resp = requests.get(url, headers=headers, params=params) result = resp.json() if result["code"] == 0: q = result["data"][0] print(f"标的：{q['s']}") print(f"最新价：{q['ld']}") print(f"开盘：{q['o']} 最高：{q['h']} 最低：{q['l']}") print(f"成交量：{q['v']} 成交额：{q['tu']}") else: print("请求失败：", result["msg"]) except Exception as e: print("请求异常：", str(e)) 响应示例 json { "code": 0, "msg": "ok", "data": [ { "s": "AAPL.US", "ld": "225.215", "o": "226.27", "h": "226.92", "l": "224.44", "v": "16742235", "tu": "3774688301.452", "t": 1731689407000 } ] } 字段说明（data[] 元素） 字段 类型 说明 s string 标的代码 ld string/number 最新价 p string/number 前收 o/h/l string/number 开 / 高 / 低 v/tu string/number 成交量 / 成交额 ch/chp string/number 涨跌额 / 涨跌幅（上游无则不出现） ts int 交易状态：0正常、1停牌、2退市、3熔断，不是时间戳 t long 行情时间戳（毫秒） 这里有个容易踩的坑：ts 是交易状态而不是时间，真正的时间是 t。 3.2 历史 K 线：GET /api/v1/candles 用于补历史 K 线、绘图或回测，始终查上游、不走缓存，返回按时间升序排列。正在走的当前 K 线请订阅 WebSocket candle，不要轮询本接口。 请求参数 参数 类型 必填 说明 symbols string 是 逗号分隔，单次最多 20 个，超限返回8006 interval enum 否，默认5m 1m/5m/15m/30m/60m/4h/1d/1w/1M；1h视为60m limit int 否，默认 40 1~500，超限返回8007 et long 否 截止时间（Unix 毫秒），缺省取最新 Python 示例 python import requests import datetime url = "https://api.deeptick.cc/api/v1/candles" headers = {"deeptoken": "your_api_key"} params = { "symbols": "300750.SZ", # 宁德时代 "interval": "1d", "limit": 10 } try: resp = requests.get(url, headers=headers, params=params) result = resp.json() if result["code"] == 0: for item in result["data"]: print(f"标的：{item['s']} 周期：{item['i']}") for bar in item["d"]: dt = datetime.datetime.fromtimestamp(bar["t"] / 1000) print( f"{dt:%Y-%m-%d} 开:{bar['o']} 高:{bar['h']} " f"低:{bar['l']} 收:{bar['c']} 量:{bar['v']} 额:{bar['tu']}" ) else: print("请求失败：", result["msg"]) except Exception as e: print("请求异常：", str(e)) 响应示例 json { "code": 0, "msg": "ok", "data": [ { "s": "300750.SZ", "i": "1d", "d": [ { "t": 1723100000000, "o": "535.0", "h": "536.0", "l": "534.5", "c": "534.5", "v": "104799385", "tu": "56119888070.5" } ] } ] } 字段说明 字段 类型 说明 s string 标的代码 i enum 周期 d array K 线数组，时间升序 d[].t long 该周期时间戳（毫秒） d[].o/h/l/c string/number 开 / 高 / 低 / 收 d[].v/d[].tu string/number 成交量 / 成交额 四、WebSocket 实时推送 连接地址： text wss://api.deeptick.cc/ws/v1 一次 subscribe 必须是同一种 type（quote 或 tick），不能混写。topic 格式： 报价：quote.{market}.{symbol}，如 quote.US.AAPL、quote.HK.700 逐笔：tick.{market}.{symbol}，如 tick.US.AAPL Python 示例（需 pip install websockets） python import asyncio import json import websockets TOKEN = "your_api_key" async def main(): async with websockets.connect( "wss://api.deeptick.cc/ws/v1", additional\_headers={"deeptoken": TOKEN} ) as ws: print(await ws.recv()) # welcome await ws.send(json.dumps({ "type": "subscribe", "topics": [ "quote.US.AAPL", "tick.US.AAPL", "quote.HK.700", "tick.HK.700" ], "request\_id": "r1" })) while True: frame = json.loads(await ws.recv()) ftype = frame.get("type") data = frame.get("data", {}) if ftype == "quote": print(f"[QUOTE] {data.get('s')} 最新价:{data.get('ld')} " f"开:{data.get('o')} 高:{data.get('h')} 低:{data.get('l')} " f"量:{data.get('v')} 额:{data.get('tu')}") elif ftype == "tick": print(f"[TICK] {data.get('s')} 价格:{data.get('ld')} " f"数量:{data.get('v')} 时间:{data.get('t')}") if __name__ == "__main__": asyncio.run(main()) quote 数据帧 json { "type": "quote", "topic": "quote.US.AAPL", "ts": 1723100000000, "data": { "s": "AAPL.US", "ld": "225.215", "o": "226.27", "h": "226.92", "l": "224.44", "v": "16742235", "tu": "3774688301.452", "t": 1731689407000 } } tick 数据帧 json { "type": "tick", "topic": "tick.US.AAPL", "ts": 1723100000000, "data": { "s": "AAPL.US", "ld": "192.41", "v": "100", "t": 1723100000000 } } 外层字段 字段 类型 说明 type enum 帧类型：tick/quote/depth/candle/trade topic string 订阅 topic 原文 ts long 行情时间戳（毫秒） quote data 字段：s、ld、p、o/h/l、v/tu、ch/chp、ts（交易状态）、t（时间戳），含义同 3.1。 tick data 字段：s 标的、ld 成交价、v 成交数量、t 成交时间（毫秒）。 五、接入踩坑与注意事项 ts 不是时间：报价里的 ts 是交易状态（0 正常 / 1 停牌 / 2 退市 / 3 熔断），时间字段是 t。 数值类型不固定：字段可能是 string 也可能是 number，落库或计算前统一 float() 转换，展示保留 2 位小数。 订阅不能混 type：一次 subscribe 只能全是 quote 或全是 tick，混写会失败。 REST 单标的限制：/api/v1/quote 只接受 1 个 symbol，批量场景请用 WebSocket 订阅。 K 线上限：单次最多 20 个标的、limit 最大 500，超限分别返回 8006、8007。 频率控制：免费版建议每秒不超过 5 次请求，高频需求升级套餐。 错误处理：code != 0 时读 msg；网络超时做重试，无数据时用缓存降级。 市场差异：A 股人民币、港股港币、美股美元，交易时间也不同，展示层需针对性适配。 六、总结 整体接入思路可以归纳为一条链路： 快照类需求（看板、列表）→ REST /api/v1/quote 历史类需求（绘图、回测）→ REST /api/v1/candles 实时类需求（盯盘、成交明细）→ WebSocket quote.* / tick.* REST 负责「拉」，WebSocket 负责「推」，两者组合即可覆盖绝大多数行情场景。代码层面重点处理好三件事：字段类型统一、时间戳转换、错误码与重试。 免责声明：本文代码用于技术接入演示，不构成投资建议。行情数据的使用、展示、存储和再分发应遵守 DeepTick 服务条款及相关市场的数据授权要求。 GitHub：https://github.com/deeptick-cc/deeptick-realtime-market-data-api 
