feat(provider): TET窄试点B——P0四方法Fetcher化(_ex新接口,strict fail-fast),老接口零改动 [vps]

sanguo_portfolio/providers/fetchers/__init__.py

@@ -0,0 +1,42 @@
+"""TET Fetcher 包(窄试点B): P0 四方法三段式(strict fail-fast)。
+
+设计: docs/design/architecture/provider-tet-design.md
+用法: LocalUnifiedProvider 的 ``*_ex`` 方法内部委托本包(对外 API 签名零变化,
+策略层无感);也可直接调 ``Fetcher.fetch(ctx=provider, **kwargs)``(未来 MCP 出口)。
+
+Phase 1(本包)只加新接口不动老接口——老接口照常可用;Phase 2 策略 session copy
+策略副本改调 ``_ex`` 在 NAS 回测对照;Phase 3 验证通过后老接口内部改委托 Fetcher。
+"""
+from .base import (
+    ConstituentQueryParams,
+    DataSchemaError,
+    FundamentalsQueryParams,
+    PanelQueryParams,
+    PriceQueryParams,
+    validate_df_schema,
+)
+from .constituent import ConstituentFetcher
+from .fundamentals import FundamentalsFetcher
+from .price import PanelFetcher, PriceFetcher
+
+# 字典注册(KISS,不用 OpenBB 的 Registry/entry_points)
+FETCHERS = {
+    "price": PriceFetcher,
+    "closes_panel": PanelFetcher,
+    "constituent": ConstituentFetcher,
+    "fundamentals": FundamentalsFetcher,
+}
+
+__all__ = [
+    "PriceFetcher",
+    "PanelFetcher",
+    "ConstituentFetcher",
+    "FundamentalsFetcher",
+    "FETCHERS",
+    "PriceQueryParams",
+    "PanelQueryParams",
+    "ConstituentQueryParams",
+    "FundamentalsQueryParams",
+    "DataSchemaError",
+    "validate_df_schema",
+]

sanguo_portfolio/providers/fetchers/base.py

@@ -0,0 +1,235 @@
+"""TET Fetcher 基础设施(窄试点B;设计: docs/design/architecture/provider-tet-design.md)。
+
+三段式契约(源自 OpenBB Fetcher,本项目化):
+- ``transform_query``: pydantic 入参严格校验(fail-fast)。老接口对非法参数静默
+  返空表/取默认,问题下沉到回测跑完才发现;新接口非法即 ``ValidationError``。
+- ``extract_data``: 唯一 IO 入口(读本地 dbbardata/parquet;复用 provider 惰性
+  连接)。SQL 原样搬迁自 LocalUnifiedProvider 对应方法,行为零变化。
+- ``transform_data``: DataFrame 级 schema 校验 fail-fast(核心列缺失/全空即报错)。
+
+与 OpenBB 的有意差异:
+1. **DataFrame 级校验,非逐行 pydantic Data 模型** —— get_closes_panel 5128只×
+   全历史逐行构造模型性能不可接受;TET 精髓是「IO 集中 + 严格校验」,不是逐行模型。
+2. **extract 读本地不打网络**(本项目落库导向)。
+3. ``high_limit``/``low_limit``/``paused`` 的缺失补默认是**显式契约** (2026-08-15
+   用户定案),不是「兜底」——补 NaN 会让 bullet_trade 把 NaN 当停牌取消全部订单
+   (见 unified-provider-paused-nan-bug)。schema 校验不把这些可选列当脏数据。
+
+已知妥协(Phase 3 切换时归一): qfq 复权因子的第二次读库 (bs_adjust_factor)
+发生在 transform_data 内(经 ctx helper)——因子计算依赖 tail 后的 dates,拆到
+extract 需两次往返;试点期注释标记,不为此过度设计。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Optional, Sequence, Union
+
+from pydantic import BaseModel, ConfigDict, field_validator
+
+# fq 合法值(与老接口 get_price/get_closes_panel 取值一致)
+FQ_VALUES = ("raw", "qfq", "pre", "前复权")
+# get_price 日线 frequency 合法值(老接口其余 frequency 返空 DataFrame=静默;
+# _ex 接口 fail-fast 报错——15m 等周期请用 get_closes_panel_ex)
+PRICE_FREQ_VALUES = ("daily", "day", "1d", "d")
+
+
+class DataSchemaError(ValueError):
+    """transform_data 检出脏数据(fail-fast)。核心列缺失/全空/类型不符。"""
+
+
+def _norm_date(v: Optional[Union[str, datetime]]) -> Optional[str]:
+    """str/datetime → ``YYYY-MM-DD``;非法类型/格式 raise ValueError。"""
+    if v is None:
+        return None
+    if isinstance(v, datetime):
+        return v.strftime("%Y-%m-%d")
+    if isinstance(v, str):
+        s = v.strip()[:10]
+        if len(s) == 10 and s[4] == "-" and s[7] == "-":
+            try:
+                datetime.strptime(s, "%Y-%m-%d")
+                return s
+            except ValueError:
+                pass
+    raise ValueError(f"非法日期(期望 YYYY-MM-DD 或 datetime): {v!r}")
+
+
+class _QueryBase(BaseModel):
+    """TET Query 基类: 未知参数直接报错(extra=forbid,防拼写错误静默吞参)。"""
+
+    model_config = ConfigDict(extra="forbid", validate_assignment=True)
+
+
+class PriceQueryParams(_QueryBase):
+    """get_price_ex 入参。"""
+
+    security: Union[str, Sequence[str]]
+    start_date: Optional[str] = None
+    end_date: Optional[str] = None
+    frequency: str = "daily"
+    fields: Optional[Sequence[str]] = None
+    skip_paused: bool = False
+    fq: str = "raw"
+    count: Optional[int] = None
+    panel: bool = True
+    fill_paused: bool = True
+
+    @field_validator("security")
+    @classmethod
+    def _v_security(cls, v):
+        secs = [v] if isinstance(v, str) else list(v)
+        if not secs:
+            raise ValueError("security 不能为空(老接口静默返空表,_ex fail-fast)")
+        for s in secs:
+            if not isinstance(s, str) or not s.strip():
+                raise ValueError(f"security 含非法代码: {s!r}")
+        return secs
+
+    @field_validator("start_date", "end_date")
+    @classmethod
+    def _v_dates(cls, v):
+        return _norm_date(v)
+
+    @field_validator("frequency")
+    @classmethod
+    def _v_frequency(cls, v):
+        f = str(v or "").lower()
+        if f not in PRICE_FREQ_VALUES:
+            raise ValueError(
+                f"frequency={v!r} 不支持(_ex 只做日线;15m 用 get_closes_panel_ex)"
+            )
+        return f
+
+    @field_validator("fq")
+    @classmethod
+    def _v_fq(cls, v):
+        if v not in FQ_VALUES:
+            raise ValueError(f"fq={v!r} 非法,合法值: {FQ_VALUES}")
+        return v
+
+    @field_validator("count")
+    @classmethod
+    def _v_count(cls, v):
+        if v is not None and v < 1:
+            raise ValueError(f"count={v!r} 必须 >=1")
+        return v
+
+
+class PanelQueryParams(_QueryBase):
+    """get_closes_panel_ex 入参。"""
+
+    symbols: Sequence[str]
+    start: Union[str, datetime]
+    end: Union[str, datetime]
+    interval: str = "d"
+    fq: str = "raw"
+
+    @field_validator("symbols")
+    @classmethod
+    def _v_symbols(cls, v):
+        syms = list(v)
+        if not syms:
+            raise ValueError("symbols 不能为空")
+        for s in syms:
+            if not isinstance(s, str) or not s.strip():
+                raise ValueError(f"symbols 含非法代码: {s!r}")
+        return syms
+
+    @field_validator("start", "end")
+    @classmethod
+    def _v_dates(cls, v):
+        return _norm_date(v)
+
+    @field_validator("interval")
+    @classmethod
+    def _v_interval(cls, v):
+        s = str(v or "")
+        if not s or len(s) > 16 or not all(c.isalnum() or c == "_" for c in s):
+            raise ValueError(f"interval={v!r} 非法(字母数字下划线,<=16)")
+        return s
+
+    @field_validator("fq")
+    @classmethod
+    def _v_fq(cls, v):
+        if v not in FQ_VALUES:
+            raise ValueError(f"fq={v!r} 非法,合法值: {FQ_VALUES}")
+        return v
+
+
+class ConstituentQueryParams(_QueryBase):
+    """get_constituent_ex 入参(date 参数保留但忽略——constituent_unified 并集模型无时点)。"""
+
+    index: str
+    date: Optional[str] = None
+
+    @field_validator("index")
+    @classmethod
+    def _v_index(cls, v):
+        if not isinstance(v, str) or not v.strip():
+            raise ValueError("index 不能为空")
+        return v.strip()
+
+    @field_validator("date")
+    @classmethod
+    def _v_date(cls, v):
+        return _norm_date(v)
+
+
+class FundamentalsQueryParams(_QueryBase):
+    """get_fundamentals_df_ex 入参。"""
+
+    stocks: Sequence[str]
+    date: Optional[str] = None
+    fields: Optional[Sequence[str]] = None
+
+    @field_validator("stocks")
+    @classmethod
+    def _v_stocks(cls, v):
+        stocks = list(v)
+        if not stocks:
+            raise ValueError("stocks 不能为空")
+        for s in stocks:
+            if not isinstance(s, str) or not s.strip():
+                raise ValueError(f"stocks 含非法代码: {s!r}")
+        return stocks
+
+    @field_validator("date")
+    @classmethod
+    def _v_date(cls, v):
+        return _norm_date(v)
+
+    @field_validator("fields")
+    @classmethod
+    def _v_fields(cls, v):
+        if v is None:
+            return None
+        fs = list(v)
+        for f in fs:
+            if not isinstance(f, str) or not f.strip():
+                raise ValueError(f"fields 含非法字段: {f!r}")
+        return fs
+
+
+def validate_df_schema(
+    df: Any,
+    *,
+    required: Sequence[str],
+    non_empty: Sequence[str] = (),
+    context: str = "",
+) -> None:
+    """DataFrame 级 schema 校验(fail-fast)。
+
+    - 空 DataFrame 是**合法**情况(标的在库中无数据=缺失,策略已有处理;脏数据
+      指「有行但核心列空/缺列」)。
+    - ``required``: 列必须存在(dbbardata schema 变化/SQL 拼错在此拦截)。
+    - ``non_empty``: 核心数值列不允许全空(全空=源数据损坏,老接口会静默流出
+      NaN 列,下游 bool(NaN) 坑)。
+    """
+    if df is None or len(df) == 0:
+        return
+    missing = [c for c in required if c not in df.columns]
+    if missing:
+        raise DataSchemaError(f"{context}: 缺失列 {missing}(df.columns={list(df.columns)})")
+    for c in non_empty:
+        if df[c].isna().all():
+            raise DataSchemaError(f"{context}: 核心列 {c!r} 全空(源数据损坏?)")

sanguo_portfolio/providers/fetchers/constituent.py

@@ -0,0 +1,69 @@
+"""成份股 Fetcher: get_constituent_ex(constituent_unified 并集,治幸存者偏差)。
+
+TET 三段式(窄试点B)。extract SQL 原样搬自 LocalUnifiedProvider.get_index_stocks。
+"""
+from __future__ import annotations
+
+import sqlite3
+from typing import TYPE_CHECKING, Any, List
+
+from .base import ConstituentQueryParams, DataSchemaError
+
+if TYPE_CHECKING:
+    from ..local_unified_provider import LocalUnifiedProvider
+
+
+class ConstituentFetcher:
+    """``get_constituent_ex``: 并集模型(in_current=1 OR was_removed=1),契约同老接口。
+
+    ⚠️ 并集语义(与老接口一致): 表无 date 列,含已退市/被踢(治「纯当前」幸存者
+    偏差)但有轻微前视;``date`` 参数保留但忽略。
+    """
+
+    @staticmethod
+    def transform_query(**kwargs: Any) -> ConstituentQueryParams:
+        return ConstituentQueryParams(**kwargs)
+
+    @staticmethod
+    def extract_data(query: ConstituentQueryParams, ctx: "LocalUnifiedProvider") -> List[tuple]:
+        """唯一 IO(constituent_unified)。SQL 读失败不再静默吞(老接口 log+返[],
+        _ex fail-fast——表缺/库锁是数据层问题,应暴露)。"""
+        idx = (
+            query.index.split(".")[0]
+            if "." in query.index else query.index
+        )
+        conn = ctx._connect()
+        try:
+            return conn.execute(
+                "SELECT code FROM constituent_unified WHERE index_code=? "
+                "AND (in_current=1 OR was_removed=1)",
+                (idx,),
+            ).fetchall()
+        except sqlite3.Error as exc:
+            raise DataSchemaError(
+                f"get_constituent_ex[{idx}]: constituent_unified 查询失败: {exc}"
+            ) from exc
+
+    @staticmethod
+    def transform_data(
+        query: ConstituentQueryParams,
+        ctx: "LocalUnifiedProvider",
+        raw: List[tuple],
+    ) -> List[str]:
+        """码清洗: 6 位数字过滤 → jq 格式(原样搬 get_index_stocks :503-510)。"""
+        from ..local_unified_provider import dbbardata_to_jq
+
+        out: List[str] = []
+        for (code,) in raw:
+            code_str = str(code).strip()
+            if len(code_str) != 6 or not code_str.isdigit():
+                continue
+            exc_name = "SSE" if code_str.startswith("6") else "SZSE"
+            out.append(dbbardata_to_jq(code_str, exc_name))
+        return out
+
+    @classmethod
+    def fetch(cls, ctx: "LocalUnifiedProvider", **kwargs: Any) -> List[str]:
+        query = cls.transform_query(**kwargs)
+        raw = cls.extract_data(query, ctx)
+        return cls.transform_data(query, ctx, raw)

sanguo_portfolio/providers/fetchers/fundamentals.py

@@ -0,0 +1,80 @@
+"""财务/估值 Fetcher: get_fundamentals_df_ex。
+
+TET 三段式(窄试点B)。extract 原样搬自 LocalUnifiedProvider.get_fundamentals_df
+(:582-625,fields 短路 + ThreadPool 并发逐只行构建)。
+"""
+from __future__ import annotations
+
+import os
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
+
+import pandas as pd
+
+from .base import FundamentalsQueryParams, validate_df_schema
+
+if TYPE_CHECKING:
+    from ..local_unified_provider import LocalUnifiedProvider
+
+# 与老接口一致: 小列表顺序(避线程池开销), 大列表 ThreadPool 并发(本地文件 I/O 安全)
+_FUND_POOL_THRESHOLD = 64
+
+
+class FundamentalsFetcher:
+    """``get_fundamentals_df_ex``: 多股 fundamentals,列对齐 _FUNDAMENTAL_COLUMNS,契约同老接口。
+
+    数据源路由(同老接口): pe/pb/ps/pcf←valuation_baostock(baostock 权威覆盖)
+    + 市值/三表←static akshare(委托 LocalParquetProvider)。
+    """
+
+    @staticmethod
+    def transform_query(**kwargs: Any) -> FundamentalsQueryParams:
+        return FundamentalsQueryParams(**kwargs)
+
+    @staticmethod
+    def extract_data(
+        query: FundamentalsQueryParams, ctx: "LocalUnifiedProvider"
+    ) -> List[Dict[str, Any]]:
+        """唯一 IO(parquet/估值): fields→need 短路 + 逐只行构建(原样搬 :607-618)。
+
+        注: _build_fundamental_row 内部读 parquet + 委托 LPP——整段视为 IO 段。
+        """
+        date_str = query.date or datetime.now().strftime("%Y-%m-%d")
+        need = ctx._fields_to_need(query.fields) if query.fields else None
+
+        def _one(jq_code: str) -> Dict[str, Any]:
+            return ctx._build_fundamental_row(jq_code, date_str, need)
+
+        stocks: Sequence[str] = query.stocks
+        if len(stocks) <= _FUND_POOL_THRESHOLD:
+            return [_one(s) for s in stocks]
+        workers = min(8, os.cpu_count() or 4)
+        with ThreadPoolExecutor(max_workers=workers) as ex:
+            return list(ex.map(_one, stocks))
+
+    @staticmethod
+    def transform_data(
+        query: FundamentalsQueryParams,
+        ctx: "LocalUnifiedProvider",
+        rows: List[Dict[str, Any]],
+    ) -> pd.DataFrame:
+        """组装+schema 校验(原样搬 :619-625)。code 列是行身份,缺失即脏。"""
+        from ..local_parquet_provider import _FUNDAMENTAL_COLUMNS
+
+        df = pd.DataFrame(rows, columns=_FUNDAMENTAL_COLUMNS)
+        validate_df_schema(
+            df, required=["code"], context="get_fundamentals_df_ex rows",
+        )
+        if "code" in df.columns:
+            df = df.set_index("code", drop=False)
+        if query.fields:
+            keep = ["code"] + [f for f in query.fields if f in df.columns]
+            df = df[keep]
+        return df
+
+    @classmethod
+    def fetch(cls, ctx: "LocalUnifiedProvider", **kwargs: Any) -> pd.DataFrame:
+        query = cls.transform_query(**kwargs)
+        rows = cls.extract_data(query, ctx)
+        return cls.transform_data(query, ctx, rows)

sanguo_portfolio/providers/fetchers/price.py

@@ -0,0 +1,251 @@
+"""行情 Fetcher: get_price_ex(单/多标的日线) + get_closes_panel_ex(批量宽表)。
+
+TET 三段式(窄试点B)。extract SQL 原样搬自 LocalUnifiedProvider.get_price /
+get_closes_panel(e7f9426 生产版,UNION ALL 340× 索引优化保留),合法参数下输出
+与老接口逐值一致(等价性由 tests/portfolio/test_fetchers.py::TestEquivalence 保证)。
+"""
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
+
+import pandas as pd
+
+from .base import PanelQueryParams, PriceQueryParams, validate_df_schema
+
+if TYPE_CHECKING:  # ctx duck-typing: LocalUnifiedProvider(运行时延迟 import 防循环)
+    from ..local_unified_provider import LocalUnifiedProvider
+
+# 复用 local_unified_provider 的 SQL 字面量防注入校验(行为一致)
+_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+_INTERVAL_RE = re.compile(r"^[A-Za-z0-9_]+$")
+
+
+def _safe_date_literal(s: str) -> str:
+    if not isinstance(s, str) or not _DATE_RE.match(s):
+        raise ValueError(f"Invalid date (expect YYYY-MM-DD): {s!r}")
+    return f"'{s}'"
+
+
+def _safe_interval_literal(s: str) -> str:
+    if not isinstance(s, str) or not _INTERVAL_RE.match(s) or len(s) > 16:
+        raise ValueError(f"Invalid interval: {s!r}")
+    return f"'{s}'"
+
+
+class PriceFetcher:
+    """``get_price_ex``: 日线 OHLCV,契约同老 get_price(panel=False 长表/fields 补默认)。"""
+
+    @staticmethod
+    def transform_query(**kwargs: Any) -> PriceQueryParams:
+        return PriceQueryParams(**kwargs)
+
+    @staticmethod
+    def extract_data(query: PriceQueryParams, ctx: "LocalUnifiedProvider") -> Dict[str, pd.DataFrame]:
+        """唯一 IO(①dbbardata): 逐只查 raw 日线列(原样搬 get_price :206-215)。"""
+        from ..local_unified_provider import jq_to_dbbardata
+
+        conn = ctx._connect()
+        start_str = query.start_date or "1990-01-01"
+        end_str = query.end_date or datetime.now().strftime("%Y-%m-%d")
+        q = (
+            "SELECT datetime, open_price, high_price, low_price, close_price, "
+            "volume, turnover FROM dbbardata WHERE symbol=? AND exchange=? "
+            "AND interval='d' AND substr(datetime,1,10)>=? AND substr(datetime,1,10)<=? "
+            "ORDER BY datetime"
+        )
+        frames: Dict[str, pd.DataFrame] = {}
+        for jq_code in query.security:
+            sym, exc = jq_to_dbbardata(jq_code)
+            frames[jq_code] = pd.read_sql(
+                q, conn, params=(sym, exc, start_str, end_str)
+            )
+        return frames
+
+    @staticmethod
+    def transform_data(
+        query: PriceQueryParams,
+        ctx: "LocalUnifiedProvider",
+        raw: Dict[str, pd.DataFrame],
+    ) -> pd.DataFrame:
+        """清洗+schema 校验(原样搬 get_price :216-269)。qfq 因子二次读库=已知妥协(见 base 模块注释)。"""
+        from ..local_unified_provider import _build_qfq_factor, _jq_to_bs_code
+
+        conn = ctx._connect()
+        frames: Dict[str, pd.DataFrame] = {}
+        for jq_code, df in raw.items():
+            # 缺数据标的: schema 校验跳过(空 df 合法),但 raw 列结构必须对
+            if not df.empty:
+                validate_df_schema(
+                    df,
+                    required=["datetime", "close_price"],
+                    non_empty=["close_price"],
+                    context=f"get_price_ex[{jq_code}] raw",
+                )
+            if df.empty:
+                frames[jq_code] = df
+                continue
+            # dbbardata datetime 混合格式(纯日期/带时间),pandas 2.3 严格模式要 mixed
+            df["datetime"] = pd.to_datetime(df["datetime"], format="mixed")
+            df = df.set_index("datetime")
+            df.index.name = None
+            if query.count:
+                df = df.tail(query.count)
+            if query.fq in ("qfq", "pre", "前复权") and not df.empty:
+                factor = _build_qfq_factor(
+                    _jq_to_bs_code(jq_code), conn, pd.Series(df.index)
+                )
+                for col in ("open_price", "high_price", "low_price", "close_price"):
+                    df[col] = df[col].values * factor.values
+            df = df.rename(columns={
+                "open_price": "open", "high_price": "high",
+                "low_price": "low", "close_price": "close",
+            })
+            # 显式契约(非兜底,2026-08-15 用户定案): paused 缺失=False(补 NaN 会被
+            # bullet_trade bool(NaN)=True 当停牌全撤单); high_limit/low_limit 缺失按
+            # close±10% 估算(与 get_current_tick 同口径); 其余可选列缺失补 NaN。
+            if query.fields:
+                for f in query.fields:
+                    if f in df.columns:
+                        continue
+                    if f == "paused":
+                        df[f] = False
+                    elif f == "high_limit" and "close" in df.columns:
+                        df[f] = (df["close"] * 1.1).round(2)
+                    elif f == "low_limit" and "close" in df.columns:
+                        df[f] = (df["close"] * 0.9).round(2)
+                    else:
+                        df[f] = float("nan")
+                df = df[[f for f in query.fields if f in df.columns]]
+            # 出口 schema: 清洗后 close 必须在(qfq/fields 裁剪不应丢核心列)
+            validate_df_schema(
+                df, required=[], non_empty=["close"] if "close" in df.columns else [],
+                context=f"get_price_ex[{jq_code}] out",
+            )
+            frames[jq_code] = df
+
+        if not frames or all(f.empty for f in frames.values()):
+            return pd.DataFrame()
+        if not query.panel:
+            parts: List[pd.DataFrame] = []
+            for jq_code, df in frames.items():
+                if df.empty:
+                    continue
+                d = df.reset_index()
+                if "index" in d.columns and "time" not in d.columns:
+                    d = d.rename(columns={"index": "time"})
+                elif "datetime" in d.columns:
+                    d = d.rename(columns={"datetime": "time"})
+                d.insert(0, "code", jq_code)
+                parts.append(d)
+            return pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
+        if len(frames) == 1:
+            return next(iter(frames.values()))
+        return pd.concat(frames, axis=1)
+
+    @classmethod
+    def fetch(cls, ctx: "LocalUnifiedProvider", **kwargs: Any) -> pd.DataFrame:
+        query = cls.transform_query(**kwargs)
+        raw = cls.extract_data(query, ctx)
+        return cls.transform_data(query, ctx, raw)
+
+
+class PanelFetcher:
+    """``get_closes_panel_ex``: 批量 close 宽表,契约同老 get_closes_panel。"""
+
+    @staticmethod
+    def transform_query(**kwargs: Any) -> PanelQueryParams:
+        return PanelQueryParams(**kwargs)
+
+    @staticmethod
+    def extract_data(
+        query: PanelQueryParams, ctx: "LocalUnifiedProvider"
+    ) -> pd.DataFrame:
+        """唯一 IO(①dbbardata): chunk=400 UNION ALL per symbol(原样搬 :331-346)。
+
+        为何 UNION ALL 而非 OR chain: 大 OR 链打不动复合索引(实证全表扫 10s vs
+        UNION ALL 0.03s,340×);参数 2/symbol×400=800<999 上限。
+        """
+        from ..local_unified_provider import jq_to_dbbardata
+
+        pairs: List[tuple[str, str, str]] = []  # (input_code, db_symbol, db_exchange)
+        for s in query.symbols:
+            sym, exc = jq_to_dbbardata(str(s))
+            pairs.append((str(s), sym, exc))
+
+        conn = ctx._connect()
+        start_lit = _safe_date_literal(query.start or "1990-01-01")
+        end_lit = _safe_date_literal(query.end or datetime.now().strftime("%Y-%m-%d"))
+        interval_lit = _safe_interval_literal(query.interval)
+
+        CHUNK_SIZE = 400
+        chunk_frames: List[pd.DataFrame] = []
+        for i in range(0, len(pairs), CHUNK_SIZE):
+            chunk = pairs[i:i + CHUNK_SIZE]
+            sub_template = (
+                "SELECT datetime, symbol, close_price FROM dbbardata "
+                f"WHERE symbol=? AND exchange=? AND interval={interval_lit} "
+                f"AND substr(datetime,1,10)>={start_lit} "
+                f"AND substr(datetime,1,10)<={end_lit}"
+            )
+            q = " UNION ALL ".join([sub_template] * len(chunk))
+            params: List[Any] = []
+            for _inp, sym, exc in chunk:
+                params.extend([sym, exc])
+            chunk_frames.append(pd.read_sql(q, conn, params=params))
+        return pd.concat(chunk_frames, ignore_index=True) if chunk_frames else pd.DataFrame()
+
+    @staticmethod
+    def transform_data(
+        query: PanelQueryParams,
+        ctx: "LocalUnifiedProvider",
+        raw: pd.DataFrame,
+    ) -> pd.DataFrame:
+        """清洗+schema 校验(原样搬 :348-390)。qfq 批量因子读库经 ctx helper(已知妥协)。"""
+        from ..local_unified_provider import jq_to_dbbardata
+
+        # pairs 重建(transform_data 纯函数需要的映射,不读库)
+        pairs: List[tuple[str, str, str]] = []
+        for s in query.symbols:
+            sym, exc = jq_to_dbbardata(str(s))
+            pairs.append((str(s), sym, exc))
+
+        if raw.empty:
+            return pd.DataFrame(
+                {inp: pd.Series(dtype=float) for inp in query.symbols},
+                index=pd.DatetimeIndex([]),
+            )
+        validate_df_schema(
+            raw, required=["datetime", "symbol", "close_price"],
+            non_empty=["close_price"], context="get_closes_panel_ex raw",
+        )
+        raw["datetime"] = pd.to_datetime(raw["datetime"], format="mixed")
+
+        sym_to_input: Dict[str, str] = {}
+        for inp, sym, _exc in pairs:
+            sym_to_input.setdefault(sym, inp)
+        raw["symbol"] = raw["symbol"].map(sym_to_input).fillna(raw["symbol"])
+        raw = raw.drop_duplicates(subset=["datetime", "symbol"], keep="last")
+        wide = raw.pivot(index="datetime", columns="symbol", values="close_price")
+
+        missing_cols = [inp for inp in query.symbols if inp not in wide.columns]
+        if missing_cols:
+            wide = pd.concat(
+                [wide, pd.DataFrame(float("nan"), index=wide.index, columns=missing_cols)],
+                axis=1,
+            )
+        wide = wide[query.symbols]
+
+        if query.fq in ("qfq", "pre", "前复权"):
+            wide = ctx._apply_qfq_batch(wide, pairs, ctx._connect())
+
+        wide = wide.sort_index()
+        wide.index.name = None
+        return wide
+
+    @classmethod
+    def fetch(cls, ctx: "LocalUnifiedProvider", **kwargs: Any) -> pd.DataFrame:
+        query = cls.transform_query(**kwargs)
+        raw = cls.extract_data(query, ctx)
+        return cls.transform_data(query, ctx, raw)

sanguo_portfolio/providers/local_unified_provider.py

@@ -1005,3 +1005,64 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
         ).fetchall()
         codes = [dbbardata_to_jq(sym, exc) for sym, exc in rows if sym and exc]
         return pd.DataFrame({"code": codes, "display_name": codes})
+
+    # ==================== TET _ex 接口(窄试点B, Phase 1) ====================
+    # 设计: docs/design/architecture/provider-tet-design.md
+    # 契约: 合法参数下输出与老接口逐值一致(等价性测试保证);**非法参数 fail-fast
+    # 报错**(老接口静默返空/取默认)——这是有意的 strict 新行为。
+    # Phase 2: 策略 session copy 策略副本改调 _ex 在 NAS 回测对照;
+    # Phase 3: 验证通过后,老接口内部改为委托 Fetcher(届时本段注释更新)。
+
+    def get_price_ex(
+        self,
+        security: Union[str, List[str]],
+        start_date: Optional[Union[str, datetime]] = None,
+        end_date: Optional[Union[str, datetime]] = None,
+        frequency: str = "daily",
+        fields: Optional[List[str]] = None,
+        skip_paused: bool = False,
+        fq: str = "raw",
+        count: Optional[int] = None,
+        panel: bool = True,
+        fill_paused: bool = True,
+    ) -> pd.DataFrame:
+        """TET 版 get_price(日线)。差异 vs 老接口: 非法 frequency/fq/count/日期 → ValueError。"""
+        from .fetchers.price import PriceFetcher
+        return PriceFetcher.fetch(
+            self, security=security, start_date=start_date, end_date=end_date,
+            frequency=frequency, fields=fields, skip_paused=skip_paused,
+            fq=fq, count=count, panel=panel, fill_paused=fill_paused,
+        )
+
+    def get_closes_panel_ex(
+        self,
+        symbols: List[str],
+        start: Union[str, datetime],
+        end: Union[str, datetime],
+        interval: str = "d",
+        fq: str = "raw",
+    ) -> pd.DataFrame:
+        """TET 版 get_closes_panel(批量 close 宽表)。差异 vs 老接口: 空 symbols/非法日期·interval·fq → ValueError。"""
+        from .fetchers.price import PanelFetcher
+        return PanelFetcher.fetch(
+            self, symbols=symbols, start=start, end=end, interval=interval, fq=fq,
+        )
+
+    def get_constituent_ex(
+        self,
+        index: str,
+        date: Optional[Union[str, datetime]] = None,
+    ) -> List[str]:
+        """TET 版 get_constituent(constituent_unified 并集)。差异 vs 老接口: 查询失败 raise(老接口 log+返[])。"""
+        from .fetchers.constituent import ConstituentFetcher
+        return ConstituentFetcher.fetch(self, index=index, date=date)
+
+    def get_fundamentals_df_ex(
+        self,
+        stocks: List[str],
+        date: Optional[Union[str, datetime]] = None,
+        fields: Optional[List[str]] = None,
+    ) -> pd.DataFrame:
+        """TET 版 get_fundamentals_df(多股财务/估值)。差异 vs 老接口: 空 stocks/非法字段 → ValueError。"""
+        from .fetchers.fundamentals import FundamentalsFetcher
+        return FundamentalsFetcher.fetch(self, stocks=stocks, date=date, fields=fields)

tests/portfolio/test_fetchers.py

@@ -0,0 +1,244 @@
+"""TET Fetcher(窄试点B)测试: 等价性 + strict fail-fast + Optional 契约。
+
+三组覆盖(方案 2026-08-15 用户定案):
+1. **TestEquivalence**: 合法参数下 `_ex` vs 老接口输出逐值一致(extract SQL
+   原样搬迁的回归保障;Phase 2 真策略回测对照前的第一道防线)。
+2. **TestStrictFailFast**: 非法参数(老接口静默返空/取默认)→ ValueError;
+   脏数据(核心列全空)→ DataSchemaError。这是 _ex 的有意新契约。
+3. **TestOptionalContract**: high_limit/low_limit/paused 缺失补默认=显式契约
+   (非兜底;补 NaN 会被 bullet_trade bool(NaN)=True 当停牌全撤单)。
+
+fixture 模式照 test_local_unified_provider.py(tmp_path sqlite 自建表,零网络)。
+"""
+from __future__ import annotations
+
+import sqlite3
+
+import pandas as pd
+import pytest
+from pydantic import ValidationError
+
+from sanguo_portfolio.providers.fetchers import DataSchemaError
+from sanguo_portfolio.providers.local_unified_provider import LocalUnifiedProvider
+
+
+@pytest.fixture
+def unified_provider(tmp_path):
+    """dbbardata(含脏数据股) + bs_adjust_factor + constituent_unified。"""
+    db = tmp_path / "quant_trading.db"
+    c = sqlite3.connect(str(db))
+    c.execute(
+        "CREATE TABLE dbbardata(symbol TEXT, exchange TEXT, datetime TEXT, "
+        "interval TEXT, volume REAL, turnover REAL, open_interest REAL, "
+        "open_price REAL, high_price REAL, low_price REAL, close_price REAL)"
+    )
+    rows = [
+        # 600519: 除权日 2024-06-19 close 1000→900 跳水(qfq 用)
+        ("600519", "SSE", "2024-06-18 00:00:00", "d", 1000, 1e6, 0,
+         1000.0, 1010.0, 990.0, 1000.0),
+        ("600519", "SSE", "2024-06-19 00:00:00", "d", 1000, 1e6, 0,
+         900.0, 910.0, 890.0, 900.0),
+        ("600519", "SSE", "2024-06-20 00:00:00", "d", 1000, 1e6, 0,
+         910.0, 920.0, 900.0, 910.0),
+        # 000001 深市 2 天(panel 多标的/宽表用)
+        ("000001", "SZSE", "2024-06-18 00:00:00", "d", 2e6, 2e7, 0,
+         10.0, 10.5, 9.8, 10.2),
+        ("000001", "SZSE", "2024-06-19 00:00:00", "d", 2e6, 2e7, 0,
+         10.3, 10.6, 10.1, 10.5),
+        # 600999 脏数据: 有行但 close 全 NULL(strict 校验靶)
+        ("600999", "SSE", "2024-06-18 00:00:00", "d", 100, 1e3, 0,
+         None, None, None, None),
+    ]
+    c.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", rows)
+    c.execute(
+        "CREATE TABLE bs_adjust_factor(code TEXT, dividOperateDate TEXT, "
+        "foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL)"
+    )
+    c.execute(
+        "INSERT INTO bs_adjust_factor VALUES('sh.600519','2024-06-19',0.9,0,0)"
+    )
+    c.execute(
+        "CREATE TABLE constituent_unified(index_code TEXT, code TEXT, "
+        "in_current INTEGER, was_removed INTEGER, code_name TEXT)"
+    )
+    c.executemany(
+        "INSERT INTO constituent_unified VALUES(?,?,?,?,?)",
+        [
+            ("000300", "600519", 1, 0, "贵州茅台"),
+            ("000300", "000001", 0, 1, "平安银行"),  # 被踢(治偏差)
+            ("000300", "bad-code", 1, 0, "非法码"),  # transform 清洗靶
+        ],
+    )
+    c.commit()
+    c.close()
+    # data_dir 指向空目录: fundamentals 无 static parquet → 全 NaN 行(两接口同路径)
+    (tmp_path / "static").mkdir(exist_ok=True)
+    return LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+
+
+# ======================== 1. 等价性: _ex vs 老接口 ========================
+class TestEquivalence:
+    """合法参数下逐值一致(assert_frame_equal 默认容差 rtol=1e-5)。"""
+
+    def test_get_price_raw(self, unified_provider):
+        old = unified_provider.get_price(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-20")
+        new = unified_provider.get_price_ex(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-20")
+        pd.testing.assert_frame_equal(old, new)
+
+    def test_get_price_qfq(self, unified_provider):
+        old = unified_provider.get_price(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-20", fq="qfq")
+        new = unified_provider.get_price_ex(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-20", fq="qfq")
+        pd.testing.assert_frame_equal(old, new)
+
+    def test_get_price_count_and_fields(self, unified_provider):
+        kwargs = dict(security="600519.XSHG", count=2,
+                     fields=["close", "high_limit", "paused"], panel=False)
+        old = unified_provider.get_price(**kwargs)
+        new = unified_provider.get_price_ex(**kwargs)
+        pd.testing.assert_frame_equal(old, new)
+
+    def test_get_price_multi_panel(self, unified_provider):
+        secs = ["600519.XSHG", "000001.XSHE"]
+        old = unified_provider.get_price(
+            secs, start_date="2024-06-18", end_date="2024-06-20", panel=False)
+        new = unified_provider.get_price_ex(
+            secs, start_date="2024-06-18", end_date="2024-06-20", panel=False)
+        pd.testing.assert_frame_equal(old, new)
+
+    def test_get_price_missing_symbol_empty_both(self, unified_provider):
+        # 库中无数据的标的: 两接口都返空 DataFrame(空=合法缺失)
+        old = unified_provider.get_price("300999.XSHE", start_date="2024-06-18")
+        new = unified_provider.get_price_ex("300999.XSHE", start_date="2024-06-18")
+        assert old.empty and new.empty
+
+    def test_get_closes_panel_raw_and_missing_col(self, unified_provider):
+        syms = ["600519.XSHG", "000001.XSHE", "300999.XSHE"]  # 含缺失标的
+        old = unified_provider.get_closes_panel(
+            syms, start="2024-06-18", end="2024-06-20")
+        new = unified_provider.get_closes_panel_ex(
+            syms, start="2024-06-18", end="2024-06-20")
+        pd.testing.assert_frame_equal(old, new)
+        assert new["300999.XSHE"].isna().all()  # 缺失标的补全 NaN 列(契约一致)
+
+    def test_get_closes_panel_qfq(self, unified_provider):
+        old = unified_provider.get_closes_panel(
+            ["600519.XSHG"], start="2024-06-18", end="2024-06-20", fq="qfq")
+        new = unified_provider.get_closes_panel_ex(
+            ["600519.XSHG"], start="2024-06-18", end="2024-06-20", fq="qfq")
+        pd.testing.assert_frame_equal(old, new)
+
+    def test_get_constituent(self, unified_provider):
+        old = unified_provider.get_constituent("000300.XSHG")
+        new = unified_provider.get_constituent_ex("000300.XSHG")
+        assert old == new
+        assert "600519.XSHG" in new and "000001.XSHE" in new  # 并集(含被踢)
+        assert not any("bad" in s for s in new)  # 非法码清洗一致
+
+    def test_get_fundamentals_df(self, unified_provider):
+        # 空 static 目录: 两接口同路径返全 NaN 行,验证组装/index/fields 过滤一致
+        old = unified_provider.get_fundamentals_df(["600519.XSHG"], date="2024-06-20")
+        new = unified_provider.get_fundamentals_df_ex(["600519.XSHG"], date="2024-06-20")
+        pd.testing.assert_frame_equal(old, new)
+        old2 = unified_provider.get_fundamentals_df(
+            ["600519.XSHG", "000001.XSHE"], date="2024-06-20", fields=["market_cap"])
+        new2 = unified_provider.get_fundamentals_df_ex(
+            ["600519.XSHG", "000001.XSHE"], date="2024-06-20", fields=["market_cap"])
+        pd.testing.assert_frame_equal(old2, new2)
+
+
+# ======================== 2. strict fail-fast ========================
+class TestStrictFailFast:
+    """非法参数/脏数据直接报错(_ex 有意新契约;老接口静默)。"""
+
+    def test_price_bad_frequency(self, unified_provider):
+        # 老接口对 '1m' 静默返空 DataFrame;_ex 报错(15m 应走 panel_ex)
+        with pytest.raises(ValidationError, match="frequency"):
+            unified_provider.get_price_ex("600519.XSHG", frequency="1m")
+
+    def test_price_bad_fq(self, unified_provider):
+        with pytest.raises(ValidationError, match="fq"):
+            unified_provider.get_price_ex("600519.XSHG", fq="xx")
+
+    def test_price_bad_count(self, unified_provider):
+        with pytest.raises(ValidationError, match="count"):
+            unified_provider.get_price_ex("600519.XSHG", count=0)
+
+    def test_price_bad_date(self, unified_provider):
+        with pytest.raises(ValidationError, match="日期"):
+            unified_provider.get_price_ex("600519.XSHG", start_date="2024/06/18")
+
+    def test_price_empty_security(self, unified_provider):
+        with pytest.raises(ValidationError, match="security"):
+            unified_provider.get_price_ex([])
+
+    def test_price_unknown_kwarg_forbidden(self, unified_provider):
+        # 拼错参数名直接报错(_ex 显式签名 → TypeError;直调 Fetcher → pydantic
+        # extra=forbid ValidationError;老接口 **kwargs 静默吞)
+        with pytest.raises((ValidationError, TypeError)):
+            unified_provider.get_price_ex("600519.XSHG", start_dat="2024-06-18")
+
+    def test_price_dirty_data_close_all_null(self, unified_provider):
+        # 600999 有行但 close 全 NULL: 老接口静默流 NaN,_ex fail-fast
+        with pytest.raises(DataSchemaError, match="close_price"):
+            unified_provider.get_price_ex(
+                "600999.XSHG", start_date="2024-06-18", end_date="2024-06-20")
+
+    def test_panel_empty_symbols(self, unified_provider):
+        with pytest.raises(ValidationError, match="symbols"):
+            unified_provider.get_closes_panel_ex([], start="2024-06-18", end="2024-06-20")
+
+    def test_panel_bad_date(self, unified_provider):
+        with pytest.raises(ValidationError, match="日期"):
+            unified_provider.get_closes_panel_ex(
+                ["600519.XSHG"], start="20240618", end="2024-06-20")
+
+    def test_panel_bad_interval(self, unified_provider):
+        with pytest.raises(ValidationError, match="interval"):
+            unified_provider.get_closes_panel_ex(
+                ["600519.XSHG"], start="2024-06-18", end="2024-06-20", interval="d; DROP TABLE")
+
+    def test_panel_dirty_data(self, unified_provider):
+        with pytest.raises(DataSchemaError):
+            unified_provider.get_closes_panel_ex(
+                ["600999.XSHG"], start="2024-06-18", end="2024-06-20")
+
+    def test_constituent_empty_index(self, unified_provider):
+        with pytest.raises(ValidationError, match="index"):
+            unified_provider.get_constituent_ex("")
+
+    def test_fundamentals_empty_stocks(self, unified_provider):
+        with pytest.raises(ValidationError, match="stocks"):
+            unified_provider.get_fundamentals_df_ex([])
+
+
+# ======================== 3. Optional 契约(非兜底) ========================
+class TestOptionalContract:
+    """high_limit/low_limit/paused 缺失补默认=2026-08-15 用户定案的显式契约。"""
+
+    def test_high_limit_estimated_not_nan(self, unified_provider):
+        df = unified_provider.get_price_ex(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-18",
+            fields=["close", "high_limit", "low_limit"],
+        )
+        assert abs(df["high_limit"].iloc[0] - round(1000.0 * 1.1, 2)) < 1e-6
+        assert abs(df["low_limit"].iloc[0] - round(1000.0 * 0.9, 2)) < 1e-6
+
+    def test_paused_false_not_nan(self, unified_provider):
+        df = unified_provider.get_price_ex(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-18",
+            fields=["close", "paused"],
+        )
+        # bool(NaN)=True 会被当停牌;契约=显式 False
+        assert not df["paused"].iloc[0]
+
+    def test_unknown_field_nan(self, unified_provider):
+        # 未知可选列(如聚宽特有字段)缺失 → NaN(声明行为,策略自查)
+        df = unified_provider.get_price_ex(
+            "600519.XSHG", start_date="2024-06-18", end_date="2024-06-18",
+            fields=["close", "acc_net_value"],
+        )
+        assert df["acc_net_value"].isna().all()