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/PanWatch
PanWatch 是一款覆盖 A股、港股与美股的 AI 盯盘工具，它的行情、K线、资金流向等数据都来自一个可插拔的数据源层。本文带你从零走完「新增一个 Vendor 接入行情 API」的完整流程：理解架构、写抓取逻辑、注册权威清单、配置主备优先级、补齐测试，并让它在 PanWatch 的数据源页真正跑起来。
一、为什么 PanWatch 需要可插拔的数据源架构
一个行情应用最难维护的不是功能，而是数据从哪来。腾讯、新浪、东财、雅虎……每家接口的字段、编码、限流策略都不一样，而且随时可能挂掉。
PanWatch 的解法是"一条路径，两层设计"：上层负责故障转移和缓存，下层每个数据源（Vendor）只管一件事——把某家 API 的原始响应解析成标准类型，内部不做任何兜底。这样新增一家源，只需要动一个文件和几行注册代码，不用改核心调度逻辑。
这套架构集中在独立包 packages/marketdata 中，零 Web/数据库依赖，可单独复用，也可嵌入任意项目。
二、PanWatch 行情抓取的整体调用链
先用一句话记住四层结构，后面每一步都对应其中一层：
| 层级 | 文件 | 职责 |
|---|---|---|
| 对象式入口 | client.py | 对外提供quotes()/klines()等 API |
| 主备调度 | engine.py | 按优先级故障转移 + TTL 缓存 + 指标 |
| 数据源抓取 | vendors/ | 一家源"怎么抓 + 怎么解析" |
| 统一 HTTP | http.py | 节流、退避重试、代理、来源标记 |
调用方向是单向的：MarketData → Engine → Vendor → market_get。Engine 拿到哪个 vendor，就调哪个 vendor，vendor 只用market_get发请求。你开发新源时，几乎只碰第 3、4 两层。
三、第一步：实现一个 Vendor 类（只需一个方法）
每个数据类型都有对应的标记基类，定义在 vendors/base.py：QuoteVendor（报价）、KlineVendor（K线）、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）——遵循系统代理设置； - 编码与解析（
encoding="gbk"、parse="json"）——自动解码。
text = market_get( _URL + list_param, host_key=_HOST, # 节流用的主机标识 headers=_HEADERS, # 部分源必带 Referer + UA parse="text", encoding="gbk", # 新浪响应是 GBK 编码 retries=2, timeout=8, 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.py：
def 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(config=DbConfigProvider())，各 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），仅供参考
