chore: establish Strategy Lab source baseline (development, not release)

6 files changed, 1209 insertions, 0 deletions

diff --git a/worker/Dockerfile b/worker/Dockerfile
new file mode 100644
index 0000000..f23bf61
--- /dev/null
+++ b/worker/Dockerfile
@@ -0,0 +1,20 @@
+# Strategy Lab Python worker image
+# - fetch: network-enabled; backtest: --network none
+# - non-root uid 10001, read-only rootfs at runtime, /output writable only
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+RUN adduser --uid 10001 --disabled-password --gecos "" worker
+
+COPY requirements-worker.txt /tmp/requirements-worker.txt
+RUN pip install --no-cache-dir -r /tmp/requirements-worker.txt
+
+COPY worker /app/worker
+
+USER worker
+
+# /input request, /data dataset objects (read-only mounts); /output writable result dir
+CMD ["python", "-m", "worker.main"]
diff --git a/worker/__init__.py b/worker/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/worker/__init__.py
diff --git a/worker/backtest.py b/worker/backtest.py
new file mode 100644
index 0000000..489ee9a
--- /dev/null
+++ b/worker/backtest.py
@@ -0,0 +1,357 @@
+"""Backtrader runner. User `Strategy` code executes with named feeds and
+next-bar fills. Recording happens in two places:
+  - a sibling Recorder strategy (never trades) records per-bar cash/equity/closes;
+  - a generated subclass of the user's Strategy intercepts notify_order /
+    notify_trade (documented hooks) to record orders and fills, then calls the
+    user's own hook methods.
+"""
+from __future__ import annotations
+
+import ast
+import math
+import resource
+import statistics
+import time
+from pathlib import Path
+
+import backtrader as bt
+import pandas as pd
+
+ENGINE_VERSION = bt.__version__
+EXECUTION_ASSUMPTIONS = {
+    "fill_timing": "signal bar -> order; order fills at NEXT bar open (no same-bar lookahead)",
+    "slippage": "percentage of fill price (broker set_slippage_perc)",
+    "commission": "percentage of order value, charged at execution",
+    "indicator_warmup": "indicators are computed in strategy code; bars before warmup carry no signal",
+    "cn_stock_etf_rules": "broker enforces at submission: buys rounded down to lot=100 "
+                          "(reject if <100), sells rejected if they would exceed the "
+                          "current position (no naked short) or exceed position minus "
+                          "same-symbol pending buys (no T+0 sell of same-bar buys). "
+                          "With next-bar-open daily fills, a sell submitted on the fill "
+                          "day executes next trading day, which is exactly the A-share "
+                          "T+1 rule at daily granularity.",
+    "unmodeled": "suspension days, price-limit halts, intraday sequencing, "
+                 "liquidity/market impact are NOT simulated; no lot-size "
+                 "difference between odd lots of pre-held position on sell",
+    "etf_rules": "A-share listed ETFs follow the same lot=100 buy rule; ETF sell "
+                 "lots in the real market can differ slightly (approximation noted)",
+}
+
+
+class CnDailyRulesBroker(bt.brokers.BackBroker):
+    """Broker rules for cn stock/ETF daily simulation, enforced at submission.
+
+    - index feeds (asset_type=index) are nontradable research proxies: any
+      order on them is rejected at the BROKER level (they only serve as
+      benchmark feeds).
+    - buys: size rounded down to a whole lot of 100; rejected if < 100 or not
+      an integer number of lots after rounding.
+    - sells: rejected when they would exceed the tradable position
+      (no naked short; pending same-symbol buys are excluded to prevent T+0
+      sell of shares that are only settled by a same-cycle buy).
+    """
+
+    def __init__(self, index_feeds=(), rule_assets=(), lot: int = 100):
+        super().__init__()
+        self.index_feeds = set(index_feeds)
+        self.rule_assets = set(rule_assets)
+        self.lot = int(lot)
+        self.rejections = []  # {symbol, reason}
+
+    def _reject(self, order, reason: str):
+        self.rejections.append({"symbol": order.data._name, "reason": reason})
+        order.reject()
+        self.notify(order)
+        return order
+
+    def submit(self, order, check=True):
+        feed = order.data._name
+        if feed in self.index_feeds:
+            return self._reject(order, "index_order_rejected")
+        if feed in self.rule_assets:
+            size = order.size
+            if not float(size).is_integer():
+                order.size = int(size)  # floor for buys handled below
+            if order.isbuy():
+                pending_buy = 0
+                for o in (*self.submitted, *self.pending):
+                    if o.isbuy() and o.data._name == feed and o.status not in \
+                            (o.Rejected, o.Cancelled, o.Completed):
+                        pending_buy += abs(int(o.size or 0)) if not o.executed.size else 0
+                base = int(size)
+                lots = (base + pending_buy) // self.lot * self.lot
+                adjusted = max(0, lots - pending_buy)
+                if adjusted < self.lot:
+                    return self._reject(order, "below_min_lot_100")
+                if adjusted != size:
+                    order.prereject_size = size  # audit trail of requested size
+                    order.size = adjusted
+                    order.executed.remsize = adjusted  # _execute reads remsize
+            elif order.issell():
+                pos = self.getposition(order.data).size
+                pending_buy = 0
+                for o in (*self.submitted, *self.pending):
+                    if o.isbuy() and o.data._name == feed and o.status not in \
+                            (o.Rejected, o.Cancelled, o.Completed):
+                        pending_buy += abs(int(o.size or 0))
+                allowed = max(0, pos - pending_buy)
+                if allowed <= 0:
+                    return self._reject(order, "no_position_to_sell")
+                if abs(order.size) > allowed:
+                    order.prereject_size = abs(order.size)
+                    order.size = -allowed  # trim to owned shares (no short)
+                    order.executed.remsize = order.size
+        return super().submit(order, check=check)
+
+
+def fail(code: str, message: str, details=None, t0: float = 0.0) -> dict:
+    return {
+        "status": "failed",
+        "error": {"code": code, "message": message, "details": details},
+        "engine": {"name": "backtrader", "version": ENGINE_VERSION},
+        "elapsed_ms": int((time.monotonic() - t0) * 1000),
+        "warnings": [],
+    }
+
+
+def _safe_object_path(data_root: str, relpath: str) -> Path:
+    root = Path(data_root).resolve()
+    p = (root / relpath).resolve()
+    if not str(p).startswith(str(root) + "/"):
+        raise ValueError(f"path traversal rejected: {relpath}")
+    if p.is_symlink():
+        raise ValueError(f"symlink rejected in data root: {relpath}")
+    if not p.is_file():
+        raise FileNotFoundError(relpath)
+    return p
+
+
+def _load_frame(path: Path) -> pd.DataFrame:
+    df = pd.read_parquet(path) if path.suffix == ".parquet" else pd.read_csv(path)
+    for c in ("date", "open", "high", "low", "close", "volume"):
+        if c not in df.columns:
+            raise ValueError(f"dataset object {path.name} missing column '{c}'")
+    df["date"] = pd.to_datetime(df["date"])
+    if df["date"].duplicated().any():
+        df = df.drop_duplicates("date", keep="last")
+    if not df["date"].is_monotonic_increasing:
+        df = df.sort_values("date")
+    return df.set_index("date").sort_index()
+
+
+def _compile_user_strategy(code: str):
+    try:
+        ast.parse(code)
+    except SyntaxError as e:
+        raise SyntaxError(f"strategy syntax error at line {e.lineno}: {e.msg}")
+    # ONE namespace serves as both exec globals and locals. With separate
+    # dicts, module-level imports/definitions land in the locals dict while
+    # method bodies resolve names via the globals dict -> NameError in methods
+    # (e.g. `import os` at top level, then os.path in __init__).
+    # The sandbox is the Docker container itself; do not fake Python globals.
+    ns: dict = {"__builtins__": __builtins__}
+    exec(compile(code, "<strategy-source>", "exec"), ns)
+    cls = ns.get("Strategy")
+    if not (isinstance(cls, type) and issubclass(cls, bt.Strategy)):
+        raise ValueError("source must define a class named 'Strategy' subclassing backtrader.Strategy")
+    return cls
+
+
+def run_backtest(request: dict) -> dict:
+    t0 = time.monotonic()
+    cfg = request.get("config") or {}
+    manifest = request.get("dataset_manifest") or {}
+    warnings = list(manifest.get("warnings") or [])
+    if manifest.get("_synthetic"):
+        warnings.append("dataset is a synthetic fixture (_synthetic=true); result is for tests only")
+
+    try:
+        user_cls = _compile_user_strategy(request["code"])
+    except SyntaxError as e:
+        return fail("strategy_syntax", str(e), t0=t0)
+    except ValueError as e:
+        return fail("strategy_invalid", str(e), t0=t0)
+
+    frames = {}
+    try:
+        for obj in manifest.get("objects", []):
+            p = _safe_object_path(request["data_root"], obj["path"])
+            frames[p.resolve().name] = _load_frame(p)
+        if not frames:
+            return fail("data_missing", "dataset manifest has no objects", t0=t0)
+    except FileNotFoundError as e:
+        return fail("data_missing", f"missing dataset object: {e}", t0=t0)
+    except ValueError as e:
+        return fail("data_invalid", str(e), t0=t0)
+
+    feed_frames = [(obj, frames[Path(obj["path"]).resolve().name])
+                   for obj in manifest["objects"] if Path(obj["path"]).resolve().name in frames]
+    symbols = [obj["instrument"]["symbol"] for obj, _ in feed_frames]
+
+    rec = {"orders": [], "fills": [], "closed_trades": [], "rows": []}
+
+    def RecordingStrategy():
+        class R(user_cls):
+            plotinfo = dict(plot=False, subplot=False)
+
+            def notify_order(self, order):
+                rec["orders"].append({
+                    "date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                    "symbol": order.data._name,
+                    "side": "buy" if order.isbuy() else "sell",
+                    "order_state": order.getstatusname(),
+                })
+                if order.status == order.Completed:
+                    ex = order.executed
+                    # turnover: actual executed QUANTITY x executed PRICE.
+                    # (Backtrader's `ex.value` for sells reports cost basis, not
+                    # executed turnover — honest accounting must use size*price.)
+                    rec["fills"].append({
+                        "date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                        "symbol": order.data._name,
+                        "side": "buy" if order.isbuy() else "sell",
+                        "quantity": abs(float(ex.size)),
+                        "price": float(ex.price),
+                        "commission": float(ex.comm),
+                        "value": abs(float(ex.size) * float(ex.price)),
+                    })
+                user_cls.notify_order(self, order)
+
+            def notify_trade(self, trade):
+                if trade.isclosed:
+                    rec["closed_trades"].append({
+                        "symbol": trade.data._name,
+                        "pnl": float(trade.pnl),
+                        "close_date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                    })
+                user_cls.notify_trade(self, trade)
+        return R
+
+    class RowsRecorder(bt.Strategy):
+        plotinfo = dict(plot=False, subplot=False)
+
+        def next(self):
+            closes = {d._name: float(d.close[0]) for d in self.datas if len(d) > 0}
+            rec["rows"].append({
+                "date": self.datas[0].datetime.date(0).strftime("%Y-%m-%d"),
+                "cash": float(self.broker.getcash()),
+                "equity": float(self.broker.getvalue()),
+                "closes": closes,
+                "positions": {d._name: self.getposition(d).size for d in self.datas},
+            })
+
+    cerebro = bt.Cerebro(stdstats=False)
+    for obj, frame in feed_frames:
+        cerebro.adddata(bt.feeds.PandasData(dataname=frame, plot=False,
+                                            timeframe=bt.TimeFrame.Days),
+                        name=obj["instrument"]["symbol"])
+    asset_types = {obj["instrument"]["symbol"]:
+                   obj["instrument"].get("asset_type") for obj, _ in feed_frames}
+    index_feeds = [s for s, a in asset_types.items() if a == "index"]
+    rule_assets = [s for s, a in asset_types.items() if a in ("stock", "etf")]
+    broker = CnDailyRulesBroker(index_feeds=index_feeds, rule_assets=rule_assets)
+    cerebro.setbroker(broker)
+    initial_capital = float(cfg.get("capital", 100000.0))
+    cerebro.broker.setcash(initial_capital)
+    cerebro.broker.setcommission(commission=float(cfg.get("commission", 0.0003)))
+    cerebro.broker.set_slippage_perc(perc=float(cfg.get("slippage", 0.001)))
+    cerebro.addstrategy(RowsRecorder)
+    cerebro.addstrategy(RecordingStrategy(), **(cfg.get("parameters") or {}))
+
+    try:
+        strat_list = cerebro.run(runonce=False, preload=False)
+    except Exception as exc:  # user-code runtime errors surface honestly
+        return fail("runtime_error", f"{type(exc).__name__}: {exc}", t0=t0)
+
+    reasons = {
+        "index_order_rejected": "index feeds are nontradable proxies; broker rejected the order",
+        "below_min_lot_100": "buy below minimum lot 100; broker rejected the order",
+        "no_position_to_sell": "sell would exceed owned shares (no naked short); broker rejected the order",
+    }
+    by_reason: dict[tuple[str, str], int] = {}
+    for rej in broker.rejections:
+        by_reason[(rej["symbol"], rej["reason"])] = by_reason.get((rej["symbol"], rej["reason"]), 0) + 1
+    for (sym, reason), n in by_reason.items():
+        rec["orders"].append({"date": None, "symbol": sym, "side": None,
+                              "order_state": "rejected", "reject_reason": reasons.get(reason, reason),
+                              "count": n})
+        warnings.append(f"broker rejected {n} order(s) for {sym}: {reasons.get(reason, reason)}")
+
+    warnings.append("execution assumptions: "
+                    + "; ".join(f"{k}: {v}" for k, v in EXECUTION_ASSUMPTIONS.items()))
+    rows = rec["rows"]
+
+    benchmark_symbol = cfg.get("benchmark_symbol")
+    bmk_rows = [r["closes"].get(benchmark_symbol) for r in rows if benchmark_symbol]
+    bmk_valid = [v for v in bmk_rows if v is not None]
+    bmk_base = bmk_valid[0] if bmk_valid else None
+    eq, bench_missing = [], False
+    for r in rows:
+        e = {"date": r["date"], "equity": r["equity"], "cash": r["cash"],
+             "closes": r["closes"], "positions": r["positions"]}
+        if benchmark_symbol:
+            close = r["closes"].get(benchmark_symbol)
+            if close is None:
+                bench_missing = True
+                e["benchmark"] = None
+            elif bmk_base:
+                # normalized to initial capital so equity and benchmark share a
+                # common start (raw price is also available in e["closes"])
+                e["benchmark"] = (close / bmk_base) * initial_capital
+            else:
+                e["benchmark"] = None
+        eq.append(e)
+    if bmk_rows and benchmark_symbol and bmk_base:
+        warnings.append(f"benchmark '{benchmark_symbol}' series normalized to initial capital "
+                        f"{initial_capital:g} (raw closes available in equity[].closes)")
+    if bench_missing and benchmark_symbol:
+        warnings.append(f"benchmark_symbol '{benchmark_symbol}' not present in dataset; benchmark omitted")
+
+    equity_vals = [r["equity"] for r in rows]
+    total_return = annual_return = sharpe = max_dd = final = None
+    if len(equity_vals) >= 2 and equity_vals[0] > 0:
+        final = equity_vals[-1]
+        total_return = final / equity_vals[0] - 1.0
+        n = len(equity_vals)
+        annual_return = (1.0 + total_return) ** (252.0 / n) - 1.0
+        peak, max_dd = -float("inf"), 0.0
+        for v in equity_vals:
+            peak = max(peak, v)
+            max_dd = max(max_dd, (peak - v) / peak)
+        rets = [b / a - 1.0 for a, b in zip(equity_vals, equity_vals[1:])]
+        std = statistics.pstdev(rets)
+        sharpe = (statistics.fmean(rets) / std) * (252 ** 0.5) if std > 1e-12 else None
+    elif equity_vals:
+        final = equity_vals[-1]
+    clean = lambda v: v if v is None or math.isfinite(v) else None
+    metrics = {
+        "total_return": clean(total_return),
+        "annual_return": clean(annual_return),
+        "max_drawdown": (None if max_dd is None else -max_dd),  # negative loss fraction; 0.0 is valid
+        "sharpe": clean(sharpe),
+        # trade_count = number of closed round-trip trades (开了又平完成一次), not fills
+        "trade_count": len(rec["closed_trades"]),
+        "final_equity": clean(final),
+    }
+
+    return {
+        "status": "succeeded",
+        "engine": {"name": "backtrader", "version": ENGINE_VERSION},
+        "initial_cash": initial_capital,
+        "equity": eq,
+        "orders": rec["orders"],
+        "trades": rec["fills"],  # executions/fills: {date,symbol,side,quantity,price,commission,value}
+        "closed_trades": rec["closed_trades"],
+        "metrics": metrics,
+        "logs": [
+            f"named feeds: {', '.join(symbols)}",
+            f"bars in equity series: {len(rows)}",
+            f"final equity: {metrics['final_equity']}",
+        ],
+        "warnings": warnings,
+        "execution_assumptions": EXECUTION_ASSUMPTIONS,
+        "elapsed_ms": int((time.monotonic() - t0) * 1000),
+        "peak_rss_kb": int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss),
+        "data_manifest_hash": manifest.get("hash"),
+        "_synthetic": bool(request.get("_synthetic")) or bool(manifest.get("_synthetic")),
+    }
diff --git a/worker/data.py b/worker/data.py
new file mode 100644
index 0000000..841a1eb
--- /dev/null
+++ b/worker/data.py
@@ -0,0 +1,405 @@
+"""AKShare data adapter plus bounded provider suggestion search.
+
+No silent provider/adjustment fallbacks. No synthetic/fabricated data.
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+import akshare as ak  # verified import at module load
+import pandas as pd
+import requests
+
+from .normalize import COLUMN_MAP, map_columns
+
+SUPPORTED_ASSET_TYPES = {"stock", "etf", "index"}
+SUPPORTED_FREQUENCIES = {"daily"}
+SUPPORTED_ADJUSTMENTS = {"none", "qfq", "hfq"}
+IDENTITY_EXCHANGES = {"SH", "SZ", "BJ"}
+
+
+SUPPORTED_SOURCES = {"eastmoney", "tencent", "sina", "auto"}
+_REQUEST_TIMEOUT_SECS = 15.0
+
+
+class DataError(ValueError):
+    def __init__(self, code: str, message: str, details: Any = None):
+        super().__init__(message)
+        self.code = code
+        self.details = details
+
+
+class FetchResult(tuple):
+    """(normalized_df, endpoint, params) with .provider and .raw extras.
+
+    Cache identity and manifest MUST use the actual provider/endpoint that
+    served the data (auto fallback records it here).
+    """
+
+    def __new__(cls, df, endpoint, params, raw, provider):
+        obj = super().__new__(cls, (df, endpoint, params))
+        obj.raw = raw
+        obj.provider = provider
+        return obj
+
+    @property
+    def df(self):
+        return self[0]
+
+    @property
+    def endpoint(self):
+        return self[1]
+
+    @property
+    def params(self):
+        return self[2]
+
+
+def split_identity(instrument: dict) -> tuple[str, str]:
+    """-> (exchange, code). Accepts 'SH#600000' or bare code with market identity."""
+    symbol = str(instrument["symbol"])
+    if "#" in symbol:
+        exch, code = symbol.split("#", 1)
+        if exch not in IDENTITY_EXCHANGES:
+            raise DataError("bad_identity", f"unknown exchange prefix: {exch}")
+    else:
+        code = symbol
+        market = instrument.get("market", "cn")
+        if market != "cn":
+            raise DataError("unsupported_market", f"market not supported by this adapter: {market}")
+        if len(code) == 6 and code[0] in "369":
+            exch = "SH"
+        else:
+            exch = "SZ"
+    if not code.isdigit() or len(code) != 6:
+        raise DataError("bad_identity", "A-share code must be a 6-digit number")
+    return exch, code
+
+
+def _fetch_eastmoney(instrument: dict, start: str, end: str,
+                     frequency: str, adjustment: str) -> FetchResult:
+    exch, code = split_identity(instrument)
+    s, e = start.replace("-", ""), end.replace("-", "")
+    asset = instrument["asset_type"]
+    if asset == "stock":
+        endpoint = "stock_zh_a_hist"
+        params = {"symbol": code, "period": "daily", "start_date": s, "end_date": e,
+                  "adjust": "" if adjustment == "none" else adjustment}
+        raw = ak.stock_zh_a_hist(**params)
+    elif asset == "etf":
+        endpoint = "fund_etf_hist_em"
+        params = {"symbol": code, "period": "daily", "start_date": s, "end_date": e,
+                  "adjust": "" if adjustment == "none" else adjustment}
+        raw = ak.fund_etf_hist_em(**params)
+    else:
+        endpoint = "index_zh_a_hist"
+        params = {"symbol": code, "period": "daily", "start_date": s, "end_date": e}
+        raw = ak.index_zh_a_hist(**params)
+    if not isinstance(raw, pd.DataFrame):
+        raise DataError("bad_provider_response", f"{endpoint} did not return a DataFrame")
+    df = _to_canonical(raw, f"{exch}#{code}")
+    return FetchResult(df, endpoint, params, raw, "eastmoney")
+
+
+def _fetch_tencent(instrument: dict, start: str, end: str,
+                   frequency: str, adjustment: str) -> FetchResult:
+    exch, code = split_identity(instrument)
+    asset = instrument["asset_type"]
+    s, e = start.replace("-", ""), end.replace("-", "")
+    tx_symbol = f"{exch.lower()}{code}"
+    if asset == "stock":
+        endpoint = "stock_zh_a_hist_tx"
+        params = {"symbol": tx_symbol, "start_date": s, "end_date": e,
+                  "adjust": "" if adjustment == "none" else adjustment}
+        raw = ak.stock_zh_a_hist_tx(**params, timeout=_REQUEST_TIMEOUT_SECS)
+    elif asset == "index":
+        if frequency != "daily":
+            raise DataError("unsupported_frequency", "tencent source supports daily only")
+        endpoint = "stock_zh_index_daily_tx"
+        params = {"symbol": tx_symbol, "start_date": s, "end_date": e}
+        raw = ak.stock_zh_index_daily_tx(**params)
+    else:
+        raise DataError("source_unavailable", "tencent source has no listed-ETF daily adapter")
+    if not isinstance(raw, pd.DataFrame):
+        raise DataError("bad_provider_response", f"{endpoint} did not return a DataFrame")
+    symbol = f"{exch}#{code}"
+    df = _to_canonical(raw, symbol)
+    warnings = []
+    if asset == "index":
+        warnings.append("tencent index data is labeled 前复权 by the provider; "
+                        "adjustment-factors mixed across sources are not supported")
+    df.attrs["source_warnings"] = warnings
+    return FetchResult(df, endpoint, params, raw, "tencent")
+
+
+def _fetch_sina(instrument: dict, start: str, end: str,
+                frequency: str, adjustment: str) -> FetchResult:
+    exch, code = split_identity(instrument)
+    asset = instrument["asset_type"]
+    if frequency != "daily":
+        raise DataError("unsupported_frequency", "sina source supports daily only")
+    if asset != "etf":
+        raise DataError("source_unavailable",
+                        f"sina source has no {asset} daily adapter in this worker")
+    if adjustment != "none":
+        raise DataError("unsupported_adjustment",
+                        "sina ETF klines are unadjusted only (no adjust parameter); "
+                        f"adjustment '{adjustment}' cannot be served by sina")
+    endpoint = "fund_etf_hist_sina"
+    params = {"symbol": f"{exch.lower()}{code}"}
+    raw = ak.fund_etf_hist_sina(**params)
+    if not isinstance(raw, pd.DataFrame):
+        raise DataError("bad_provider_response", f"{endpoint} did not return a DataFrame")
+    df = _to_canonical(raw, f"{exch}#{code}")
+    # sina has no date-range parameter: slice the full history locally to the
+    # requested window (ISO strings compare lexicographically like dates)
+    df = df[(df["date"] >= start) & (df["date"] <= end)].reset_index(drop=True)
+    df.attrs["source_warnings"] = [
+        "sina returns the full trading history without date parameters; sliced "
+        "locally to the requested range; data is unadjusted (no adjust parameter)",
+        "sina volume unit is 股 (shares); verified to be 100x the 手 (lots) "
+        "convention used by lot-based providers on the same session "
+        "(2026-09-17 evidence, docs/recovery-01-plan.md)",
+    ]
+    return FetchResult(df, endpoint, params, raw, "sina")
+
+
+def _to_canonical(raw: pd.DataFrame, canonical_symbol: str) -> pd.DataFrame:
+    df = map_columns(raw.copy())
+    if not df.empty and "date" in df.columns:
+        df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d")
+    df["symbol"] = canonical_symbol
+    return df
+
+
+def fetch_source(instrument: dict, start: str, end: str, frequency: str,
+                 adjustment: str, fields: list[str] | None = None,
+                 source: str = "auto") -> FetchResult:
+    if frequency not in SUPPORTED_FREQUENCIES:
+        raise DataError("unsupported_frequency", f"frequency '{frequency}' not supported; supported: daily")
+    if adjustment not in SUPPORTED_ADJUSTMENTS:
+        raise DataError("unsupported_adjustment", f"adjustment '{adjustment}' not supported")
+    if instrument.get("asset_type") not in {"stock", "etf", "index"}:
+        raise DataError("unsupported_asset_type", f"asset_type '{instrument.get('asset_type')}' not supported")
+    if source not in SUPPORTED_SOURCES:
+        raise DataError("bad_source", f"source '{source}' not supported; supported: {sorted(SUPPORTED_SOURCES)}")
+    if instrument["asset_type"] == "index" and adjustment != "none":
+        raise DataError("unsupported_adjustment", "indexes have no adjustment factors; adjustment must be 'none'")
+
+    def run(provider: str) -> FetchResult:
+        if provider == "eastmoney":
+            return _fetch_eastmoney(instrument, start, end, frequency, adjustment)
+        if provider == "sina":
+            return _fetch_sina(instrument, start, end, frequency, adjustment)
+        return _fetch_tencent(instrument, start, end, frequency, adjustment)
+
+    if source != "auto":
+        return run(source)
+    # auto: eastmoney first, transparent same-symbol fallback, honestly labeled.
+    # tencent has no listed-ETF daily adapter, so ETFs fall back to sina.
+    fallback = "sina" if instrument["asset_type"] == "etf" else "tencent"
+    try:
+        return run("eastmoney")
+    except Exception as em_err:
+        try:
+            res = run(fallback)
+        except Exception as fb_err:
+            raise DataError(
+                "provider_unavailable",
+                f"eastmoney failed ({em_err}); {fallback} fallback failed ({fb_err})")
+        res.df.attrs["source_warnings"] = list(getattr(res.df, "attrs", {}).get("source_warnings", [])) + [
+            f"provider_fallback: eastmoney attempt failed ({type(em_err).__name__}); "
+            f"served by {fallback} for the SAME symbol; sources may differ in "
+            "adjustment method and units (check provider warnings)"
+        ]
+        return res
+
+
+def fetch_source_with_warnings(instrument: dict, start: str, end: str, frequency: str,
+                               adjustment: str, fields: list[str] | None = None,
+                               source: str = "auto"):
+    """fetch_source plus source warnings pulled off the returned frame."""
+    res = fetch_source(instrument, start, end, frequency, adjustment, fields, source)
+    src_warnings = list(res.df.attrs.get("source_warnings", []))
+    return res, src_warnings
+
+
+# ---- provider suggestion search (bounded, direct HTTP, no full catalogs) ----
+#
+# Root-cause fix for the production search stall: the legacy path fetched the
+# ENTIRE stock catalog and the ENTIRE ETF snapshot on every query. Instead we
+# hit the actual suggestion endpoints of the same data providers used by
+# fetch — bounded HTTP calls with short timeouts, then validated identity
+# classification per item. Asset class comes ONLY from the provider's own
+# classification field (Eastmoney `Classify` / Tencent hint tail token),
+# never guessed from the numeric code shape.
+
+SUGGEST_TIMEOUT_SECS = 4.0  # per-source hard timeout, well under overall budget
+
+_EASTMONEY_URL = "https://searchapi.eastmoney.com/api/suggest/get"
+_TENCENT_URL = "https://smartbox.gtimg.cn/s3/"
+
+# Live-probed provider-stated classes (2026-09, see docs/search-fix.md):
+# Eastmoney Classify: AStock→stock, Index→index, Fund→fund (ambiguous:
+# same SecurityType/Classify covers ETF AND LOF), Bond/HK/OTCFUND/... unsupported.
+# Tencent hint tail: GP-A→stock, ETF→etf, ZS→index, LOF/others unsupported.
+_EASTMONEY_CLASSIFY = {"AStock": "stock", "Index": "index", "Fund": "etf"}
+_TENCENT_TAGS = {"GP-A": "stock", "ETF": "etf", "ZS": "index"}
+_MKTNUM_EXCHANGE = {"1": "SH", "0": "SZ"}
+_TENCENT_EXCHANGE = {"sh": "SH", "sz": "SZ"}
+
+_HEADERS = {"User-Agent": "Mozilla/5.0 (strategy-lab instrument search)"}
+
+
+def _http_json(url: str, params: dict) -> dict:
+    r = requests.get(url, params=params, timeout=SUGGEST_TIMEOUT_SECS, headers=_HEADERS)
+    r.raise_for_status()
+    return r.json()
+
+
+def _http_text(url: str, params: dict) -> str:
+    r = requests.get(url, params=params, timeout=SUGGEST_TIMEOUT_SECS, headers=_HEADERS)
+    r.raise_for_status()
+    return r.text
+
+
+def _search_eastmoney(q: str, limit: int) -> list[dict]:
+    """Bounded direct eastmoney suggest call. Returns validated items only."""
+    payload = _http_json(
+        _EASTMONEY_URL,
+        {"input": q, "type": 14, "count": max(5, min(limit, 100))},
+    )
+    table = payload.get("QuotationCodeTable") or {}
+    if table.get("Status") != 0:
+        raise RuntimeError(f"eastmoney suggest status={table.get('Status')!r}")
+    rows = table.get("Data") or []
+    items = []
+    for row in rows[:limit]:
+        code = str(row.get("Code", "")).strip()
+        name = str(row.get("Name", "")).strip()
+        classify = str(row.get("Classify", "")).strip()
+        asset = _EASTMONEY_CLASSIFY.get(classify)
+        exchange = _MKTNUM_EXCHANGE.get(str(row.get("MktNum", "")).strip())
+        # identity must be fully validated by the provider reply itself; the
+        # asset class is provider-stated, never inferred from the code digits
+        if not code or not exchange or asset is None:
+            continue
+        items.append({
+            "symbol": code,
+            "canonical_symbol": f"{exchange}#{code}",
+            "market": "cn",
+            "asset_type": asset,
+            "name": name,
+            "currency": "CNY",
+            "source": "eastmoney",
+        })
+    return items
+
+
+def _search_tencent(q: str, limit: int) -> list[dict]:
+    """Bounded direct tencent smartbox call. Returns validated items only."""
+    text = _http_text(_TENCENT_URL, {"q": q, "t": "all"})
+    m = re.search(r'v_hint="(.*)"', text)
+    if not m:
+        return []  # empty suggestion is a valid provider-no-match reply
+    body = m.group(1)
+    # provider encodes non-ASCII as \uXXXX escapes; decode JSON-style safely
+    try:
+        body = json.loads(f'"{body}"')
+    except ValueError:
+        body = decode_unicode_escapes(body)
+    items = []
+    for entry in body.split("^"):
+        parts = entry.strip().split("~")
+        if len(parts) < 5:
+            continue
+        mkt, code, name, _py, tag = parts[0].lower(), parts[1], parts[2], parts[3], parts[4].strip()
+        asset = _TENCENT_TAGS.get(tag)
+        exchange = _TENCENT_EXCHANGE.get(mkt)
+        # class and exchange both provider-stated; LOF/bonds/bj are excluded
+        if not code or not exchange or asset is None:
+            continue
+        items.append({
+            "symbol": code,
+            "canonical_symbol": f"{exchange}#{code}",
+            "market": "cn",
+            "asset_type": asset,
+            "name": name,
+            "currency": "CNY",
+            "source": "tencent",
+        })
+    return items[:limit]
+
+
+def decode_unicode_escapes(s: str) -> str:
+    return re.sub(r"\\u([0-9a-fA-F]{4})", lambda m: chr(int(m.group(1), 16)), s)
+
+
+def search_instruments(q: str, limit: int = 50) -> dict:
+    """Bounded direct provider suggestion search.
+
+    Per-source HTTP timeout 4s; one source failing is visible but not fatal if
+    the other serves results. Both failing -> honest failed status, never an
+    empty success. Each item carries its actual provenance source.
+    """
+    out = {"source": "provider_suggest", "status": "failed",
+           "items": [], "error": None, "providers": {}}
+    errors: list[str] = []
+    sources_used: list[str] = []
+    results: dict[str, list[dict]] = {}
+
+    try:
+        results["eastmoney"] = _search_eastmoney(q, limit)
+        sources_used.append("eastmoney")
+        out["providers"]["eastmoney"] = "ok"
+    except Exception as exc:
+        errors.append(f"eastmoney: {type(exc).__name__}: {exc}")
+        out["providers"]["eastmoney"] = f"failed: {type(exc).__name__}"
+
+    try:
+        results["tencent"] = _search_tencent(q, limit)
+        sources_used.append("tencent")
+        out["providers"]["tencent"] = "ok"
+    except Exception as exc:
+        errors.append(f"tencent: {type(exc).__name__}: {exc}")
+        out["providers"]["tencent"] = f"failed: {type(exc).__name__}"
+
+    if not sources_used:
+        out["error"] = {"code": "providers_unavailable",
+                        "message": "all providers failed: " + "; ".join(errors)}
+        return out
+
+    tencent_by_code = {it["symbol"]: it for it in results.get("tencent", [])}
+    tencent_ok = "tencent" in results
+    seen: set[tuple[str, str, str]] = set()
+    items = []
+    for provider in ("eastmoney", "tencent"):
+        for it in results.get(provider, []):
+            # Eastmoney's Classify "Fund" is ambiguous (verified live: ETF and
+            # LOF share the same Classify/SecurityType). Only emit Eastmoney
+            # fund items when Tencent — whose tags distinguish ETF from LOF —
+            # serves the same code; otherwise the LOF guarantee cannot hold.
+            if provider == "eastmoney" and it["asset_type"] == "etf":
+                cross = tencent_by_code.get(it["symbol"])
+                if not (tencent_ok and cross and cross["asset_type"] == "etf"):
+                    continue
+            key = (it["canonical_symbol"], it["asset_type"], it["source"])
+            if key in seen:
+                continue
+            seen.add(key)
+            items.append(it)
+            if len(items) >= limit:
+                break
+        if len(items) >= limit:
+            break
+
+    out["status"] = "ready"
+    out["items"] = items
+    out["source"] = "provider_suggest"
+    out["error"] = ({"code": "partial_providers",
+                     "message": "some providers failed: " + "; ".join(errors)}) if errors else None
+    if errors:
+        out["warnings"] = errors
+    return out
diff --git a/worker/main.py b/worker/main.py
new file mode 100644
index 0000000..bedeb03
--- /dev/null
+++ b/worker/main.py
@@ -0,0 +1,267 @@
+"""Strategy Lab Python worker CLI.
+
+Subcommands:
+  fetch    --request /input/request.json --output /output     (network OK)
+  backtest --request /input/request.json --output /output     (network none)
+  search   --query <text> [--limit 50]                        (network OK)
+  probe    --output <dir>                                     (real AKShare evidence)
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import signal
+import sys
+import time
+from datetime import date, timedelta
+from pathlib import Path
+
+import akshare as ak
+import pandas as pd
+
+from . import backtest as bt_runner
+from . import data as data_mod
+from .normalize import (NORMALIZATION_VERSION, SCHEMA_VERSION, build_manifest_entry,
+                        compute_object_hash, coverage_summary, normalize_frame)
+
+MAX_INSTRUMENTS = 5
+MAX_YEARS = 15
+BASE_FIELDS = ["open", "high", "low", "close", "volume"]
+ALLOWED_FIELDS = set(BASE_FIELDS) | {"amount", "turnover"}
+EXTRA_MAP = {"amount": {"成交额", "amount"}, "turnover": {"换手率", "turnover"}}
+ALLOWED_SOURCES = {"eastmoney", "tencent", "sina", "auto"}
+FETCH_WALL_SECS = int(os.environ.get("STRATEGY_LAB_FETCH_WALL_SECS", "180"))
+
+
+def _load_json(path: str):
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _dump_json(path: Path, obj) -> None:
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(obj, f, ensure_ascii=False, indent=1, sort_keys=True, default=str)
+
+
+def _slug(instrument: dict) -> str:
+    return instrument["symbol"].lower().replace("#", "_")
+
+
+def _validate_request(req: dict) -> list[str]:
+    errors = []
+    instruments = req.get("instruments") or []
+    if not instruments or len(instruments) > MAX_INSTRUMENTS:
+        errors.append(f"1..{MAX_INSTRUMENTS} instruments required")
+    if req.get("frequency") != "daily":
+        errors.append("frequency must be 'daily'")
+    if req.get("adjustment") not in {"none", "qfq", "hfq"}:
+        errors.append("adjustment must be none|qfq|hfq")
+    s, e = req.get("start_date"), req.get("end_date")
+    try:
+        d0, d1 = date.fromisoformat(s), date.fromisoformat(e)
+        if d0 > d1:
+            errors.append("start_date after end_date")
+        if d1 - d0 > timedelta(days=365 * MAX_YEARS):
+            errors.append(f"range exceeds {MAX_YEARS} years POC limit")
+    except (TypeError, ValueError):
+        errors.append("start_date/end_date must be ISO dates")
+    for f in req.get("fields", BASE_FIELDS):
+        if f not in ALLOWED_FIELDS:
+            errors.append(f"field '{f}' not supported; supported: {sorted(ALLOWED_FIELDS)}")
+    if req.get("source", "auto") not in ALLOWED_SOURCES:
+        errors.append(f"source '{req.get('source')}' not supported; supported: {sorted(ALLOWED_SOURCES)}")
+    return errors
+
+
+def _fetch_timeout(signum, frame):
+    raise TimeoutError("fetch wall-clock limit exceeded (bounded network call)")
+
+
+def _write_raw(out: Path, slug: str, endpoint: str, params: dict, raw, fetched_at: str) -> tuple[str, str]:
+    """Immutable raw provider response object. Returns (objects-relative name, content hash)."""
+    payload = {"endpoint": endpoint, "params": params, "fetched_at": fetched_at,
+               "data": json.loads(raw.to_json(orient="records", force_ascii=False))}
+    data = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
+    digest = hashlib.sha256(data).hexdigest()
+    name = f"raw_{slug}_{digest[:12]}.json"
+    (out / "objects").mkdir(parents=True, exist_ok=True)
+    (out / "objects" / name).write_bytes(data)
+    return name, digest
+
+
+def run_fetch(request_path: str, output_dir: str) -> int:
+    signal.signal(signal.SIGALRM, _fetch_timeout)
+    signal.alarm(FETCH_WALL_SECS)
+    req = _load_json(request_path)
+    out = Path(output_dir)
+    out.mkdir(parents=True, exist_ok=True)
+    t0 = time.monotonic()
+    errors = _validate_request(req)
+    result = {"status": "failed", "errors": errors, "warnings": []} if errors else None
+    if result:
+        _dump_json(out / "result.json", result)
+        return 1
+
+    instruments = req["instruments"]
+    objects, warnings = [], []
+    ok = True
+    for inst in instruments:
+        w = list(inst.get("warnings", []))
+        try:
+            res, src_warn = data_mod.fetch_source_with_warnings(
+                inst, req["start_date"], req["end_date"], req["frequency"],
+                req["adjustment"], req.get("fields") or BASE_FIELDS,
+                source=req.get("source", "auto"))
+            df, endpoint, params = res
+            raw = res.raw
+            w.extend(src_warn)
+        except data_mod.DataError as exc:
+            objects.append({"instrument": inst, "error": {"code": exc.code, "message": str(exc)}})
+            warnings.append(f"{inst['symbol']}: {exc}")
+            ok = False
+            continue
+        except Exception as exc:
+            objects.append({"instrument": inst,
+                            "error": {"code": "provider_error", "message": f"{type(exc).__name__}: {exc}"}})
+            warnings.append(f"{inst['symbol']}: provider error {exc}")
+            ok = False
+            continue
+
+        fetched_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+        requested_flds = req.get("fields") or BASE_FIELDS
+        kept = [c for c in df.columns if c in ("date", "symbol", "open", "high", "low", "close", "volume")
+                or c in {alt for f in requested_flds if f in EXTRA_MAP for alt in EXTRA_MAP[f]}]
+        df = df[kept]
+        norm = normalize_frame(df, inst["symbol"])
+        raw_name, raw_hash = _write_raw(out, _slug(inst), res.endpoint, res.params, raw, fetched_at)
+        csv_path = f"objects/{_slug(inst)}.csv"
+        (out / "objects").mkdir(exist_ok=True)
+        norm.to_csv(out / csv_path, index=False)
+        cov = coverage_summary(norm)
+        if cov["actual_start"] != req["start_date"] or cov["actual_end"] != req["end_date"]:
+            w.append(f"actual coverage {cov['actual_start']}..{cov['actual_end']} differs from request; gaps kept")
+        entry = build_manifest_entry(
+            norm,
+            instrument={**inst, "symbol": inst["symbol"].upper()},
+            provider=res.provider, endpoint=res.endpoint, params=res.params,
+            adjustment=req["adjustment"],
+            requested_start=req["start_date"], requested_end=req["end_date"],
+            raw_object_hash=raw_hash, warnings=w, path=csv_path,
+            fetched_at=fetched_at, akshare_version=ak.__version__,
+        )
+        objects.append(entry)
+
+    if not ok:
+        result = {"status": "failed", "errors": [f"{o['instrument']['symbol']}: {o.get('error', {}).get('message')}"
+                                                 for o in objects if "error" in o],
+                  "warnings": warnings}
+        _dump_json(out / "result.json", result)
+        return 1
+
+    manifest = {
+        "schema_version": SCHEMA_VERSION,
+        "normalization_version": NORMALIZATION_VERSION,
+        "fetched_at": objects[0]["fetched_at"],
+        "frequency": req["frequency"],
+        "adjustment": req["adjustment"],
+        "objects": objects,
+        "immutable": True,
+    }
+    content = json.dumps(objects, sort_keys=True, ensure_ascii=False)
+    manifest["hash"] = hashlib.sha256(content.encode()).hexdigest()
+
+    result = {
+        "status": "ready",
+        "manifest": manifest,
+        "preview": {"columns": [c for c in objects[0]["columns"] if c != "symbol"],
+                     "row_count": objects[0]["row_count"],
+                     "rows": json.loads(pd.read_csv(out / objects[0]["path"]).tail(20).to_json(orient="records")),
+                     "coverage": coverage_summary(pd.read_csv(out / objects[0]["path"]))},
+        "manifest_hash": manifest["hash"],
+        "cache_key": manifest["hash"],
+        "warnings": warnings,
+        "elapsed_ms": int((time.monotonic() - t0) * 1000),
+        "akshare_version": ak.__version__,
+    }
+    _dump_json(out / "result.json", result)
+    return 0
+
+
+def run_backtest(request_path: str, output_dir: str) -> int:
+    req = _load_json(request_path)
+    out = Path(output_dir)
+    res = bt_runner.run_backtest(req)
+    _dump_json(out / "result.json", res)
+    return 0 if res.get("status") == "succeeded" else 1
+
+
+def run_search(query: str, limit: int) -> int:
+    """Bounded provider suggestion search. Prints a JSON envelope with an
+    explicit status; failed upstream is a real error, never empty success."""
+    envelope = data_mod.search_instruments(query, limit)
+    print(json.dumps(envelope, ensure_ascii=False))
+    return 0 if envelope.get("status") == "ready" else 1
+
+
+def run_probe(output_dir: str) -> int:
+    """Real AKShare connectivity probe for qa evidence. No fabrication."""
+    out = Path(output_dir)
+    out.mkdir(parents=True, exist_ok=True)
+    targets = [
+        {"endpoint": "stock_zh_a_hist", "symbol": "600000", "asset_type": "stock",
+         "call": lambda: ak.stock_zh_a_hist(symbol="600000", period="daily",
+                                            start_date="20260101", end_date="20260930", adjust="qfq")},
+        {"endpoint": "fund_etf_hist_em", "symbol": "510300", "asset_type": "etf",
+         "call": lambda: ak.fund_etf_hist_em(symbol="510300", period="daily",
+                                             start_date="20260101", end_date="20260930", adjust="")},
+        {"endpoint": "stock_zh_index_daily_em", "symbol": "000300", "asset_type": "index",
+         "call": lambda: ak.stock_zh_index_daily_em(symbol="sh000300")},
+        {"endpoint": "stock_info_a_code_name", "symbol": "<catalog>", "asset_type": "catalog",
+         "call": lambda: ak.stock_info_a_code_name()},
+    ]
+    report = {"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+              "akshare_version": ak.__version__, "probes": []}
+    for t in targets:
+        rec = {"endpoint": t["endpoint"], "symbol": t["symbol"], "asset_type": t["asset_type"]}
+        try:
+            df = t["call"]()
+            rec.update(status="ok", rows=int(len(df)), columns=list(df.columns),
+                       head=[json.loads(df.head(2).to_json(orient="records", force_ascii=False)),
+                             ][0])
+        except Exception as exc:
+            rec.update(status="failed", error=f"{type(exc).__name__}: {exc}")
+        report["probes"].append(rec)
+    _dump_json(out / "akshare-probe.json", report)
+    ok = sum(1 for p in report["probes"] if p["status"] == "ok")
+    print(json.dumps({"ok": ok, "total": len(targets)}))
+    return 0
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(prog="python -m worker.main")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    for name in ("fetch", "backtest"):
+        p = sub.add_parser(name)
+        p.add_argument("--request", required=True)
+        p.add_argument("--output", required=True)
+    ps = sub.add_parser("search")
+    ps.add_argument("--query", required=True)
+    ps.add_argument("--limit", type=int, default=50)
+    pp = sub.add_parser("probe")
+    pp.add_argument("--output", required=True)
+    args = ap.parse_args(argv)
+    if args.cmd == "fetch":
+        return run_fetch(args.request, args.output)
+    if args.cmd == "backtest":
+        return run_backtest(args.request, args.output)
+    if args.cmd == "search":
+        return run_search(args.query, args.limit)
+    if args.cmd == "probe":
+        return run_probe(args.output)
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/worker/normalize.py b/worker/normalize.py
new file mode 100644
index 0000000..4cb373d
--- /dev/null
+++ b/worker/normalize.py
@@ -0,0 +1,160 @@
+"""Normalization: provider frames -> canonical OHLCV schema + content hashes."""
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+import pandas as pd
+
+SCHEMA_VERSION = "1"
+NORMALIZATION_VERSION = "1"
+
+# Canonical instrument identity format: "<EXCHANGE>#<CODE>", e.g. SH#600000.
+# Prices are never substituted across canonical symbols.
+
+# AKShare (Chinese) -> canonical column names. Non-mapped fields are preserved
+# verbatim as raw provider fields (units/precision untouched).
+COLUMN_MAP = {
+    "日期": "date",
+    "开盘": "open",
+    "收盘": "close",
+    "最高": "high",
+    "最低": "low",
+    "成交量": "volume",
+}
+
+REQUIRED = ["date", "open", "high", "low", "close", "volume"]
+
+
+class NormalizeError(ValueError):
+    def __init__(self, code: str, message: str, details: Any = None):
+        super().__init__(message)
+        self.code = code
+        self.details = details
+
+
+def map_columns(df: pd.DataFrame) -> pd.DataFrame:
+    return df.rename(columns=COLUMN_MAP)
+
+
+def normalize_frame(df: pd.DataFrame, symbol: str) -> pd.DataFrame:
+    if df is None or df.empty:
+        raise NormalizeError("empty_response", "provider returned no rows")
+    missing_req = {c for c in REQUIRED if c not in df.columns}
+    if missing_req:
+        raise NormalizeError(
+            "missing_columns",
+            f"provider frame missing standard columns: {sorted(missing_req)}",
+            {"columns": list(df.columns)},
+        )
+    out = df.copy()
+    out["date"] = pd.to_datetime(out["date"], errors="coerce")
+    if out["date"].isna().any():
+        bad = out.loc[out["date"].isna()].index.tolist()
+        raise NormalizeError(
+            "bad_date", "unparseable date values", {"rows": [int(i) for i in bad[:5]]}
+        )
+    for col in ("open", "high", "low", "close", "volume"):
+        out[col] = pd.to_numeric(out[col], errors="coerce")
+        if out[col].isna().any():
+            rows = out.loc[out[col].isna()].index.tolist()
+            first = out.loc[rows[0], "date"].date().isoformat()
+            raise NormalizeError(
+                "missing_ohlcv",
+                f"missing or non-numeric '{col}' on {first} — gap kept, never imputed",
+                {"column": col, "row_count": len(rows), "rows": [int(i) for i in rows[:5]]},
+            )
+    out["date"] = out["date"].dt.strftime("%Y-%m-%d")
+    out = out.sort_values("date", kind="mergesort")
+    out = out.drop_duplicates(subset="date", keep="last")
+    # canonical identity, prices never substituted across symbols
+    out["symbol"] = symbol
+    return out.reset_index(drop=True)
+
+
+def frame_records(df: pd.DataFrame) -> list[dict]:
+    return json.loads(df.to_json(orient="records", force_ascii=False))
+
+
+def compute_object_hash(df: pd.DataFrame) -> str:
+    """Content hash over canonical records; independent of request/user ids."""
+    recs = frame_records(df)
+    return hashlib.sha256(
+        json.dumps(recs, sort_keys=True, ensure_ascii=False).encode("utf-8")
+    ).hexdigest()
+
+
+def coverage_summary(df_or_dates) -> dict:
+    if isinstance(df_or_dates, pd.DataFrame):
+        dates = pd.to_datetime(df_or_dates["date"]).tolist()
+    else:
+        dates = pd.to_datetime(list(df_or_dates)).tolist()
+    if not dates:
+        return {"actual_start": None, "actual_end": None, "segments": [], "gaps": []}
+    segs = [[dates[0], dates[0]]]
+    gaps: list[dict] = []
+    for prev, cur in zip(dates, dates[1:]):
+        delta = cur - prev
+        if delta.days == 1:
+            segs[-1][1] = cur
+        else:
+            segs.append([cur, cur])
+            gaps.append({"after": prev.date().isoformat(), "before": cur.date().isoformat(),
+                         "calendar_days": delta.days - 1})
+    fmt = lambda d: d.date().isoformat()
+    return {
+        "actual_start": fmt(dates[0]),
+        "actual_end": fmt(dates[-1]),
+        "segments": [[fmt(a), fmt(b)] for a, b in segs],
+        "gaps": gaps,
+    }
+
+
+def build_manifest_entry(
+    df: pd.DataFrame,
+    *,
+    instrument: dict,
+    provider: str,
+    endpoint: str,
+    params: dict,
+    adjustment: str,
+    requested_start: str,
+    requested_end: str,
+    raw_object_hash: str,
+    warnings: list,
+    schema_version: str = SCHEMA_VERSION,
+    normalization_version: str = NORMALIZATION_VERSION,
+    path: str,
+    fetched_at: str,
+    akshare_version: str,
+) -> dict:
+    cov = coverage_summary(df)
+    return {
+        "instrument": {
+            "symbol": instrument["symbol"],
+            "market": instrument["market"],
+            "asset_type": instrument["asset_type"],
+            **({"name": instrument["name"]} if instrument.get("name") else {}),
+        },
+        "object_hash": compute_object_hash(df),
+        "path": path,  # internal-only: backend rewrites before client exposure
+        "raw_object_hash": raw_object_hash,
+        "provider": provider,
+        "endpoint": endpoint,
+        "params": params,
+        "akshare_version": akshare_version,
+        "fetched_at": fetched_at,
+        "schema_version": schema_version,
+        "normalization_version": normalization_version,
+        "adjustment": adjustment,
+        "requested_start": requested_start,
+        "requested_end": requested_end,
+        "actual_start": cov["actual_start"],
+        "actual_end": cov["actual_end"],
+        "coverage": {"segments": cov["segments"], "gaps": cov["gaps"]},
+        "row_count": int(len(df)),
+        "columns": list(df.columns),
+        "warnings": list(warnings),
+        "immutable": True,
+    }