feat(fuyao): 日K dump 链路强化与财务数据集适配

- 财务接入(股本除外): 三大报表多期 + 指标单期, 字段映射至项目 canonical 列
- metrics: bps 由估值 pb_mrq 反推, eps_basic 取自利润表, report 期由最新披露期反查
- financial_sync 报告期合并升级为逐列填空: 多源数据并集共存, 新值覆盖缺列保留
- 前序未入库工作一并提交: 10 年/10d 日K dump、除权因子事件推导与涨跌停自检

backend/app/plugins/fuyao/client.py

@@ -1,12 +1,17 @@
 """扶摇(同花顺金融数据 API) HTTP 客户端。
 
-职责: 认证、统一信封解包、分页拉取快照。不知道 provider / services 层。
+职责: 认证、统一信封解包、快照分页、单标的日K、市场 dump 下载。不知道 provider / services 层。
 文档: https://fuyao.aicubes.cn/docs — REST + X-api-key, 响应信封 {code, message, data}。
+
+时间字段口径: 所有 *ms 字段(含 start/end 入参与 date_ms/ex_date_ms 出参)均为
+北京时间零点对应的 epoch ms(= UTC 前一日 16:00), 由 provider 层统一 +8h 换算。
 """
+
 from __future__ import annotations
 
 import logging
 import time
+from pathlib import Path
 
 import httpx
 
@@ -60,7 +65,9 @@ class FuyaoClient:
         return payload.get("data") or {}
 
     # ---- 快照 ----
-    def snapshot_page(self, limit: int = _SNAPSHOT_PAGE_SIZE, offset: int = 0) -> tuple[list[dict], int]:
+    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}; 官方文档示例为
@@ -107,3 +114,131 @@ class FuyaoClient:
         if not out:
             raise FuyaoError("全市场快照为空")
         return out, server_ts
+
+    # ---- 历史日K ----
+    def historical_kline(
+        self, thscode: str, start_ms: int, end_ms: int, adjust: str = "none"
+    ) -> list[dict]:
+        """单标的日K(interval=1d 固定)。单次窗口 ≤10 年, 超出由调用方分片。
+
+        adjust 必须显式传 "none" 取原始价 — 服务端默认是 forward(前复权),
+        官方前复权序列事件间存在逐日漂移, 项目内禁止使用。
+        返回 data.item 原始行: {date_ms, open_price, high_price, low_price,
+        close_price, volume(股), turnover(元)}。
+        """
+        data = self._get(
+            "/api/a-share/prices/historical",
+            {
+                "thscode": thscode,
+                "interval": "1d",
+                "adjust": adjust,
+                "start": int(start_ms),
+                "end": int(end_ms),
+            },
+        )
+        rows = data.get("item")
+        return rows if isinstance(rows, list) else []
+
+    # ---- 财务 ----
+    # 端点均单标的(thscode 不接受逗号)。取数模式二选一: limit=最近N期 或 start/end 区间,
+    # 这里只用 limit。period=quarterly 覆盖每个季度末(含年报期), 与项目"各报告期累积"口径一致。
+    _STATEMENT_ENDPOINTS = {
+        "income": "income-statements",
+        "balance_sheet": "balance-sheets",
+        "cash_flow": "cash-flow-statements",
+    }
+
+    def financial_statements(
+        self, stmt: str, thscode: str, limit: int = 1
+    ) -> list[dict]:
+        """单标的财务报表多期序列。stmt: income | balance_sheet | cash_flow。
+
+        返回 data.item 原始行: 共有元数据(thscode/period/fiscal_year/fiscal_period/
+        report_date_ms/period_end_ms/currency) + 各表字段。行内 null 表示该期未披露。
+        """
+        endpoint = self._STATEMENT_ENDPOINTS.get(stmt)
+        if endpoint is None:
+            raise FuyaoError(f"未知财务报表类型: {stmt}")
+        data = self._get(
+            f"/api/a-share/financials/{endpoint}",
+            {"thscode": thscode, "period": "quarterly", "limit": max(1, min(20, limit))},
+        )
+        rows = data.get("item")
+        return rows if isinstance(rows, list) else []
+
+    def financial_indicators(self, thscode: str, report: str) -> list[dict]:
+        """单标的单报告期财务指标(report 格式 yyyy-N, N=1..4 对应一季报..年报)。
+
+        返回 data.abilities 原始列表 [{ability, indicators: [{index_id, value}]}];
+        value 为保留原始精度的数值字符串(百分制指标即百分点数), 缺失为 null。
+        未披露报告期实测返回 code=5003(文档写 3002, 以实测为准) → 经 _get 抛 FuyaoError,
+        由调用方按"该期无数据"处理。
+        """
+        data = self._get(
+            "/api/a-share/financials/indicators",
+            {"thscode": thscode, "report": report},
+        )
+        abilities = data.get("abilities")
+        return abilities if isinstance(abilities, list) else []
+
+    def valuations_snapshot(self, thscodes: list[str]) -> list[dict]:
+        """批量估值快照(pe_ttm/pe_mrq/pb_mrq/ps_ttm/pcf_ttm), 数值为最新口径。
+
+        服务端单次上限 100 只(超出 code=1003), 分批由调用方负责。
+        返回 data.item 原始行。
+        """
+        data = self._get(
+            "/api/a-share/valuations/snapshot",
+            {"thscodes": ",".join(thscodes[:100])},
+        )
+        rows = data.get("item")
+        return rows if isinstance(rows, list) else []
+
+    def price_snapshot_batch(self, thscodes: list[str]) -> list[dict]:
+        """按 thscodes 批量行情快照(最新价等), 用于估值推导的分母。
+
+        与全市场分页快照同一端点; thscodes 显式传入时不分页。
+        返回 data.item 原始行。
+        """
+        data = self._get(
+            "/api/a-share/prices/snapshot",
+            {"thscodes": ",".join(thscodes[:100])},
+        )
+        rows = data.get("item")
+        return rows if isinstance(rows, list) else []
+
+    # ---- 市场 dump ----
+    def dump_download_url(self, dump_kind: str) -> dict:
+        """获取 dump 预签名下载信息(约 300s 有效)。
+
+        dump_kind: adjustment-factors | daily-k-10d | daily-k。
+        返回 {presigned_url, presigned_url_expires_at, expires_in_seconds};
+        release 版本号(如 20260828)嵌在 presigned_url 的 releases/<date>/ 路径中,
+        供调用方做缓存版本管理。
+        """
+        return self._get(f"/api/dump/market-dumps/{dump_kind}/download-url", {})
+
+    def download_dump(self, dump_kind: str, dest: Path) -> Path:
+        """下载 dump 到 dest(先写 .part 临时文件, 成功后原子改名)。失败抛 FuyaoError。
+
+        预签名 URL 指向对象存储, 请求不得携带 X-api-key 头 → 用独立裸请求,
+        不经过持有认证头的 self._http。
+        """
+        url = str(self.dump_download_url(dump_kind).get("presigned_url") or "")
+        if not url:
+            raise FuyaoError(f"dump {dump_kind} 未返回预签名 URL")
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        tmp = dest.with_name(dest.name + ".part")
+        try:
+            with httpx.stream("GET", url, timeout=120.0, follow_redirects=True) as resp:
+                if resp.status_code != 200:
+                    raise FuyaoError(f"dump {dump_kind} 下载失败 HTTP {resp.status_code}")
+                with open(tmp, "wb") as fh:
+                    for chunk in resp.iter_bytes(1 << 20):
+                        fh.write(chunk)
+            tmp.replace(dest)
+        except httpx.HTTPError as e:
+            raise FuyaoError(f"dump {dump_kind} 下载网络失败: {e}") from e
+        finally:
+            tmp.unlink(missing_ok=True)
+        return dest

backend/app/plugins/fuyao/plugin.yaml

@@ -7,8 +7,8 @@ display_name: "fuyao"
 runtime: none
 entry: app.plugins.fuyao.provider:FuyaoProvider
 check: app.plugins.fuyao.provider:availability
-datasets: [realtime]
+datasets: [realtime, daily, adj_factor, financial]
 api_key_env: FUYAO_API_KEY   # 声明后设置页提供 Key 输入框(先探后存, secrets.json 优先)
-description: "同花顺官方 REST 数据 API。当前提供 A 股全市场实时快照(分页拉取);日K/分钟/财务未接入,自动回退 TickFlow。"
+description: "同花顺官方 REST 数据 API。A 股实时快照、日K(原始价: 近端走 10d dump, 深窗口走 10 年全量 dump 一次下载秒级筛选, 兜底单标的接口)、除权因子(事件 dump + 本地日K dump 配价推导, 涨跌停自检, 秒级)、财务(利润表/资产负债表/现金流量表/指标, 字段映射至项目口径, bps 由估值反推; 股本无上游接口, 指标历史期建议切回 TickFlow 补齐);分钟未接入,自动回退 TickFlow。"
 install_hint: "点击卡片中的输入框配置 API Key(https://fuyao.aicubes.cn 申请),或在 .env 中配置 FUYAO_API_KEY"
 homepage: "https://fuyao.aicubes.cn"

backend/app/plugins/fuyao/provider.py

@@ -3,35 +3,77 @@
 方法签名对齐 custom.GenericHTTPProvider(service 分流点按这套签名调用),
 注入 custom loader 注册表后, 各 service 无需改动即可路由到本 provider。
 
-当前实现数据集: realtime (A 股全市场快照, 分页)。
-未声明 daily / minute / financial → provider_has_dataset 为 False, 自动回退 tickflow。
+实现数据集:
+  - realtime     A 股全市场快照 (分页)
+  - daily        A 股日K, 原始价; 近端窗口走 daily-k-10d 全市场 dump(1 次请求),
+                 深窗口走单标的 historical 接口(≤10 年/次自动分片)
+  - adj_factor   A 股除权因子; adjustment-factors 事件 dump + 自家原始日K前收盘,
+                 按交易所公式推导单事件比值, 涨跌停自检
+  - financial    财务五表(股本除外): 三表多期序列 + 指标单期, 字段映射为 TickFlow
+                 canonical 列名, 扶摇独有字段原名透传为扩展列; bps 由估值 pb_mrq
+                 反推; shares 无上游接口恒空
+未声明 minute → provider_has_dataset 为 False, 自动回退 tickflow。
 
-单位口径 (CONTRIBUTING §3.1, 不可凭字段名推断):
+单位与口径 (CONTRIBUTING §3.1, 不可凭字段名推断):
   - 扶摇 price_change_ratio_pct 为百分数数值 (1.74 = +1.74%), 本项目 realtime
     change_pct 契约为小数制 (0.0174 = 1.74%) → 此处显式 / 100。
-  - volume 单位股、turnover 单位元, 与内部契约一致, 直接透传。
+  - 扶摇 volume 单位为股, 本项目日K/实时契约均为手 → 统一 floor(股/100)。
+  - turnover 单位元, 与内部一致, 直接透传。
+  - 日K取数 adjust=none 锁定: 官方 forward 序列事件间有逐日漂移(2026-08 实测),
+    项目内前复权一律由 indicators.pipeline 用本地因子计算。
+  - ex_factor 为单事件比值(非累积), 累积链由 pipeline._apply_adj_factor 构建。
 """
+
 from __future__ import annotations
 
+import calendar
 import contextlib
 import logging
+import math
+import re
 import time
+from collections.abc import Callable
 from dataclasses import dataclass, field
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
 
+import polars as pl
+
+from app.data_providers.normalizer import DAILY_COLS, normalize_daily
+from app.indicators.pipeline import filter_halt_days
 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",)
+_DATASETS = ("realtime", "daily", "adj_factor", "financial")
 
 API_KEY_ENV = "FUYAO_API_KEY"
 SECRETS_FIELD = "fuyao_api_key"  # UI 配置的 Key 存 secrets.json, 优先级高于 .env
 
+# 扶摇 *ms 时间字段为北京时间零点(= UTC 前一日 16:00), +8h 后按 UTC 解析即得交易日
+_SH_MS = 28_800_000
+_HIST_MAX_SPAN_MS = 3650 * 86_400_000  # historical 单次窗口上限 10 年, 超出由本层分片
+_HIST_INTERVAL_S = 0.12  # 单标的请求节流(实测 200+ 连发未触发 4001 限频)
+_FINANCIAL_HISTORY_PERIODS = 8  # 财务首装全量历史: 最近 8 期季报(约 2 年)
+_VALUATION_BATCH = 100  # 估值/价格快照端点单次上限 100 只
+# 项目财务表名 → 扶摇报表端点名
+_STATEMENT_ENDPOINTS = {
+    "income": "income-statements",
+    "balance_sheet": "balance-sheets",
+    "cash_flow": "cash-flow-statements",
+}
+_ADJ_DUMP_KIND = "adjustment-factors"
+_DAILY10_DUMP_KIND = "daily-k-10d"
+_DAILY_DUMP_KIND = "daily-k"  # 10 年全量日K dump(约 172MB), 深窗口一次下载覆盖全市场
+_RECENT_DUMP_DAYS = 12  # 窗口跨度 ≤ 此天数时优先走 10d dump(覆盖 ≈10 个交易日)
+_PREV_CLOSE_BACKDAYS = 30  # 推导因子时向前找"除权日前收盘"的回看天数(容忍长期停牌)
+
 
 def get_api_key() -> str:
     from app import secrets_store
+
     return secrets_store.get_env_backed_secret(SECRETS_FIELD, API_KEY_ENV)
 
 
@@ -39,7 +81,8 @@ def availability() -> tuple[bool, str]:
     """loader 启动自检: API Key 已配置(secrets.json 或 .env)才注册为可切换数据源。不抛异常。"""
     if get_api_key():
         return True, "ok"
-    return False, f"未配置 {API_KEY_ENV}(可在设置页数据源卡片中直接填写)"
+    # 状态行会拼在「未配置」标签之后, 文案不再重复"未配置"字样
+    return False, f"缺少 API Key(可在下方输入框直接填写,或配置环境变量 {API_KEY_ENV})"
 
 
 def probe_api_key(api_key: str) -> tuple[bool, str]:
@@ -85,6 +128,128 @@ def _first(row: dict, *names: str):
     return None
 
 
+def _date_of_ms(value) -> date | None:
+    """扶摇 *ms(上海零点) → 交易日。None/非法值返回 None, 不伪造。"""
+    if value is None:
+        return None
+    try:
+        ms = int(value)
+    except (TypeError, ValueError):
+        return None
+    return datetime.fromtimestamp((ms + _SH_MS) // 1000, tz=UTC).date()
+
+
+def _ms_of_date(d: date) -> int:
+    """交易日 → 扶摇 start/end 入参口径的 ms(该日上海零点的 epoch ms, 不依赖本机时区)。"""
+    return (calendar.timegm(d.timetuple()) - 28_800) * 1000
+
+
+def _iso_of_ms(value) -> str | None:
+    """扶摇 *ms → ISO 日期字符串(项目财务表 period_end/announce_date 的存储口径)。"""
+    d = _date_of_ms(value)
+    return d.isoformat() if d is not None else None
+
+
+def _report_quarter(fiscal_period) -> int | None:
+    """fiscal_period(Q1..Q4/FY) → 指标接口 report 参数的季号 N(1..4)。"""
+    if not fiscal_period:
+        return None
+    text = str(fiscal_period).strip().upper()
+    if text == "FY":
+        return 4
+    try:
+        return int(text.lstrip("Q"))
+    except ValueError:
+        return None
+
+
+def _ref_price(
+    prev_close: float, dividend: float, bonus: float, allot: float, allot_price: float
+) -> float | None:
+    """交易所除权参考价: (P - D + AR·AP) / (1 + S + AR), 四舍五入(half-up)保留 2 位。
+
+    half-up 是交易所口径; 银行家舍入会让约半数事件在第 2 位小数上偏离(对拍实证)。
+    """
+    denom = 1.0 + bonus + allot
+    if denom <= 0:
+        return None
+    x = (prev_close - dividend + allot * allot_price) / denom
+    return math.floor(x * 100 + 0.5) / 100
+
+
+def _price_limit(symbol: str) -> float:
+    """按代码前缀给涨跌停幅度(自检容差用): 创业板/科创板 20%, 北交所 30%, 主板 10%。"""
+    code = symbol.split(".")[0]
+    if code.startswith(("300", "301", "688", "689")):
+        return 0.20
+    if code.startswith(("8", "4", "92")):
+        return 0.30
+    return 0.10
+
+
+def _release_of(url: str) -> str:
+    """从预签名 URL 提取 release 版本号(releases/<date>/ 路径), 提不到返回 unknown。"""
+    m = re.search(r"releases/(\d+)/", url or "")
+    return m.group(1) if m else "unknown"
+
+
+def _cache_dir() -> Path:
+    from app.config import settings
+
+    d = settings.data_dir / "cache" / "fuyao"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def _kline_rows(symbol: str, bars: list[dict]) -> list[dict]:
+    """historical/dump 原始行(价格元, volume 股) → 内部行(volume 手)。"""
+    out = []
+    for b in bars:
+        v = _to_float(b.get("volume"))
+        out.append(
+            {
+                "symbol": symbol,
+                "date": _date_of_ms(b.get("date_ms")),
+                "open": _to_float(b.get("open_price")),
+                "high": _to_float(b.get("high_price")),
+                "low": _to_float(b.get("low_price")),
+                "close": _to_float(b.get("close_price")),
+                "volume": math.floor(v / 100.0) if v is not None else None,
+                "amount": _to_float(b.get("turnover")),
+            }
+        )
+    return out
+
+
+def _tail_ok(end_d: date, covered_max: date) -> bool:
+    """请求终点是否被覆盖到 covered_max: 周末/节假日的自然缺口(≤3 天)不算缺失。"""
+    if end_d <= covered_max:
+        return True
+    return (end_d - covered_max).days <= 3 and end_d.weekday() >= 5
+
+
+def _dump_covers(dump: pl.DataFrame, start_d: date, end_d: date) -> bool:
+    """dump 日期范围是否覆盖请求窗口: 起点必须落在 dump 内; 终点允许周末自然缺口。"""
+    if dump.is_empty() or "date_ms" not in dump.columns:
+        return False
+    dates = pl.from_epoch(dump["date_ms"].cast(pl.Int64) + _SH_MS, time_unit="ms").dt.date()
+    dmin, dmax = dates.min(), dates.max()
+    return start_d >= dmin and _tail_ok(end_d, dmax)
+
+
+def _dump_date_range(path: Path) -> tuple[date | None, date | None]:
+    """lazy 读 parquet 的 date_ms 边界(走元数据/少量行组, 不整读大文件)。"""
+    row = (
+        pl.scan_parquet(path)
+        .select(
+            pl.from_epoch(pl.col("date_ms").min() + _SH_MS, time_unit="ms").dt.date().alias("dmin"),
+            pl.from_epoch(pl.col("date_ms").max() + _SH_MS, time_unit="ms").dt.date().alias("dmax"),
+        )
+        .collect()
+    )
+    return row["dmin"][0], row["dmax"][0]
+
+
 def _map_snapshot_row(row: dict, fetched_ms: int) -> dict | None:
     """扶摇快照行 → 内部 realtime record。字段缺失时按依赖推导, 不伪造数据。
 
@@ -108,6 +273,8 @@ def _map_snapshot_row(row: dict, fetched_ms: int) -> dict | None:
         # 与 quote_service 的推导同口径: 小数制, 不乘 100
         change_pct = change_amount / prev
 
+    volume = _to_float(row.get("volume"))
+
     return {
         "symbol": symbol,
         "name": row.get("name"),  # 快照无名称, 由下游维表关联
@@ -116,11 +283,11 @@ def _map_snapshot_row(row: dict, fetched_ms: int) -> dict | None:
         "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")),
+        "volume": math.floor(volume / 100.0) if volume is not None else None,  # 股 → 手
         "amount": _to_float(row.get("turnover")),
         "change_pct": change_pct,
         "change_amount": change_amount,
-        "amplitude": None,      # 快照未提供, 不启发式计算
+        "amplitude": None,  # 快照未提供, 不启发式计算
         "turnover_rate": None,  # 需股本口径 (§3.4), 交给 enriched 管道用历史股本计算
         "timestamp": fetched_ms,
         "session": None,
@@ -136,18 +303,76 @@ class FuyaoProvider:
     def __init__(self) -> None:
         self.config = _FuyaoConfig()
         self._client: FuyaoClient | None = None
+        self._dump_memo: dict[str, pl.DataFrame] = {}
+        self._dump_path_memo: dict[str, Path] = {}
 
     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
+        self._dump_memo.clear()
+        self._dump_path_memo.clear()
 
     def _get_client(self) -> FuyaoClient:
         if self._client is None:
             self._client = fuyao_client.FuyaoClient(api_key=get_api_key())
         return self._client
 
+    # ---- dump 缓存 ----
+    def _ensure_dump_path(self, dump_kind: str, cache_prefix: str) -> Path:
+        """确保最新 release 的 dump 已落盘, 返回缓存路径(大文件不整读进内存)。
+
+        release 号取自预签名 URL 的 releases/<date>/ 路径; 新 release 落盘后清理旧版缓存。
+        """
+        memo = self._dump_path_memo.get(dump_kind)
+        if memo is not None and memo.exists():
+            return memo
+        client = self._get_client()
+        info = client.dump_download_url(dump_kind)
+        release = _release_of(str(info.get("presigned_url") or ""))
+        dest = _cache_dir() / f"{cache_prefix}__{release}.parquet"
+        if not dest.exists():
+            client.download_dump(dump_kind, dest)
+            for old in dest.parent.glob(f"{cache_prefix}__*.parquet"):
+                if old.name != dest.name:
+                    old.unlink(missing_ok=True)
+            logger.info("扶摇 dump %s(release %s)已下载: %s", dump_kind, release, dest.name)
+        self._dump_path_memo[dump_kind] = dest
+        return dest
+
+    def _ensure_dump(self, dump_kind: str, cache_prefix: str) -> pl.DataFrame:
+        """小体量 dump(快照 10d / 因子)整读 + 进程内 memo, 避免重复打接口/读盘。"""
+        memo = self._dump_memo.get(dump_kind)
+        if memo is not None:
+            return memo
+        df = pl.read_parquet(self._ensure_dump_path(dump_kind, cache_prefix))
+        self._dump_memo[dump_kind] = df
+        return df
+
+    def _ensure_daily_big_dump(self, start_d: date) -> Path | None:
+        """10 年全量日K dump(约 172MB)。只要求覆盖窗口起点; 末端缺口由 10d dump 补。
+
+        已有缓存覆盖起点就直接复用, 不追新 release(避免深窗口高频触发时日日重下
+        172MB — 旧 release 的中段历史不会变, 尾部新鲜度交给 1MB 的 10d dump)。
+        """
+        for f in sorted(_cache_dir().glob("daily_k__*.parquet"), reverse=True):
+            try:
+                dmin, _ = _dump_date_range(f)
+            except Exception:  # 缓存损坏不致命, 换下一个/重拉
+                continue
+            if dmin is not None and start_d >= dmin:
+                return f
+        try:
+            path = self._ensure_dump_path(_DAILY_DUMP_KIND, "daily_k")
+        except FuyaoError as e:
+            logger.warning("扶摇 10 年 dump 不可用, 回退单标的接口: %s", e)
+            return None
+        dmin, _ = _dump_date_range(path)
+        if dmin is None or start_d < dmin:
+            return None  # 窗口比 10 年更早 → 单标的兜底
+        return path
+
     # ---- realtime ----
     def get_realtime(self) -> list[dict]:
         """全市场实时快照 → 内部 realtime records。失败软返回空列表(不阻断轮询)。"""
@@ -175,11 +400,636 @@ class FuyaoProvider:
         logger.info("扶摇实时行情拉取完成: %d 条(丢弃 %d 行)", len(records), dropped)
         return records
 
+    # ---- daily ----
+    def get_daily(
+        self,
+        symbols: list[str],
+        start_time: datetime | None,
+        end_time: datetime | None,
+        asset_type: str = "stock",
+        on_chunk_done: Callable[[int, int], None] | None = None,
+    ) -> pl.DataFrame:
+        """A 股日K → 内部契约。原始价(adjust=none 锁定)、volume 股→手(floor /100)。
+
+        取数三档(逐级降级, 全部失败才空手而归):
+        - 近端窗口(跨度 ≤12 天): daily-k-10d dump(约 1MB), 1 次请求覆盖全部标的;
+        - 深窗口: daily-k 10 年全量 dump(约 172MB, 缓存覆盖起点即复用, 不追新 release),
+          末端缺口由 10d dump 补尾 — 全市场深回填从"逐票 34 分钟"降为"一次下载+秒级筛选";
+        - 兜底: 单标的 historical 接口(窗口早于 dump 覆盖 / dump 不可用; 10 年自动分片,
+          逐标的节流 + 进度回调)。
+        """
+        if not symbols or asset_type != "stock":
+            return pl.DataFrame()
+        end_dt = end_time or datetime.now()
+        start_dt = start_time or (end_dt - timedelta(days=365))
+        start_d, end_d = start_dt.date(), end_dt.date()
+
+        if (end_d - start_d).days <= _RECENT_DUMP_DAYS:
+            try:
+                dump = self._ensure_dump(_DAILY10_DUMP_KIND, "daily_k_10d")
+                if _dump_covers(dump, start_d, end_d):
+                    df = self._daily_from_dump(dump, set(symbols), start_d, end_d)
+                    if on_chunk_done:
+                        on_chunk_done(1, 1)
+                    logger.info("扶摇日K(10d dump)完成: %d 行 [%s ~ %s]", df.height, start_d, end_d)
+                    return df
+                logger.info("扶摇 10d dump 未覆盖窗口 [%s ~ %s], 尝试 10 年 dump", start_d, end_d)
+            except FuyaoError as e:
+                logger.warning("扶摇日K 10d dump 不可用, 尝试 10 年 dump: %s", e)
+
+        df = self._daily_from_big_dump(set(symbols), start_d, end_d)
+        if df is not None:
+            if on_chunk_done:
+                on_chunk_done(1, 1)
+            logger.info("扶摇日K(10 年 dump)完成: %d 行 [%s ~ %s]", df.height, start_d, end_d)
+            return df
+        df = self._daily_from_api(symbols, start_d, end_d, on_chunk_done)
+        logger.info("扶摇日K(单标的接口)完成: %d 行 [%s ~ %s]", df.height, start_d, end_d)
+        return df
+
+    def _daily_from_dump(
+        self, dump: pl.DataFrame, symset: set[str], start_d: date, end_d: date
+    ) -> pl.DataFrame:
+        df = dump.with_columns(
+            pl.from_epoch(pl.col("date_ms") + _SH_MS, time_unit="ms").dt.date().alias("date")
+        )
+        return self._map_daily_dump(df, symset, start_d, end_d)
+
+    def _map_daily_dump(
+        self, df: pl.DataFrame, symset: set[str], start_d: date, end_d: date
+    ) -> pl.DataFrame:
+        """dump 原始行 → 内部日K契约(股→手、字段重命名、停牌过滤)。"""
+        if "adjusted" in df.columns:
+            df = df.filter(pl.col("adjusted") == "none")  # 防御: dump 变为复权口径时拒绝落库
+        out = (
+            df.filter(
+                (pl.col("date") >= start_d)
+                & (pl.col("date") <= end_d)
+                & pl.col("thscode").is_in(sorted(symset))
+            )
+            .with_columns(
+                (pl.col("volume") / 100.0).floor().alias("volume"),  # 股 → 手
+                pl.col("thscode").alias("symbol"),
+            )
+            .rename(
+                {
+                    "open_price": "open",
+                    "high_price": "high",
+                    "low_price": "low",
+                    "close_price": "close",
+                    "turnover": "amount",
+                }
+            )
+        )
+        out = filter_halt_days(out)
+        cols = [c for c in DAILY_COLS if c in out.columns]
+        return out.select(cols).sort(["symbol", "date"]) if not out.is_empty() else out.select(cols)
+
+    def _daily_from_big_dump(
+        self, symset: set[str], start_d: date, end_d: date
+    ) -> pl.DataFrame | None:
+        """深窗口主路径: 10 年全量 dump(lazy 按需筛) + 必要时 10d dump 补尾。
+
+        覆盖不了(窗口早于 10 年 / dump 拉取失败)返回 None, 由调用方走单标的接口。
+        """
+        path = self._ensure_daily_big_dump(start_d)
+        if path is None:
+            return None
+        _, dmax = _dump_date_range(path)
+        big_hi = min(end_d, dmax)
+        # 窗口/标的过滤下推到 lazy 计划, 只物化需要的行(全量 10 年 ≈ 13.6M 行)
+        window = (
+            pl.scan_parquet(path)
+            .with_columns(
+                pl.from_epoch(pl.col("date_ms") + _SH_MS, time_unit="ms").dt.date().alias("date")
+            )
+            .filter(
+                (pl.col("date") >= start_d)
+                & (pl.col("date") <= big_hi)
+                & pl.col("thscode").is_in(sorted(symset))
+            )
+            .collect()
+        )
+        parts = [self._map_daily_dump(window, symset, start_d, big_hi)]
+        if not _tail_ok(end_d, dmax):
+            # 末端缺口(如 10 年 dump 是旧 release, end 是最近交易日): 10d dump 补尾
+            try:
+                ten = self._ensure_dump(_DAILY10_DUMP_KIND, "daily_k_10d")
+                ten_dates = pl.from_epoch(
+                    ten["date_ms"].cast(pl.Int64) + _SH_MS, time_unit="ms"
+                ).dt.date()
+                ten_min, ten_max = ten_dates.min(), ten_dates.max()
+                if (
+                    ten_min is not None
+                    and ten_min <= dmax + timedelta(days=1)
+                    and _tail_ok(end_d, ten_max)
+                ):
+                    tail_start = max(start_d, dmax + timedelta(days=1))
+                    parts.append(self._daily_from_dump(ten, symset, tail_start, end_d))
+                else:
+                    return None  # 中段或尾部仍有缺口 → 单标的兜底, 不交缺口数据
+            except FuyaoError as e:
+                logger.warning("扶摇 10d dump 补尾失败: %s", e)
+                return None
+        non_empty = [p for p in parts if not p.is_empty()]
+        if not non_empty:
+            return pl.DataFrame()
+        out = pl.concat(non_empty, how="vertical_relaxed")
+        return out.unique(subset=["symbol", "date"], keep="last").sort(["symbol", "date"])
+
+    def _daily_from_api(
+        self,
+        symbols: list[str],
+        start_d: date,
+        end_d: date,
+        on_chunk_done: Callable[[int, int], None] | None,
+    ) -> pl.DataFrame:
+        frames: list[pl.DataFrame] = []
+        for i, sym in enumerate(symbols):
+            rows = self._historical_bars(sym, start_d, end_d)
+            time.sleep(_HIST_INTERVAL_S)
+            if rows:
+                df = normalize_daily(_kline_rows(sym, rows), default_symbol=sym, source=self.name)
+                if not df.is_empty():
+                    frames.append(df)
+            if on_chunk_done:
+                on_chunk_done(i + 1, len(symbols))
+        return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame()
+
+    def _historical_bars(self, symbol: str, start_d: date, end_d: date) -> list[dict]:
+        """按 ≤10 年窗口分片拉取单标的原始日K。中途失败软返回已得行, 不抛出。"""
+        out: list[dict] = []
+        s = _ms_of_date(start_d)
+        e = _ms_of_date(end_d)
+        while s <= e:
+            chunk_end = min(e, s + _HIST_MAX_SPAN_MS - 1)
+            try:
+                out.extend(self._get_client().historical_kline(symbol, s, chunk_end, adjust="none"))
+            except FuyaoError as err:
+                logger.warning("扶摇日K拉取失败 %s [%s ~ %s]: %s", symbol, start_d, end_d, err)
+                break
+            if s + _HIST_MAX_SPAN_MS <= e:
+                time.sleep(_HIST_INTERVAL_S)
+            s = chunk_end + 1
+        return out
+
+    # ---- adj_factor ----
+    def get_adj_factors(
+        self,
+        symbols: list[str],
+        start_time: datetime | None,
+        end_time: datetime | None,
+        asset_type: str = "stock",
+        on_chunk_done: Callable[[int, int], None] | None = None,
+    ) -> pl.DataFrame:
+        """A 股除权因子 → 内部契约(symbol/trade_date/ex_factor, 单事件比值非累积)。
+
+        数据链: adjustment-factors 全量事件 dump → 清洗(时区 +8h / 滤全零 / 同日成分合并)
+        → 前收盘价从本地日K dump 一次取齐(缺价标的回退单标的接口) → 交易所公式推导
+        → 涨跌停自检剔除异常。
+        未来已公告事件无前收盘, 留给滚动增量窗口(15 天)自然补上。
+        """
+        schema = {"symbol": pl.String, "trade_date": pl.Date, "ex_factor": pl.Float64}
+        if not symbols or asset_type != "stock":
+            return pl.DataFrame(schema=schema)
+        try:
+            events = self._load_adj_events(set(symbols), start_time, end_time)
+        except FuyaoError as e:
+            logger.warning("扶摇除权因子 dump 加载失败: %s", e)
+            return pl.DataFrame(schema=schema)
+        if events.is_empty():
+            return pl.DataFrame(schema=schema)
+
+        out_rows: list[dict] = []
+        syms = sorted(events["symbol"].unique().to_list())
+        # 前收盘价优先从本地日K dump 一次取齐(全市场配价从逐标的 ~13 分钟降为秒级),
+        # 大 dump 不可用或个别标的缺价时回退单标的接口(节流保留)。
+        bounds = events.group_by("symbol").agg(
+            pl.col("ex_date").min().alias("first_ex"),
+            pl.col("ex_date").max().alias("last_ex"),
+        )
+        closes_by_sym = self._closes_from_dumps(bounds)
+        fallbacks = 0
+        for i, sym in enumerate(syms):
+            evs = events.filter(pl.col("symbol") == sym).sort("ex_date")
+            first_ex: date = evs["ex_date"][0]
+            last_ex: date = evs["ex_date"][-1]
+            closes = (closes_by_sym or {}).get(sym)
+            if not closes or min(closes) > first_ex:
+                # dump 无该标的 / 覆盖不到首个事件前 → 单标的接口兜底
+                closes = self._fetch_closes(
+                    sym, first_ex - timedelta(days=_PREV_CLOSE_BACKDAYS), last_ex
+                )
+                time.sleep(_HIST_INTERVAL_S)
+                fallbacks += 1
+            if not closes:
+                logger.warning("扶摇除权因子: %s 原始日K为空, 跳过其 %d 个事件", sym, evs.height)
+                continue
+            days = sorted(closes)
+            for ev in evs.iter_rows(named=True):
+                exd: date = ev["ex_date"]
+                prev_days = [d for d in days if d < exd]
+                if not prev_days:
+                    continue  # 日K窗口未覆盖的远古事件(如 10 年分片边界之外)
+                p = closes[prev_days[-1]]
+                ref = _ref_price(p, ev["dividend"], ev["bonus"], ev["allot"], ev["allot_price"])
+                if ref is None or ref <= 0:
+                    logger.warning(
+                        "扶摇除权因子: %s %s 参考价无法计算(P=%s D=%s S=%s AR=%s AP=%s), 跳过",
+                        sym,
+                        exd,
+                        p,
+                        ev["dividend"],
+                        ev["bonus"],
+                        ev["allot"],
+                        ev["allot_price"],
+                    )
+                    continue
+                factor = p / ref
+                ex_days = [d for d in days if d >= exd]
+                if ex_days:  # 涨跌停自检: 错误因子会令复权后除权日涨跌幅超出板限
+                    ret = closes[ex_days[0]] / (p / factor) - 1.0
+                    if abs(ret) > _price_limit(sym) + 0.02:
+                        logger.warning(
+                            "扶摇除权因子自检剔除 %s %s: 复权后除权日涨跌幅 %.1f%% 超出涨跌停",
+                            sym,
+                            exd,
+                            ret * 100,
+                        )
+                        continue
+                out_rows.append({"symbol": sym, "trade_date": exd, "ex_factor": factor})
+            if on_chunk_done:
+                on_chunk_done(i + 1, len(syms))
+        if closes_by_sym is not None:
+            logger.info(
+                "扶摇除权因子: 本地 dump 配价 %d/%d 标的, 回退接口 %d 标的",
+                len(closes_by_sym),
+                len(syms),
+                fallbacks,
+            )
+        if not out_rows:
+            return pl.DataFrame(schema=schema)
+        return (
+            pl.DataFrame(out_rows, schema=schema)
+            .unique(subset=["symbol", "trade_date"], keep="last")
+            .sort(["symbol", "trade_date"])
+        )
+
+    def _load_adj_events(
+        self,
+        symset: set[str],
+        start_time: datetime | None,
+        end_time: datetime | None,
+    ) -> pl.DataFrame:
+        """事件 dump → 清洗 → 窗口过滤。返回 symbol/ex_date/dividend/bonus/allot/allot_price。
+
+        清洗规则(对拍实证, 见 2026-08 验证记录):
+        - ex_date_ms 为上海零点戳, +8h 转日期;
+        - 全零事件行(疑似特殊事件, dump 未给成分)过滤;
+        - 同日拆行(如分红/送转各一行)按成分合并后推导, 顺序不可反;
+        - 配股但配股价缺失 → 无法推导, 过滤。
+        """
+        today = datetime.now().date()
+        df = self._ensure_dump(_ADJ_DUMP_KIND, "adj_factors").rename({"thscode": "symbol"})
+        df = (
+            df.with_columns(
+                pl.from_epoch(pl.col("ex_date_ms") + _SH_MS, time_unit="ms")
+                .dt.date()
+                .alias("ex_date"),
+                pl.col("dividend_per_share").fill_null(0.0),
+                pl.col("per_share_bonus").fill_null(0.0),
+                pl.col("allotment_ratio").fill_null(0.0),
+                pl.col("allotment_price").fill_null(0.0),
+            )
+            .filter(
+                (pl.col("ex_date") <= today)  # 未来已公告事件无前收盘, 留给滚动增量
+                & (
+                    (pl.col("dividend_per_share") != 0)
+                    | (pl.col("per_share_bonus") != 0)
+                    | (pl.col("allotment_ratio") != 0)
+                )
+            )
+            .group_by("symbol", "ex_date")
+            .agg(
+                pl.col("dividend_per_share").sum().alias("dividend"),
+                pl.col("per_share_bonus").sum().alias("bonus"),
+                pl.col("allotment_ratio").sum().alias("allot"),
+                pl.col("allotment_price").max().alias("allot_price"),  # 同日拆行共享配股价
+            )
+            .filter(~((pl.col("allot") > 0) & (pl.col("allot_price") <= 0)))
+            .filter(pl.col("symbol").is_in(sorted(symset)))
+        )
+        if start_time is not None:
+            df = df.filter(pl.col("ex_date") >= start_time.date())
+        if end_time is not None:
+            df = df.filter(pl.col("ex_date") <= end_time.date())
+        return df.select("symbol", "ex_date", "dividend", "bonus", "allot", "allot_price")
+
+    # ---- financial ----
+    # 字段映射: 扶摇原始字段 → 项目 canonical 列名(TickFlow 口径, 前端财务页与回测
+    # FUNDAMENTAL_FACTORS 按此消费)。映射表之外的扶摇独有字段以原名透传为扩展列。
+    _INCOME_FIELD_MAP = {
+        "operating_income": "revenue",
+        "operating_costs": "operating_cost",
+        "sales_fee": "selling_expense",
+        "manage_fee": "admin_expense",
+        "research_and_development_expenses": "rd_expense",
+        "operating_profit": "operating_profit",
+        "interest_expenses": "financial_expense",  # 近似口径: 扶摇只给利息费用
+        "profit_total": "total_profit",
+        "income_tax_expense": "income_tax",
+        "net_profit": "net_income",
+        "parent_holder_net_profit": "net_income_attributable",
+        "basic_eps": "basic_eps",
+    }
+    _BALANCE_FIELD_MAP = {
+        "assets_total": "total_assets",
+        "total_current_assets": "total_current_assets",
+        "non_current_nets_total": "total_non_current_assets",
+        "cash": "cash_and_equivalents",
+        "accounts_receivable": "accounts_receivable",
+        "total_debt": "total_liabilities",
+        "holder_equity_total": "total_equity",
+    }
+    _CASHFLOW_FIELD_MAP = {
+        "act_cash_flow_net": "net_operating_cash_flow",
+        "invest_cash_flow_net": "net_investing_cash_flow",
+        "financing_cash_flow_net": "net_financing_cash_flow",
+        "pay_fixed_assets_etc_cash": "capex",
+        "cash_equivalents_net_addition": "net_cash_change",
+    }
+    # 官方指标 index_id → canonical。归母净利同比近似 tickflow net_income_yoy;
+    # 实测 index_id 与文档有出入(calculate_ 前缀等), 以实测为准。
+    _METRICS_FIELD_MAP = {
+        "index_weighted_avg_roe": "roe",
+        "total_assets_net_ratio": "roa",
+        "sale_gross_margin": "gross_margin",
+        "sale_net_interest_ratio": "net_margin",
+        "assets_debt_ratio": "debt_to_asset_ratio",
+        "calculate_operating_income_yoy_growth_ratio": "revenue_yoy",
+        "calculate_parent_holder_net_profit_yoy_growth_ratio": "net_income_yoy",
+        "operating_cash_flow_net_divide_income": "operating_cash_to_revenue",
+        "inventory_turnover_ratio": "inventory_turnover",
+    }
+
+    def get_financials(
+        self,
+        table: str,
+        symbols: list[str],
+        latest_only: bool = True,
+    ) -> pl.DataFrame:
+        """拉取财务数据, 映射为 canonical 列(symbol/period_end/announce_date/指标)。
+
+        - 三大报表: 单股单请求, latest_only 决定最近 1 期还是 8 期季报;
+        - metrics: 指标接口为单股单期, 恒只拉最新一期(bps 由估值快照 pb_mrq 反推,
+          eps_basic 顺带取自利润表); 历史各期建议切回 TickFlow 同步补齐 —
+          报告期合并写入会让两源数据共存, 互不覆盖;
+        - shares: 扶摇无股本接口, 恒返回空(已有存量靠合并写入保留)。
+        """
+        if table == "shares":
+            logger.info("扶摇无股本接口, shares 表跳过 (已有数据保留)")
+            return pl.DataFrame()
+        if table in _STATEMENT_ENDPOINTS:
+            field_map = {
+                "income": self._INCOME_FIELD_MAP,
+                "balance_sheet": self._BALANCE_FIELD_MAP,
+                "cash_flow": self._CASHFLOW_FIELD_MAP,
+            }[table]
+            return self._financial_statements(table, field_map, symbols, latest_only)
+        if table == "metrics":
+            return self._financial_metrics(symbols)
+        return pl.DataFrame()
+
+    def _financial_statements(
+        self,
+        stmt: str,
+        field_map: dict[str, str],
+        symbols: list[str],
+        latest_only: bool,
+    ) -> pl.DataFrame:
+        client = self._get_client()
+        limit = 1 if latest_only else _FINANCIAL_HISTORY_PERIODS
+        rows_out: list[dict] = []
+        for i, sym in enumerate(symbols):
+            if i:
+                time.sleep(_HIST_INTERVAL_S)
+            try:
+                rows = client.financial_statements(stmt, sym, limit=limit)
+            except FuyaoError as e:
+                logger.warning("扶摇财务 %s %s 失败: %s", stmt, sym, e)
+                continue
+            for r in rows:
+                row: dict = {
+                    "symbol": sym,
+                    "period_end": _iso_of_ms(r.get("period_end_ms")),
+                    "announce_date": _iso_of_ms(r.get("report_date_ms")),
+                }
+                for src, dst in field_map.items():
+                    row[dst] = _to_float(r.get(src))
+                # 扶摇独有字段以原名透传为扩展列 (canonical 之外的增量信息)
+                for src, value in r.items():
+                    if src not in field_map and src not in row and isinstance(value, (int, float)):
+                        row[src] = value
+                rows_out.append(row)
+        return pl.DataFrame(rows_out) if rows_out else pl.DataFrame()
+
+    def _financial_metrics(self, symbols: list[str]) -> pl.DataFrame:
+        client = self._get_client()
+        # 指标接口按 report(yyyy-N) 单期查询 → 先用利润表 limit=1 反查每股最新披露期
+        latest: dict[str, dict] = {}
+        for i, sym in enumerate(symbols):
+            if i:
+                time.sleep(_HIST_INTERVAL_S)
+            try:
+                rows = client.financial_statements("income", sym, limit=1)
+            except FuyaoError as e:
+                logger.warning("扶摇财务 income %s 失败: %s", sym, e)
+                continue
+            if rows:
+                latest[sym] = rows[0]
+        if not latest:
+            return pl.DataFrame()
+        bps_by_sym = self._derive_bps(sorted(latest))
+        rows_out: list[dict] = []
+        for sym, r in latest.items():
+            quarter = _report_quarter(r.get("fiscal_period"))
+            report = f"{r.get('fiscal_year')}-{quarter}" if quarter else None
+            row: dict = {
+                "symbol": sym,
+                "period_end": _iso_of_ms(r.get("period_end_ms")),
+                "announce_date": _iso_of_ms(r.get("report_date_ms")),
+                "eps_basic": _to_float(r.get("basic_eps")),
+                "bps": bps_by_sym.get(sym),
+            }
+            if report:
+                try:
+                    abilities = client.financial_indicators(sym, report)
+                except FuyaoError as e:
+                    logger.warning("扶摇指标 %s %s 失败: %s", sym, report, e)
+                    abilities = []
+                for ability in abilities:
+                    for ind in ability.get("indicators") or []:
+                        index_id = ind.get("index_id")
+                        if not index_id:
+                            continue
+                        value = _to_float(ind.get("value"))
+                        if value is not None:
+                            row[self._METRICS_FIELD_MAP.get(index_id, index_id)] = value
+            rows_out.append(row)
+        return pl.DataFrame(rows_out) if rows_out else pl.DataFrame()
+
+    def _derive_bps(self, symbols: list[str]) -> dict[str, float]:
+        """估值快照 pb_mrq 与行情快照最新价同源同刻 → bps = price / pb_mrq。
+
+        与财报口径 bps 可能差几个百分点(上游权益基准不完全透明), 用于补齐
+        metrics.bps 使回测 pb_latest 因子可用。接口失败只影响 bps 列, 不致命。
+        """
+        client = self._get_client()
+        pb: dict[str, float] = {}
+        price: dict[str, float] = {}
+        for i in range(0, len(symbols), _VALUATION_BATCH):
+            if i:
+                time.sleep(_HIST_INTERVAL_S)
+            batch = symbols[i : i + _VALUATION_BATCH]
+            for fetch, store in (
+                (client.valuations_snapshot, pb),
+                (client.price_snapshot_batch, price),
+            ):
+                time.sleep(_HIST_INTERVAL_S)
+                try:
+                    for r in fetch(batch):
+                        value = _to_float(r.get("pb_mrq" if store is pb else "last_price"))
+                        code = r.get("thscode")
+                        if code and value is not None and value != 0:
+                            store[code] = value
+                except FuyaoError as e:
+                    logger.warning("扶摇 bps 推导快照失败(%d 只): %s", len(batch), e)
+        return {
+            sym: price[sym] / pb[sym]
+            for sym in symbols
+            if sym in pb and pb[sym] and sym in price
+        }
+
+    def _fetch_closes(self, symbol: str, start_d: date, end_d: date) -> dict[date, float]:
+        rows = self._historical_bars(symbol, start_d, end_d)
+        out: dict[date, float] = {}
+        for r in rows:
+            d = _date_of_ms(r.get("date_ms"))
+            c = _to_float(r.get("close_price"))
+            if d is not None and c is not None:
+                out[d] = c
+        return out
+
+    def _closes_from_dumps(
+        self, bounds: pl.DataFrame
+    ) -> dict[str, dict[date, float]] | None:
+        """从本地日K dump 一次取齐全部事件标的的收盘价。
+
+        每标的开窗 [first_ex-30d, last_ex](与单标的接口同窗): 10 年大 dump 为
+        主体, 10d dump 叠加补末端新鲜度(除权日当天的自检需要 ex 日收盘)。
+        返回 symbol → {date: close}; 大 dump 不可用/窗口早于其覆盖/读盘失败
+        → None, 由调用方整轮回退单标的接口。
+        """
+        try:
+            lo = bounds["first_ex"].min() - timedelta(days=_PREV_CLOSE_BACKDAYS)
+            path = self._ensure_daily_big_dump(lo)
+            if path is None:
+                return None
+            sources = [self._closes_scan(pl.scan_parquet(path))]
+            ten = self._dump_memo.get(_DAILY10_DUMP_KIND)
+            if ten is not None:
+                sources.append(self._closes_scan(ten.lazy()))
+            else:
+                try:
+                    p10 = self._ensure_dump_path(_DAILY10_DUMP_KIND, "daily_k_10d")
+                except FuyaoError as e:
+                    logger.info("扶摇 10d dump 不可用, 配价仅用 10 年 dump: %s", e)
+                else:
+                    sources.append(self._closes_scan(pl.scan_parquet(p10)))
+            win = bounds.select(
+                "symbol",
+                (pl.col("first_ex") - pl.duration(days=_PREV_CLOSE_BACKDAYS)).alias("lo"),
+                "last_ex",
+            )
+            # concat 顺序 = 叠加优先级: 同 (symbol, date) 时 10d dump(更新鲜)覆盖大 dump
+            df = (
+                pl.concat(sources, how="vertical_relaxed")
+                .join(win.lazy(), on="symbol", how="inner")
+                .filter(
+                    (pl.col("date") >= pl.col("lo")) & (pl.col("date") <= pl.col("last_ex"))
+                )
+                .unique(subset=["symbol", "date"], keep="last", maintain_order=True)
+                .collect()
+            )
+            return {
+                f["symbol"][0]: dict(
+                    zip(f["date"].to_list(), f["close"].to_list(), strict=True)
+                )
+                for f in df.partition_by("symbol")
+            }
+        except Exception as e:  # 缓存损坏等不致命: 回退逐标的接口
+            logger.warning("扶摇除权因子本地配价失败, 回退单标的接口: %s", e)
+            return None
+
+    @staticmethod
+    def _closes_scan(lf: pl.LazyFrame) -> pl.LazyFrame:
+        """dump 行 → (symbol, date, close) lazy 投影; adjusted 列存在时锁 none。"""
+        if "adjusted" in lf.collect_schema().names():
+            lf = lf.filter(pl.col("adjusted") == "none")
+        return lf.select(
+            pl.col("thscode").alias("symbol"),
+            pl.from_epoch(pl.col("date_ms").cast(pl.Int64) + _SH_MS, time_unit="ms")
+            .dt.date()
+            .alias("date"),
+            pl.col("close_price").cast(pl.Float64).alias("close"),
+        ).filter(pl.col("close").is_not_null())
+
     # ---- 测试(设置页试拉) ----
     def test_dataset(self, dataset: str, symbols: list[str] | None = None) -> dict:
+        if dataset in ("daily", "adj_factor"):
+            syms = [s for s in (symbols or [])][:3] or ["000001.SZ"]
+            try:
+                if dataset == "daily":
+                    df = self.get_daily(syms, datetime.now() - timedelta(days=30), datetime.now())
+                else:
+                    df = self.get_adj_factors(
+                        syms, datetime.now() - timedelta(days=365), datetime.now()
+                    )
+            except FuyaoError as e:
+                return {"provider": self.name, "dataset": dataset, "rows": 0, "error": str(e)}
+            head = df.head(5).to_dicts()
+            for row in head:  # date/datetime → ISO 字符串, 保证 JSON 可序列化
+                for k, v in list(row.items()):
+                    if isinstance(v, (date, datetime)):
+                        row[k] = v.isoformat()
+            return {
+                "provider": self.name,
+                "dataset": dataset,
+                "rows": df.height,
+                "columns": df.columns,
+                "preview": head,
+            }
+        if dataset == "financial":
+            syms = [s for s in (symbols or [])][:1] or ["600519.SH"]
+            try:
+                df = self.get_financials("metrics", syms, latest_only=True)
+            except FuyaoError as e:
+                return {"provider": self.name, "dataset": dataset, "rows": 0, "error": str(e)}
+            head = df.head(5).to_dicts()
+            return {
+                "provider": self.name,
+                "dataset": dataset,
+                "rows": df.height,
+                "columns": df.columns,
+                "preview": head,
+            }
         if dataset != "realtime":
-            return {"provider": self.name, "dataset": dataset, "rows": 0,
-                    "error": f"扶摇插件未接入 {dataset} 数据集(自动回退 TickFlow)"}
+            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:

backend/app/services/financial_sync.py

@@ -155,6 +155,13 @@ def _sync_table(
 
 
 def _merge_report_history(*frames: pl.DataFrame) -> pl.DataFrame:
+    """按 (symbol, period_end) 合并各报告期, 同期多行逐列取最新非空值。
+
+    语义(区分"覆盖"与"填空"): 每列独立取 announce_date 最新的非空值 —
+    新同步行有值则覆盖旧值, 新行缺的列(如 fuyao 不提供的字段)由旧行补齐,
+    实现多数据源并集共存。历史报告期不可变, 合并不会引入过期数据。
+    无 announce_date 的帧按输入顺序, 后写优先(与旧行为 keep="last" 一致)。
+    """
     valid = [
         frame
         for frame in frames
@@ -166,11 +173,15 @@ def _merge_report_history(*frames: pl.DataFrame) -> pl.DataFrame:
         pl.concat(valid, how="diagonal_relaxed")
         .filter(pl.col("symbol").is_not_null() & pl.col("period_end").is_not_null())
     )
-    # 同一 (symbol, period_end) 多条时保留 announce_date 最新一条 (业绩修正以最新公告为准)。
-    if "announce_date" in merged.columns:
-        merged = merged.sort(["symbol", "period_end", "announce_date"], nulls_last=True)
-    return merged.unique(subset=["symbol", "period_end"], keep="last").sort(
-        ["symbol", "period_end"]
+    sort_keys = ["symbol", "period_end"] + (
+        ["announce_date"] if "announce_date" in merged.columns else []
+    )
+    merged = merged.sort(sort_keys, nulls_last=True)
+    value_cols = [c for c in merged.columns if c not in ("symbol", "period_end")]
+    return (
+        merged.group_by("symbol", "period_end")
+        .agg([pl.col(c).drop_nulls().last() for c in value_cols])
+        .sort(["symbol", "period_end"])
     )
 
 

backend/tests/test_fuyao_financial.py

@@ -0,0 +1,242 @@
+"""fuyao 财务适配测试 (不依赖真实网络)。
+
+覆盖: 三大报表字段映射 (canonical 列名 + 扩展列透传 + ISO 日期口径)、
+latest_only 分档 (limit 1 vs 8)、metrics 组装 (eps_basic 顺带 / bps 估值反推 /
+指标 index_id 映射与未知 id 透传 / 单股指标失败不弃行)、shares 恒空、
+报告期合并写入的逐列填空语义 (并集共存, 新行缺列不覆盖旧值)。
+"""
+
+from __future__ import annotations
+
+import polars as pl
+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
+from app.services.financial_sync import _merge_report_history
+
+
+class _FakeFinClient:
+    """财务端点假客户端: 记录调用入参, 按表返回预置行。"""
+
+    def __init__(
+        self,
+        statements: dict[str, list[dict]] | None = None,
+        indicators: dict[str, list[dict]] | None = None,
+        indicator_error: Exception | None = None,
+        valuations: list[dict] | None = None,
+        prices: list[dict] | None = None,
+    ):
+        self.statements = statements or {}
+        self.indicators = indicators or {}
+        self.indicator_error = indicator_error
+        self.valuations = valuations or []
+        self.prices = prices or []
+        self.stmt_calls: list[tuple] = []
+        self.ind_calls: list[str] = []
+
+    def financial_statements(self, stmt, thscode, limit=1):
+        self.stmt_calls.append((stmt, thscode, limit))
+        return [dict(r, thscode=thscode) for r in self.statements.get(stmt, [])]
+
+    def financial_indicators(self, thscode, report):
+        self.ind_calls.append(f"{thscode}@{report}")
+        if self.indicator_error:
+            raise self.indicator_error
+        return self.indicators.get(report, [])
+
+    def valuations_snapshot(self, thscodes):
+        return [r for r in self.valuations if r.get("thscode") in thscodes]
+
+    def price_snapshot_batch(self, thscodes):
+        return [r for r in self.prices if r.get("thscode") in thscodes]
+
+
+def _provider_with(monkeypatch, fake):
+    monkeypatch.setattr(
+        fp, "fuyao_client", type("M", (), {"FuyaoClient": lambda **kw: fake})
+    )
+    monkeypatch.setattr(fp, "get_api_key", lambda: "test-key")
+    monkeypatch.setattr(fp, "_HIST_INTERVAL_S", 0)
+    return FuyaoProvider()
+
+
+# period_end_ms: 2026-06-30 上海零点; report_date_ms: 2026-08-15 上海零点
+_INCOME_ROW = {
+    "period": "quarterly",
+    "fiscal_year": 2026,
+    "fiscal_period": "Q2",
+    "report_date_ms": 1786723200000,
+    "period_end_ms": 1782748800000,
+    "currency": "CNY",
+    "operating_income": 90703260964.48,
+    "net_profit": 46033330566.78,
+    "parent_holder_net_profit": 44516880421.86,
+    "basic_eps": 35.57,
+    "operating_expenses": 50000000000.0,  # 扶摇独有 → 扩展列
+}
+
+
+def test_income_mapping_canonical_columns(monkeypatch):
+    fake = _FakeFinClient(statements={"income": [_INCOME_ROW]})
+    provider = _provider_with(monkeypatch, fake)
+    df = provider.get_financials("income", ["600519.SH"], latest_only=True)
+    row = df.to_dicts()[0]
+    assert row["symbol"] == "600519.SH"
+    assert row["period_end"] == "2026-06-30"
+    assert row["announce_date"] == "2026-08-15"
+    assert row["revenue"] == pytest.approx(_INCOME_ROW["operating_income"])
+    assert row["net_income"] == pytest.approx(_INCOME_ROW["net_profit"])
+    assert row["net_income_attributable"] == pytest.approx(
+        _INCOME_ROW["parent_holder_net_profit"]
+    )
+    assert row["basic_eps"] == 35.57
+    # 扩展列: 扶摇独有数字字段原名透传; 字符串元数据不透传
+    assert row["operating_expenses"] == 50000000000.0
+    assert "thscode" not in df.columns and "ticker" not in df.columns
+    # 原始名不残留 (已映射字段)
+    assert "operating_income" not in df.columns and "net_profit" not in df.columns
+
+
+def test_statements_limit_latest_vs_history(monkeypatch):
+    fake = _FakeFinClient(statements={"income": [_INCOME_ROW]})
+    provider = _provider_with(monkeypatch, fake)
+    provider.get_financials("income", ["600519.SH"], latest_only=True)
+    assert fake.stmt_calls == [("income", "600519.SH", 1)]
+    provider.get_financials("income", ["600519.SH"], latest_only=False)
+    assert fake.stmt_calls[-1] == ("income", "600519.SH", fp._FINANCIAL_HISTORY_PERIODS)
+
+
+def test_balance_and_cashflow_mapping(monkeypatch):
+    balance = {
+        "period_end_ms": 1782748800000,
+        "report_date_ms": 1786723200000,
+        "assets_total": 309050784569.31,
+        "total_debt": 46954432394.95,
+        "holder_equity_total": 262096352174.36,
+    }
+    cashflow = {
+        "period_end_ms": 1782748800000,
+        "report_date_ms": 1786723200000,
+        "act_cash_flow_net": 92000000000.0,
+        "invest_cash_flow_net": -3000000000.0,
+        "pay_dividends_profits_interest_cash": 64000000000.0,  # 扶摇独有 → 扩展列
+    }
+    fake = _FakeFinClient(
+        statements={"balance_sheet": [balance], "cash_flow": [cashflow]}
+    )
+    provider = _provider_with(monkeypatch, fake)
+    bal = provider.get_financials("balance_sheet", ["600519.SH"]).to_dicts()[0]
+    assert bal["total_assets"] == pytest.approx(balance["assets_total"])
+    assert bal["total_liabilities"] == pytest.approx(balance["total_debt"])
+    assert bal["total_equity"] == pytest.approx(balance["holder_equity_total"])
+    cf = provider.get_financials("cash_flow", ["600519.SH"]).to_dicts()[0]
+    assert cf["net_operating_cash_flow"] == pytest.approx(92000000000.0)
+    assert cf["net_investing_cash_flow"] == pytest.approx(-3000000000.0)
+    assert cf["pay_dividends_profits_interest_cash"] == pytest.approx(64000000000.0)
+
+
+_METRICS_ABILITIES = [
+    {
+        "ability": "profitability",
+        "indicators": [
+            {"index_id": "index_weighted_avg_roe", "value": "16.7500"},
+            {"index_id": "sale_gross_margin", "value": "89.5552"},
+        ],
+    },
+    {
+        "ability": "growth",
+        # 实测 index_id 与文档有出入; 未映射 id 原名透传
+        "indicators": [{"index_id": "fixed_asset_invest_expansion_ratio", "value": "2.12587300"}],
+    },
+    {
+        "ability": "solvency",
+        "indicators": [{"index_id": "earned_interest_multiple", "value": None}],  # null → 不写列
+    },
+]
+
+
+def test_metrics_assembly(monkeypatch):
+    fake = _FakeFinClient(
+        statements={"income": [_INCOME_ROW]},
+        indicators={"2026-2": _METRICS_ABILITIES},
+        valuations=[{"thscode": "600519.SH", "pb_mrq": 6.455055}],
+        prices=[{"thscode": "600519.SH", "last_price": 1297.4}],
+    )
+    provider = _provider_with(monkeypatch, fake)
+    df = provider.get_financials("metrics", ["600519.SH"], latest_only=True)
+    row = df.to_dicts()[0]
+    assert row["period_end"] == "2026-06-30"
+    assert row["announce_date"] == "2026-08-15"
+    assert row["eps_basic"] == 35.57  # 顺带取自利润表
+    assert row["bps"] == pytest.approx(1297.4 / 6.455055)  # 估值反推
+    assert row["roe"] == pytest.approx(16.75)  # 字符串 → float
+    assert row["gross_margin"] == pytest.approx(89.5552)
+    assert row["fixed_asset_invest_expansion_ratio"] == pytest.approx(2.125873)
+    assert "earned_interest_multiple" not in df.columns  # 全空指标不成列
+    assert fake.ind_calls == ["600519.SH@2026-2"]  # report 由利润表最新期反推
+
+
+def test_metrics_indicator_failure_keeps_row(monkeypatch):
+    """指标端点单股失败 (如未披露期 code=5003) → 行仍写入 (eps/bps 保留)。"""
+    fake = _FakeFinClient(
+        statements={"income": [_INCOME_ROW]},
+        indicator_error=fc.FuyaoError("code=5003"),
+        valuations=[{"thscode": "600519.SH", "pb_mrq": 6.455055}],
+        prices=[{"thscode": "600519.SH", "last_price": 1297.4}],
+    )
+    provider = _provider_with(monkeypatch, fake)
+    df = provider.get_financials("metrics", ["600519.SH"], latest_only=True)
+    row = df.to_dicts()[0]
+    assert row["symbol"] == "600519.SH"
+    assert row["eps_basic"] == 35.57
+    assert "roe" not in df.columns
+
+
+def test_metrics_skips_symbol_without_income(monkeypatch):
+    fake = _FakeFinClient(statements={"income": []})
+    provider = _provider_with(monkeypatch, fake)
+    assert provider.get_financials("metrics", ["600519.SH"]).is_empty()
+
+
+def test_shares_returns_empty(monkeypatch):
+    provider = _provider_with(monkeypatch, _FakeFinClient())
+    assert provider.get_financials("shares", ["600519.SH"]).is_empty()
+
+
+def test_merge_fills_missing_cells_from_old_rows():
+    """逐列填空: 同报告期新行缺的列由旧行补齐, 有值则覆盖 (并集共存语义)。"""
+    old = pl.DataFrame({
+        "symbol": ["600519.SH", "600519.SH"],
+        "period_end": ["2026-03-31", "2026-06-30"],
+        "announce_date": ["2026-04-20", "2026-08-10"],
+        "diluted_eps": [68.1, 70.2],       # tickflow 提供, fuyao 没有
+        "net_income": [280.0, 460.0],
+    })
+    new = pl.DataFrame({
+        "symbol": ["600519.SH"],
+        "period_end": ["2026-06-30"],
+        "announce_date": ["2026-08-15"],   # 更晚公告 → 该期以新行为基准
+        "net_income": [461.5],             # 修正值覆盖
+        # diluted_eps 缺失 → 由旧行 70.2 补齐
+        "bps": [200.99],                   # fuyao 扩展列, 旧行没有
+    })
+    merged = _merge_report_history(old, new).to_dicts()
+    assert len(merged) == 2
+    q2 = next(r for r in merged if r["period_end"] == "2026-06-30")
+    assert q2["net_income"] == pytest.approx(461.5)   # 新值覆盖
+    assert q2["diluted_eps"] == pytest.approx(70.2)   # 旧行补齐
+    assert q2["bps"] == pytest.approx(200.99)         # 扩展列并入
+    q1 = next(r for r in merged if r["period_end"] == "2026-03-31")
+    assert q1["diluted_eps"] == pytest.approx(68.1)   # 未触碰期原样保留
+    # 旧公告覆盖新公告的倒序场景: announce 早的行不覆盖晚的
+    reversed_new = pl.DataFrame({
+        "symbol": ["600519.SH"],
+        "period_end": ["2026-06-30"],
+        "announce_date": ["2026-08-01"],
+        "net_income": [999.0],
+    })
+    q2b = _merge_report_history(old, reversed_new).to_dicts()[1]
+    # 公告更晚的 old 行 (08-10) 胜出, 早公告的新行不覆盖 → 业绩修正以最新公告为准
+    assert q2b["net_income"] == pytest.approx(460.0)

docs/custom-data-source.md

@@ -296,7 +296,9 @@ cp docs/examples/custom-data-source/mock_source.yaml data/data_sources/mock_sour
 
 分钟K (minute):
   symbol = 股票代码
-  datetime = 时间戳 (YYYY-MM-DD HH:MM:SS)
+  # datetime 必须是北京时间墙钟 (如 2026-08-28 09:35:00), 不要返回 UTC;
+  # 入口守卫会自动纠偏 UTC 特征帧, 但契约仍要求源头写对
+  datetime = 北京时间墙钟 (YYYY-MM-DD HH:MM:SS)
   open / high / low / close = OHLC
   volume = 成交量
   amount = 成交额

docs/plugin-development.md

@@ -147,7 +147,7 @@ class MyProvider:
 
     def get_minute(self, symbols, start_time, end_time, asset_type="stock",
                    on_chunk_done=None, freq="1m") -> pl.DataFrame:
-        """分钟K: [symbol, datetime, open, high, low, close, volume, amount]"""
+        """分钟K: [symbol, datetime(北京墙钟), open, high, low, close, volume, amount]"""
 
     def get_realtime(self) -> list[dict]:
         """全市场实时快照 → list[dict]。失败软返回 [], 不抛异常(不阻断轮询线程)。"""
@@ -164,6 +164,21 @@ class MyProvider:
         返回 error 字段说明会回退 TickFlow。"""
 ```
 
+### get_minute 的 datetime 时区契约
+
+`datetime` 必须是**北京时间墙钟**（naive，如 `2026-08-28 09:35:00`），与日K的
+`date` 语义对齐；不要返回 UTC 或带时区的时间。前端分时图按交易时段时轴
+（09:30–11:30 / 13:00–15:00）映射每根K线，UTC 口径的帧会导致全部点位落在时轴外、
+分时图空白。
+
+入口守卫（`kline_sync._enforce_minute_beijing_wallclock`）对所有分钟源强制归一：
+带时区 → 自动换算成北京墙钟；naive 但整体呈 UTC 特征（如 01:30）→ 自动 +8 纠偏并
+记日志；完全无法识别的口径 → 拒收并回退 TickFlow。契约仍要求源头写对，守卫只是兜底。
+
+可选类属性 `minute_history_days = 5` 声明 1 分钟历史深度（交易日）；未声明视为
+深历史（TickFlow 基准）。浅源（如 stock-sdk 免费分时仅保留最近 5 个交易日）声明后，
+个股分时档位自动收窄为可行选项并默认 5 日，深源默认 20 日。
+
 ### 异常语义
 
 | 方法 | 失败行为 |
@@ -213,7 +228,7 @@ class MyConfig:
 插件 PR 必须带契约测试(CONTRIBUTING §9), **不依赖真实网络与 API Key**——用假
 Client/桥接注入。以 `backend/tests/test_fuyao_provider.py` 为范本, 至少覆盖:
 
-1. 字段映射与单位转换: 百分数→小数制、缺失字段按口径推导、缺失字段置 None 不伪造
+1. 字段映射与单位转换: 百分数→小数制、volume 股→手、*ms 零点戳时区换算、缺失字段按口径推导、缺失字段置 None 不伪造
 2. 接口响应结构变体: 实测结构 vs 官方文档示例双兼容(供应商文档与实际不一致是常态)
 3. 分页: 多页合并、空页终止、页数上限
 4. 软失败: 接口报错返回 []; 整页 schema 变化有告警而非静默空数据
@@ -229,10 +244,10 @@ uv run --extra dev python -m ruff check app/plugins/<your_plugin>/ tests/test_<y
 ## 现有插件参考
 
 - **`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 探测)
-  - `tests/test_fuyao_provider.py` — 32 个契约测试, 是新插件的测试范本
+  - 提供 `realtime`(A 股全市场快照, 分页拉取)、`daily`(原始价日K三档: 近端窗口走 daily-k-10d dump, 深窗口走 daily-k 10 年全量 dump(172MB 一次下载、缓存复用、10d 补尾), 兜底单标的接口按 10 年自动分片)、`adj_factor`(事件 dump + 前收盘价从本地日K dump 一次取齐、缺价标的回退单标的接口, 按交易所公式推导单事件比值, 涨跌停自检; 全市场配价从逐标的 ~13 分钟降为秒级); Key 在设置页卡片直接配置(先探后存), 或 `.env` 配 `FUYAO_API_KEY`
+  - `client.py` — httpx 客户端(X-api-key 认证 + 统一信封解包 + 分页 + 页间隔限频 + 单标的日K + dump 预签名下载, S3 下载不带 Key 头)
+  - `provider.py` — Provider 实现(实测/文档双字段名映射、百分数→小数制、volume 股→手、上海零点戳 +8h 时区、dump 按 release 版本缓存、软失败、Key 探测)
+  - `tests/test_fuyao_provider.py` — 73 个契约测试, 是新插件的测试范本
 - **`backend/app/plugins/stocksdk/`** — Node 型插件, 通过 subprocess 桥接调用 stock-sdk
   - `bridge.py` — Python↔Node 桥接 + availability 检测
   - `bridge.mjs` — Node 端(并发池、重试、SDK 解析)