diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py
index ce256bc..c10009e 100644
--- a/backend/app/api/settings.py
+++ b/backend/app/api/settings.py
@@ -401,6 +401,11 @@ class DataProvidersIn(BaseModel):
     financial_data_provider: str | None = None
 
 
+class PluginKeyIn(BaseModel):
+    plugin: str
+    api_key: str
+
+
 class DataSourceJobTimeoutPrefs(BaseModel):
     data_source_job_timeout_s: int = Field(ge=60)
     data_source_long_job_timeout_s: int = Field(ge=60)
@@ -536,6 +541,53 @@ def list_data_sources() -> dict:
     }
 
 
+@router.post("/plugin-key")
+def save_plugin_key(req: PluginKeyIn) -> dict:
+    """保存插件 API Key(先探后存, 对齐 /tickflow-key 语义)。
+
+    流程: probe_plugin_key 用候选 Key 实探 → 有效才写 secrets.json
+    ({plugin}_api_key, 优先级高于 .env) → load_all 重扫, 插件即刻变为可切换。
+    """
+    from app.data_providers import custom as custom_sources
+
+    name = req.plugin.strip().lower()
+    key = req.api_key.strip()
+    if not key:
+        return {"ok": False, "error": "key empty"}
+    ok, message = custom_sources.probe_plugin_key(name, key)
+    if not ok:
+        return {"ok": False, "reason": "invalid", "error": message}
+    secrets_store.save({f"{name}_api_key": key})
+    custom_sources.load_all()
+    status = next((p for p in custom_sources.list_plugins() if p["name"] == name), None)
+    return {
+        "ok": True,
+        "api_key_masked": secrets_store.mask(key),
+        "plugin_available": bool(status and status.get("available")),
+        "plugin": status,
+    }
+
+
+@router.delete("/plugin-key/{name}")
+def clear_plugin_key(name: str) -> dict:
+    """清除插件的界面配置 Key(secrets.json);.env 里的同名变量仍然生效。"""
+    from app.data_providers import custom as custom_sources
+
+    manifest = custom_sources.plugin_manifest(name)
+    if manifest is None or not custom_sources.is_builtin(name):
+        raise HTTPException(status_code=404, detail=f"插件 '{name}' 不存在")
+    if not manifest.get("api_key_env"):
+        raise HTTPException(status_code=400, detail=f"插件 '{name}' 不支持在界面配置 Key")
+    secrets_store.clear(f"{name.lower()}_api_key")
+    custom_sources.load_all()
+    status = next((p for p in custom_sources.list_plugins() if p["name"] == name), None)
+    return {
+        "ok": True,
+        "plugin_available": bool(status and status.get("available")),
+        "plugin": status,
+    }
+
+
 @router.post("/data-sources/reload")
 def reload_data_sources() -> dict:
     """重新加载 data_sources/*.yaml。"""
diff --git a/backend/app/data_providers/custom/__init__.py b/backend/app/data_providers/custom/__init__.py
index 28405ac..dc19531 100644
--- a/backend/app/data_providers/custom/__init__.py
+++ b/backend/app/data_providers/custom/__init__.py
@@ -13,6 +13,8 @@ from app.data_providers.custom.loader import (
     list_sources,
     load_all,
     names,
+    plugin_manifest,
+    probe_plugin_key,
     provider_has_dataset,
     save_config,
     uninstall_plugin,
@@ -32,6 +34,8 @@ __all__ = [
     "list_sources",
     "load_all",
     "names",
+    "plugin_manifest",
+    "probe_plugin_key",
     "provider_has_dataset",
     "save_config",
     "uninstall_plugin",
diff --git a/backend/app/data_providers/custom/loader.py b/backend/app/data_providers/custom/loader.py
index e610cc4..4f3aa1d 100644
--- a/backend/app/data_providers/custom/loader.py
+++ b/backend/app/data_providers/custom/loader.py
@@ -103,6 +103,33 @@ def plugin_dir_of(name: str) -> Path:
     return plugins_dir() / (name or "")
 
 
+def probe_plugin_key(name: str, api_key: str) -> tuple[bool, str]:
+    """调用插件的 probe_api_key(key) 探测候选 Key(不落盘)。
+
+    约定: 声明了 api_key_env 的插件, 其 entry 模块提供模块级
+    probe_api_key(key) -> (ok, reason)。未声明或未提供 → (False, 原因)。
+    """
+    manifest = plugin_manifest(name)
+    if manifest is None:
+        return False, f"插件 '{name}' 不存在"
+    if not manifest.get("api_key_env"):
+        return False, f"插件 '{name}' 不支持在界面配置 Key"
+    entry = str(manifest.get("entry") or "")
+    if ":" not in entry:
+        return False, f"插件 '{name}' entry 非法"
+    try:
+        module = importlib.import_module(entry.split(":", 1)[0])
+    except Exception as e:
+        return False, f"插件模块加载失败: {e}"
+    probe = getattr(module, "probe_api_key", None)
+    if probe is None:
+        return False, f"插件 '{name}' 未提供 Key 探测"
+    try:
+        return probe(api_key)
+    except Exception as e:
+        return False, f"探测失败: {e}"
+
+
 def install_plugin(name: str) -> tuple[bool, str]:
     """安装指定插件的依赖。根据 runtime 执行 npm install / pip install。
 
@@ -509,6 +536,10 @@ def _register_one_plugin(manifest: dict) -> None:
     if not name or not _NAME_RE.match(name):
         logger.warning("插件清单缺少合法 name: %r", name)
         return
+    # hidden: 已加载但对 UI 隐藏 (功能未完成/暂不开放), 不注册不展示
+    if manifest.get("hidden"):
+        logger.info("插件 %s 标记为 hidden, 跳过注册", name)
+        return
     runtime = str(manifest.get("runtime", "none")).lower()
     # 委托检测: 调用插件自己的 check 函数 (node 型/python 型各自实现)
     available, reason = _call_check(manifest.get("check"))
@@ -521,6 +552,7 @@ def _register_one_plugin(manifest: dict) -> None:
         "status": reason,
         "description": manifest.get("description", ""),
         "install_hint": manifest.get("install_hint", ""),
+        "api_key_env": manifest.get("api_key_env", ""),
     }
     if not available:
         return  # 依赖没装: 不注册, 但状态已记录供 UI 显示
diff --git a/backend/app/plugins/fuyao/__init__.py b/backend/app/plugins/fuyao/__init__.py
new file mode 100644
index 0000000..408e0bf
--- /dev/null
+++ b/backend/app/plugins/fuyao/__init__.py
@@ -0,0 +1 @@
+"""扶摇(同花顺金融数据 API)内置数据源插件。"""
diff --git a/backend/app/plugins/fuyao/client.py b/backend/app/plugins/fuyao/client.py
new file mode 100644
index 0000000..821f97b
--- /dev/null
+++ b/backend/app/plugins/fuyao/client.py
@@ -0,0 +1,108 @@
+"""扶摇(同花顺金融数据 API) HTTP 客户端。
+
+职责: 认证、统一信封解包、分页拉取快照。不知道 provider / services 层。
+文档: https://fuyao.aicubes.cn/docs — REST + X-api-key, 响应信封 {code, message, data}。
+"""
+from __future__ import annotations
+
+import logging
+import time
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+BASE_URL = "https://fuyao.aicubes.cn"
+
+# A 股约 5400 只, 500/页约 11 页; 50 页上限防御 count 异常导致的死循环。
+_SNAPSHOT_PAGE_SIZE = 500
+_SNAPSHOT_MAX_PAGES = 50
+_PAGE_INTERVAL_S = 0.15  # 页间隔, 降低触发限频 (code=4001) 的概率
+
+
+class FuyaoError(Exception):
+    """扶摇接口错误(配置缺失 / 网络失败 / 信封 code != 0)。"""
+
+
+class FuyaoClient:
+    """扶摇 REST 客户端 (线程安全: httpx.Client 可并发复用)。"""
+
+    def __init__(self, api_key: str, base_url: str = BASE_URL, timeout: float = 20.0) -> None:
+        if not api_key:
+            raise FuyaoError("未配置 FUYAO_API_KEY")
+        self.last_server_ts = 0  # 最近一页响应里的服务端时间戳(ms), 供行情归属
+        self._http = httpx.Client(
+            base_url=base_url,
+            headers={"X-api-key": api_key},
+            timeout=timeout,
+        )
+
+    def close(self) -> None:
+        self._http.close()
+
+    # ---- 内部 ----
+    def _get(self, path: str, params: dict) -> dict:
+        """GET + 信封解包。code != 0 时抛 FuyaoError(含 code 与 message)。"""
+        try:
+            resp = self._http.get(path, params=params)
+        except httpx.HTTPError as e:
+            raise FuyaoError(f"网络请求失败: {e}") from e
+        if resp.status_code != 200:
+            raise FuyaoError(f"HTTP {resp.status_code}: {path}")
+        try:
+            payload = resp.json()
+        except ValueError as e:
+            raise FuyaoError(f"响应不是 JSON: {path}") from e
+        code = payload.get("code")
+        if code not in (0, "0", None):
+            raise FuyaoError(f"扶摇接口错误 code={code}: {payload.get('message', '')} ({path})")
+        return payload.get("data") or {}
+
+    # ---- 快照 ----
+    def snapshot_page(self, limit: int = _SNAPSHOT_PAGE_SIZE, offset: int = 0) -> tuple[list[dict], int]:
+        """拉取一页 A 股全市场快照。返回 (rows, total), total 为全市场总数。
+
+        实测响应(2026-08): data={timestamp, total, item}; 官方文档示例为
+        data={count, data}。两者都兼容, 以实测为准。
+        """
+        data = self._get("/api/a-share/prices/snapshot", {"limit": limit, "offset": offset})
+        try:
+            self.last_server_ts = int(data.get("timestamp") or 0)
+        except (TypeError, ValueError):
+            self.last_server_ts = 0
+        rows = data.get("item")
+        if not isinstance(rows, list):
+            rows = data.get("data") if isinstance(data.get("data"), list) else []
+        raw_total = data.get("total")
+        if raw_total is None:
+            raw_total = data.get("count") or 0
+        try:
+            total = int(raw_total or 0)
+        except (TypeError, ValueError):
+            total = 0
+        return rows, total
+
+    def snapshot_all(self) -> tuple[list[dict], int]:
+        """分页拉取全市场快照。返回 (rows, 服务端时间戳ms)。
+
+        服务端时间戳用于行情归属; 缺失时返回 0, 由调用方退回本地时间。
+        空数据 / 中途失败时抛 FuyaoError。
+        """
+        out: list[dict] = []
+        server_ts = 0
+        offset = 0
+        for page in range(_SNAPSHOT_MAX_PAGES):
+            if page > 0:
+                time.sleep(_PAGE_INTERVAL_S)
+            rows, total = self.snapshot_page(offset=offset)
+            if not rows:
+                break
+            out.extend(rows)
+            if not server_ts:
+                server_ts = self.last_server_ts
+            if total and len(out) >= total:
+                break
+            offset += len(rows)
+        if not out:
+            raise FuyaoError("全市场快照为空")
+        return out, server_ts
diff --git a/backend/app/plugins/fuyao/plugin.yaml b/backend/app/plugins/fuyao/plugin.yaml
new file mode 100644
index 0000000..aef54dd
--- /dev/null
+++ b/backend/app/plugins/fuyao/plugin.yaml
@@ -0,0 +1,14 @@
+# 扶摇 — 同花顺官方金融数据 API 插件 (https://fuyao.aicubes.cn)
+# 纯 HTTP REST 源, 无额外依赖 (runtime: none)。
+# 需在 .env 或环境变量配置 FUYAO_API_KEY, 未配置时插件灰显不可切换。
+
+name: fuyao
+display_name: "fuyao"
+runtime: none
+entry: app.plugins.fuyao.provider:FuyaoProvider
+check: app.plugins.fuyao.provider:availability
+datasets: [realtime]
+api_key_env: FUYAO_API_KEY   # 声明后设置页提供 Key 输入框(先探后存, secrets.json 优先)
+hidden: true                  # 优化完成前不在数据源页展示; 删除此行即可恢复
+description: "同花顺官方 REST 数据 API。当前提供 A 股全市场实时快照(分页拉取);日K/分钟/财务未接入,自动回退 TickFlow。"
+install_hint: "点击卡片中的输入框配置 API Key(https://fuyao.aicubes.cn 申请),或在 .env 中配置 FUYAO_API_KEY"
diff --git a/backend/app/plugins/fuyao/provider.py b/backend/app/plugins/fuyao/provider.py
new file mode 100644
index 0000000..f0c7379
--- /dev/null
+++ b/backend/app/plugins/fuyao/provider.py
@@ -0,0 +1,195 @@
+"""扶摇(同花顺金融数据 API)内置数据源 provider。
+
+方法签名对齐 custom.GenericHTTPProvider(service 分流点按这套签名调用),
+注入 custom loader 注册表后, 各 service 无需改动即可路由到本 provider。
+
+当前实现数据集: realtime (A 股全市场快照, 分页)。
+未声明 daily / minute / financial → provider_has_dataset 为 False, 自动回退 tickflow。
+
+单位口径 (CONTRIBUTING §3.1, 不可凭字段名推断):
+  - 扶摇 price_change_ratio_pct 为百分数数值 (1.74 = +1.74%), 本项目 realtime
+    change_pct 契约为小数制 (0.0174 = 1.74%) → 此处显式 / 100。
+  - volume 单位股、turnover 单位元, 与内部契约一致, 直接透传。
+"""
+from __future__ import annotations
+
+import contextlib
+import logging
+import time
+from dataclasses import dataclass, field
+
+from app.plugins.fuyao import client as fuyao_client
+from app.plugins.fuyao.client import FuyaoClient, FuyaoError
+
+logger = logging.getLogger(__name__)
+
+# 只声明真实提供的数据集; 其余数据集 provider_has_dataset 返回 False → 回退 tickflow
+_DATASETS = ("realtime",)
+
+API_KEY_ENV = "FUYAO_API_KEY"
+SECRETS_FIELD = "fuyao_api_key"  # UI 配置的 Key 存 secrets.json, 优先级高于 .env
+
+
+def get_api_key() -> str:
+    from app import secrets_store
+    return secrets_store.get_env_backed_secret(SECRETS_FIELD, API_KEY_ENV)
+
+
+def availability() -> tuple[bool, str]:
+    """loader 启动自检: API Key 已配置(secrets.json 或 .env)才注册为可切换数据源。不抛异常。"""
+    if get_api_key():
+        return True, "ok"
+    return False, f"未配置 {API_KEY_ENV}(可在设置页数据源卡片中直接填写)"
+
+
+def probe_api_key(api_key: str) -> tuple[bool, str]:
+    """用候选 Key 实探一次快照接口(先探后存, 对齐 /tickflow-key 语义)。不落盘。"""
+    client = None
+    try:
+        client = fuyao_client.FuyaoClient(api_key=api_key, timeout=10.0)
+        client.snapshot_page(limit=1)
+        return True, "ok"
+    except FuyaoError as e:
+        return False, f"Key 无效或网络失败: {e}"
+    finally:
+        if client is not None:
+            with contextlib.suppress(Exception):
+                client.close()
+
+
+@dataclass
+class _FuyaoConfig:
+    """轻量 config shim, 让 custom loader 的 provider_has_dataset 能识别本 provider。"""
+
+    name: str = "fuyao"
+    display_name: str = "fuyao"
+    datasets: dict = field(default_factory=lambda: dict.fromkeys(_DATASETS))
+    path: None = None
+    builtin: bool = True
+
+
+def _to_float(value) -> float | None:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _first(row: dict, *names: str):
+    """按优先级取第一个非 None 字段。实测字段名与官方文档示例不一致, 两者兼容。"""
+    for n in names:
+        if row.get(n) is not None:
+            return row.get(n)
+    return None
+
+
+def _map_snapshot_row(row: dict, fetched_ms: int) -> dict | None:
+    """扶摇快照行 → 内部 realtime record。字段缺失时按依赖推导, 不伪造数据。
+
+    实测字段(2026-08): high_price / low_price / prev_price;
+    官方文档示例: highest_price / lowest_price / prev_close_price。两者都取。
+    """
+    symbol = row.get("thscode")
+    if not symbol:
+        return None
+    last = _to_float(row.get("last_price"))
+    prev = _to_float(_first(row, "prev_price", "prev_close_price"))
+
+    # 百分数 (1.74 = +1.74%) → 小数制 (0.0174), 契约见模块 docstring
+    pct = _to_float(row.get("price_change_ratio_pct"))
+    change_pct = pct / 100.0 if pct is not None else None
+
+    change_amount = _to_float(row.get("price_change"))
+    if change_amount is None and last is not None and prev is not None:
+        change_amount = last - prev
+    if change_pct is None and change_amount is not None and prev not in (None, 0):
+        # 与 quote_service 的推导同口径: 小数制, 不乘 100
+        change_pct = change_amount / prev
+
+    return {
+        "symbol": symbol,
+        "name": row.get("name"),  # 快照无名称, 由下游维表关联
+        "last_price": last,
+        "prev_close": prev,
+        "open": _to_float(row.get("open_price")),
+        "high": _to_float(_first(row, "high_price", "highest_price")),
+        "low": _to_float(_first(row, "low_price", "lowest_price")),
+        "volume": _to_float(row.get("volume")),
+        "amount": _to_float(row.get("turnover")),
+        "change_pct": change_pct,
+        "change_amount": change_amount,
+        "amplitude": None,      # 快照未提供, 不启发式计算
+        "turnover_rate": None,  # 需股本口径 (§3.4), 交给 enriched 管道用历史股本计算
+        "timestamp": fetched_ms,
+        "session": None,
+    }
+
+
+class FuyaoProvider:
+    """扶摇数据源。realtime = A 股全市场快照(quote_service 全市场模式轮询调用)。"""
+
+    name = "fuyao"
+    builtin = True
+
+    def __init__(self) -> None:
+        self.config = _FuyaoConfig()
+        self._client: FuyaoClient | None = None
+
+    def close(self) -> None:  # loader.load_all 重建注册表时会对每个 provider 调 close
+        if self._client is not None:
+            with contextlib.suppress(Exception):
+                self._client.close()
+            self._client = None
+
+    def _get_client(self) -> FuyaoClient:
+        if self._client is None:
+            self._client = fuyao_client.FuyaoClient(api_key=get_api_key())
+        return self._client
+
+    # ---- realtime ----
+    def get_realtime(self) -> list[dict]:
+        """全市场实时快照 → 内部 realtime records。失败软返回空列表(不阻断轮询)。"""
+        try:
+            rows, server_ts = self._get_client().snapshot_all()
+        except FuyaoError as e:
+            logger.warning("扶摇实时行情拉取失败: %s", e)
+            return []
+
+        # 优先用服务端时间戳(行情归属); 缺失时退回本地时间
+        fetched_ms = server_ts or int(time.time() * 1000)
+
+        records = []
+        dropped = 0
+        for row in rows:
+            rec = _map_snapshot_row(row, fetched_ms)
+            if rec is not None:
+                records.append(rec)
+            else:
+                dropped += 1
+        if dropped and not records:
+            # 整页都识别不出 thscode → 大概率接口 schema 变了, 明确告警而非静默空数据
+            logger.warning("扶摇快照 %d 行全部缺少 thscode 字段, 疑似接口结构变化", dropped)
+            return []
+        logger.info("扶摇实时行情拉取完成: %d 条(丢弃 %d 行)", len(records), dropped)
+        return records
+
+    # ---- 测试(设置页试拉) ----
+    def test_dataset(self, dataset: str, symbols: list[str] | None = None) -> dict:
+        if dataset != "realtime":
+            return {"provider": self.name, "dataset": dataset, "rows": 0,
+                    "error": f"扶摇插件未接入 {dataset} 数据集(自动回退 TickFlow)"}
+        try:
+            rows, count = self._get_client().snapshot_page(limit=5)
+        except FuyaoError as e:
+            return {"provider": self.name, "dataset": "realtime", "rows": 0, "error": str(e)}
+        fetched_ms = int(time.time() * 1000)
+        head = [r for r in (_map_snapshot_row(row, fetched_ms) for row in rows) if r][:5]
+        return {
+            "provider": self.name,
+            "dataset": "realtime",
+            "rows": count or len(head),
+            "columns": list(head[0].keys()) if head else [],
+            "preview": head,
+        }
diff --git a/backend/app/secrets_store.py b/backend/app/secrets_store.py
index 8cc4ca8..e0de664 100644
--- a/backend/app/secrets_store.py
+++ b/backend/app/secrets_store.py
@@ -99,6 +99,17 @@ def get_ai_config_int(key: str, default: int) -> int:
     return int(getattr(settings, key, default) or default)
 
 
+def get_env_backed_secret(field: str, env_name: str) -> str:
+    """取环境变量后备的密钥(插件 API Key 等):secrets.json 优先,否则环境变量。
+
+    与 get_tickflow_key 同优先级语义:UI 写入 secrets.json 后即覆盖 .env。
+    """
+    val = load().get(field)
+    if val:
+        return str(val).strip()
+    return os.environ.get(env_name, "").strip()
+
+
 def mask(key: str, prefix: int = 4, suffix: int = 4) -> str:
     """脱敏显示。"""
     if not key:
diff --git a/backend/tests/test_fuyao_provider.py b/backend/tests/test_fuyao_provider.py
new file mode 100644
index 0000000..6bfdfed
--- /dev/null
+++ b/backend/tests/test_fuyao_provider.py
@@ -0,0 +1,401 @@
+"""FuyaoProvider 契约与单位标准化测试。
+
+不依赖真实网络: 用假 FuyaoClient 返回样例快照页, 验证字段映射、
+百分数→小数制转换 (CONTRIBUTING §3.1)、分页合并、软失败、
+能力声明 (未声明数据集回退 tickflow) 与设置页试拉。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.plugins.fuyao import client as fc
+from app.plugins.fuyao import provider as fp
+from app.plugins.fuyao.provider import FuyaoProvider
+
+
+class _FakeClient:
+    """按调用次数返回预置页, 记录调用供分页断言。snapshot_all 同真实客户端语义。"""
+
+    def __init__(self, pages: list[list[dict]], count: int, error: Exception | None = None, server_ts: int = 0):
+        self.pages = pages
+        self.count = count
+        self.error = error
+        self.server_ts = server_ts
+        self.calls: list[dict] = []
+        self.last_server_ts = server_ts
+
+    def snapshot_page(self, limit=500, offset=0):
+        self.calls.append({"limit": limit, "offset": offset})
+        if self.error:
+            raise self.error
+        if not self.pages:
+            return [], self.count
+        # 按调用次数取页(provider 每轮 offset += len(rows), 页序与调用序一致)
+        idx = min(len(self.calls) - 1, len(self.pages) - 1)
+        return list(self.pages[idx]), self.count
+
+    def snapshot_all(self):
+        """与真实客户端同语义的分页循环 (无页间隔, 测试用)。返回 (rows, server_ts)。"""
+        out: list[dict] = []
+        offset = 0
+        for _ in range(50):
+            rows, count = self.snapshot_page(offset=offset)
+            if not rows:
+                break
+            out.extend(rows)
+            if count and len(out) >= count:
+                break
+            offset += len(rows)
+        if not out:
+            raise fc.FuyaoError("全市场快照为空")
+        return out, self.server_ts
+
+    def close(self):
+        pass
+
+
+def _row(thscode: str = "600519.SH", **over):
+    """实测快照行结构(2026-08): high_price/low_price/prev_price 命名。"""
+    row = {
+        "thscode": thscode,
+        "last_price": 1480.0,
+        "price_change": 25.0,
+        "price_change_ratio_pct": 1.72,
+        "open_price": 1460.0,
+        "high_price": 1490.5,
+        "low_price": 1455.0,
+        "prev_price": 1455.0,
+        "volume": 1234500,
+        "turnover": 1.83e9,
+    }
+    row.update(over)
+    return row
+
+
+def _provider_with(monkeypatch, pages, count=None, error=None, **fake_kwargs):
+    fake = _FakeClient(pages, count if count is not None else sum(len(p) for p in pages), error, **fake_kwargs)
+    monkeypatch.setattr(fp, "fuyao_client", type("M", (), {"FuyaoClient": lambda **kw: fake}))
+    monkeypatch.setattr(fp, "get_api_key", lambda: "test-key")
+    return FuyaoProvider(), fake
+
+
+# ---- 单位与字段映射 ----
+
+def test_snapshot_units_and_field_mapping(monkeypatch):
+    """核心口径: price_change_ratio_pct 百分数 → change_pct 小数制 (1.72 → 0.0172)。"""
+    provider, _ = _provider_with(monkeypatch, [[_row()]])
+    records = provider.get_realtime()
+    assert len(records) == 1
+    r = records[0]
+    assert r["symbol"] == "600519.SH"
+    assert r["change_pct"] == pytest.approx(0.0172)
+    assert r["change_amount"] == pytest.approx(25.0)
+    assert r["prev_close"] == 1455.0
+    assert r["open"] == 1460.0
+    assert r["high"] == 1490.5
+    assert r["low"] == 1455.0
+    assert r["volume"] == 1234500
+    assert r["amount"] == 1.83e9
+    assert r["timestamp"] > 0
+    # 快照不提供的字段必须为 None, 不启发式伪造
+    assert r["name"] is None
+    assert r["amplitude"] is None
+    assert r["turnover_rate"] is None
+
+
+def test_missing_pct_derives_decimal_from_change_amount(monkeypatch):
+    """涨跌幅缺失时按 change_amount/prev_close 推导, 仍为小数制。"""
+    provider, _ = _provider_with(monkeypatch, [[_row(price_change_ratio_pct=None)]])
+    r = provider.get_realtime()[0]
+    assert r["change_pct"] == pytest.approx(25.0 / 1455.0)
+    assert r["change_amount"] == pytest.approx(25.0)
+
+
+def test_missing_change_amount_derived_from_prices(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[_row(price_change=None)]])
+    r = provider.get_realtime()[0]
+    assert r["change_amount"] == pytest.approx(1480.0 - 1455.0)
+
+
+def test_row_without_thscode_dropped(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[_row(), {"last_price": 1.0}, _row("000001.SZ")]])
+    records = provider.get_realtime()
+    assert [r["symbol"] for r in records] == ["600519.SH", "000001.SZ"]
+
+
+def test_all_rows_unrecognized_returns_empty_with_no_fake_data(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[{"foo": "bar"}, {"baz": 1}]])
+    assert provider.get_realtime() == []
+
+
+# ---- 客户端信封解析 (实测结构 vs 文档示例) ----
+
+def _patch_http(monkeypatch, payload, status_code=200):
+    class _Resp:
+        def json(self):
+            return payload
+    _Resp.status_code = status_code
+
+    class _Http:
+        def get(self, path, params=None):
+            return _Resp()
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(fc.httpx, "Client", lambda **kw: _Http())
+
+
+def test_client_parses_real_world_envelope(monkeypatch):
+    """实测信封(2026-08): data={timestamp, total, item}。"""
+    _patch_http(monkeypatch, {
+        "code": 0, "message": "success",
+        "data": {"timestamp": 1787542612000, "total": 2,
+                 "item": [_row(), _row("000001.SZ")]},
+    })
+    c = fc.FuyaoClient(api_key="k")
+    rows, total = c.snapshot_page()
+    assert total == 2 and len(rows) == 2
+    assert c.last_server_ts == 1787542612000
+
+
+def test_client_parses_documented_envelope(monkeypatch):
+    """官方文档示例信封: data={count, data}。"""
+    _patch_http(monkeypatch, {
+        "code": 0, "message": "OK",
+        "data": {"count": 3, "data": [_row()]},
+    })
+    c = fc.FuyaoClient(api_key="k")
+    rows, total = c.snapshot_page()
+    assert total == 3 and len(rows) == 1
+
+
+def test_client_raises_on_error_code(monkeypatch):
+    _patch_http(monkeypatch, {"code": 4001, "message": "rate limited", "data": None})
+    c = fc.FuyaoClient(api_key="k")
+    with pytest.raises(fc.FuyaoError, match="4001"):
+        c.snapshot_page()
+
+
+# ---- 字段名兼容与服务端时间戳 ----
+
+def test_doc_style_field_names_fallback(monkeypatch):
+    """文档示例字段名 (highest_price/lowest_price/prev_close_price) 也能映射。"""
+    row = {
+        "thscode": "600519.SH", "last_price": 1480.0, "price_change": 25.0,
+        "price_change_ratio_pct": 1.72, "open_price": 1460.0,
+        "highest_price": 1490.5, "lowest_price": 1455.0, "prev_close_price": 1455.0,
+        "volume": 1234500, "turnover": 1.83e9,
+    }
+    provider, _ = _provider_with(monkeypatch, [[row]])
+    r = provider.get_realtime()[0]
+    assert r["high"] == 1490.5
+    assert r["low"] == 1455.0
+    assert r["prev_close"] == 1455.0
+
+
+def test_realtime_uses_server_timestamp(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[_row()]], server_ts=1787542612000)
+    assert provider.get_realtime()[0]["timestamp"] == 1787542612000
+
+
+def test_realtime_falls_back_to_local_time_without_server_ts(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[_row()]], server_ts=0)
+    assert provider.get_realtime()[0]["timestamp"] > 0
+
+
+# ---- 分页 ----
+
+def test_snapshot_pagination_merges_pages(monkeypatch):
+    page1 = [_row(f"{600000 + i}.SH") for i in range(2)]
+    page2 = [_row(f"{688000 + i}.SH") for i in range(1)]
+    provider, fake = _provider_with(monkeypatch, [page1, page2], count=3)
+    records = provider.get_realtime()
+    assert len(records) == 3
+    assert len(fake.calls) == 2
+    assert fake.calls[1]["offset"] == 2
+
+
+def test_snapshot_stops_when_page_empty(monkeypatch):
+    provider, fake = _provider_with(monkeypatch, [[_row()], []], count=0)
+    assert len(provider.get_realtime()) == 1
+    assert len(fake.calls) == 2
+
+
+# ---- 软失败 ----
+
+def test_realtime_error_returns_empty_list(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[]], error=fc.FuyaoError("扶摇接口错误 code=4001: 频率超限"))
+    assert provider.get_realtime() == []
+
+
+def test_client_requires_api_key():
+    with pytest.raises(fc.FuyaoError):
+        fc.FuyaoClient(api_key="")
+
+
+# ---- 能力声明与注册 ----
+
+def test_datasets_declaration_realtime_only():
+    """只声明 realtime; 其他数据集 provider_has_dataset 必须为 False (回退 tickflow)。"""
+    config = FuyaoProvider().config
+    assert "realtime" in config.datasets
+    assert "daily" not in config.datasets
+    assert "minute" not in config.datasets
+    assert "financial" not in config.datasets
+
+
+# ---- API Key 解析 (secrets.json > .env, 对齐 tickflow 语义) ----
+
+def test_get_api_key_secrets_store_takes_priority(monkeypatch):
+    from app import secrets_store
+    monkeypatch.delenv(fp.API_KEY_ENV, raising=False)
+    monkeypatch.setattr(secrets_store, "load", lambda: {fp.SECRETS_FIELD: "sk-from-ui"})
+    assert fp.get_api_key() == "sk-from-ui"
+
+
+def test_get_api_key_falls_back_to_env(monkeypatch):
+    from app import secrets_store
+    monkeypatch.setenv(fp.API_KEY_ENV, "sk-from-env")
+    monkeypatch.setattr(secrets_store, "load", lambda: {})
+    assert fp.get_api_key() == "sk-from-env"
+
+
+def test_availability_accepts_secrets_store_key(monkeypatch):
+    from app import secrets_store
+    monkeypatch.delenv(fp.API_KEY_ENV, raising=False)
+    monkeypatch.setattr(secrets_store, "load", lambda: {fp.SECRETS_FIELD: "sk-from-ui"})
+    assert fp.availability() == (True, "ok")
+
+
+def test_availability_requires_env_key(monkeypatch):
+    from app import secrets_store
+    monkeypatch.delenv(fp.API_KEY_ENV, raising=False)
+    monkeypatch.setattr(secrets_store, "load", lambda: {})
+    ok, reason = fp.availability()
+    assert ok is False and fp.API_KEY_ENV in reason
+
+
+# ---- 先探后存 (probe_api_key) ----
+
+def _patch_client_cls(monkeypatch, fake):
+    monkeypatch.setattr(fp, "fuyao_client", type("M", (), {"FuyaoClient": lambda **kw: fake}))
+
+
+def test_probe_api_key_ok(monkeypatch):
+    _patch_client_cls(monkeypatch, _FakeClient([[_row()]], 1))
+    ok, reason = fp.probe_api_key("sk-candidate")
+    assert ok is True and reason == "ok"
+
+
+def test_probe_api_key_invalid_key(monkeypatch):
+    _patch_client_cls(monkeypatch, _FakeClient([[]], 0, error=fc.FuyaoError("扶摇接口错误 code=1001: 无效 api key")))
+    ok, reason = fp.probe_api_key("sk-bad")
+    assert ok is False and "无效" in reason
+
+
+def test_loader_probe_plugin_key_dispatch(monkeypatch):
+    import app.plugins.fuyao.provider as provider_mod
+    from app.data_providers.custom import loader
+
+    monkeypatch.setattr(provider_mod, "probe_api_key", lambda key: (True, "ok") if key == "good" else (False, "bad"))
+    assert loader.probe_plugin_key("fuyao", "good") == (True, "ok")
+    assert loader.probe_plugin_key("fuyao", "bad") == (False, "bad")
+
+
+def test_loader_probe_plugin_key_unsupported_plugin():
+    from app.data_providers.custom import loader
+    # stock-sdk 未声明 api_key_env → 不支持界面配 Key
+    ok, reason = loader.probe_plugin_key("stocksdk", "x")
+    assert ok is False and "不支持" in reason
+    ok, reason = loader.probe_plugin_key("no_such_plugin", "x")
+    assert ok is False and "不存在" in reason
+
+
+# ---- 保存/清除端点 (直接调用 handler, 先探后存语义) ----
+
+def test_save_plugin_key_invalid_key_not_persisted(monkeypatch):
+    from app.api import settings as settings_api
+    from app.data_providers import custom as custom_sources
+
+    saved: dict = {}
+    monkeypatch.setattr(custom_sources, "probe_plugin_key", lambda n, k: (False, "Key 无效"))
+    monkeypatch.setattr(settings_api.secrets_store, "save", lambda updates: saved.update(updates) or updates)
+    out = settings_api.save_plugin_key(settings_api.PluginKeyIn(plugin="fuyao", api_key="bad"))
+    assert out["ok"] is False and out["reason"] == "invalid"
+    assert saved == {}  # 无效 Key 不落盘
+
+
+def test_save_plugin_key_valid_persists_and_rescans(monkeypatch):
+    from app.api import settings as settings_api
+    from app.data_providers import custom as custom_sources
+
+    saved: dict = {}
+    reloaded = []
+    monkeypatch.setattr(custom_sources, "probe_plugin_key", lambda n, k: (True, "ok"))
+    monkeypatch.setattr(settings_api.secrets_store, "save", lambda updates: saved.update(updates) or updates)
+    monkeypatch.setattr(settings_api.secrets_store, "mask", lambda key, prefix=4, suffix=4: "abcd••••wxyz")
+    monkeypatch.setattr(custom_sources, "load_all", lambda: reloaded.append(1))
+    monkeypatch.setattr(custom_sources, "list_plugins", lambda: [{"name": "fuyao", "available": True}])
+    out = settings_api.save_plugin_key(settings_api.PluginKeyIn(plugin="fuyao", api_key="good-key"))
+    assert out["ok"] is True
+    assert saved == {"fuyao_api_key": "good-key"}  # 字段名与 provider.SECRETS_FIELD 一致
+    assert out["plugin_available"] is True
+    assert reloaded == [1]  # 保存后重扫, 插件即刻可用
+
+
+def test_clear_plugin_key(monkeypatch):
+    from app.api import settings as settings_api
+    from app.data_providers import custom as custom_sources
+
+    cleared: list = []
+    monkeypatch.setattr(custom_sources, "is_builtin", lambda n: n == "fuyao")
+    monkeypatch.setattr(settings_api.secrets_store, "clear", lambda *keys: cleared.extend(keys))
+    monkeypatch.setattr(custom_sources, "load_all", lambda: None)
+    monkeypatch.setattr(custom_sources, "list_plugins", lambda: [{"name": "fuyao", "available": False}])
+    out = settings_api.clear_plugin_key("fuyao")
+    assert out["ok"] is True and out["plugin_available"] is False
+    assert cleared == ["fuyao_api_key"]
+
+
+def test_manifest_declares_realtime_dataset():
+    from app.data_providers.custom import loader
+    manifest = loader.plugin_manifest("fuyao")
+    assert manifest is not None
+    assert manifest["entry"] == "app.plugins.fuyao.provider:FuyaoProvider"
+    assert "realtime" in (manifest.get("datasets") or [])
+    assert manifest.get("runtime") == "none"
+    assert manifest.get("api_key_env") == fp.API_KEY_ENV
+
+
+def test_hidden_plugin_not_registered():
+    """hidden: true 的插件不注册、不在数据源页展示 (优化完成前隐藏 fuyao)。"""
+    from app.data_providers.custom import loader
+    manifest = loader.plugin_manifest("fuyao")
+    assert manifest.get("hidden") is True
+    loader._register_one_plugin(manifest)
+    assert "fuyao" not in loader._PLUGIN_STATUS
+    assert "fuyao" not in loader._PROVIDERS
+
+
+# ---- 设置页试拉 ----
+
+def test_test_dataset_realtime_preview(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[_row(), _row("000001.SZ")]], count=5400)
+    out = provider.test_dataset("realtime")
+    assert out["provider"] == "fuyao"
+    assert out["rows"] == 5400
+    assert out["preview"][0]["symbol"] == "600519.SH"
+    assert out["preview"][0]["change_pct"] == pytest.approx(0.0172)
+
+
+def test_test_dataset_unsupported_dataset_reports_fallback(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[]])
+    out = provider.test_dataset("minute")
+    assert "error" in out and "回退" in out["error"]
+
+
+def test_close_is_idempotent(monkeypatch):
+    provider, _ = _provider_with(monkeypatch, [[_row()]])
+    provider.close()
+    provider.close()
diff --git a/docs/plugin-development.md b/docs/plugin-development.md
index 840c2dc..9d04a4f 100644
--- a/docs/plugin-development.md
+++ b/docs/plugin-development.md
@@ -25,10 +25,25 @@ 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]     # 支持的数据集
+api_key_env: MY_SOURCE_API_KEY           # (可选)声明后设置页提供 Key 输入框
+hidden: false                            # (可选)true = 已加载但对设置页隐藏,不注册不展示
 description: "数据源描述"
 install_hint: "pip install xxx"          # 未装依赖时显示的安装提示
 ```
 
+#### api_key_env(界面配置 API Key)
+
+声明 `api_key_env` 的插件可以在设置页的数据源卡片中直接填写 Key, 对齐
+TickFlow 的「先探后存」语义:
+
+1. entry 模块需提供模块级 `probe_api_key(key) -> (ok, reason)`,
+   后端用候选 Key 实探一次, **无效不落盘**
+2. 有效则写入 `data/user_data/secrets.json` 的 `{name}_api_key` 字段
+   (0600 权限, 优先级高于 `.env` / 环境变量)
+3. 保存后自动重载数据源注册表, 插件即刻变为可切换
+4. 插件取 Key 用 `secrets_store.get_env_backed_secret("{name}_api_key", api_key_env)`,
+   保证 secrets.json 与 .env 两条配置路径一致
+
 ### runtime 字段说明
 
 | runtime | 含义 | 典型场景 |
@@ -105,6 +120,12 @@ class MyConfig:
 
 ## 现有插件参考
 
+- **`backend/app/plugins/fuyao/`** — 同花顺官方 REST 数据源(runtime: none, 纯 HTTP 零依赖)
+  - 当前提供 `realtime`(A 股全市场快照, 分页拉取); Key 在设置页卡片直接配置(先探后存), 或 `.env` 配 `FUYAO_API_KEY`
+  - `client.py` — httpx 客户端(X-api-key 认证 + 统一信封解包 + 分页)
+  - `provider.py` — Provider 实现(字段映射、百分数→小数制单位转换、软失败、Key 探测)
+  - 单位口径注意: 扶摇 `price_change_ratio_pct` 为百分数数值(1.74 = +1.74%),
+    内部 `change_pct` 契约为小数制, provider 内显式 / 100(见 CONTRIBUTING §3.1)
 - **`backend/app/plugins/stocksdk/`** — Node 型插件, 通过 subprocess 桥接调用 stock-sdk
   - `bridge.py` — Python↔Node 桥接 + availability 检测
   - `bridge.mjs` — Node 端(并发池、重试、SDK 解析)
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts
index 75c334d..6ae9cc3 100644
--- a/frontend/src/lib/api.ts
+++ b/frontend/src/lib/api.ts
@@ -1376,6 +1376,7 @@ export interface PluginDataSourceItem {
   status: string           // 可用性原因 (供 UI 显示)
   description: string
   install_hint: string     // 未装依赖时显示的安装命令
+  api_key_env?: string     // 声明后设置页提供 Key 输入框 (先探后存)
 }
 
 export interface DataSourceLoadError {
@@ -1400,6 +1401,16 @@ export interface DataSourceTestResult {
   preview: Record<string, unknown>[]
 }
 
+/** 插件 Key 保存结果 (先探后存: 无效 Key 返回 ok=false 且不落盘) */
+export interface PluginKeyResult {
+  ok: boolean
+  reason?: string
+  error?: string
+  api_key_masked?: string
+  plugin_available?: boolean
+  plugin?: PluginDataSourceItem | null
+}
+
 export interface DatasetConfig {
   url: string
   method: string
@@ -1595,6 +1606,18 @@ export const api = {
       `/api/settings/plugins/${encodeURIComponent(name)}/install`,
       { method: 'DELETE' },
     ),
+  savePluginKey: (plugin: string, apiKey: string) => {
+    // 先探后存: 后端会用候选 Key 实探一次, 探测超时 10s + 余量
+    const controller = new AbortController()
+    const timer = setTimeout(() => controller.abort(), 30_000)
+    return request<PluginKeyResult>('/api/settings/plugin-key', {
+      method: 'POST',
+      body: JSON.stringify({ plugin, api_key: apiKey }),
+      signal: controller.signal,
+    }).finally(() => clearTimeout(timer))
+  },
+  clearPluginKey: (plugin: string) =>
+    request<PluginKeyResult>(`/api/settings/plugin-key/${encodeURIComponent(plugin)}`, { method: 'DELETE' }),
   testDataSource: (
     provider: string,
     dataset: string,