feat(adj): 除权因子独立同步 + 事件明细扩列 + 按需窗口优化

- 扩列: adj_factor 表扩至 8 列(dividend/bonus/allot/allot_price/prev_close 明细, 可空), fuyao 从事件 dump 原始成分回填; merge 变更检测按 ex_factor diff>1e-9, 仅明细回填不计入受影响列表; 字段说明补齐
- 独立同步: /api/pipeline/adj-factor/run 端点(共用任务机制: 单飞+重任务槽+行情暂停), Data 页除权因子卡片设置齿轮弹窗(生效数据源/能力徽标/TickFlow 无明细提示), 因子变更个股自动 enriched 局部重算
- 按需优化: 独立同步事件窗口钳到本地日K起点(更早事件在前复权 cum/total 两侧抵消, 对已存价格零影响); fuyao 配价由全有或全无改为逐只按需(远古事件只拖累自己, 不再整轮回退单标的接口); 实测全市场因子阶段 15min+ → ~11s, 配价 3676/3676 走本地 dump 零回退
- 盘后管道步骤条显示各能力路由徽标(矩阵驱动); README 补桌面端 macOS 安装注意事项
- 测试: 明细列 merge 语义 4 例 / 独立同步联动 5 例 / 远古事件不连累全场回归; 全量回归通过(2 例历史遗留失败与本改动无关, 基线同样失败)

README.md

@@ -509,6 +509,7 @@ docker run -d --name tsp -p 3018:3018 -v ${PWD}/data:/app/data ghcr.io/shy3130/t
 | **B · Compose 本地构建**              | 跑自己改过的代码 / 全套挂载 | Docker                                                                               |
 | **C · 本机 AI 代部署**                | 完全不想碰命令行            | 任一本机 AI 编程助手                                                                 |
 | **D · Dev 模式**                      | 二次开发                    | Python ≥ 3.11 · Node ≥ 20 · [uv](https://docs.astral.sh/uv/) · pnpm(`npm i -g pnpm`) |
+| **E · 桌面客户端**                    | 想要原生桌面窗口、免装环境  | Windows 10+ / macOS(Apple Silicon)                                                   |
 
 ### 方式 A:GHCR 现成镜像(免本地构建,多数用户推荐)
 
@@ -573,6 +574,24 @@ cp .env.example .env       # 按需填 TICKFLOW_API_KEY(留空 = None 模式)
 
 自动检查 / 下载依赖、释放端口、同时起前后端。后端 → <http://localhost:3018> · 前端 → <http://localhost:3011>。
 
+### 方式 E:桌面客户端(Windows / macOS)
+
+从 [Releases](https://github.com/shy3130/tick-stock-panel/releases/latest) 下载对应安装包:Windows 双击 `TSP-Setup-x64-*.exe` 按向导安装即可;macOS 下载 `TSP-macos-arm64-*.dmg`。
+
+<details>
+<summary><b>🍎 macOS 安装注意事项(点开查看)</b></summary>
+
+- **仅支持 Apple Silicon(M 系列芯片)**,暂无 Intel 版本。
+- 打开 dmg 后,请先把 **TSP.app 拖入「应用程序(Applications)」再运行**;不要直接在 dmg 挂载窗口里双击运行 —— 挂载卷是只读的,会导致启动即退出。
+- 首次打开会被 macOS Gatekeeper 拦截(安装包暂未做开发者签名与公证):前往 **系统设置 → 隐私与安全性**,下滑到「已阻止使用 "TSP"」→ 点 **仍要打开**,再按提示确认一次。
+- 若提示 **「"TSP" 已损坏,无法打开」**,或完成上一步后仍一闪退出:打开「终端」执行下面的命令,然后再启动(`xattr` 为 macOS 自带命令,作用是移除下载文件上的隔离标记,不会改动系统设置):
+
+```bash
+xattr -dr com.apple.quarantine /Applications/TSP.app
+```
+
+</details>
+
 ### 跑起来后的第一次使用
 
 1. **设置 → 凭据与能力** → 点 **重新检测**,确认档位标签与能力路由矩阵

backend/app/api/data.py

@@ -751,9 +751,14 @@ _TABLE_FIELD_DESC: dict[str, dict[str, str]] = {
     },
     "adj_factor": {
         "symbol": "股票代码",
-        "timestamp": "除权除息时间戳(ms)",
         "trade_date": "除权除息日",
-        "ex_factor": "复权因子",
+        "ex_factor": "单事件等比因子 = 除权前收盘价 / 除权参考价, 按日期连乘得累计复权因子",
+        # 事件明细列: 表结构固定 8 列, 明细由提供事件的数据源(扶摇)回填, TickFlow 源为 null
+        "dividend": "每股现金分红(税前, 元)。可空, 仅事件明细数据源回填",
+        "bonus": "每股送转比例(10送3 = 0.3)。可空, 仅事件明细数据源回填",
+        "allot": "每股配股比例。可空, 仅事件明细数据源回填",
+        "allot_price": "配股价(元/股)。可空, 仅事件明细数据源回填",
+        "prev_close": "除权前一日收盘价(元), 因子推导基准。可空, 仅事件明细数据源回填",
     },
     "instruments": {
         "symbol": "股票代码",

backend/app/api/pipeline.py

@@ -41,6 +41,19 @@ async def trigger_pipeline_job(repo, capset, quote_service=None) -> dict:
     若已有任务在跑,**返回该任务 id 而不是开新任务**(防止并发拉数据撞限流)。
     卡死判定按「进度停滞」而非总时长(慢带宽下长任务不会被误杀), 见 reap_stale。
     """
+    return await _submit_job(
+        repo, quote_service,
+        lambda progress: (lambda: daily_pipeline.run_now(repo, capset, on_progress=progress)),
+    )
+
+
+async def _submit_job(repo, quote_service, make_run) -> dict:
+    """任务化执行公共机制: 单飞 create → 重任务槽 → 行情暂停 → 缓存失效。
+
+    make_run(progress) 返回实际执行闭包(阻塞 IO + CPU, 在 executor 里跑)。
+    盘后管道与独立除权因子同步共用 —— 两者写同一批 parquet, 必须共享
+    单飞与重任务槽防并发覆写。
+    """
     # 检测卡死的 running job (如 reload 后孤儿 task / 网络读无限阻塞)。
     # reap_stale 会在 /run 和 /jobs/{id} 轮询端点都调用,保证卡死后能自愈。
     job_store.reap_stale()
@@ -50,27 +63,29 @@ async def trigger_pipeline_job(repo, capset, quote_service=None) -> dict:
     if not is_new:
         return {"job_id": job_id, "reused": True}
 
+    def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None,
+                 skip_log: bool = False) -> None:
+        job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
+
+    run = make_run(progress)
+
     # 在 executor 里跑同步任务(pipeline 内部都是阻塞 IO + CPU)
     async def task() -> None:
         # 重任务执行槽: 防僵尸并发(reap 后线程仍活时新任务不得并行写 parquet)
         if not try_acquire_run_slot(job_id):
             job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
             return
-        # 管道运行期间暂停实时行情取数, 防止覆写同一批 parquet 竞态
+        # 任务运行期间暂停实时行情取数, 防止覆写同一批 parquet 竞态
         qs = quote_service
         try:
             loop = asyncio.get_event_loop()
 
-            def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None,
-                         skip_log: bool = False) -> None:
-                job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
-
             def _run() -> dict:
                 try:
                     if qs:
                         with qs.paused():
-                            return daily_pipeline.run_now(repo, capset, on_progress=progress)
-                    return daily_pipeline.run_now(repo, capset, on_progress=progress)
+                            return run()
+                    return run()
                 finally:
                     repo.refresh_cache()
 
@@ -82,7 +97,7 @@ async def trigger_pipeline_job(repo, capset, quote_service=None) -> dict:
             # 拉取线程在分块回调处自行退出, 这里无需(也无法)再写状态。
             logger.warning("pipeline job %s cancelled", job_id)
         except Exception as e:  # noqa: BLE001
-            logger.exception("pipeline failed")
+            logger.exception("pipeline job failed")
             job_store.fail(job_id, str(e))
             invalidate_storage_cache()
         finally:
@@ -92,6 +107,30 @@ async def trigger_pipeline_job(repo, capset, quote_service=None) -> dict:
     return {"job_id": job_id, "reused": False}
 
 
+@router.post("/adj-factor/run")
+async def run_adj_factor_now(request: Request) -> dict:
+    """独立同步除权因子(全历史 + 受影响个股 enriched 局部重算), 任务化+进度轮询。
+
+    与盘后管道共用单飞与重任务槽 (两者写同一批 parquet); 门控与矩阵
+    usable 同口径: TickFlow 档位不足且未路由到其他可用源时 400。
+    """
+    from app.services import preferences
+    from app.tickflow.capabilities import Cap
+
+    capset = request.app.state.capabilities
+    if not (capset.has(Cap.ADJ_FACTOR) or preferences.get_adj_factor_provider() != "tickflow"):
+        raise HTTPException(
+            status_code=400,
+            detail="除权因子能力不可用: 请先在 设置 → 数据源 中配置可用数据源",
+        )
+    repo = request.app.state.repo
+    return await _submit_job(
+        repo,
+        getattr(request.app.state, "quote_service", None),
+        lambda progress: (lambda: daily_pipeline.run_adj_factor_sync(repo, capset, on_progress=progress)),
+    )
+
+
 @router.get("/jobs/{job_id}")
 def get_job(job_id: str) -> dict:
     # 每次轮询都检查卡死 job — 前端持续轮询, 进度停滞超阈值后必定自愈,

backend/app/jobs/daily_pipeline.py

@@ -800,6 +800,94 @@ def run_now(
     return result
 
 
+def run_adj_factor_sync(
+    repo: KlineRepository,
+    capset: CapabilitySet,
+    on_progress: ProgressCb | None = None,
+) -> dict:
+    """独立同步除权因子(数据页单独按钮)。
+
+    与盘后管道 Step 1.5 的差异: 管道只拉日K范围或 15 天兜底窗口;
+    这里 start_time=None 拉**全历史** —— 用于切换数据源后的因子修复、
+    以及事件明细列扩表后的存量回填 (明细回填不改变 ex_factor,
+    sync_adj_factor 的变化判定不会因此触发 enriched 重算)。
+
+    下游联动与管道同语义: adj_factor 视图刷新 → 受影响个股 enriched
+    局部重算(全部日期) → kline_enriched 视图刷新 → 两表缓存失效。
+    enriched 尚未构建时跳过重算 (全量计算是盘后管道的职责)。
+    """
+    from datetime import datetime
+
+    emit = on_progress or _noop
+    adj_provider = _prefs.get_adj_factor_provider()
+    if not (capset.has(Cap.ADJ_FACTOR) or adj_provider != "tickflow"):
+        return {
+            "adj_factor_symbols": 0,
+            "adj_written": 0,
+            "enriched_days": 0,
+            "skipped_stages": ["sync_adj"],
+            "reason": "无可用除权因子数据源",
+        }
+
+    emit("resolve_universe", 5, "解析标的池…")
+    universe = _resolve_universe(capset, repo)
+    emit("resolve_universe", 10, f"标的池规模:{len(universe)} 只")
+
+    def _adj_chunk_progress(cur: int, tot: int) -> None:
+        emit("sync_adj", 10 + int(55 * cur / tot),
+             f"除权因子批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True)
+
+    # 按需窗口: 钳到本地日K起点。前复权里早于任一根已存K线的事件在
+    # cum/total 两侧同时出现、恰好抵消, 对所有已存价格零影响 (见
+    # indicators.pipeline._apply_adj_factor) — 拉它们只会为远古事件
+    # 逐只回补历史K线, 全市场一次曾多花 ~15 分钟。
+    store_min = repo.earliest_daily_date()
+    start = datetime(store_min.year, store_min.month, store_min.day) if store_min else None
+    window_label = f"自 {store_min.isoformat()}" if store_min else "全历史"
+    logger.info(
+        "standalone adj sync: %d symbols, event window %s",
+        len(universe), window_label,
+    )
+    emit("sync_adj", 10, f"获取除权因子 [{window_label}, {len(universe)} 只]…")
+    written_adj, affected = kline_sync.sync_adj_factor(
+        universe, repo, capset,
+        start_time=start, end_time=datetime.now(),
+        on_chunk_done=_adj_chunk_progress,
+    )
+    affected = sorted(set(affected))
+    _refresh_single_view(repo, "adj_factor")
+    emit("sync_adj", 65,
+         f"除权因子完成,新增 {written_adj} 行,因子变更 {len(affected)} 只")
+    _invalidate("adj_factor")
+
+    # enriched 局部重算: 与管道「无新日期 + 有新除权因子」分支同语义 (只重算受影响个股)
+    enriched_dir = repo.store.data_dir / "kline_daily_enriched"
+    enriched_exists = enriched_dir.exists() and any(enriched_dir.glob("date=*"))
+    written_enriched = 0
+    if affected and enriched_exists:
+        def _enriched_batch_progress(cur: int, tot: int) -> None:
+            emit("compute_enriched", 65 + int(30 * cur / tot),
+                 f"计算指标 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True)
+
+        emit("compute_enriched", 65, f"增量计算 enriched ({len(affected)} 只个股)…")
+        logger.info("compute_enriched: adj_factor incremental (standalone), %d symbols", len(affected))
+        written_enriched = run_pipeline(symbols=affected, on_batch_done=_enriched_batch_progress)
+        _refresh_single_view(repo, "kline_enriched")
+        emit("compute_enriched", 95, f"enriched 完成,{len(affected)} 只个股")
+        _invalidate("enriched")
+    elif affected:
+        logger.info("standalone adj sync: enriched 未构建, 跳过局部重算 (交由盘后管道全量计算)")
+
+    emit("done", 100, "完成")
+    return {
+        "universe_size": len(universe),
+        "adj_written": written_adj,
+        "adj_factor_symbols": len(affected),
+        "enriched_days": written_enriched,
+        "skipped_stages": [],
+    }
+
+
 def warn_if_enriched_too_thin(data_dir: Path) -> int:
     """enriched 总覆盖天数; 低于常见指标暖机窗口时 WARN 引导全量回填 (#303)。
 

backend/app/plugins/fuyao/provider.py

@@ -358,13 +358,26 @@ class FuyaoProvider:
         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 补。
+    def _ensure_daily_big_dump(self, start_d: date | None) -> Path | None:
+        """10 年全量日K dump(约 172MB)。要求覆盖窗口起点时校验; 末端缺口由 10d dump 补。
+
+        start_d=None 表示不校验覆盖(配价场景): 只要拿到 dump 本身即可,
+        个别覆盖不到的标的由 get_adj_factors 逐只回退 — 避免"全市场最早的
+        一个远古事件早于 dump 起点"就把整轮配价打回逐标的接口。
 
         已有缓存覆盖起点就直接复用, 不追新 release(避免深窗口高频触发时日日重下
         172MB — 旧 release 的中段历史不会变, 尾部新鲜度交给 1MB 的 10d dump)。
         """
-        for f in sorted(_cache_dir().glob("daily_k__*.parquet"), reverse=True):
+        cached = sorted(_cache_dir().glob("daily_k__*.parquet"), reverse=True)
+        if start_d is None:
+            if cached:
+                return cached[0]
+            try:
+                return self._ensure_dump_path(_DAILY_DUMP_KIND, "daily_k")
+            except FuyaoError as e:
+                logger.warning("扶摇 10 年 dump 不可用, 回退单标的接口: %s", e)
+                return None
+        for f in cached:
             try:
                 dmin, _ = _dump_date_range(f)
             except Exception:  # 缓存损坏不致命, 换下一个/重拉
@@ -685,14 +698,25 @@ class FuyaoProvider:
         asset_type: str = "stock",
         on_chunk_done: Callable[[int, int], None] | None = None,
     ) -> pl.DataFrame:
-        """A 股除权因子 → 内部契约(symbol/trade_date/ex_factor, 单事件比值非累积)。
+        """A 股除权因子 → 内部契约(symbol/trade_date/ex_factor, 单事件比值非累积)
+        + 事件明细列(dividend/bonus/allot/allot_price/prev_close, 可空):
+        供等差显示投影(Σ未来每股分红)与官方金额全精度因子链重建使用,
+        明细取自事件 dump 原始成分, 不再在推导后丢弃。
 
         数据链: adjustment-factors 全量事件 dump → 清洗(时区 +8h / 滤全零 / 同日成分合并)
         → 前收盘价从本地日K dump 一次取齐(缺价标的回退单标的接口) → 交易所公式推导
         → 涨跌停自检剔除异常。
         未来已公告事件无前收盘, 留给滚动增量窗口(15 天)自然补上。
         """
-        schema = {"symbol": pl.String, "trade_date": pl.Date, "ex_factor": pl.Float64}
+        schema = {
+            "symbol": pl.String, "trade_date": pl.Date, "ex_factor": pl.Float64,
+            "dividend": pl.Float64, "bonus": pl.Float64, "allot": pl.Float64,
+            "allot_price": pl.Float64, "prev_close": pl.Float64,
+        }
+
+        def _f(x) -> float | None:
+            return None if x is None else float(x)
+
         if not symbols or asset_type != "stock":
             return pl.DataFrame(schema=schema)
         try:
@@ -760,7 +784,12 @@ class FuyaoProvider:
                             ret * 100,
                         )
                         continue
-                out_rows.append({"symbol": sym, "trade_date": exd, "ex_factor": factor})
+                out_rows.append({
+                    "symbol": sym, "trade_date": exd, "ex_factor": factor,
+                    "dividend": _f(ev["dividend"]), "bonus": _f(ev["bonus"]),
+                    "allot": _f(ev["allot"]), "allot_price": _f(ev["allot_price"]),
+                    "prev_close": _f(p),
+                })
             if on_chunk_done:
                 on_chunk_done(i + 1, len(syms))
         if closes_by_sym is not None:
@@ -1057,12 +1086,12 @@ class FuyaoProvider:
 
         每标的开窗 [first_ex-30d, last_ex](与单标的接口同窗): 10 年大 dump 为
         主体, 10d dump 叠加补末端新鲜度(除权日当天的自检需要 ex 日收盘)。
-        返回 symbol → {date: close}; 大 dump 不可用/窗口早于其覆盖/读盘失败
-        → None, 由调用方整轮回退单标的接口。
+        返回 symbol → {date: close}; 大 dump 不可用/读盘失败 → None 整轮回退。
+        覆盖不到首个事件的个别标的不在此 dict 中(或首价晚于 first_ex),
+        由 get_adj_factors 逐只回退单标的接口 — 按需, 不全有或全无。
         """
         try:
-            lo = bounds["first_ex"].min() - timedelta(days=_PREV_CLOSE_BACKDAYS)
-            path = self._ensure_daily_big_dump(lo)
+            path = self._ensure_daily_big_dump(None)
             if path is None:
                 return None
             sources = [self._closes_scan(pl.scan_parquet(path))]

backend/app/services/kline_sync.py

@@ -15,6 +15,7 @@ import time
 import uuid
 from collections.abc import Callable
 from datetime import date, datetime, timedelta
+from pathlib import Path
 from typing import Any
 
 import polars as pl
@@ -496,6 +497,13 @@ def sync_daily_by_quotes(repo: KlineRepository) -> int:
     return daily_df.height
 
 
+# 除权事件明细列 (可选, 缺省为 null): 等差显示投影 (原始价 - Σ未来每股分红,
+# 分红按其后送转比例缩放) 与官方金额全精度因子链重建使用。
+# 存量 all.parquet 只有规范三列, 合并用 how="diagonal" 自动补 null;
+# 读取方 (_apply_adj_factor / minute_adjust) 只按规范三列消费, 不受影响。
+ADJ_DETAIL_COLS = ("dividend", "bonus", "allot", "allot_price", "prev_close")
+
+
 def _normalize_adj_factor(raw) -> pl.DataFrame:
     """Normalize SDK ex_factors response to symbol/trade_date/ex_factor."""
     if raw is None or len(raw) == 0:
@@ -534,10 +542,68 @@ def _normalize_adj_factor(raw) -> pl.DataFrame:
             df = df.with_columns(pl.col("trade_date").cast(pl.Date, strict=False))
     if "ex_factor" in df.columns:
         df = df.with_columns(pl.col("ex_factor").cast(pl.Float64, strict=False))
-    cols = [c for c in ["symbol", "trade_date", "ex_factor"] if c in df.columns]
-    if len(cols) < 3:
+    if not all(c in df.columns for c in ("symbol", "trade_date", "ex_factor")):
         return pl.DataFrame()
-    return df.select(cols).drop_nulls()
+    # 明细列存在则透传 (缺省不影响规范三列); drop_nulls 只作用于规范三列 ——
+    # 明细列合法可为 null (如纯送转事件的 dividend)。
+    df = df.with_columns([
+        pl.col(c).cast(pl.Float64, strict=False)
+        for c in ADJ_DETAIL_COLS if c in df.columns
+    ])
+    cols = ["symbol", "trade_date", "ex_factor"] + [
+        c for c in ADJ_DETAIL_COLS if c in df.columns
+    ]
+    return df.select(cols).drop_nulls(subset=["symbol", "trade_date", "ex_factor"])
+
+
+def _align_adj_factor_frame(df: pl.DataFrame) -> pl.DataFrame:
+    """除权因子帧对齐为统一 schema: 规范三列 + 明细列 (缺则补 null 列)。"""
+    df = df.with_columns(
+        pl.col("trade_date").cast(pl.Date, strict=False),
+        pl.col("ex_factor").cast(pl.Float64, strict=False),
+        *[pl.col(c).cast(pl.Float64, strict=False)
+          for c in ADJ_DETAIL_COLS if c in df.columns],
+    )
+    df = df.with_columns([
+        pl.lit(None, dtype=pl.Float64).alias(c)
+        for c in ADJ_DETAIL_COLS if c not in df.columns
+    ])
+    return df.select(["symbol", "trade_date", "ex_factor", *ADJ_DETAIL_COLS])
+
+
+def _merge_adj_factor_store(out: Path, new_data: pl.DataFrame) -> tuple[int, list[str]]:
+    """合并写入 adj_factor 的 all.parquet, 返回 (新增行数, 因子发生变化的 symbol 列表)。
+
+    how="diagonal" 兼容存量三列 schema (旧行明细自动为 null);
+    unique keep="last" → 重复 (symbol, trade_date) 以本次同步为准,
+    重跑一次同步即完成存量数据的明细回填。
+
+    变化判定只看 ex_factor (新键 + 因子值差异 > 1e-9): 仅明细列回填
+    (dividend/prev_close 等补 null) 不改变复权价格 → 不计入变化列表,
+    enriched 局部重算因此不会被明细回填触发全市场重算。
+    """
+    out.parent.mkdir(parents=True, exist_ok=True)
+    new_data = _align_adj_factor_frame(new_data)
+    if not out.exists():
+        _atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
+        return new_data.height, sorted(new_data["symbol"].unique().to_list())
+    existing = pl.read_parquet(out)
+    before = existing.height
+    # ex_factor 变化判定 (明细回填不算): 与存量按 (symbol, trade_date) 对齐,
+    # 旧值缺失 (新事件) 或值差超容差即视为变化。
+    diffed = new_data.select(["symbol", "trade_date", "ex_factor"]).join(
+        existing.select(["symbol", "trade_date", "ex_factor"]),
+        on=["symbol", "trade_date"], how="left", suffix="_old",
+    ).filter(
+        pl.col("ex_factor_old").is_null()
+        | ((pl.col("ex_factor") - pl.col("ex_factor_old")).abs() > 1e-9)
+    )
+    changed_symbols = sorted(set(diffed["symbol"].to_list()))
+    merged = pl.concat([existing, new_data], how="diagonal").unique(
+        subset=["symbol", "trade_date"], keep="last",
+    ).sort(["symbol", "trade_date"])
+    _atomic_write_parquet(merged, out)
+    return merged.height - before, changed_symbols
 
 
 def sync_adj_factor(symbols: list[str], repo: KlineRepository,
@@ -550,6 +616,8 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
 
     支持增量: 传 start_time/end_time 只拉取该时间范围内的新除权事件。
     返回 (写入行数, 受影响的 symbol 列表) — 供 enriched 局部重算使用。
+    受影响 = 因子新增或值变化 (ex_factor diff) 的标的; 仅事件明细列
+    (dividend 等) 回填不改变复权价, 不计入受影响列表。
     """
     if not symbols:
         return 0, []
@@ -577,20 +645,11 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
                 else:
                     return 0, []
             else:
-                affected = new_data["symbol"].unique().to_list()
                 factor_dir = "adj_factor_etf" if asset_type == "etf" else "adj_factor"
-                out = repo.store.data_dir / factor_dir / "all.parquet"
-                out.parent.mkdir(parents=True, exist_ok=True)
-                if out.exists():
-                    existing = pl.read_parquet(out)
-                    before = existing.height
-                    merged = pl.concat([existing, new_data]).unique(
-                        subset=["symbol", "trade_date"], keep="last",
-                    ).sort(["symbol", "trade_date"])
-                    _atomic_write_parquet(merged, out)
-                    return merged.height - before, affected
-                _atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
-                return new_data.height, affected
+                added, affected = _merge_adj_factor_store(
+                    repo.store.data_dir / factor_dir / "all.parquet", new_data
+                )
+                return added, affected
         # 自定义源未配置 adj_factor → 回退 TickFlow
 
     if not capset.has(Cap.ADJ_FACTOR):
@@ -646,28 +705,11 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
 
     new_data = pl.concat(all_dfs, how="diagonal_relaxed") if len(all_dfs) > 1 else all_dfs[0]
 
-    # 提取受影响的 symbol 列表(合并前)
-    affected = new_data["symbol"].unique().to_list()
-
     factor_dir = "adj_factor_etf" if asset_type == "etf" else "adj_factor"
-    out = repo.store.data_dir / factor_dir / "all.parquet"
-    out.parent.mkdir(parents=True, exist_ok=True)
-
-    if out.exists():
-        existing = pl.read_parquet(out)
-        before = existing.height
-        merged = pl.concat([existing, new_data]).unique(
-            subset=["symbol", "trade_date"], keep="last",
-        ).sort(["symbol", "trade_date"])
-        _atomic_write_parquet(merged, out)
-        added = merged.height - before
-        logger.info("adj_factor merged: %d total (+%d new), %d/%d symbols",
-                     merged.height, added, new_data.height, len(symbols))
-        return added, affected
-    else:
-        _atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
-        logger.info("adj_factor synced: %d rows (%d symbols)", new_data.height, len(symbols))
-        return new_data.height, affected
+    added, affected = _merge_adj_factor_store(repo.store.data_dir / factor_dir / "all.parquet", new_data)
+    logger.info("adj_factor merged: +%d new rows, %d changed symbols, %d fetched / %d symbols",
+                added, len(affected), new_data.height, len(symbols))
+    return added, affected
 
 
 # ===== 分钟 K 同步 =====

backend/tests/test_adj_factor_event_columns.py

@@ -0,0 +1,175 @@
+"""adj_factor 事件明细扩列: 归一化透传 / 存量三列 schema 合并兼容 / 消费方不受影响。
+
+新增列 dividend/bonus/allot/allot_price/prev_close 全部可选可空:
+- fuyao 等提供明细的源 → 落库供等差显示投影与全精度因子链重建;
+- 存量 all.parquet 只有规范三列 → diagonal 合并自动补 null, 读取方零改动;
+- 重跑一次同步即完成存量数据明细回填 (unique keep="last")。
+"""
+
+from datetime import date
+
+import polars as pl
+import pytest
+
+from app.indicators.pipeline import _apply_adj_factor
+from app.services.kline_sync import (
+    ADJ_DETAIL_COLS,
+    _align_adj_factor_frame,
+    _atomic_write_parquet,
+    _merge_adj_factor_store,
+    _normalize_adj_factor,
+)
+
+
+def _detail_row(symbol="600519.SH", td=date(2026, 6, 26), f=1.023663, **detail):
+    return {"symbol": symbol, "trade_date": td, "ex_factor": f, **detail}
+
+
+# ── _normalize_adj_factor: 明细列透传 ───────────────────────────────
+
+
+def test_normalize_keeps_detail_columns_when_present():
+    raw = pl.DataFrame([
+        _detail_row(dividend=28.019, bonus=0.0, allot=0.0, allot_price=0.0, prev_close=1212.1)
+    ])
+    out = _normalize_adj_factor(raw)
+    assert out.columns == ["symbol", "trade_date", "ex_factor", *ADJ_DETAIL_COLS]
+
+
+def test_normalize_legacy_three_column_frame_unaffected():
+    raw = pl.DataFrame([_detail_row()])
+    out = _normalize_adj_factor(raw)
+    assert out.columns == ["symbol", "trade_date", "ex_factor"]
+
+
+def test_normalize_preserves_null_detail_values():
+    """明细合法可为 null (如纯送转事件的 dividend), 不得被 drop_nulls 丢行。"""
+    raw = pl.DataFrame([
+        _detail_row(symbol="000812.SZ", f=1.10, dividend=None, bonus=0.1)
+    ])
+    out = _normalize_adj_factor(raw)
+    assert out.height == 1
+    assert out["dividend"][0] is None
+
+
+# ── _align / _merge: 存量三列 schema 兼容与明细回填 ─────────────────
+
+
+def test_align_adds_missing_detail_columns_as_null():
+    aligned = _align_adj_factor_frame(pl.DataFrame([_detail_row()]))
+    assert aligned.columns == ["symbol", "trade_date", "ex_factor", *ADJ_DETAIL_COLS]
+    assert all(aligned[c][0] is None for c in ADJ_DETAIL_COLS)
+
+
+def test_merge_backfills_details_into_legacy_schema(tmp_path):
+    """旧行明细为 null、重同步的行以明细覆盖、新事件行计数正确。"""
+    out = tmp_path / "all.parquet"
+    legacy = pl.DataFrame([
+        _detail_row(symbol="600519.SH", td=date(2026, 6, 26), f=1.023663),
+        _detail_row(symbol="000001.SZ", td=date(2026, 6, 26), f=1.005),
+    ])
+    _atomic_write_parquet(legacy, out)
+
+    new = pl.DataFrame([
+        # 同事件重同步: 携带明细 → 覆盖旧行
+        _detail_row(symbol="600519.SH", td=date(2026, 6, 26), f=1.023663,
+                    dividend=28.019, bonus=0.0, allot=0.0, allot_price=0.0,
+                    prev_close=1212.1),
+        # 新事件
+        _detail_row(symbol="000001.SZ", td=date(2025, 12, 19), f=1.017028,
+                    dividend=23.959, bonus=0.0, allot=0.0, allot_price=0.0,
+                    prev_close=1431.0),
+    ])
+    added, changed = _merge_adj_factor_store(out, new)
+    assert added == 1  # 仅新事件计入
+    # 变化判定只看 ex_factor: 600519 同因子仅回填明细 → 不算变化;
+    # 000001 新事件 → 变化
+    assert changed == ["000001.SZ"]
+
+    merged = pl.read_parquet(out)
+    assert merged.columns == ["symbol", "trade_date", "ex_factor", *ADJ_DETAIL_COLS]
+    mt = merged.filter(
+        (pl.col("symbol") == "600519.SH") & (pl.col("trade_date") == date(2026, 6, 26))
+    ).row(0, named=True)
+    assert mt["dividend"] == pytest.approx(28.019)  # 重同步回填明细
+    old = merged.filter(
+        (pl.col("symbol") == "000001.SZ") & (pl.col("trade_date") == date(2026, 6, 26))
+    ).row(0, named=True)
+    assert old["dividend"] is None  # 未重同步的旧行明细保持 null
+    assert merged.height == 3
+
+
+# ── 变化判定: enriched 局部重算的触发精度 ───────────────────────────
+
+
+def test_merge_detail_only_backfill_reports_no_changed_symbols(tmp_path):
+    """扩表后重拉全历史: 因子值全部未变, 只是补明细 → 不触发任何 enriched 重算。"""
+    out = tmp_path / "all.parquet"
+    legacy = pl.DataFrame([
+        _detail_row(symbol="600519.SH", td=date(2026, 6, 26), f=1.023663),
+        _detail_row(symbol="000001.SZ", td=date(2025, 12, 19), f=1.017028),
+    ])
+    _atomic_write_parquet(legacy, out)
+    backfill = pl.DataFrame([
+        _detail_row(symbol="600519.SH", td=date(2026, 6, 26), f=1.023663, dividend=28.019),
+        _detail_row(symbol="000001.SZ", td=date(2025, 12, 19), f=1.017028, dividend=23.959),
+    ])
+    added, changed = _merge_adj_factor_store(out, backfill)
+    assert added == 0
+    assert changed == []  # 关键: 明细回填不改变复权价, 不触发重算
+
+
+def test_merge_changed_factor_value_reports_symbol(tmp_path):
+    """同键因子值修正 (如切换数据源后微差) → 该标的计入变化, 行数不增。"""
+    out = tmp_path / "all.parquet"
+    legacy = pl.DataFrame([_detail_row(symbol="600519.SH", f=1.023663)])
+    _atomic_write_parquet(legacy, out)
+    corrected = pl.DataFrame([
+        _detail_row(symbol="600519.SH", f=1.0237, dividend=28.019)  # 值差 > 1e-9
+    ])
+    added, changed = _merge_adj_factor_store(out, corrected)
+    assert added == 0
+    assert changed == ["600519.SH"]
+
+
+def test_merge_tiny_factor_diff_within_tolerance_not_changed(tmp_path):
+    """值差 ≤ 1e-9 视为未变 (浮点噪声保护)。"""
+    out = tmp_path / "all.parquet"
+    legacy = pl.DataFrame([_detail_row(symbol="600519.SH", f=1.023663)])
+    _atomic_write_parquet(legacy, out)
+    noisy = pl.DataFrame([_detail_row(symbol="600519.SH", f=1.0236630000001)])
+    added, changed = _merge_adj_factor_store(out, noisy)
+    assert added == 0
+    assert changed == []
+
+
+def test_merge_first_write_all_symbols_changed(tmp_path):
+    """库不存在时全部落库, 全部标的都算变化 (下游局部重算语义自然成立)。"""
+    out = tmp_path / "all.parquet"
+    first = pl.DataFrame([
+        _detail_row(symbol="600519.SH", f=1.023663),
+        _detail_row(symbol="000001.SZ", td=date(2025, 12, 19), f=1.017028),
+    ])
+    added, changed = _merge_adj_factor_store(out, first)
+    assert added == 2
+    assert changed == ["000001.SZ", "600519.SH"]
+
+
+# ── 消费方回归: 带明细列的 factors 不影响复权计算 ───────────────────
+
+
+def test_apply_adj_factor_tolerates_detail_columns():
+    raw = pl.DataFrame({
+        "symbol": ["A"] * 3,
+        "date": [date(2026, 1, 2), date(2026, 1, 5), date(2026, 1, 6)],
+        "open": [10.0] * 3, "high": [10.5] * 3, "low": [9.5] * 3,
+        "close": [10.0, 9.0, 9.9],
+    })
+    factors = pl.DataFrame([
+        _detail_row(symbol="A", td=date(2026, 1, 5), f=1.25,
+                    dividend=2.0, bonus=0.0, allot=0.0, allot_price=0.0, prev_close=10.0)
+    ])
+    out = _apply_adj_factor(raw, factors)
+    assert out["close"][0] == pytest.approx(10.0 / 1.25)  # 事件前按因子折算
+    assert out["close"][1] == pytest.approx(9.0)          # 事件起锚定原始价
+    assert out["close"][2] == pytest.approx(9.9)

backend/tests/test_adj_factor_standalone_sync.py

@@ -0,0 +1,134 @@
+"""独立除权因子同步 (run_adj_factor_sync) 的联动语义回归。
+
+覆盖: 事件窗口按本地日K起点钳制 / 受影响个股 enriched 局部重算 /
+明细回填 (无因子变化) 不触发重算 / enriched 未构建时跳过 / 无能力时跳过。
+纯逻辑, 不触网。
+"""
+from __future__ import annotations
+
+from datetime import date, datetime
+from types import SimpleNamespace
+
+from app.jobs import daily_pipeline
+from app.services import kline_sync
+
+
+def _repo(tmp_path, earliest_daily=None):
+    return SimpleNamespace(
+        store=SimpleNamespace(data_dir=tmp_path),
+        refresh_cache=lambda: None,
+        earliest_daily_date=lambda: earliest_daily,
+    )
+
+
+def _capset(has_adj=True):
+    return SimpleNamespace(has=lambda key: has_adj)
+
+
+def _patch_common(monkeypatch, tmp_path, *, written=2, affected=("600519.SH",),
+                  with_enriched=True, provider="tickflow"):
+    """公共桩: 标的池/失效/视图刷新/因子同步/enriched 重算全部可观测。"""
+    monkeypatch.setattr(daily_pipeline, "_resolve_universe",
+                        lambda capset, repo=None: ["600519.SH", "000001.SZ"])
+    invalidated: list[str | None] = []
+    monkeypatch.setattr(daily_pipeline, "_invalidate", invalidated.append)
+    refreshed: list[str] = []
+    monkeypatch.setattr(daily_pipeline, "_refresh_single_view",
+                        lambda repo, name: refreshed.append(name))
+
+    captured_sync: dict = {}
+
+    def fake_sync(symbols, repo, capset, start_time=None, end_time=None, on_chunk_done=None):
+        captured_sync["symbols"] = symbols
+        captured_sync["start_time"] = start_time
+        return written, list(affected)
+
+    monkeypatch.setattr(kline_sync, "sync_adj_factor", fake_sync)
+
+    captured_recompute: dict = {}
+
+    def fake_run_pipeline(data_dir=None, symbols=None, new_dates_only=False, on_batch_done=None):
+        captured_recompute["symbols"] = symbols
+        return 7
+
+    monkeypatch.setattr(daily_pipeline, "run_pipeline", fake_run_pipeline)
+    monkeypatch.setattr(daily_pipeline._prefs, "get_adj_factor_provider", lambda: provider)
+
+    if with_enriched:
+        (tmp_path / "kline_daily_enriched" / "date=2026-01-05").mkdir(parents=True)
+    return captured_sync, captured_recompute, invalidated, refreshed
+
+
+def test_full_history_window_and_partial_recompute(monkeypatch, tmp_path):
+    """无本地日K时全历史拉取 + 因子变更个股的 enriched 全日期局部重算 + 双表联动。"""
+    sync, recompute, invalidated, refreshed = _patch_common(monkeypatch, tmp_path)
+
+    events: list[tuple[str, int, str]] = []
+
+    def progress(stage, pct, msg, **kw):
+        events.append((stage, pct, msg))
+
+    result = daily_pipeline.run_adj_factor_sync(_repo(tmp_path), _capset(), on_progress=progress)
+
+    assert sync["start_time"] is None, "无本地日K起点时不钳制, 仍拉全历史"
+    assert sync["symbols"] == ["600519.SH", "000001.SZ"]
+    assert recompute["symbols"] == ["600519.SH"], "只重算因子变化个股 (与管道 adj 增量分支同语义)"
+    assert invalidated == ["adj_factor", "enriched"]
+    assert refreshed == ["adj_factor", "kline_enriched"]
+    assert result["adj_factor_symbols"] == 1
+    assert result["enriched_days"] == 7
+    assert result["skipped_stages"] == []
+    assert events[-1][0] == "done" and events[-1][1] == 100
+
+
+def test_window_clamped_to_local_daily_horizon(monkeypatch, tmp_path):
+    """有本地日K时事件窗口钳到其起点 — 更早事件对已存价格零影响 (前复权 cum/total 抵消),
+    拉它们只会为远古事件逐只回补历史K线。"""
+    sync, _, _, _ = _patch_common(monkeypatch, tmp_path)
+
+    repo = _repo(tmp_path, earliest_daily=date(2016, 5, 10))
+    daily_pipeline.run_adj_factor_sync(repo, _capset())
+
+    assert sync["start_time"] == datetime(2016, 5, 10, 0, 0)
+
+
+def test_detail_backfill_does_not_recompute_enriched(monkeypatch, tmp_path):
+    """扩表后重拉全历史但因子值全未变 (仅明细回填) → 不触发 enriched 重算。"""
+    _sync, recompute, invalidated, _ = _patch_common(
+        monkeypatch, tmp_path, written=0, affected=())
+
+    result = daily_pipeline.run_adj_factor_sync(_repo(tmp_path), _capset())
+
+    assert "symbols" not in recompute, "无因子变化时不得调 run_pipeline"
+    assert invalidated == ["adj_factor"], "enriched 未动则不失效"
+    assert result["adj_factor_symbols"] == 0
+    assert result["enriched_days"] == 0
+
+
+def test_skips_recompute_when_enriched_not_built(monkeypatch, tmp_path):
+    """enriched 尚未构建 (首次使用前) → 跳过局部重算, 交由盘后管道全量计算。"""
+    _sync, recompute, invalidated, _ = _patch_common(
+        monkeypatch, tmp_path, with_enriched=False)
+
+    result = daily_pipeline.run_adj_factor_sync(_repo(tmp_path), _capset())
+
+    assert "symbols" not in recompute
+    assert invalidated == ["adj_factor"]
+    assert result["adj_factor_symbols"] == 1  # 因子本身已同步入库
+
+
+def test_no_capability_returns_skipped(monkeypatch, tmp_path):
+    """TickFlow 无除权能力且未路由到其他源 → 跳过, 不发起任何同步。"""
+    def boom(*a, **k):  # pragma: no cover - 触发即失败
+        raise AssertionError("无能力时不得触网")
+
+    monkeypatch.setattr(kline_sync, "sync_adj_factor", boom)
+    monkeypatch.setattr(daily_pipeline._prefs, "get_adj_factor_provider", lambda: "tickflow")
+    monkeypatch.setattr(daily_pipeline, "_invalidate", lambda table=None: None)
+    monkeypatch.setattr(daily_pipeline, "_resolve_universe",
+                        lambda capset, repo=None: ["600519.SH"])
+
+    result = daily_pipeline.run_adj_factor_sync(_repo(tmp_path), _capset(has_adj=False))
+
+    assert result["skipped_stages"] == ["sync_adj"]
+    assert result["adj_factor_symbols"] == 0

backend/tests/test_fuyao_provider.py

@@ -1079,6 +1079,12 @@ def test_tail_ok_weekend_tolerance():
 
 # ---- adj_factor: 推导 ----
 
+# get_adj_factors 输出契约: 规范三列 + 事件明细列 (等差显示/全精度链重建用)
+_ADJ_OUT_COLS = [
+    "symbol", "trade_date", "ex_factor",
+    "dividend", "bonus", "allot", "allot_price", "prev_close",
+]
+
 
 def test_adj_dividend_factor_derivation(monkeypatch):
     """纯分红: P=32.8, D=0.68 → ref=32.12 → factor=32.8/32.12 (与 tickflow 对拍值一致)。"""
@@ -1092,6 +1098,20 @@ def test_adj_dividend_factor_derivation(monkeypatch):
     assert row["ex_factor"] == pytest.approx(32.8 / 32.12)
 
 
+def test_adj_output_keeps_event_details(monkeypatch):
+    """扩表: 输出保留事件明细 (每股分红/送转/配股/前收盘), 供等差显示与全精度链重建。"""
+    events = [("600519.SH", date(2026, 6, 12), 0.68, 0.0, 0.0, 0.0)]
+    bars = {"600519.SH": [_bar(date(2026, 6, 11), 32.8), _bar(date(2026, 6, 12), 32.0)]}
+    provider = _adj_provider(monkeypatch, events, bars)
+    df = provider.get_adj_factors(["600519.SH"], datetime(2026, 1, 1), datetime(2026, 12, 31))
+    row = df.row(0, named=True)
+    assert row["dividend"] == pytest.approx(0.68)
+    assert row["bonus"] == 0.0
+    assert row["allot"] == 0.0
+    assert row["allot_price"] == 0.0
+    assert row["prev_close"] == pytest.approx(32.8)
+
+
 def test_adj_bonus_and_allotment_formula(monkeypatch):
     """送转+配股混合: P=5.23, AR=0.4, AP=3.36 → ref=(5.23+1.344)/1.4=4.6957→4.70(half-up)。"""
     events = [("300176.SZ", date(2026, 8, 21), 0.0, 0.0, 0.4, 3.36)]
@@ -1194,7 +1214,7 @@ def test_adj_output_schema_sorted_and_deduped(monkeypatch):
     }
     provider = _adj_provider(monkeypatch, events, bars)
     df = provider.get_adj_factors(["000001.SZ", "600519.SH"], None, None)
-    assert df.columns == ["symbol", "trade_date", "ex_factor"]
+    assert df.columns == _ADJ_OUT_COLS
     assert df.schema["trade_date"] == pl.Date and df.schema["ex_factor"] == pl.Float64
     assert df.height == 3
     assert df.sort(["symbol", "trade_date"]).equals(df)  # 已排序
@@ -1204,7 +1224,7 @@ def test_adj_output_schema_sorted_and_deduped(monkeypatch):
 def test_adj_etf_and_empty_symbols_return_empty(monkeypatch):
     provider = _adj_provider(monkeypatch, [], {})
     empty = provider.get_adj_factors([], None, None)
-    assert empty.is_empty() and empty.columns == ["symbol", "trade_date", "ex_factor"]
+    assert empty.is_empty() and empty.columns == _ADJ_OUT_COLS
     etf = provider.get_adj_factors(["510300.SH"], None, None, asset_type="etf")
     assert etf.is_empty()
 
@@ -1218,7 +1238,7 @@ def test_adj_dump_unavailable_returns_empty(monkeypatch):
 
     provider._ensure_dump = _boom  # type: ignore[assignment]
     df = provider.get_adj_factors(["600519.SH"], None, None)
-    assert df.is_empty() and df.columns == ["symbol", "trade_date", "ex_factor"]
+    assert df.is_empty() and df.columns == _ADJ_OUT_COLS
 
 
 def test_adj_progress_callback_per_symbol(monkeypatch):
@@ -1289,6 +1309,32 @@ def test_adj_missing_symbol_in_dump_falls_back_to_http(monkeypatch, tmp_path):
     assert set(df["symbol"].to_list()) == {"000001.SZ", "600519.SH"}
 
 
+def test_adj_ancient_event_does_not_drag_universe_to_http(monkeypatch, tmp_path):
+    """一只远古事件(早于 dump 覆盖起点)只拖累自己; dump 覆盖内的标的零请求。
+
+    回归: 配价曾是"全有或全无" — 全市场最早事件早于 dump 起点时整轮放弃
+    dump, 全部标的逐只回补历史K线 (全市场一次 ~15 分钟)。
+    """
+    events = [
+        ("000001.SZ", date(2026, 6, 12), 0.3, 0.0, 0.0, 0.0),
+        ("600519.SH", date(2005, 6, 30), 1.0, 0.0, 0.0, 0.0),  # 早于 dump 起点
+    ]
+    big_rows = [
+        _dump_bar("000001.SZ", date(2026, 5, 10), 10.2),
+        _dump_bar("000001.SZ", date(2026, 6, 11), 10.625),
+        _dump_bar("000001.SZ", date(2026, 6, 12), 9.6),
+    ]
+    provider = _bigdump_provider(monkeypatch, tmp_path, big_rows)
+    provider._dump_memo[fp._ADJ_DUMP_KIND] = _adj_dump(events)
+    provider._get_client().bars["600519.SH"] = [
+        _bar(date(2005, 6, 29), 50.0),
+        _bar(date(2005, 6, 30), 45.0),
+    ]
+    df = provider.get_adj_factors(["000001.SZ", "600519.SH"], None, None)
+    assert [c["thscode"] for c in provider._get_client().calls] == ["600519.SH"]
+    assert "000001.SZ" in set(df["symbol"].to_list())  # dump 路径产出未被连累
+
+
 def test_adj_ten_day_dump_overlays_fresh_ex_close(monkeypatch, tmp_path):
     """大 dump 缺除权日收盘(发布滞后)时由 10d dump 叠加补, 涨跌停自检仍生效。"""
     events = [("000001.SZ", date(2026, 8, 28), 8.0, 0.0, 0.0, 0.0)]  # 异常大额分红

backend/tests/test_pipeline_capacity.py

@@ -166,6 +166,7 @@ def test_in_process_backtests_share_capacity(capacity, monkeypatch, kind):
 
 @pytest.mark.parametrize("endpoint,body,module_name,function_name,result", [
     ("/api/pipeline/run", {}, "app.jobs.daily_pipeline", "run_now", {}),
+    ("/api/pipeline/adj-factor/run", {}, "app.jobs.daily_pipeline", "run_adj_factor_sync", {}),
     ("/api/kline/extend_history", {"value": 1, "unit": "month"},
      "app.services.extend_history", "run_extend_history", {}),
     ("/api/kline/repair_daily", {"start_date": "2026-01-01"},

docs/custom-data-source.md

@@ -118,6 +118,10 @@ datasets:
 | `trade_date` | 除权日期 |
 | `ex_factor` | 复权因子 |
 
+可选明细列 (提供即落库, 缺省为空): `dividend` 每股现金分红、`bonus` 每股送转比例、
+`allot`/`allot_price` 配股比例与配股价、`prev_close` 除权前收盘 — 供等差显示投影
+与全精度因子链重建。
+
 ### realtime 必填
 
 | 内部字段 | 含义 |

docs/plugin-development.md

@@ -147,7 +147,9 @@ class MyProvider:
 
     def get_adj_factors(self, symbols, start_time, end_time, asset_type="stock",
                         on_chunk_done=None) -> pl.DataFrame:
-        """除权因子: [symbol, trade_date, ex_factor]"""
+        """除权因子: [symbol, trade_date, ex_factor]
+        (+可选明细列 dividend/bonus/allot/allot_price/prev_close, 均可空 —
+        提供时用于等差显示投影与全精度因子链重建)"""
 
     def get_minute(self, symbols, start_time, end_time, asset_type="stock",
                    on_chunk_done=None, freq="1m") -> pl.DataFrame:

frontend/src/components/data/ActiveJobCard.tsx

@@ -115,13 +115,17 @@ export function ActiveJobCard({ job }: { job: PipelineJob }) {
         const skipped = new Set(job.result.skipped_stages ?? [])
         const cell = (stage: string | null, v: string) =>
           stage && skipped.has(stage) ? '跳过' : v
+        // 按结果里实际存在的字段渲染: 独立除权因子同步没有日K/分钟K字段, 不显示 "0 天"
+        const pills: { label: string; value: string }[] = []
+        if (job.result.universe_size != null) pills.push({ label: '标的池', value: String(job.result.universe_size) })
+        if (job.result.daily_days != null) pills.push({ label: '日 K', value: cell(null, `${job.result.daily_days} 天`) })
+        if (job.result.adj_factor_symbols != null) pills.push({ label: '除权因子', value: cell('sync_adj', `${job.result.adj_factor_symbols} 只`) })
+        if (job.result.enriched_days != null) pills.push({ label: 'enriched', value: cell(null, `${job.result.enriched_days} 行`) })
+        if (job.result.minute_rows != null) pills.push({ label: '分钟K', value: cell('sync_minute', `${job.result.minute_rows} 行`) })
+        if (pills.length === 0) return null
         return (
           <div className="mt-3 grid grid-cols-2 md:grid-cols-5 gap-3 text-xs">
-            <Pill label="标的池" value={job.result.universe_size ?? '—'} />
-            <Pill label="日 K" value={cell(null, `${job.result.daily_days ?? 0} 天`)} />
-            <Pill label="除权因子" value={cell('sync_adj', `${job.result.adj_factor_symbols ?? 0} 只`)} />
-            <Pill label="enriched" value={cell(null, `${job.result.enriched_days ?? 0} 行`)} />
-            <Pill label="分钟K" value={cell('sync_minute', `${job.result.minute_rows ?? 0} 行`)} />
+            {pills.map(p => <Pill key={p.label} {...p} />)}
           </div>
         )
       })()}

frontend/src/components/data/AdjFactorSyncConfig.tsx

@@ -0,0 +1,76 @@
+import { useState } from 'react'
+import { useQueryClient } from '@tanstack/react-query'
+import { Loader2, Download, Info } from 'lucide-react'
+import { api } from '@/lib/api'
+import { QK } from '@/lib/queryKeys'
+import { MissingCapChip } from '@/lib/capability-labels'
+
+// 除权因子独立同步面板 (数据画像卡齿轮弹窗, 模式与 MinuteSyncConfig 一致)。
+// hasCap: 路由矩阵判定除权因子能力可用 (生效源含插件/自定义源);
+// providerDisplay: 生效源展示名, 路由为 TickFlow 时为 null。
+export function AdjFactorSyncConfig({ hasCap, providerDisplay, onJobStart }: {
+  hasCap: boolean
+  providerDisplay: string | null
+  onJobStart?: (jobId: string) => void
+}) {
+  const qc = useQueryClient()
+  const [starting, setStarting] = useState(false)
+
+  const handleSync = () => {
+    if (!hasCap || starting) return
+    setStarting(true)
+    api.pipelineAdjFactorRun().then((res) => {
+      qc.invalidateQueries({ queryKey: QK.pipelineJobs })
+      qc.invalidateQueries({ queryKey: QK.dataStatus })
+      // 通知主页面跟踪 job 进度 (顶部 ActiveJobCard 显示实时进度+日志)
+      if (res.job_id && onJobStart) onJobStart(res.job_id)
+    }).finally(() => setStarting(false))
+  }
+
+  return (
+    <div className="px-4 pb-4 pt-3 border-t border-accent/20 space-y-3">
+      {/* 生效数据源 */}
+      <div className="flex items-center justify-between">
+        <span className="text-xs text-foreground font-medium">生效数据源</span>
+        <div className="flex items-center gap-2">
+          <span className="text-[11px] text-secondary font-mono">{providerDisplay ?? 'TickFlow'}</span>
+          {!hasCap && <MissingCapChip capKey="adj_factor" />}
+        </div>
+      </div>
+
+      {/* 手动同步 (全历史, 与盘后管道共用单飞任务槽) */}
+      <div className="pt-3 border-t border-border space-y-2">
+        <div className="flex items-center gap-1.5">
+          <Download className="h-3 w-3 text-secondary" />
+          <span className="text-[11px] text-secondary font-medium">手动同步</span>
+          <span className="text-[10px] text-muted">与盘后管道共用任务槽</span>
+        </div>
+        <button
+          onClick={handleSync}
+          disabled={!hasCap || starting}
+          className="w-full inline-flex items-center justify-center gap-1.5 px-3 py-2 rounded-btn bg-accent/90 text-foreground text-xs font-medium hover:bg-accent disabled:opacity-40 transition-colors duration-150"
+        >
+          {starting ? (
+            <><Loader2 className="h-3.5 w-3.5 animate-spin" /><span>启动中…</span></>
+          ) : (
+            <><Download className="h-3.5 w-3.5" /><span>全历史同步除权因子</span></>
+          )}
+        </button>
+        <div className="flex items-start gap-1.5 text-[10px] text-muted leading-relaxed">
+          <Info className="h-3 w-3 shrink-0 mt-px" />
+          <span>
+            拉取除权事件(等比因子 + 分红/送转/配股明细),窗口按需钳到本地日K起点 —
+            更早的事件对已存价格零影响;因子发生变化的个股自动重算前复权,
+            仅明细回填不触发重算。幂等,可重复执行。
+          </span>
+        </div>
+        {providerDisplay == null && (
+          <div className="text-[10px] text-muted leading-relaxed">
+            当前路由 TickFlow:只返回等比因子,无事件明细。需要明细时,在
+            设置 → 数据源 将「除权因子」切换到提供明细的数据源后重跑。
+          </div>
+        )}
+      </div>
+    </div>
+  )
+}

frontend/src/components/data/SectionTitle.tsx

@@ -40,6 +40,8 @@ export function HistoryRow({ job, onClick }: { job: any; onClick: () => void })
           const r = job.result as Record<string, any>
           const parts: string[] = []
           if (r.daily_days != null) parts.push(`日K ${r.daily_days}日`)
+          // 独立除权因子同步任务 (无日K字段): 摘要以除权因子开头便于区分
+          if (r.adj_factor_symbols != null && r.daily_days == null) parts.push(`除权因子 ${r.adj_factor_symbols}只`)
           if (r.enriched_days != null) parts.push(`enriched ${r.enriched_days}行`)
           if (r.minute_rows != null) parts.push(`分钟K ${r.minute_rows}行`)
           if (r.earliest_after && r.earliest_before) {

frontend/src/lib/api.ts

@@ -3173,6 +3173,10 @@ export const api = {
   pipelineRun: () => request<{ job_id: string; reused: boolean }>(
     '/api/pipeline/run', { method: 'POST' },
   ),
+  /** 独立同步除权因子 (全历史 + 受影响个股 enriched 局部重算), 与管道共用任务槽 */
+  pipelineAdjFactorRun: () => request<{ job_id: string; reused: boolean }>(
+    '/api/pipeline/adj-factor/run', { method: 'POST' },
+  ),
   pipelineJob: (id: string) => request<PipelineJob>(`/api/pipeline/jobs/${id}`),
   /** 手动停止一个 running/pending 的同步任务 (协作式: 当前分块完成后线程自行退出) */
   pipelineJobCancel: (id: string) =>
@@ -4063,13 +4067,15 @@ export interface PipelineJob {
   finished_at: string | null
   duration_s: number | null
   result: {
-    universe_size: number
-    daily_days: number
-    adj_factor_symbols: number
-    enriched_days: number
+    universe_size?: number
+    daily_days?: number
+    adj_factor_symbols?: number
+    /** 独立除权因子同步: 本轮写入/合并的因子行数 */
+    adj_written?: number
+    enriched_days?: number
     index_count?: number
     index_daily_rows?: number
-    minute_rows: number
+    minute_rows?: number
     skipped_stages?: string[]
   } | null
   error: string | null

frontend/src/pages/Data.tsx

@@ -31,7 +31,7 @@ import {
   useDataStatus,
 } from '@/lib/useSharedQueries'
 import { useToggleRealtimeQuotes, useUpdateQuoteInterval } from '@/lib/useSharedMutations'
-import { MissingCapChip, routeCapUsable, routeProviderDisplay, type RouteCapId } from '@/lib/capability-labels'
+import { MissingCapChip, routeCap, routeCapUsable, routeProviderDisplay, type RouteCapId } from '@/lib/capability-labels'
 import { QK } from '@/lib/queryKeys'
 import { PageHeader } from '@/components/PageHeader'
 import { useAdjFactorSyncGate } from '@/components/AdjFactorSyncGate'
@@ -48,6 +48,7 @@ import { ExtendHistoryPanel } from '@/components/data/ExtendHistoryPanel'
 import { RepairDailyPanel } from '@/components/data/RepairDailyPanel'
 import { EnrichedRebuildPanel } from '@/components/data/EnrichedRebuildPanel'
 import { MinuteSyncConfig } from '@/components/data/MinuteSyncConfig'
+import { AdjFactorSyncConfig } from '@/components/data/AdjFactorSyncConfig'
 import { RegimeConfigCard } from '@/components/data/RegimeConfigCard'
 import { PipelineScopeConfig } from '@/components/data/PipelineScopeConfig'
 import { PageSettingsModal, getCardVisibility, getCardOrder, type CardKey } from '@/components/data/PageSettingsModal'
@@ -261,13 +262,14 @@ export function Data() {
   const hasMinuteCap = usableOr('minute', !!tfCaps?.['kline.minute.batch'])
   const indexAuto = prefs.data?.pipeline_pull_index ?? true
   const etfAuto = prefs.data?.pipeline_pull_etf ?? false
-  const pipelineSteps = [
-    '日K',
-    ...(hasAdjCap ? ['复权'] : []),
-    '指标',
-    ...(indexAuto ? ['指数'] : []),
-    ...(etfAuto ? ['ETF'] : []),
-    ...((hasMinuteCap && minuteAuto) ? ['分钟K'] : []),
+  // 管道步骤 + 各步能力路由 (路由徽标数据源 = 能力矩阵, 单一权威; 指标为本地计算无路由)
+  const pipelineSteps: { label: string; cap?: RouteCapId }[] = [
+    { label: '日K', cap: 'daily' },
+    ...(hasAdjCap ? [{ label: '复权', cap: 'adj_factor' as const }] : []),
+    { label: '指标' },
+    ...(indexAuto ? [{ label: '指数', cap: 'daily' as const }] : []),
+    ...(etfAuto ? [{ label: 'ETF', cap: 'daily' as const }] : []),
+    ...((hasMinuteCap && minuteAuto) ? [{ label: '分钟K', cap: 'minute' as const }] : []),
   ]
 
   // 数据画像卡片显隐(由页面设置弹窗控制,存 localStorage)
@@ -456,6 +458,8 @@ export function Data() {
             customProvider={routeProviderDisplay(matrix.data, 'adj_factor')}
             auto
             onShowFields={() => setSchemaTable('adj_factor')}
+            onSettings={() => setOpenSettings(v => v === 'adj_factor' ? null : 'adj_factor')}
+            settingsOpen={openSettings === 'adj_factor'}
           />
         )
       case 'enriched':
@@ -741,15 +745,27 @@ export function Data() {
               </div>
             ) : (
               <div className="space-y-2">
-                <div className="flex items-center gap-1.5 text-[10px] text-muted pb-2 border-b border-border/50">
+                <div className="flex items-center gap-1.5 text-[10px] text-muted pb-2 border-b border-border/50 flex-wrap">
                   <span className="text-accent/60 font-medium">盘前</span>
                   <span>个股维表</span>
                   <span className="text-border">→</span>
                   <span className="text-accent/60 font-medium">盘后</span>
                   {pipelineSteps.map((step, i) => (
-                    <span key={step} className="contents">
+                    <span key={step.label} className="contents">
                       {i > 0 && <span className="text-border">→</span>}
-                      <span>{step}</span>
+                      <span
+                        className="inline-flex items-center gap-0.5"
+                        title={step.cap
+                          ? `路由: ${routeCap(matrix.data, step.cap)?.effective_display ?? '—'}`
+                          : '本地计算'}
+                      >
+                        {step.label}
+                        {step.cap && (
+                          <span className="text-muted/60">
+                            ·{routeCap(matrix.data, step.cap)?.effective_display ?? '…'}
+                          </span>
+                        )}
+                      </span>
                     </span>
                   ))}
                 </div>
@@ -1177,6 +1193,18 @@ export function Data() {
         )}
       </AnimatePresence>
 
+      <AnimatePresence>
+        {openSettings === 'adj_factor' && (
+          <SettingsModal title="除权因子 · 同步设置" onClose={() => setOpenSettings(null)}>
+            <AdjFactorSyncConfig
+              hasCap={hasAdjCap}
+              providerDisplay={routeProviderDisplay(matrix.data, 'adj_factor')}
+              onJobStart={(jobId) => { setActiveJobId(jobId); setOpenSettings(null) }}
+            />
+          </SettingsModal>
+        )}
+      </AnimatePresence>
+
       {/* 停止同步二次确认弹窗 */}
       <AnimatePresence>
         {showStopConfirm && (