亲测最好用的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 1.开源项目 XTick行情API提供了全面、准确、稳定的行情数据，帮助开发者和研究者构建创新的交易和分析工具，满足金融行业的需求，进行深入的市场分析和模型验证。 项目网址：http://www.xtick.top/ GitHub地址 https://github.com/xticktop/xtick API接口文档 API接口分为订阅数据、行情数据、财务数据三个部分。行情数据支持盘中实时更新。 除了订阅接口是Websocket API，其余接口为Http API接口且均支持GET和POST方法，下面以GET请求示例。 订阅数据接口 在GitHub上，已实现Java版本和Python版本的订阅代码，请先下载代码直接调用。 暂时无法在飞书文档外展示此内容 订阅数据按照证券交易所订阅推送，包括上交所、深交所、北交所、港交所（只支持部分股票）。 数据为实时推送，发数据非常快，客户端接受到数据后，最好做异步处理，将接受数据和数据处理分开，避免接受数据阻塞。 订阅方法： 订阅数据：订阅为Websocket API，请在Github上下载开源项目，参考XTickWebSocketClient.java中已实现的订阅功能。 入参1：authCodes 枚举取值如下： tick.SZ - 订阅深交所A股的tick数据。 tick.SH - 订阅上交所A股的tick数据。 tick.BJ - 订阅北交所A股的tick数据。 tick.HK - 订阅港交所港股的tick数据。 time.SZ - 订阅深交所A股的k线数据，包括time、1m。 time.SH - 订阅上交所A股的k线数据，包括time、1m。 time.BJ - 订阅北交所A股的k线数据，包括time、1m。 time.HK - 订阅港交所港股的k线数据，包括time、1m。 入参2：token 登录XTick网站，注册获取 取消订阅：http://api.xtick.top/doc/unsubscribe?token=043fbdcba7f3f3ab332ffff123456789 入参：token 登录XTick网站，注册获取 行情数据接口 请求方法： 请求地址：http://api.xtick.top/doc/market?type=1&code=000001&period=tick&fq=none&startDate=2025-03-25&endDate=2025-03-25&token=043fbdcba7f3f3ab332ffff123456789 备注：行情数据支持交易日内盘内实时更新。 入参1：type 股票类别 沪深京A股type=1，港股type=3; **入参2**：**code** 股票代码 比如平安银行为000001 **入参3**：**period** 用于表示要获取的周期，枚举取值如下： tick - 分笔数据 1m - 1分钟线 5m - 5分钟线 15m - 15分钟线 30m - 30分钟线 1h - 1小时线 1d - 日线 1w - 周线 1mon - 月线 1q - 季度线 1hy - 半年线 1y - 年线 参数4：fq 除权方式，用于K线数据复权计算，对tick等其他周期数据无效，枚举取值如下： none 不复权 front 前复权 back 后复权 front_ratio 等比前复权 back_ratio 等比后复权 参数5：时间范围，用于指定数据请求范围，表示的范围是[<b>startDate</b> ,<span> </span><b>endDate</b>]区间（包含前后边界）。 特别说明：period为tick类型，则单次请求时间跨度最大为一天，即startDate和endDate日期需设置为同一天。 period为分钟类型（包括1m、5m、15m、30m、1h），则单次请求时间跨度最大为一月，即**endDate - startDate不超过30天。** startDate - 起始时间，日期格式：2025-03-25 endDate- 结束时间，日期格式：2025-03-25 入参6：token 登录XTick网站，注册获取 财务数据接口 请求方法： 请求地址：http://api.xtick.top/doc/financial?type=1&code=000001&report=Pershareindex&startDate=2020-03-25&endDate=2025-03-25&token=043fbdcba7f3f3ab332ffff123456789 入参1：type 股票类别 沪深京A股type=1，港股type=3; **入参2**：**code** 股票代码 比如平安银行为000001 **入参3**：**report** 用于表示要获取的财务报表，枚举取值如下： Balance - 资产负债表 Income - 利润表 CashFlow - 现金流量表 Capital - 股本表 Holdernum - 股东数 Top10holder - 十大股东 Top10flowholder - 十大流通股东 Pershareindex - 每股指标 **参数4：**时间范围，用于指定数据请求范围，表示的范围是[<b>startDate</b> ,<span> </span><b>endDate</b>]区间（包含前后边界）。 startDate - 起始时间，日期格式：2025-03-25 endDate- 结束时间，日期格式：2025-03-25 入参5：token 登录XTick网站，注册获取 
