diff --git a/README.md b/README.md
index 35de9ba..3ac93af 100644
--- a/README.md
+++ b/README.md
@@ -139,15 +139,22 @@ print(factor_backtest.benchmark_stats(benchmark_returns))
 
 # 下游稳定交付：显式提供代码版本、数据快照和时区，不在核心层写数据库。
 from quant_engine.artifact import build_research_run_artifact
+from quant_engine.data_adapter import prepare_asset_return_snapshot
 from quant_engine.risk import estimate_covariance_snapshot
 
 risk_date = factor_backtest.position_weights.index[-1].date()
+market_snapshot = prepare_asset_return_snapshot(
+    qtdb_daily_long,
+    source="qtdb_pro.hq_daily",
+    source_snapshot_id="<upstream-ingestion-snapshot-id>",
+    adjustment="qfq",
+)
 risk_snapshot = estimate_covariance_snapshot(
-    daily_asset_returns,
+    market_snapshot.returns,
     as_of_date=risk_date,
     lookback_sessions=252,
     min_observations=120,
-    data_snapshot_id="<risk-return-data-snapshot-id>",
+    data_snapshot_id=market_snapshot.data_snapshot_id,
 )
 
 artifact = build_research_run_artifact(
@@ -158,7 +165,7 @@ artifact = build_research_run_artifact(
     strategy_version="1.0.0",
     engine_version="1.2.0",
     code_revision="<git-sha>",
-    data_snapshot_id="<data-snapshot-id>",
+    data_snapshot_id=market_snapshot.data_snapshot_id,
     calendar="CN-A",
     timezone="Asia/Shanghai",
     started_at="2026-08-21T10:00:00+08:00",
diff --git a/docs/OPEN_SOURCE_REFERENCES.md b/docs/OPEN_SOURCE_REFERENCES.md
index f0fa4f9..645184c 100644
--- a/docs/OPEN_SOURCE_REFERENCES.md
+++ b/docs/OPEN_SOURCE_REFERENCES.md
@@ -46,6 +46,16 @@ component risk 闭合到年化组合波动，percentage contribution 闭合到 1
 pairwise 样本集合产生含义不一致的矩阵。snapshot ID 对窗口数据、缺失掩码、上游数据
 快照身份和估计参数做 SHA-256，追加未来数据不会改变历史快照。
 
+市场适配层现以 `AssetReturnSnapshot` 固化 simple-return 输入：上游 ingestion snapshot ID、
+数据源、价格字段、复权口径、规范化价格值和缺失掩码共同形成内容寻址 ID；不前向填充
+停牌/缺失价格。该 ID 同时传入协方差快照和研究运行工件，避免同一研究链出现两套数据
+身份。
+
+可选 shrinkage adapter 的评估结论是“保留边界，暂不实现”：当前运行依赖没有声明
+scikit-learn，本切片也不修改版本或锁文件。未来只有在依赖治理接受后，才以延迟导入
+直接调用 scikit-learn 的 `LedoitWolf` / `OAS`，并让估计器名称、库版本与参数进入
+snapshot identity；不复制成熟统计公式，也不让环境中偶然存在的包改变 baseline 行为。
+
 ## hikyuu 的定位
 
 [hikyuu](https://github.com/fasiondog/hikyuu) 的 SG / MM / CN / PG 部件化思想、
diff --git a/docs/handoff/2026-08-21-research-artifact-contract.md b/docs/handoff/2026-08-21-research-artifact-contract.md
index d2fb347..2257e6d 100644
--- a/docs/handoff/2026-08-21-research-artifact-contract.md
+++ b/docs/handoff/2026-08-21-research-artifact-contract.md
@@ -40,9 +40,14 @@
 - `covariance_as_of_date` 不得晚于 `trade_date`；无正组合方差时拒绝产物。
 - `estimate_covariance_snapshot` 从显式数据快照的日收益生成无前视、complete-case、
   SHA-256 可复现的 per-period sample covariance；不包含 I/O 或未来行。
+- `prepare_asset_return_snapshot` 从规范化长表行情生成不前向填充的 simple daily returns；
+  显式 ingestion snapshot ID、源/字段/复权口径、价格值和缺失掩码共同形成
+  `asset-returns-v1:<sha256>`，并把同一 ID 传给 covariance 与 run artifact。
+- shrinkage 适配器本轮不实现：scikit-learn 尚非声明依赖，未来只允许薄适配
+  `LedoitWolf` / `OAS`，不复制公式、不依赖环境偶然安装状态。
 
 ## Next action
 
 保持 Draft PR #5，不绕过堆叠顺序合并；下游 `research_results` / `research_platform`
-已在各自 Draft 分支兼容 1.0.0 / 1.1.0，下一阶段让市场数据适配器提供带稳定
-`data_snapshot_id` 的资产日收益，再评估可选 scikit-learn shrinkage adapter。
+继续在现有 Draft 分支消费同一数据 lineage。下一阶段优先把 ingestion snapshot ID 从
+真实 ELT 元数据接入调用方，再在依赖治理通过后单独交付可选 shrinkage adapter。
diff --git a/src/quant_engine/data_adapter.py b/src/quant_engine/data_adapter.py
index 041b6c3..28b3496 100644
--- a/src/quant_engine/data_adapter.py
+++ b/src/quant_engine/data_adapter.py
@@ -6,22 +6,27 @@
 - execution.py 需要**宽表**（date × stock_code）prices / volumes
 - Tushare 字段命名：`ts_code / vol(手) / amount(千元) / pct_chg`，且**无 vwap 字段**
 
-本模块提供 6 个纯函数，让新模块直接吃 qtdb_pro 真实数据：
+本模块提供可组合的数据适配函数，让新模块直接吃 qtdb_pro 真实数据：
 1. `long_to_wide()` — 长表 → 宽表（date × stock_code）
 2. `wide_to_long()` — 宽表 → 长表
 3. `rename_tushare_columns()` — 列名映射（ts_code→stock_code, vol→volume 等）
 4. `add_vwap_proxy()` — vwap 代理（Tushare 无 vwap 字段）
 5. `apply_adj_factor()` — 复权（hq_daily × hq_adj_factor 前复权）
 6. `prepare_stock_series()` — 单股提取（alpha_factors 输入）
-7. `prepare_execution_inputs()` — execution 输入（prices + volumes 宽表）
-8. `load_qtdb_daily()` — 便捷加载（qtdb_pro.hq_daily + 可选复权）
+7. `prepare_asset_return_snapshot()` — 带稳定 lineage 的资产日收益
+8. `prepare_execution_inputs()` — execution 输入（prices + volumes 宽表）
+9. `load_qtdb_daily()` — 便捷加载（qtdb_pro.hq_daily + 可选复权）
 
 全部纯 pandas/numpy，零新依赖，mypy strict 兼容。
 """
 
 from __future__ import annotations
 
+import hashlib
+import json
 from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import date
 from typing import Any
 
 import numpy as np
@@ -32,12 +37,14 @@ from quant_engine.logging import get_logger
 logger = get_logger(__name__)
 
 __all__ = [
+    "AssetReturnSnapshot",
     "long_to_wide",
     "wide_to_long",
     "rename_tushare_columns",
     "add_vwap_proxy",
     "apply_adj_factor",
     "prepare_stock_series",
+    "prepare_asset_return_snapshot",
     "prepare_execution_inputs",
     "load_qtdb_daily",
 ]
@@ -57,6 +64,107 @@ TUSHARE_RENAME: dict[str, str] = {
 }
 
 
+@dataclass(frozen=True, slots=True, init=False, eq=False)
+class AssetReturnSnapshot:
+    """Immutable-by-interface daily return matrix with reproducible lineage."""
+
+    data_snapshot_id: str
+    source: str
+    source_snapshot_id: str
+    price_field: str
+    adjustment: str
+    return_method: str
+    start_date: date
+    end_date: date
+    sessions: int
+    assets: tuple[str, ...]
+    _returns: pd.DataFrame
+
+    def __init__(
+        self,
+        *,
+        data_snapshot_id: str,
+        source: str,
+        source_snapshot_id: str,
+        price_field: str,
+        adjustment: str,
+        return_method: str,
+        start_date: date,
+        end_date: date,
+        assets: tuple[str, ...],
+        returns: pd.DataFrame,
+    ) -> None:
+        for value, name in (
+            (data_snapshot_id, "data_snapshot_id"),
+            (source, "source"),
+            (source_snapshot_id, "source_snapshot_id"),
+            (price_field, "price_field"),
+            (adjustment, "adjustment"),
+            (return_method, "return_method"),
+        ):
+            if not isinstance(value, str) or not value.strip():
+                raise ValueError(f"{name} must be non-empty")
+        if returns.empty or not isinstance(returns.index, pd.DatetimeIndex):
+            raise ValueError("returns must contain a DatetimeIndex and at least one session")
+        if tuple(returns.columns) != assets:
+            raise ValueError("assets must match returns columns")
+        if start_date > end_date:
+            raise ValueError("start_date must not be after end_date")
+
+        object.__setattr__(self, "data_snapshot_id", data_snapshot_id.strip())
+        object.__setattr__(self, "source", source.strip())
+        object.__setattr__(self, "source_snapshot_id", source_snapshot_id.strip())
+        object.__setattr__(self, "price_field", price_field.strip())
+        object.__setattr__(self, "adjustment", adjustment.strip())
+        object.__setattr__(self, "return_method", return_method.strip())
+        object.__setattr__(self, "start_date", start_date)
+        object.__setattr__(self, "end_date", end_date)
+        object.__setattr__(self, "sessions", len(returns))
+        object.__setattr__(self, "assets", assets)
+        object.__setattr__(self, "_returns", returns.copy(deep=True))
+
+    @property
+    def returns(self) -> pd.DataFrame:
+        """Return an isolated copy so callers cannot mutate the snapshot."""
+        return self._returns.copy(deep=True)
+
+
+def _non_empty(value: str, name: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{name} must be non-empty")
+    return value.strip()
+
+
+def _asset_return_snapshot_id(
+    prices: pd.DataFrame,
+    *,
+    source: str,
+    source_snapshot_id: str,
+    price_field: str,
+    adjustment: str,
+) -> str:
+    values = prices.to_numpy(dtype=float, copy=True)
+    missing = np.isnan(values)
+    normalized = np.where(missing, 0.0, values).astype("<f8", copy=False)
+    metadata = {
+        "adjustment": adjustment,
+        "assets": [str(asset) for asset in prices.columns],
+        "price_field": price_field,
+        "return_method": "simple",
+        "schema": "asset-returns-v1",
+        "sessions": [timestamp.date().isoformat() for timestamp in prices.index],
+        "shape": list(values.shape),
+        "source": source,
+        "source_snapshot_id": source_snapshot_id,
+    }
+    digest = hashlib.sha256(
+        json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    )
+    digest.update(missing.astype(np.uint8, copy=False).tobytes(order="C"))
+    digest.update(normalized.tobytes(order="C"))
+    return f"asset-returns-v1:{digest.hexdigest()}"
+
+
 def long_to_wide(
     df: pd.DataFrame,
     value_col: str = "close",
@@ -283,6 +391,98 @@ def prepare_stock_series(
     return series_map
 
 
+def prepare_asset_return_snapshot(
+    df: pd.DataFrame,
+    *,
+    source: str,
+    source_snapshot_id: str,
+    price_col: str = "close",
+    adjustment: str = "none",
+    stock_col: str = "stock_code",
+    date_col: str = "trade_date",
+) -> AssetReturnSnapshot:
+    """Build deterministic simple daily returns from a long market-price table.
+
+    ``source_snapshot_id`` must identify the upstream ingestion snapshot. The
+    resulting ID additionally fingerprints canonical price values and their
+    missing mask, so changed contents cannot retain the same downstream identity.
+    Missing prices are never forward-filled.
+    """
+    normalized_source = _non_empty(source, "source")
+    normalized_source_snapshot_id = _non_empty(
+        source_snapshot_id,
+        "source_snapshot_id",
+    )
+    normalized_price_col = _non_empty(price_col, "price_col")
+    normalized_adjustment = _non_empty(adjustment, "adjustment")
+    if not isinstance(df, pd.DataFrame):
+        raise TypeError("df must be a pandas DataFrame")
+    if df.empty:
+        raise ValueError("df must contain market prices")
+    required = {date_col, stock_col, normalized_price_col}
+    missing_columns = sorted(required.difference(df.columns))
+    if missing_columns:
+        raise ValueError(f"prepare_asset_return_snapshot: missing columns={missing_columns}")
+
+    market = df[[date_col, stock_col, normalized_price_col]].copy()
+    if any(not isinstance(asset, str) or not asset.strip() for asset in market[stock_col]):
+        raise ValueError("asset labels must be non-empty strings")
+    market[stock_col] = market[stock_col].str.strip()
+    try:
+        normalized_dates = pd.to_datetime(market[date_col], errors="raise")
+    except (TypeError, ValueError) as error:
+        raise ValueError("trade dates must be valid dates") from error
+    if normalized_dates.isna().any():
+        raise ValueError("trade dates must be valid dates")
+    market[date_col] = normalized_dates.dt.normalize()
+    if market.duplicated(subset=[date_col, stock_col]).any():
+        raise ValueError("duplicate asset/session prices are not allowed")
+
+    try:
+        market[normalized_price_col] = pd.to_numeric(
+            market[normalized_price_col],
+            errors="raise",
+        )
+    except (TypeError, ValueError) as error:
+        raise ValueError("prices must be numeric") from error
+    observed_prices = market[normalized_price_col].dropna().to_numpy(dtype=float)
+    if observed_prices.size == 0 or not np.isfinite(observed_prices).all():
+        raise ValueError("prices must contain positive finite observations")
+    if (observed_prices <= 0.0).any():
+        raise ValueError("prices must contain positive finite observations")
+
+    prices = market.pivot(
+        index=date_col,
+        columns=stock_col,
+        values=normalized_price_col,
+    ).sort_index()
+    prices = prices.reindex(sorted(str(asset) for asset in prices.columns), axis="columns")
+    prices = prices.astype(float)
+    if len(prices) < 2:
+        raise ValueError("market prices must contain at least two sessions")
+    returns = prices.pct_change(fill_method=None)
+    assets = tuple(str(asset) for asset in prices.columns)
+    snapshot_id = _asset_return_snapshot_id(
+        prices,
+        source=normalized_source,
+        source_snapshot_id=normalized_source_snapshot_id,
+        price_field=normalized_price_col,
+        adjustment=normalized_adjustment,
+    )
+    return AssetReturnSnapshot(
+        data_snapshot_id=snapshot_id,
+        source=normalized_source,
+        source_snapshot_id=normalized_source_snapshot_id,
+        price_field=normalized_price_col,
+        adjustment=normalized_adjustment,
+        return_method="simple",
+        start_date=prices.index[0].date(),
+        end_date=prices.index[-1].date(),
+        assets=assets,
+        returns=returns,
+    )
+
+
 def prepare_execution_inputs(
     df: pd.DataFrame,
     stock_col: str = "stock_code",
diff --git a/tests/test_data_adapter.py b/tests/test_data_adapter.py
index df21754..87b5af5 100644
--- a/tests/test_data_adapter.py
+++ b/tests/test_data_adapter.py
@@ -7,10 +7,12 @@ import pandas as pd
 import pytest
 
 from quant_engine.data_adapter import (
+    AssetReturnSnapshot,
     add_vwap_proxy,
     apply_adj_factor,
     load_qtdb_daily,
     long_to_wide,
+    prepare_asset_return_snapshot,
     prepare_execution_inputs,
     prepare_stock_series,
     rename_tushare_columns,
@@ -298,13 +300,176 @@ def test_prepare_execution_inputs_missing_close_raises() -> None:
 
 
 def test_prepare_execution_inputs_missing_selected_price_raises() -> None:
-    df = pd.DataFrame(
-        {"stock_code": ["A"], "trade_date": ["2024-01-01"], "close": [10.0]}
-    )
+    df = pd.DataFrame({"stock_code": ["A"], "trade_date": ["2024-01-01"], "close": [10.0]})
     with pytest.raises(ValueError, match="缺 open"):
         prepare_execution_inputs(df, price_col="open")
 
 
+# ── prepare_asset_return_snapshot ────────────────────────────
+
+
+def _daily_prices() -> pd.DataFrame:
+    return pd.DataFrame(
+        {
+            "stock_code": ["B", "A", "B", "A", "B", "A"],
+            "trade_date": [
+                "2024-01-02",
+                "2024-01-01",
+                "2024-01-01",
+                "2024-01-03",
+                "2024-01-03",
+                "2024-01-02",
+            ],
+            "close": [18.0, 10.0, 20.0, 12.1, 19.8, 11.0],
+        }
+    )
+
+
+def test_prepare_asset_return_snapshot_is_stable_and_immutable_by_interface() -> None:
+    snapshot = prepare_asset_return_snapshot(
+        _daily_prices(),
+        source="qtdb_pro.hq_daily",
+        source_snapshot_id="hq-daily:2024-01-03:v1",
+        adjustment="qfq",
+    )
+
+    assert isinstance(snapshot, AssetReturnSnapshot)
+    assert snapshot.data_snapshot_id.startswith("asset-returns-v1:")
+    assert snapshot.source == "qtdb_pro.hq_daily"
+    assert snapshot.source_snapshot_id == "hq-daily:2024-01-03:v1"
+    assert snapshot.price_field == "close"
+    assert snapshot.adjustment == "qfq"
+    assert snapshot.return_method == "simple"
+    assert snapshot.start_date.isoformat() == "2024-01-01"
+    assert snapshot.end_date.isoformat() == "2024-01-03"
+    assert snapshot.sessions == 3
+    assert snapshot.assets == ("A", "B")
+
+    expected = pd.DataFrame(
+        {
+            "A": [np.nan, 0.1, 0.1],
+            "B": [np.nan, -0.1, 0.1],
+        },
+        index=pd.to_datetime(["2024-01-01", "2024-01-02", "2024-01-03"]),
+    )
+    expected.index.name = "trade_date"
+    expected.columns.name = "stock_code"
+    pd.testing.assert_frame_equal(snapshot.returns, expected)
+
+    exposed = snapshot.returns
+    exposed.iloc[1, 0] = 999.0
+    assert snapshot.returns.iloc[1, 0] == pytest.approx(0.1)
+
+
+def test_asset_return_snapshot_identity_is_order_independent_and_content_addressed() -> None:
+    kwargs = {
+        "source": "qtdb_pro.hq_daily",
+        "source_snapshot_id": "hq-daily:2024-01-03:v1",
+        "adjustment": "none",
+    }
+    baseline = prepare_asset_return_snapshot(_daily_prices(), **kwargs)
+    shuffled = prepare_asset_return_snapshot(
+        _daily_prices().sample(frac=1.0, random_state=7),
+        **kwargs,
+    )
+    changed_prices = _daily_prices().copy()
+    changed_prices.loc[changed_prices["close"] == 12.1, "close"] = 12.2
+    changed_content = prepare_asset_return_snapshot(changed_prices, **kwargs)
+    changed_source = prepare_asset_return_snapshot(
+        _daily_prices(),
+        source="qtdb_pro.hq_daily",
+        source_snapshot_id="hq-daily:2024-01-03:v2",
+        adjustment="none",
+    )
+
+    assert shuffled.data_snapshot_id == baseline.data_snapshot_id
+    assert changed_content.data_snapshot_id != baseline.data_snapshot_id
+    assert changed_source.data_snapshot_id != baseline.data_snapshot_id
+
+
+def test_prepare_asset_return_snapshot_does_not_fill_missing_prices() -> None:
+    prices = _daily_prices()
+    prices.loc[
+        (prices["stock_code"] == "A") & (prices["trade_date"] == "2024-01-02"),
+        "close",
+    ] = np.nan
+
+    snapshot = prepare_asset_return_snapshot(
+        prices,
+        source="qtdb_pro.hq_daily",
+        source_snapshot_id="hq-daily:missing-middle",
+    )
+
+    assert pd.isna(snapshot.returns.loc[pd.Timestamp("2024-01-02"), "A"])
+    assert pd.isna(snapshot.returns.loc[pd.Timestamp("2024-01-03"), "A"])
+
+
+def test_prepare_asset_return_snapshot_rejects_duplicate_sessions() -> None:
+    duplicate = pd.concat([_daily_prices(), _daily_prices().iloc[[0]]], ignore_index=True)
+
+    with pytest.raises(ValueError, match="duplicate"):
+        prepare_asset_return_snapshot(
+            duplicate,
+            source="qtdb_pro.hq_daily",
+            source_snapshot_id="hq-daily:duplicate",
+        )
+
+
+@pytest.mark.parametrize("invalid_price", [0.0, -1.0, np.inf])
+def test_prepare_asset_return_snapshot_rejects_invalid_prices(invalid_price: float) -> None:
+    prices = _daily_prices()
+    prices.loc[0, "close"] = invalid_price
+
+    with pytest.raises(ValueError, match="positive finite"):
+        prepare_asset_return_snapshot(
+            prices,
+            source="qtdb_pro.hq_daily",
+            source_snapshot_id="hq-daily:invalid-price",
+        )
+
+
+@pytest.mark.parametrize(
+    ("source", "source_snapshot_id", "adjustment"),
+    [
+        ("", "source-1", "none"),
+        ("qtdb_pro.hq_daily", "", "none"),
+        ("qtdb_pro.hq_daily", "source-1", ""),
+    ],
+)
+def test_prepare_asset_return_snapshot_requires_explicit_identity_semantics(
+    source: str,
+    source_snapshot_id: str,
+    adjustment: str,
+) -> None:
+    with pytest.raises(ValueError, match="must be non-empty"):
+        prepare_asset_return_snapshot(
+            _daily_prices(),
+            source=source,
+            source_snapshot_id=source_snapshot_id,
+            adjustment=adjustment,
+        )
+
+
+def test_asset_return_snapshot_feeds_reproducible_covariance_lineage() -> None:
+    from quant_engine.risk import estimate_covariance_snapshot
+
+    market_snapshot = prepare_asset_return_snapshot(
+        _daily_prices(),
+        source="qtdb_pro.hq_daily",
+        source_snapshot_id="hq-daily:2024-01-03:v1",
+    )
+    covariance_snapshot = estimate_covariance_snapshot(
+        market_snapshot.returns,
+        as_of_date=market_snapshot.end_date,
+        lookback_sessions=3,
+        min_observations=2,
+        data_snapshot_id=market_snapshot.data_snapshot_id,
+    )
+
+    assert covariance_snapshot.data_snapshot_id == market_snapshot.data_snapshot_id
+    assert covariance_snapshot.snapshot_id.startswith("sample-cov-v1:")
+
+
 # ── 端到端：长表 → 适配 → alpha158 + execution ──────────────