diff --git a/sanguo_portfolio/providers/fetchers/__init__.py b/sanguo_portfolio/providers/fetchers/__init__.py new file mode 100644 index 0000000..6fdfbbd --- /dev/null +++ b/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", +] diff --git a/sanguo_portfolio/providers/fetchers/base.py b/sanguo_portfolio/providers/fetchers/base.py new file mode 100644 index 0000000..fa4fb29 --- /dev/null +++ b/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} 全空(源数据损坏?)") diff --git a/sanguo_portfolio/providers/fetchers/constituent.py b/sanguo_portfolio/providers/fetchers/constituent.py new file mode 100644 index 0000000..26f5887 --- /dev/null +++ b/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) diff --git a/sanguo_portfolio/providers/fetchers/fundamentals.py b/sanguo_portfolio/providers/fetchers/fundamentals.py new file mode 100644 index 0000000..bace179 --- /dev/null +++ b/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) diff --git a/sanguo_portfolio/providers/fetchers/price.py b/sanguo_portfolio/providers/fetchers/price.py new file mode 100644 index 0000000..377a6b9 --- /dev/null +++ b/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) diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index 62605c3..94e5505 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/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) diff --git a/tests/portfolio/test_fetchers.py b/tests/portfolio/test_fetchers.py new file mode 100644 index 0000000..81f5d5b --- /dev/null +++ b/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()