PanWatch 数据源开发指南实现一个 Vendor 接入新行情 API 全流程【免费下载链接】PanWatchPanWatch — AI stock monitoring for A-shares, HK US markets, powered by TradingAgents. Portfolio insights, real-time alerts automated reports.盯盘侠覆盖 A股/港股/美股的 AI 盯盘、持仓分析、实时提醒与自动报告。项目地址: https://gitcode.com/GitHub_Trending/pa/PanWatchPanWatch 是一款覆盖 A股、港股与美股的 AI 盯盘工具它的行情、K线、资金流向等数据都来自一个可插拔的数据源层。本文带你从零走完「新增一个 Vendor 接入行情 API」的完整流程理解架构、写抓取逻辑、注册权威清单、配置主备优先级、补齐测试并让它在 PanWatch 的数据源页真正跑起来。一、为什么 PanWatch 需要可插拔的数据源架构一个行情应用最难维护的不是功能而是数据从哪来。腾讯、新浪、东财、雅虎……每家接口的字段、编码、限流策略都不一样而且随时可能挂掉。PanWatch 的解法是一条路径两层设计上层负责故障转移和缓存下层每个数据源Vendor只管一件事——把某家 API 的原始响应解析成标准类型内部不做任何兜底。这样新增一家源只需要动一个文件和几行注册代码不用改核心调度逻辑。这套架构集中在独立包 packages/marketdata 中零 Web/数据库依赖可单独复用也可嵌入任意项目。二、PanWatch 行情抓取的整体调用链先用一句话记住四层结构后面每一步都对应其中一层层级文件职责对象式入口client.py对外提供quotes()/klines()等 API主备调度engine.py按优先级故障转移 TTL 缓存 指标数据源抓取vendors/一家源怎么抓 怎么解析统一 HTTPhttp.py节流、退避重试、代理、来源标记调用方向是单向的MarketData → Engine → Vendor → market_get。Engine 拿到哪个 vendor就调哪个 vendorvendor 只用market_get发请求。你开发新源时几乎只碰第 3、4 两层。三、第一步实现一个 Vendor 类只需一个方法每个数据类型都有对应的标记基类定义在 vendors/base.pyQuoteVendor报价、KlineVendorK线、CapitalFlowVendor资金流向、EventsVendor公告等共 11 类。以接入报价为例你只需继承QuoteVendor声明三样东西name注册名要和后续配置里的vendor字段完全一致supports_markets支持的市场集合{US, HK}fetch(symbols, config)唯一的抽象方法返回标准类型列表。class SinaQuoteVendor(QuoteVendor): name sina supports_markets {US, HK} def fetch(self, symbols: list[Symbol], config: dict) - list[Quote]: # 拼接 URL → market_get 抓取 → 解析成 Quote 列表完整可运行的参考实现是 vendors/sina.py新浪美股/港股备源。注意fetch的约定失败要抛异常Engine 会捕获并转移空结果返回[]不要自己写 fallback。四、第二步用 market_get 发请求节流重试都内置了不要自己import requests。PanWatch 提供了统一工具 http.py 里的market_get它一次帮你解决了四件脏活按 host 节流min_interval_s——避免打爆同一接口退避重试retries——带抖动的指数退避代理支持trust_env/proxy——遵循系统代理设置编码与解析encodinggbk、parsejson——自动解码。text market_get( _URL list_param, host_key_HOST, # 节流用的主机标识 headers_HEADERS, # 部分源必带 Referer UA parsetext, encodinggbk, # 新浪响应是 GBK 编码 retries2, timeout8, log_label新浪报价, )抓回来的text再解析成标准 dataclass字段定义见 types.py如Quote、Bar、CapitalFlow。市场代码统一用Symbol值对象 symbol.py 归一化别再到处手写前缀逻辑。五、第三步把新 Vendor 注册到权威清单这是最容易被漏掉的一步。registry.py 里的VENDOR_CLASSES_BY_TYPE是type → vendor 的唯一真相源VENDOR_CLASSES_BY_TYPE { quote: { tencent: TencentQuoteVendor, sina: SinaQuoteVendor, # 新增一行即可 }, }往对应数据类型下加一行你的name: 你的Vendor类就完成了。为什么强调它是唯一真相源——因为PACKAGE_VENDORS_BY_TYPE合法 vendor 名集合和MarketData里 Engine 的vendors{}都由它自动派生不会出现改了引擎忘了改权威表的漂移。六、第四步通过 ConfigProvider 配置主备优先级注册只是能用优先级决定谁先被调。Engine 会按priority从小到大尝试 vendor第一个成功且非空的就返回并缓存全失败才返回空——这就是 PanWatch 的主备故障转移。配置项SourceConfig长这样节选自 README字段含义vendor你的注册名priority越小越优先enabled是否启用config透传凭证/参数token、cookie 等在独立使用时可用内置StaticConfigProvider而在 PanWatch 宿主里优先级来自数据库的DataSource表由 marketdata_client.py 中的DbConfigProvider读取并转成SourceConfig。也就是说给新源在数据源表里插一行、排好 priority它就自动进主备链了无需改代码。七、第五步写单元测试monkeypatch 即可不发真实网络PanWatch 的全部测试都mock HTTP不发真实请求。套路是monkeypatch 掉 vendor 模块里的market_get喂一段构造好的原始响应断言解析结果。参考 test_sina_quote_vendor.pydef test_sina_us_quote(monkeypatch): line var hq_str_gb_aapl ... ; monkeypatch.setattr(sv, market_get, lambda *a, **k: line) out sv.SinaQuoteVendor().fetch([Symbol.parse(AAPL, US)], {}) assert out[0].current_price 150.5在 packages/marketdata 下跑python -m pytest -q全绿即可无需联网。八、让新数据源跑起来PanWatch 宿主接线与验证代码合并后宿主是单一路径进程级单例get_market_data()持有一个MarketData(configDbConfigProvider())各 collector / Agent 直接调用它取数没有灰度 flag、没有回退分支。上线建议按这个顺序自查跑单测pytest -q保证解析正确核对字段在「数据源」页对新增的(type, provider)点「测试」按钮看真实返回是否符合预期看健康度health()会把每个 vendor 的成功率、p50 延迟、最近错误喂到健康度面板一眼看出新源是否稳定调优先级若新源更稳把它的priority调小让它晋升为主源。九、常见坑与最佳实践市场要对齐supports_markets留空 全市场否则 Engine 会按市场过滤掉不支持的请求可选依赖惰性 import像yfinance这种三方库务必放在fetch()内部import不要在模块顶层引入避免拖慢启动失败要抛、空要返回fetch内不要自己吞异常做 fallback交给 Engine 统一转移编码别猜GBK/UTF-8 用encoding显式声明别靠resp.text自动探测。小结接入一个新行情 API 在 PanWatch 里就是五步走继承对应*Vendor实现fetch→ 用market_get发请求 → 在registry.py权威清单加一行 → 配置priority进主备链 → 补一个 mock 单测。得益于抓取与调度分离的架构你只写最薄的一层故障转移、缓存、限流、指标全由 Engine 与market_get兜底新源即可无缝服务 A股、港股与美股的盯盘、提醒与自动报告。【免费下载链接】PanWatchPanWatch — AI stock monitoring for A-shares, HK US markets, powered by TradingAgents. Portfolio insights, real-time alerts automated reports.盯盘侠覆盖 A股/港股/美股的 AI 盯盘、持仓分析、实时提醒与自动报告。项目地址: https://gitcode.com/GitHub_Trending/pa/PanWatch创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考