feat(quant): consolidate research artifact contract (#7)

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,10 +391,104 @@ 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",
     date_col: str = "trade_date",
+    *,
+    price_col: str = "close",
 ) -> tuple[pd.DataFrame, pd.DataFrame]:
     """长表行情 → execution 输入（prices + volumes 宽表）。
 
@@ -294,10 +496,11 @@ def prepare_execution_inputs(
         df: 长表行情（含 close / volume 列，Tushare rename 后）
         stock_col: 股票代码列名
         date_col: 日期列名
+        price_col: 执行价字段，默认 close；防前视研究可显式选择下一交易日 open
 
     Returns:
         (prices_wide, volumes_wide)：
-        - prices_wide: date × stock_code，值=close
+        - prices_wide: date × stock_code，值=price_col
         - volumes_wide: date × stock_code，值=volume（若无 volume 列则全 1.0）
 
     Examples:
@@ -313,9 +516,9 @@ def prepare_execution_inputs(
     """
     if df.empty:
         return pd.DataFrame(), pd.DataFrame()
-    if "close" not in df.columns:
-        raise ValueError(f"prepare_execution_inputs: 缺 close 列，实际列={list(df.columns)}")
-    prices = long_to_wide(df, value_col="close", date_col=date_col, stock_col=stock_col)
+    if price_col not in df.columns:
+        raise ValueError(f"prepare_execution_inputs: 缺 {price_col} 列，实际列={list(df.columns)}")
+    prices = long_to_wide(df, value_col=price_col, date_col=date_col, stock_col=stock_col)
     if "volume" in df.columns:
         volumes = long_to_wide(df, value_col="volume", date_col=date_col, stock_col=stock_col)
     else: