tick-stock-panel/docs/plugin-development.md

数据源插件开发指南

数据源插件是可选的行情数据来源(stock-sdk、akshare 等),作为独立模块放在 backend/app/plugins/ 下。用户手动安装依赖后才可用(开发模式);不安装完全不影响主功能。

💡 Docker 部署已预装:内置插件(如 stock-sdk)的 Node 运行时与 node_modules 已在镜像构建期装好,Docker 下无需手动 npm install,开箱即用。下方"手动安装依赖"仅适用于开发模式。

快速上手

一个插件 = 一个目录 + 一个 plugin.yaml 清单:

backend/app/plugins/<your_plugin>/
├── plugin.yaml          # 清单(必需)
├── provider.py          # Provider 实现(必需)
├── ...                  # 桥接/依赖文件(按需)

plugin.yaml 字段

name: my_source                          # 唯一标识, 只允许 [a-z0-9_], 也是 provider name
display_name: "我的数据源"                 # 设置页显示名
runtime: python                          # 运行时类型: node | python | none
entry: app.plugins.my_source.provider:MyProvider   # provider 类的导入路径
check: app.plugins.my_source.bridge:availability   # 可用性检测函数(可选)
datasets: [daily, adj_factor, minute, realtime]     # 支持的数据集
description: "数据源描述"
install_hint: "pip install xxx"          # 未装依赖时显示的安装提示

runtime 字段说明

runtime	含义	典型场景
python	纯 Python 依赖, pip install	akshare、tushare
node	需要 Node.js 运行时, npm install	stock-sdk(已内置,见下)

stock-sdk 在 Docker 镜像里已预装 Node 运行时与依赖;开发模式下才需手动 npm install。 | none | 无额外依赖 | 纯 HTTP API 源 |

runtime 字段当前仅用于 UI 展示, 实际依赖检测由 check 函数负责。

check 函数

插件自己负责检测依赖是否已安装。后端启动时会调用此函数:

# app/plugins/my_source/bridge.py
def availability() -> tuple[bool, str]:
    """返回 (是否可用, 原因)。不抛异常。"""
    try:
        import akshare  # noqa: F401
        return True, "ok"
    except ImportError:
        return False, "未安装 akshare, 运行: pip install akshare"

• 可用 → 插件注册进路由表, 设置页可切换
• 不可用 → 设置页显示插件卡片但灰显, 展示 install_hint

Provider 接口契约

Provider 是一个普通 Python 类(无需继承基类), 实现以下方法签名。方法签名对齐 GenericHTTPProvider, 这样 services 层(kline_sync / quote_service 等)的路由逻辑 零改动即可路由到插件。

class MyProvider:
    name = "my_source"
    builtin = True  # 标记为内置(不可被用户编辑/删除)

    def __init__(self):
        self.config = MyConfig()  # 需有 .datasets 属性(dict, key 是数据集名)

    def close(self) -> None:
        """清理资源(load_all 重建注册表时会调)。"""

    def get_daily(self, symbols, start_time, end_time, asset_type="stock", on_chunk_done=None) -> pl.DataFrame:
        """日K: 返回 schema [symbol, date, open, high, low, close, volume, amount]"""

    def get_adj_factors(self, symbols, start_time, end_time, asset_type="stock", on_chunk_done=None) -> pl.DataFrame:
        """除权因子: 返回 schema [symbol, trade_date, ex_factor]"""

    def get_minute(self, symbols, start_time, end_time, asset_type="stock", on_chunk_done=None, freq="1m") -> pl.DataFrame:
        """分钟K: 返回 schema [symbol, datetime, open, high, low, close, volume, amount]"""

    def get_realtime(self) -> list[dict]:
        """全市场实时快照: 返回 list[dict], 每行含 symbol/last_price/prev_close/open/high/low/volume"""

    def get_instruments(self, asset_type="stock") -> list[dict]:
        """标的维表(可选): 返回 tickflow Instrument 形状的行, 供 instrument_sync 复用 flatten"""

config.datasets 的作用

provider_has_dataset(name, dataset) 通过 dataset in provider.config.datasets 判断。 这是 services 层路由的关键: 用户在设置页选了插件, 但某数据集未声明时, 该数据集 自动回退 TickFlow。

class MyConfig:
    datasets = {"daily": ..., "realtime": ...}  # key 是数据集名, value 任意

现有插件参考

• backend/app/plugins/stocksdk/ — Node 型插件, 通过 subprocess 桥接调用 stock-sdk
  • bridge.py — Python↔Node 桥接 + availability 检测
  • bridge.mjs — Node 端(并发池、重试、SDK 解析)
  • provider.py — Provider 实现(归一化、分批、错误降级)

路由机制(无需关心, 仅参考)

后端启动时, loader.py 的 _load_builtin_plugins() 扫描 plugins/ 目录:

1. 读每个子目录的 plugin.yaml
2. 调 check 函数检测可用性
3. 可用 → 动态 import entry 指向的 Provider 类 → 注册进 _PROVIDERS
4. 不可用 → 记录状态, 设置页显示但不可切换

注册后, 插件和用户 YAML 自定义源走完全相同的路由路径(services 层的 provider_has_dataset / get_provider 调用), 无需额外集成代码。