feat(kline): 分钟K原始口径落盘+共享复权因子表, 读取时即时前复权投影

解决除权后已落盘分钟数据(前复权口径)与最新数据不可比的问题:
- 新增 minute_adjust 服务: 除权因子表 + 读取路径投影, ratio =
  cum_factor(D)/total_factor 作用于价格与成交额, ratio=1 短路直读
- kline_sync/repository: 数据目录存在 .raw_basis 标记时分钟拉取改用
  adjust='none' 落盘原始 OHLC; 无标记保持旧前复权行为, 字节级兼容
- api/kline: 新增 POST /api/kline/minute-migrate 幂等迁移, 将混合口径
  分区转为原始口径后写入标记启用新架构
- 测试: 新增 test_minute_adjust (9 例); 5 个既有分钟测试适配新口径

全量后端测试通过 (2239 passed, 1 例既有偶发 flaky 与本次无关)。

backend/app/api/kline.py

@@ -18,6 +18,7 @@ from app.market_time import cn_now, cn_today, in_continuous_session
 from app.price_limits import is_risk_warning_name, price_limit_pct
 from app.db_safe import is_valid_ext_ident
 from app.services import kline_sync, trading_day
+from app.services import minute_adjust
 
 logger = logging.getLogger(__name__)
 
@@ -811,6 +812,7 @@ def get_minute_batch(request: Request, body: dict):
     def _pull(asset: str, sym_list: list[str], start: datetime) -> None:
         if not sym_list:
             return
+        raw_basis = minute_adjust.minute_basis_is_raw(repo.store.data_dir)
         df_live = kline_sync.sync_minute_batch(
             sym_list,
             start_time=start,
@@ -818,18 +820,25 @@ def get_minute_batch(request: Request, body: dict):
             batch_size=lim.batch if lim else None,
             rpm=lim.rpm if lim else None,
             asset_type=asset,
+            raw_basis=raw_basis,
         )
         if df_live.is_empty():
             return
         try:
             # 读-改-写必须持仓库写锁 (与全量分钟服务/盘后同步同一纪律, Windows 临时文件占用)。
             # 仅在拿到真实目录时落盘: data_dir 异常 (非 Path) 时跳过, 只返回本轮数据。
+            # 落盘必须是原始口径 (raw_basis=True 时 sync 已按 adjust='none' 取回);
+            # 对外响应再统一复权投影。
             minute_dir = minute_dirs[asset]
             if isinstance(minute_dir, Path):
                 with repo._write_lock:
                     kline_sync._write_minute_partition(df_live, minute_dir)
         except Exception as e:  # noqa: BLE001
             logger.warning("minute-batch 补拉落盘失败 (降级为仅返回): %s", e)
+        if raw_basis:
+            df_live = minute_adjust.apply_minute_adjustment(
+                df_live, repo.store.data_dir, asset,
+            )
         for part in df_live.partition_by("symbol", maintain_order=True):
             live_map[part["symbol"][0]] = part.sort("datetime")
 
@@ -973,6 +982,7 @@ def get_minute(
     """
     repo = request.app.state.repo
     capset = request.app.state.capabilities
+    raw_basis = minute_adjust.minute_basis_is_raw(repo.store.data_dir)
     asset_type = repo.resolve_asset_type(symbol)
     stock_info = _get_stock_info(repo, symbol) if asset_type == "stock" else _get_asset_info(repo, symbol, asset_type)
     stock_name = stock_info.get("name")
@@ -1002,7 +1012,10 @@ def get_minute(
         trade_date = cn_today()
         df = kline_sync.fetch_minute_single(
             symbol, trade_date, asset_type=asset_type, capset=capset,
+            raw_basis=raw_basis,
         )
+        if raw_basis:
+            df = minute_adjust.apply_minute_adjustment(df, repo.store.data_dir, asset_type)
         price_limit = _get_price_limit_info(
             repo, symbol, trade_date, asset_type, stock_name,
         )
@@ -1034,8 +1047,13 @@ def get_minute(
         # 时段边界)则落回下方本地优先路径。
         live_df = kline_sync.fetch_minute_single(
             symbol, trade_date, asset_type=asset_type, capset=capset,
+            raw_basis=raw_basis,
         )
         if not live_df.is_empty():
+            if raw_basis:
+                live_df = minute_adjust.apply_minute_adjustment(
+                    live_df, repo.store.data_dir, asset_type,
+                )
             return _gzip_payload(
                 request,
                 {
@@ -1084,7 +1102,12 @@ def get_minute(
     # 本地不完整或无数据 → 从当前有效分钟源实时拉取
     live_df = kline_sync.fetch_minute_single(
         symbol, trade_date, asset_type=asset_type, capset=capset,
+        raw_basis=raw_basis,
     )
+    if raw_basis and not live_df.is_empty():
+        live_df = minute_adjust.apply_minute_adjustment(
+            live_df, repo.store.data_dir, asset_type,
+        )
     return _gzip_payload(
         request,
         {
@@ -1244,7 +1267,7 @@ async def sync_minute(request: Request):
 
 @router.post("/sync_minute_single")
 async def sync_minute_single(request: Request, body: dict):
-    """手动拉取单只股票的分钟K并落库 (前复权)。
+    """手动拉取单只股票的分钟K并落库 (口径随基准标记: 未迁移=前复权, 已迁移=原始)。
 
     body: { "symbol": "000001.SZ" }
     用于个股分时图"获取数据"按钮: 本地无数据时单独拉取并持久化。
@@ -1290,6 +1313,33 @@ async def sync_minute_single(request: Request, body: dict):
     return {"status": "ok", "symbol": symbol, "rows": written}
 
 
+@router.post("/minute-migrate")
+async def minute_migrate(request: Request):
+    """存量分钟K迁移为原始口径并启用读取时复权 (幂等, 见 services/minute_adjust)。
+
+    - 用日K原始收盘价做锚点, 把历史"拉取时前复权"的分区换算回原始价;
+    已是原始价的行 (全量分钟落盘) 锚点 k≈1 自动跳过;
+    - 全部分区成功后创建 .raw_basis 标记: 此后读取自动复权投影、拉取改取原始价;
+    - 失败不标记, 可重复调用续跑 (已换算分区二次运行为 no-op)。
+    返回 { partitions, converted_symbols, skipped_no_daily, failed, marked }。
+    """
+    import asyncio
+
+    repo = request.app.state.repo
+    loop = asyncio.get_event_loop()
+    stats = await loop.run_in_executor(
+        _long_task_executor,
+        lambda: minute_adjust.migrate_minute_to_raw(
+            repo.store.data_dir, write_lock=repo._write_lock,
+        ),
+    )
+    if stats.get("marked"):
+        from app.jobs.daily_pipeline import _refresh_single_view
+        _refresh_single_view(repo, "kline_minute")
+        _refresh_single_view(repo, "kline_etf_minute")
+    return {"status": "ok", **stats}
+
+
 @router.post("/clear_minute")
 async def clear_minute(request: Request):
     """清空全部分钟K数据 (仅 kline_minute, 不影响其他数据)。

backend/app/services/kline_sync.py

@@ -20,7 +20,7 @@ import polars as pl
 from app.data_providers.base import AssetType
 from app.indicators.pipeline import filter_halt_days
 from app.market_time import CN_TZ, cn_now, cn_today
-from app.services import preferences
+from app.services import minute_adjust, preferences
 from app.tickflow.capabilities import Cap, CapabilitySet
 from app.tickflow.client import get_client
 from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches
@@ -931,10 +931,11 @@ def sync_minute_batch(
     count: int | None = None,
     batch_size: int | None = None,
     rpm: int | None = None,
-    on_chunk_done: Callable[[int, int, str], None] | None = None,
+    on_chunk_done: Callable[[int, int, str] | None] | None = None,
     segment_trading_days: int = 20,
     on_segment: Callable[[pl.DataFrame], None] | None = None,
     asset_type: AssetType = "stock",
+    raw_basis: bool = False,
 ) -> pl.DataFrame:
     """批量拉取多股分钟 K。
 
@@ -1015,12 +1016,12 @@ def sync_minute_batch(
                         start_time=_datetime_to_ms(cur_start),
                         end_time=_datetime_to_ms(cur_end),
                         count=10000,
-                        adjust="forward",
+                        adjust="none" if raw_basis else "forward",
                         as_dataframe=False, show_progress=False,
                     )
                 else:
                     raw = tf.klines.batch(chunk, period="1m", count=count or 1200,
-                                          adjust="forward",
+                                          adjust="none" if raw_basis else "forward",
                                           as_dataframe=False, show_progress=False)
             except Exception as e:  # noqa: BLE001
                 logger.warning("minute batch fetch failed for %d symbols: %s", len(chunk), e)
@@ -1345,6 +1346,7 @@ def fetch_minute_single(
     asset_type: AssetType = "stock",
     *,
     capset: CapabilitySet,
+    raw_basis: bool = False,
 ) -> pl.DataFrame:
     """实时拉取单股单日分钟 K(不写入本地)。
 
@@ -1377,7 +1379,7 @@ def fetch_minute_single(
             start_time=_datetime_to_ms(start_time),
             end_time=_datetime_to_ms(end_time),
             count=10000,
-            adjust="forward",
+            adjust="none" if raw_basis else "forward",
             as_dataframe=False, show_progress=False,
         )
     except Exception as e:
@@ -1528,8 +1530,10 @@ def sync_and_persist_minute(
     extend_backward: bool = False,
     force_full_days: bool = False,
 ) -> int:
-    """同步分钟 K 并存到 Parquet(前复权价格, SDK 端 adjust=qfq)。返回写入行数。
+    """同步分钟 K 并存到 Parquet。返回写入行数。
 
+    存储口径由基准标记决定 (services/minute_adjust): 存量未迁移 → SDK adjust=qfq
+    前复权 (旧行为); 已迁移 → adjust='none' 原始价落盘, 复权读取时投影。
     使用 start_time / end_time 区间拉取, 确保所有标的覆盖同一时间段。
     on_chunk_done(current, total) 每个 chunk 完成后回调。
     force_full_days=True 时强制回溯 days 自然日 (不增量补, 用于个股补齐历史)。
@@ -1613,6 +1617,7 @@ def sync_and_persist_minute(
         segment_trading_days=segment_days,
         on_segment=_persist,
         asset_type="stock",
+        raw_basis=minute_adjust.minute_basis_is_raw(repo.store.data_dir),
     )
 
     if written_box[0] == 0:

backend/app/services/minute_adjust.py

@@ -0,0 +1,285 @@
+"""分钟K复权投影与原始基准迁移 (与日K复权架构同构, 见 CONTRIBUTING §3.2)。
+
+三层架构:
+- 事实层: kline_minute / kline_etf_minute 存原始价 + 原始 amount (拉取 adjust='none',
+  落盘后不可变)。原始价与成交额天然同口径, 均价线在存储层即正确。
+- 因子层: 共享日K的 adj_factor (股票) / adj_factor_etf (ETF), 只追加。
+- 投影层: 读取时复权 ratio = 截至 D 日累积因子 / 全部累积因子, 价格与 amount 同乘,
+  volume 不变。amount 同乘保证 VWAP 与复权价严格同口径。当日 ratio 恒为 1
+  (今天不可能有未来事件), 盘中实时路径整体短路。
+
+基准标记 data/kline_minute/.raw_basis:
+- 存在 → 分区已是原始口径: 读取应用复权投影, TickFlow 拉取 adjust='none'。
+- 不存在 → 旧基准 (拉取时前复权, 混合口径): 读取原样返回, 拉取维持 adjust='forward'。
+- 迁移任务把全部分区换算为原始口径后创建标记; 失败不标记, 幂等可续跑。
+
+性能 (2026-09 本机实测, 133 万行/日真实分区):
+- 单股单日 240 根: 复权 +6~9ms; 当日全市场: ratio 全 1 短路, 开销为零。
+- 实现为「去重 (symbol,date) 小表 asof → hash join 贴回 → ratio=1 短路」三层。
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from datetime import date
+from pathlib import Path
+
+import polars as pl
+
+logger = logging.getLogger(__name__)
+
+PRICE_COLS = ("open", "high", "low", "close")
+ADJUSTED_COLS = ("open", "high", "low", "close", "amount")
+
+_MARKER_NAME = ".raw_basis"
+
+# 因子表内存缓存: (data_dir, 表名) -> (源文件最新 mtime, fs帧, total帧)。
+# 因子表仅数千行 (全市场一年 ~5000 事件), 读盘 ~10ms, 按 mtime 失效。
+_factor_cache: dict[tuple[str, str], tuple[float, pl.DataFrame, pl.DataFrame]] = {}
+_factor_cache_lock = threading.Lock()
+
+
+def minute_dir(data_dir: Path | str) -> Path:
+    return Path(data_dir) / "kline_minute"
+
+
+def marker_path(data_dir: Path | str) -> Path:
+    return minute_dir(data_dir) / _MARKER_NAME
+
+
+def minute_basis_is_raw(data_dir: Path | str) -> bool:
+    """分区数据是否已迁移为原始口径 (标记文件存在)。"""
+    return marker_path(data_dir).exists()
+
+
+def mark_minute_basis_raw(data_dir: Path | str) -> None:
+    """迁移完成后创建标记。目录不存在时一并创建 (空库直接切新基准)。"""
+    p = marker_path(data_dir)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.touch()
+
+
+def minute_fetch_adjust(data_dir: Path | str) -> str:
+    """TickFlow 分钟拉取的 adjust 参数: 已迁移 → 'none' (原始), 否则 'forward' (旧行为)。"""
+    return "none" if minute_basis_is_raw(data_dir) else "forward"
+
+
+def _factor_table_dir(data_dir: Path, asset_type: str) -> Path:
+    return data_dir / ("adj_factor_etf" if asset_type == "etf" else "adj_factor")
+
+
+def _load_factor_frames(data_dir: Path, asset_type: str) -> tuple[pl.DataFrame, pl.DataFrame]:
+    """读因子表并构建 (symbol, trade_date, cum) 与 (symbol, total) 两帧, 按 mtime 缓存。
+
+    返回空帧表示该资产类型无因子数据 (调用方按恒等处理)。
+    """
+    table_dir = _factor_table_dir(data_dir, asset_type)
+    files = sorted(table_dir.glob("*.parquet")) if table_dir.exists() else []
+    if not files:
+        return (pl.DataFrame(), pl.DataFrame())
+    newest = max(f.stat().st_mtime for f in files)
+    key = (str(data_dir), table_dir.name)
+    with _factor_cache_lock:
+        cached = _factor_cache.get(key)
+        if cached is not None and cached[0] == newest:
+            return cached[1], cached[2]
+
+    factors = pl.read_parquet([str(f) for f in files])
+    if factors.is_empty() or not {"symbol", "trade_date", "ex_factor"}.issubset(factors.columns):
+        frames = (pl.DataFrame(), pl.DataFrame())
+    else:
+        fs = (
+            factors.with_columns(
+                pl.col("trade_date").cast(pl.Date, strict=False),
+                pl.col("ex_factor").cast(pl.Float64, strict=False),
+            )
+            .select("symbol", "trade_date", "ex_factor")
+            .drop_nulls()
+            .unique(subset=["symbol", "trade_date"])
+            .sort(["symbol", "trade_date"])
+            .with_columns(pl.col("ex_factor").cum_prod().over("symbol").alias("cum"))
+        )
+        total = fs.group_by("symbol").agg(pl.col("cum").last().alias("total"))
+        frames = (fs.select("symbol", "trade_date", "cum"), total)
+    with _factor_cache_lock:
+        _factor_cache[key] = (newest, frames[0], frames[1])
+    return frames
+
+
+def compute_ratio_pairs(
+    df: pl.DataFrame,
+    data_dir: Path | str,
+    asset_type: str = "stock",
+) -> pl.DataFrame:
+    """为 df 中出现的去重 (symbol, 交易日) 对计算复权 ratio。
+
+    返回 schema: symbol, _d (Date), _ratio。无因子/无事件时 _ratio 恒为 1。
+    """
+    if asset_type == "index" or df.is_empty() or "datetime" not in df.columns:
+        return pl.DataFrame(schema={"symbol": pl.String, "_d": pl.Date, "_ratio": pl.Float64})
+    fs, total = _load_factor_frames(Path(data_dir), asset_type)
+    pairs = df.select(pl.col("symbol"), pl.col("datetime").dt.date().alias("_d")).unique()
+    if fs.is_empty():
+        return pairs.with_columns(pl.lit(1.0).alias("_ratio"))
+    pairs = (
+        pairs.sort(["symbol", "_d"])
+        .join_asof(fs, left_on="_d", right_on="trade_date", by="symbol", strategy="backward", check_sortedness=False)
+        .join(total, on="symbol", how="left")
+        .with_columns(
+            (pl.col("cum").fill_null(1.0) / pl.col("total").fill_null(1.0)).alias("_ratio")
+        )
+        .select("symbol", "_d", "_ratio")
+    )
+    return pairs
+
+
+def apply_minute_adjustment(
+    df: pl.DataFrame,
+    data_dir: Path | str,
+    asset_type: str = "stock",
+) -> pl.DataFrame:
+    """读取时复权投影: 价格与 amount 乘以 ratio (volume 不变), 纯函数无副作用。
+
+    - 指数无复权概念, 原样返回;
+    - 因子表为空 → ratio 恒 1, 原样返回 (fail-open, 与日K缺因子语义一致);
+    - 当日/无事件数据 ratio 全 1, 短路返回原帧 (零拷贝)。
+    """
+    if df.is_empty() or asset_type == "index":
+        return df
+    cols = [c for c in ADJUSTED_COLS if c in df.columns]
+    if not cols or "datetime" not in df.columns:
+        return df
+    pairs = compute_ratio_pairs(df, data_dir, asset_type)
+    if pairs.is_empty() or pairs["_ratio"].min() >= 1.0 - 1e-12:
+        return df
+    out = (
+        df.with_columns(pl.col("datetime").dt.date().alias("_d"))
+        .join(pairs, on=["symbol", "_d"], how="left")
+        .with_columns([pl.col(c) * pl.col("_ratio").fill_null(1.0) for c in cols])
+        .drop(["_d", "_ratio"])
+    )
+    return out
+
+
+# ================================================================
+# 存量迁移: 拉取时前复权的旧分区 → 原始口径 (日K原始收盘价锚点换算)
+# ================================================================
+
+def _daily_close_lookup(daily_dir: Path, trade_day: date) -> dict[str, float] | None:
+    """读某交易日日K分区的原始收盘价 {symbol: close}。分区缺失返回 None。"""
+    part = daily_dir / f"date={trade_day.isoformat()}" / "part.parquet"
+    if not part.exists():
+        return None
+    try:
+        df = pl.read_parquet(part, columns=["symbol", "close"])
+    except Exception as e:
+        logger.warning("迁移读日K分区失败 %s: %s", part, e)
+        return None
+    if df.is_empty():
+        return None
+    return dict(zip(df["symbol"].to_list(), df["close"].to_list(), strict=True))
+
+
+def _partition_anchor_ratio(minute_df: pl.DataFrame, daily_close: dict[str, float]) -> pl.DataFrame:
+    """按 (symbol) 计算锚点 k = 当日最后一根分钟 close / 日K原始 close。
+
+    该分区单日, (symbol, date) 即 symbol。返回 schema: symbol, _k。
+    k≈1 → 已是原始价; k 偏离 1 → 该 symbol 为拉取时前复权数据, 价格需除以 k。
+    """
+    last_close = (
+        minute_df.sort("datetime")
+        .group_by("symbol")
+        .agg(pl.col("close").last().alias("_last"))
+    )
+    anchors = (
+        pl.DataFrame({"symbol": list(daily_close.keys()), "_raw": list(daily_close.values())})
+        .join(last_close, on="symbol", how="inner")
+        .with_columns((pl.col("_last") / pl.col("_raw")).alias("_k"))
+        .filter(
+            pl.col("_k").is_finite()
+            & (pl.col("_k") > 0.2)
+            & (pl.col("_k") < 5.0)   # 锚点换算的合理界: 远超个股除权幅度视为数据异常
+        )
+        .select("symbol", "_k")
+    )
+    return anchors
+
+
+def migrate_minute_to_raw(
+    data_dir: Path | str,
+    write_lock: threading.Lock | None = None,
+    progress_cb=None,
+) -> dict:
+    """把 kline_minute / kline_etf_minute 全部分区换算为原始口径, 成功后创建基准标记。
+
+    换算法: 每 (symbol, 分区日) 的锚点 k = 最后一根分钟 close ÷ 日K原始 close;
+    已是原始价的行 k≈1 不动, 前复权价除以 k 还原为原始价。amount 本就是原始值, 不动。
+    幂等: 已换算分区二次运行 k≈1 全程 no-op; 全部成功才写标记, 失败可续跑。
+
+    返回统计: partitions / converted_symbols / skipped_no_daily / failed / marked。
+    """
+    data_dir = Path(data_dir)
+    stats = {"partitions": 0, "converted_symbols": 0, "skipped_no_daily": 0, "failed": 0, "marked": False}
+    if minute_basis_is_raw(data_dir):
+        stats["marked"] = True
+        return stats
+
+    jobs: list[tuple[Path, Path]] = [
+        (data_dir / "kline_minute", data_dir / "kline_daily"),
+        (data_dir / "kline_etf_minute", data_dir / "kline_etf_daily"),
+    ]
+    for minute_root, daily_root in jobs:
+        if not minute_root.exists():
+            continue
+        partitions = sorted(p for p in minute_root.glob("date=*") if (p / "part.parquet").exists())
+        for part_dir in partitions:
+            trade_day = date.fromisoformat(part_dir.name[5:])
+            stats["partitions"] += 1
+            try:
+                daily_close = _daily_close_lookup(daily_root, trade_day)
+                if not daily_close:
+                    # 无日K锚点的分区无法换算, 不算失败: 留待日K补齐后续跑
+                    stats["skipped_no_daily"] += 1
+                    continue
+                part = part_dir / "part.parquet"
+                df = pl.read_parquet(part)
+                if df.is_empty() or "close" not in df.columns:
+                    continue
+                anchors = _partition_anchor_ratio(df, daily_close)
+                conv = anchors.filter((pl.col("_k") - 1.0).abs() > 1e-9)
+                if not conv.is_empty():
+                    fixed = (
+                        df.join(conv, on="symbol", how="left")
+                        .with_columns(
+                            [pl.col(c) / pl.col("_k").fill_null(1.0) for c in PRICE_COLS if c in df.columns]
+                        )
+                        .drop("_k")
+                        .sort(["symbol", "datetime"])
+                    )
+                    tmp = part.with_suffix(".parquet.tmp")
+                    fixed.write_parquet(tmp)
+                    if write_lock is not None:
+                        with write_lock:
+                            tmp.replace(part)
+                    else:
+                        tmp.replace(part)
+                    stats["converted_symbols"] += conv.height
+            except Exception as e:
+                stats["failed"] += 1
+                logger.warning("分钟分区迁移失败 %s: %s", part_dir.name, e)
+            if progress_cb is not None:
+                progress_cb(stats)
+
+    if stats["failed"] == 0:
+        # 有分区但全部缺日K锚点时不下标记 (数据未真正收敛), 空库可直接切换
+        total_actionable = stats["partitions"] - stats["skipped_no_daily"]
+        if total_actionable == 0 and stats["partitions"] > 0:
+            return stats
+        mark_minute_basis_raw(data_dir)
+        stats["marked"] = True
+        logger.info(
+            "分钟K已迁移为原始口径: %d 分区, 换算 %d symbol·日, 跳过 %d",
+            stats["partitions"], stats["converted_symbols"], stats["skipped_no_daily"],
+        )
+    return stats

backend/app/tickflow/repository.py

@@ -35,6 +35,7 @@ from app.enriched_generation import (
 from app.market_time import cn_today
 from app.parquet import scan_enriched_parquet
 from app.polars_guard import guarded_collect
+from app.services.minute_adjust import apply_minute_adjustment, minute_basis_is_raw
 
 logger = logging.getLogger(__name__)
 
@@ -1608,6 +1609,20 @@ class KlineRepository:
         """按资产类型选择分钟K parquet glob。ETF 分钟数据独立存储于 kline_etf_minute。"""
         return self._etf_minute_glob if asset_type == "etf" else self._minute_glob
 
+    def _maybe_adjust_minute(self, df: pl.DataFrame, asset_type: str) -> pl.DataFrame:
+        """原始基准标记开启时应用读取时复权投影 (services/minute_adjust 三层架构)。
+
+        标记未开启 (存量未迁移) 原样返回, 行为与旧版逐字节一致; 投影异常也按原样
+        返回 (fail-open, 与日K缺因子语义一致), 不让复权层破坏数据可用性。
+        """
+        if df.is_empty() or not minute_basis_is_raw(self.store.data_dir):
+            return df
+        try:
+            return apply_minute_adjustment(df, self.store.data_dir, asset_type)
+        except Exception as e:
+            logger.warning("分钟复权投影失败, 按原始数据返回: %s", e)
+            return df
+
     def get_minute(
         self,
         symbol: str,
@@ -1616,12 +1631,13 @@ class KlineRepository:
     ) -> pl.DataFrame:
         """分钟K查询 — Polars scan_parquet + predicate pushdown。"""
         try:
-            return guarded_collect(
+            df = guarded_collect(
                 pl.scan_parquet(self._minute_glob_for(asset_type)).filter(
                     (pl.col("symbol") == symbol)
                     & (pl.col("datetime").dt.date() == trade_date)
                 ).sort("datetime")
             )
+            return self._maybe_adjust_minute(df, asset_type)
         except Exception as e:  # noqa: BLE001
             logger.warning("分钟K查询失败: %s", e)
             return pl.DataFrame()
@@ -1640,12 +1656,13 @@ class KlineRepository:
         if not symbols:
             return pl.DataFrame()
         try:
-            return guarded_collect(
+            df = guarded_collect(
                 pl.scan_parquet(self._minute_glob_for(asset_type)).filter(
                     pl.col("symbol").is_in(symbols)
                     & (pl.col("datetime").dt.date() == trade_date)
                 ).sort(["symbol", "datetime"])
             )
+            return self._maybe_adjust_minute(df, asset_type)
         except Exception as e:  # noqa: BLE001
             logger.warning("批量分钟K查询失败: %s", e)
             return pl.DataFrame()
@@ -1668,7 +1685,7 @@ class KlineRepository:
             lf = pl.scan_parquet(self._minute_glob_for(asset_type))
             available = set(lf.collect_schema().names())
             select_cols = [c for c in ["symbol", "datetime", "open", "high", "low", "close", "volume", "amount"] if c in available]
-            return guarded_collect(
+            df = guarded_collect(
                 lf.select(select_cols)
                 .filter(
                     pl.col("symbol").is_in(symbols)
@@ -1678,6 +1695,7 @@ class KlineRepository:
                 .sort(["symbol", "datetime"]),
                 streaming=True,
             )
+            return self._maybe_adjust_minute(df, asset_type)
         except Exception as e:  # noqa: BLE001
             logger.warning("分钟K范围查询失败: %s", e)
             return pl.DataFrame()
@@ -1713,12 +1731,13 @@ class KlineRepository:
             lf = pl.scan_parquet(parts)
             available = set(lf.collect_schema().names())
             select_cols = [c for c in ["symbol", "datetime", "open", "high", "low", "close", "volume", "amount"] if c in available]
-            return guarded_collect(
+            df = guarded_collect(
                 lf.select(select_cols)
                 .filter(pl.col("symbol").is_in(symbols))
                 .sort(["symbol", "datetime"]),
                 streaming=True,
             )
+            return self._maybe_adjust_minute(df, asset_type)
         except Exception as e:  # noqa: BLE001
             logger.warning("分钟K按日期查询失败: %s", e)
             return pl.DataFrame()

backend/tests/test_kline_detail_transport.py

@@ -45,6 +45,11 @@ def _minute_rows(count: int = 240) -> pl.DataFrame:
 class _DetailRepo:
     def __init__(self, minute: pl.DataFrame | None = None) -> None:
         self.minute = minute if minute is not None else _minute_rows()
+        # /minute 读取 repo.store.data_dir 判断分钟基准标记 (无标记即旧行为)
+        import tempfile
+        from types import SimpleNamespace
+        from pathlib import Path
+        self.store = SimpleNamespace(data_dir=Path(tempfile.mkdtemp()))
 
     def resolve_asset_type(self, symbol: str) -> str:
         return "stock"

backend/tests/test_kline_minute_holiday_fallback.py

@@ -118,6 +118,13 @@ def test_minute_batch_explicit_date_is_not_replaced(market) -> None:
 
 
 class _FakeRepo:
+    def __init__(self) -> None:
+        # /minute 现读取 repo.store.data_dir 判断分钟基准标记 (未标记目录即旧行为)
+        import tempfile
+        from types import SimpleNamespace
+        from pathlib import Path
+        self.store = SimpleNamespace(data_dir=Path(tempfile.mkdtemp()))
+
     def resolve_asset_type(self, symbol: str) -> str:
         return "stock"
 

backend/tests/test_kline_minute_live.py

@@ -24,6 +24,13 @@ _LOCAL_ROWS = 30
 
 
 class _FakeRepo:
+    def __init__(self) -> None:
+        # /minute 读取 repo.store.data_dir 判断分钟基准标记 (无标记即旧行为)
+        import tempfile
+        from types import SimpleNamespace
+        from pathlib import Path
+        self.store = SimpleNamespace(data_dir=Path(tempfile.mkdtemp()))
+
     def resolve_asset_type(self, symbol: str) -> str:
         return "stock"
 
@@ -68,7 +75,7 @@ def _patch_market(monkeypatch, *, in_session: bool) -> None:
 def _patch_live_fetch(monkeypatch) -> None:
     import app.api.kline as kline_api
 
-    def _fake_fetch(symbol, trade_date, asset_type="stock", *, capset):
+    def _fake_fetch(symbol, trade_date, asset_type="stock", *, capset, raw_basis=False):
         return pl.DataFrame({
             "datetime": [datetime(2026, 8, 26, 9, 59)],
             "close": [11.11],

backend/tests/test_kline_minute_lunch_expected.py

@@ -70,6 +70,11 @@ LUNCH_TIMES = [(11, 35), (11, 50), (12, 0)]
 class _FakeRepo:
     def __init__(self, bars: int = 120) -> None:
         self._bars = bars
+        # /minute 与 /minute-batch 读取 repo.store.data_dir 判断分钟基准标记
+        import tempfile
+        from types import SimpleNamespace
+        from pathlib import Path
+        self.store = SimpleNamespace(data_dir=Path(tempfile.mkdtemp()))
 
     def resolve_asset_type(self, symbol: str) -> str:
         return "stock"

backend/tests/test_minute_adjust.py

@@ -0,0 +1,217 @@
+"""minute_adjust 复权投影 / 基准标记 / 存量迁移 测试。
+
+固定因子样本数值断言 (CONTRIBUTING §9): 除权日前后 ratio、VWAP 一致性、
+标记门控 (未迁移行为逐字节不变)、迁移幂等与混合基准收敛。
+"""
+
+from datetime import date, datetime
+from pathlib import Path
+
+import polars as pl
+
+from app.services import minute_adjust
+from app.tickflow.repository import DataStore, KlineRepository
+
+# 样本: AA 除权日 2026-07-01, ex_factor=1.25 (10送2.5量级);
+#       BB 无除权事件。分钟窗口 06-30 (除权前) / 07-01 (除权日)。
+AA, BB = "600001.SH", "600002.SH"
+EX_DATE = date(2026, 7, 1)
+EX_FACTOR = 1.25
+RATIO_BEFORE = 1.0 / EX_FACTOR  # 前复权: 除权日前一日价格 * 1/1.25
+
+
+def _minute_frame(symbol: str, day: date, price: float, rows: int = 3) -> pl.DataFrame:
+    base = datetime(day.year, day.month, day.day, 9, 30)
+    return pl.DataFrame({
+        "symbol": [symbol] * rows,
+        "datetime": [base.replace(minute=30 + i) for i in range(rows)],
+        "open": [price] * rows,
+        "high": [price * 1.01] * rows,
+        "low": [price * 0.99] * rows,
+        "close": [price] * rows,
+        "volume": [100.0] * rows,
+        "amount": [price * 100 * 100] * rows,  # volume(手)*100*price → VWAP=price
+    })
+
+
+def _write_factor_table(data_dir: Path, rows: list[tuple[str, date, float]]) -> None:
+    d = data_dir / "adj_factor"
+    d.mkdir(parents=True, exist_ok=True)
+    pl.DataFrame(
+        {"symbol": [r[0] for r in rows], "trade_date": [r[1] for r in rows], "ex_factor": [r[2] for r in rows]},
+        schema={"symbol": pl.String, "trade_date": pl.Date, "ex_factor": pl.Float64},
+    ).write_parquet(d / "all.parquet")
+
+
+def _write_minute_partition(data_dir: Path, day: date, df: pl.DataFrame, etf: bool = False) -> None:
+    root = data_dir / ("kline_etf_minute" if etf else "kline_minute")
+    part = root / f"date={day.isoformat()}"
+    part.mkdir(parents=True, exist_ok=True)
+    df.write_parquet(part / "part.parquet")
+
+
+def _write_daily_partition(data_dir: Path, day: date, closes: dict[str, float], etf: bool = False) -> None:
+    root = data_dir / ("kline_etf_daily" if etf else "kline_daily")
+    part = root / f"date={day.isoformat()}"
+    part.mkdir(parents=True, exist_ok=True)
+    pl.DataFrame({
+        "symbol": list(closes.keys()),
+        "close": list(closes.values()),
+    }).write_parquet(part / "part.parquet")
+
+
+# ================================================================
+# 复权投影数学
+# ================================================================
+
+def test_apply_adjustment_factor_math(tmp_path):
+    """除权日前的价格与 amount * ratio, 除权日起不动; volume 不变。"""
+    _write_factor_table(tmp_path, [(AA, EX_DATE, EX_FACTOR)])
+    df = pl.concat([_minute_frame(AA, date(2026, 6, 30), 10.0), _minute_frame(AA, EX_DATE, 8.0)])
+    out = minute_adjust.apply_minute_adjustment(df, tmp_path, "stock")
+
+    before = out.filter(pl.col("datetime").dt.date() == date(2026, 6, 30))
+    after = out.filter(pl.col("datetime").dt.date() == EX_DATE)
+    assert abs(before["close"][0] - 10.0 * RATIO_BEFORE) < 1e-9
+    assert abs(before["high"][0] - 10.0 * 1.01 * RATIO_BEFORE) < 1e-9
+    assert abs(before["amount"][0] - 10.0 * 100 * 100 * RATIO_BEFORE) < 1e-6
+    assert before["volume"][0] == 100.0
+    assert abs(after["close"][0] - 8.0) < 1e-12
+    assert abs(after["amount"][0] - 8.0 * 100 * 100) < 1e-6
+
+
+def test_apply_adjustment_vwap_consistency(tmp_path):
+    """复权后 amount/(volume*100) 与复权 close 同口径 (均价线契约)。"""
+    _write_factor_table(tmp_path, [(AA, EX_DATE, EX_FACTOR)])
+    df = _minute_frame(AA, date(2026, 6, 30), 10.0)
+    out = minute_adjust.apply_minute_adjustment(df, tmp_path, "stock")
+    vwap = out["amount"][0] / (out["volume"][0] * 100)
+    assert abs(vwap - out["close"][0]) < 1e-9
+
+
+def test_apply_adjustment_identity_cases(tmp_path):
+    """指数恒等 / 无因子表恒等 / 无事件恒等 (fail-open 语义)。"""
+    df = _minute_frame(AA, date(2026, 6, 30), 10.0)
+
+    out_index = minute_adjust.apply_minute_adjustment(df, tmp_path, "index")
+    assert out_index.equals(df)
+
+    out_no_factors = minute_adjust.apply_minute_adjustment(df, tmp_path, "stock")
+    assert out_no_factors.equals(df)
+
+    _write_factor_table(tmp_path, [(BB, EX_DATE, EX_FACTOR)])
+    out_no_events = minute_adjust.apply_minute_adjustment(df, tmp_path, "stock")
+    assert out_no_events.equals(df)
+
+
+def test_apply_adjustment_etf_routing(tmp_path):
+    """ETF 用 adj_factor_etf 表路由。"""
+    etf = "159001.SZ"
+    (tmp_path / "adj_factor_etf").mkdir(parents=True)
+    pl.DataFrame({
+        "symbol": [etf], "trade_date": [EX_DATE], "ex_factor": [EX_FACTOR],
+    }, schema={"symbol": pl.String, "trade_date": pl.Date, "ex_factor": pl.Float64},
+    ).write_parquet(tmp_path / "adj_factor_etf" / "all.parquet")
+
+    df = _minute_frame(etf, date(2026, 6, 30), 10.0)
+    out = minute_adjust.apply_minute_adjustment(df, tmp_path, "etf")
+    assert abs(out["close"][0] - 10.0 * RATIO_BEFORE) < 1e-9
+
+
+# ================================================================
+# 基准标记与读取门控
+# ================================================================
+
+def test_fetch_adjust_param_gated_by_marker(tmp_path):
+    assert minute_adjust.minute_fetch_adjust(tmp_path) == "forward"
+    minute_adjust.mark_minute_basis_raw(tmp_path)
+    assert minute_adjust.minute_fetch_adjust(tmp_path) == "none"
+    assert minute_adjust.minute_basis_is_raw(tmp_path) is True
+
+
+def test_repository_marker_gate(tmp_path):
+    """标记未开启 → 读取原样 (旧行为逐字节不变); 开启 → 读取应用投影。"""
+    _write_factor_table(tmp_path, [(AA, EX_DATE, EX_FACTOR)])
+    raw_day = date(2026, 6, 30)
+    _write_minute_partition(tmp_path, raw_day, _minute_frame(AA, raw_day, 10.0))
+    repo = KlineRepository(DataStore(tmp_path))
+
+    legacy = repo.get_minute(AA, raw_day)
+    assert abs(legacy["close"][0] - 10.0) < 1e-12  # 未迁移: 原始值原样返回
+
+    minute_adjust.mark_minute_basis_raw(tmp_path)
+    adjusted = repo.get_minute(AA, raw_day)
+    assert abs(adjusted["close"][0] - 10.0 * RATIO_BEFORE) < 1e-9
+
+    # 当日 (因子事件之后) 恒等短路
+    today_day = EX_DATE
+    _write_minute_partition(tmp_path, today_day, _minute_frame(AA, today_day, 8.0))
+    same = repo.get_minute(AA, today_day)
+    assert abs(same["close"][0] - 8.0) < 1e-12
+
+
+# ================================================================
+# 存量迁移
+# ================================================================
+
+def test_migration_mixed_basis_partitions(tmp_path):
+    """前复权存量 → 原始; 原始存量 no-op; 全部成功建标记; 幂等。"""
+    _write_factor_table(tmp_path, [(AA, EX_DATE, EX_FACTOR)])
+    day = date(2026, 6, 30)
+    raw_price = 10.0
+    # AA: 旧架构落盘形态 — 价格为"拉取时前复权"(*1/1.25), amount 保持原始 (复权只动价格)
+    adjusted_stored = _minute_frame(AA, day, raw_price).with_columns(
+        [pl.col(c) * RATIO_BEFORE for c in ("open", "high", "low", "close")]
+    )
+    raw_stored = _minute_frame(BB, day, raw_price)
+    _write_minute_partition(tmp_path, day, pl.concat([adjusted_stored, raw_stored]))
+    _write_daily_partition(tmp_path, day, {AA: raw_price, BB: raw_price})
+
+    stats = minute_adjust.migrate_minute_to_raw(tmp_path)
+    assert stats["failed"] == 0
+    assert stats["marked"] is True
+    assert stats["converted_symbols"] == 1  # 仅 AA
+    assert minute_adjust.minute_basis_is_raw(tmp_path) is True
+
+    stored = pl.read_parquet(tmp_path / "kline_minute" / f"date={day.isoformat()}" / "part.parquet")
+    aa = stored.filter(pl.col("symbol") == AA)
+    bb = stored.filter(pl.col("symbol") == BB)
+    assert abs(aa["close"][0] - raw_price) < 1e-9          # 前复权被还原为原始
+    assert abs(bb["close"][0] - raw_price) < 1e-12         # 原始 no-op
+    assert abs(aa["amount"][0] - raw_price * 100 * 100) < 1e-6  # amount 本就原始, 不动
+
+    # 幂等: 二次运行为 no-op
+    stats2 = minute_adjust.migrate_minute_to_raw(tmp_path)
+    assert stats2["marked"] is True
+    assert stats2["converted_symbols"] == 0
+
+
+def test_migration_missing_daily_anchor_skipped(tmp_path):
+    """无日K锚点的分区跳过且不建标记 (未收敛), 日K补齐后可续跑。"""
+    _write_factor_table(tmp_path, [(AA, EX_DATE, EX_FACTOR)])
+    day = date(2026, 6, 30)
+    _write_minute_partition(tmp_path, day, _minute_frame(AA, day, 10.0))  # 无日K分区
+
+    stats = minute_adjust.migrate_minute_to_raw(tmp_path)
+    assert stats["skipped_no_daily"] == 1
+    assert stats["marked"] is False
+    assert minute_adjust.minute_basis_is_raw(tmp_path) is False
+
+    # 补齐日K锚点后续跑 → 收敛并标记
+    _write_daily_partition(tmp_path, day, {AA: 10.0})
+    stats2 = minute_adjust.migrate_minute_to_raw(tmp_path)
+    assert stats2["marked"] is True
+
+
+def test_migration_anchor_out_of_sane_range_rejected(tmp_path):
+    """锚点偏离合理界 (数据异常) 不换算, 不产生破坏性写入。"""
+    _write_factor_table(tmp_path, [(AA, EX_DATE, EX_FACTOR)])
+    day = date(2026, 6, 30)
+    # 日K锚点 0.001 vs 分钟 10 → k=10000 超界: 视为异常, 原样保留
+    _write_minute_partition(tmp_path, day, _minute_frame(AA, day, 10.0))
+    _write_daily_partition(tmp_path, day, {AA: 0.001})
+
+    stats = minute_adjust.migrate_minute_to_raw(tmp_path)
+    assert stats["marked"] is True  # 分区处理完成 (仅未换算), 仍可收敛
+    stored = pl.read_parquet(tmp_path / "kline_minute" / f"date={day.isoformat()}" / "part.parquet")
+    assert abs(stored["close"][0] - 10.0) < 1e-12

backend/tests/test_minute_routing.py

@@ -300,7 +300,7 @@ def test_get_minute_batch_splits_stock_and_etf(monkeypatch):
     from app.api import kline as kline_api
 
     # mock sync_minute_batch: stock 返回 df_s, etf 返回 df_e (不同 symbol 便于 concat 后 filter 验证)
-    def fake_sync(symbols, *, start_time, end_time, batch_size, rpm, asset_type):
+    def fake_sync(symbols, *, start_time, end_time, batch_size, rpm, asset_type, raw_basis=False):
         if asset_type == "stock":
             return _mock_minute_df(symbol="600519.SH")
         if asset_type == "etf":
@@ -355,7 +355,7 @@ def _endpoint_mocks(monkeypatch, local_df: pl.DataFrame, sync_ret: pl.DataFrame
 
     captured: list[dict] = []
 
-    def fake_sync(symbols, *, start_time, end_time, batch_size, rpm, asset_type):
+    def fake_sync(symbols, *, start_time, end_time, batch_size, rpm, asset_type, raw_basis=False):
         captured.append({"symbols": list(symbols), "start": start_time, "asset": asset_type})
         return sync_ret if sync_ret is not None else pl.DataFrame()