From e12c66acd52912db3142550b4650e9c86b670606 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Thu, 23 Jul 2026 08:05:09 +0800 Subject: [PATCH] =?UTF-8?q?feat(portfolio):=20LocalUnifiedProvider=20?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E8=BD=AC=E6=8D=A2+=E5=A4=8D=E6=9D=83?= =?UTF-8?q?=E5=9B=A0=E5=AD=90(Task0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../providers/local_unified_provider.py | 120 +++++++++++++++ .../portfolio/test_local_unified_provider.py | 145 ++++++++++++++++++ 2 files changed, 265 insertions(+) create mode 100644 sanguo_portfolio/providers/local_unified_provider.py create mode 100644 tests/portfolio/test_local_unified_provider.py diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py new file mode 100644 index 0000000..8f0318d --- /dev/null +++ b/sanguo_portfolio/providers/local_unified_provider.py @@ -0,0 +1,120 @@ +"""LocalUnifiedProvider: 读方案A 权威数据层, 零 online, 治幸存者偏差(spec §6)。 + +数据源(全本地 VPS ``C:\\sanguo_vnpy_v2\\data\\``): +- 日线: ``dbbardata('d')`` raw + ``bs_adjust_factor`` 算前复权(§14.7) +- 成份股: ``constituent_unified`` 并集(治偏差,无 date 时点) +- 估值 pe/pb/ps/pcf: ``valuation_baostock/.parquet``(baostock 权威) +- 市值/股本: ``static/valuation`` akshare parquet(baostock 无市值列) +- 三表: ``static/{balance,income,cashflow}`` akshare parquet(委托 LocalParquetProvider) + +零 online: 不 import baostock 调 online。Mac 测试用 sqlite+parquet fixture。 +""" +from __future__ import annotations + +import logging +import os +import sqlite3 +from datetime import datetime +from typing import Any, Dict, List, Optional, Union + +import numpy as np +import pandas as pd + +# bullet-trade 可能未装,容错 import DataProvider(照 local_parquet_provider 模式) +try: + from bullet_trade.data.providers.base import DataProvider # type: ignore +except ImportError: # Mac dev 环境未装,允许模块加载 + class DataProvider: # type: ignore[no-redef] + name: str = "base" + + +logger = logging.getLogger(__name__) + +# VPS 默认路径(Windows); Mac 测试通过 config["db_path"] / config["data_dir"] 覆盖 +_DEFAULT_DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +_DEFAULT_DATA_DIR = r"C:\sanguo_vnpy_v2\data" + +# jq 后缀 ↔ dbbardata exchange +_JQ_SUFFIX_TO_EXC = {"XSHG": "SSE", "XSHE": "SZSE", "SH": "SSE", "SZ": "SZSE"} +_EXC_TO_JQ_SUFFIX = {"SSE": "XSHG", "SZSE": "XSHE"} + + +def jq_to_dbbardata(jq_code: str) -> tuple[str, str]: + """``600519.XSHG`` → ``("600519", "SSE")``。 + + 纯 6 位按 6 开头 = sh / 0,3 开头 = sz 推断。 + """ + s = (jq_code or "").strip() + if "." not in s: + if len(s) == 6: + return s, ("SSE" if s.startswith("6") else "SZSE") + return s, "SSE" + code, suffix = s.split(".", 1) + return code, _JQ_SUFFIX_TO_EXC.get(suffix.upper(), "SSE") + + +def dbbardata_to_jq(symbol: str, exchange: str) -> str: + """``("600519", "SSE")`` → ``"600519.XSHG"``。""" + jq_suffix = _EXC_TO_JQ_SUFFIX.get(str(exchange).upper(), "XSHG") + return f"{symbol}.{jq_suffix}" + + +def _jq_to_bs_code(jq_code: str) -> str: + """``600519.XSHG`` → ``"sh.600519"``(``bs_adjust_factor.code`` 格式)。""" + sym, exc = jq_to_dbbardata(jq_code) + prefix = "sh" if exc == "SSE" else "sz" + return f"{prefix}.{sym}" + + +def _build_qfq_factor( + bs_code: str, + conn: sqlite3.Connection, + dates: pd.Series, +) -> pd.Series: + """构造每个 date 的前复权因子(asof 语义)。 + + 规则: + - 找 ``<= d`` 的最大 dividOperateDate 的 foreAdjustFactor + - 全部事件 > d(早于所有事件)→ 用最早的 factor + - 全部事件 <= d(晚于所有事件)→ 用最新的 factor + - 无事件 → 全 1.0 + + ``qfq[t] = raw[t] * factor[t]`` + """ + rows = conn.execute( + "SELECT dividOperateDate, foreAdjustFactor FROM bs_adjust_factor " + "WHERE code=? ORDER BY dividOperateDate", + (bs_code,), + ).fetchall() + dates_ts = pd.to_datetime(dates) + if not rows: + return pd.Series([1.0] * len(dates_ts), index=dates_ts) + ev_dates = pd.to_datetime([r[0] for r in rows]) + factors = [float(r[1]) for r in rows] + out: List[float] = [] + for d in dates_ts: + mask = ev_dates <= d + if mask.any(): + # <= d 的最大事件 = 最后一个 True + idx = int(np.where(mask)[0][-1]) + else: + # 全部 > d → 用最早(第一个) + idx = 0 + out.append(factors[idx]) + return pd.Series(out, index=dates_ts) + + +class LocalUnifiedProvider(DataProvider): # type: ignore[misc] + """占位类(Task0 骨架; Task1-4 填充方法)。 + + 读方案A 权威数据层, 零 online, 治幸存者偏差(spec §6 使用层)。 + """ + + name: str = "sanguo_local_unified" + requires_live_data: bool = False + + def __init__(self, config: Optional[Dict[str, Any]] = None) -> None: + cfg = config or {} + self.db_path: str = cfg.get("db_path", _DEFAULT_DB) + self.data_dir: str = cfg.get("data_dir", _DEFAULT_DATA_DIR) + self._conn: Optional[sqlite3.Connection] = None diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py new file mode 100644 index 0000000..5be1804 --- /dev/null +++ b/tests/portfolio/test_local_unified_provider.py @@ -0,0 +1,145 @@ +"""LocalUnifiedProvider 单元测试(spec §6 使用层)。 + +Mac 本地 TDD: sqlite tmp_path + tmp parquet fixture,零 VPS 依赖,零网络。 +覆盖: +- ``jq_to_dbbardata`` / ``dbbardata_to_jq`` / ``_jq_to_bs_code`` 代码转换 +- ``_build_qfq_factor`` 复权因子构造(asof) +- ``get_price`` dbbardata('d') raw + fq='qfq' 前复权 + panel=False 长表 +- ``get_index_stocks`` constituent_unified 并集(治偏差,无 date 时点) +- ``get_fundamentals_df`` pe/pb/ps/pcf←valuation_baostock + 市值←static akshare + 三表委托 LocalParquetProvider +- 辅助方法 get_trade_days / get_all_securities / get_security_info / get_current_tick / get_split_dividend +""" +from __future__ import annotations + +import sqlite3 +from typing import Any, Dict, List + +import pandas as pd +import pytest + +from sanguo_portfolio.providers.local_unified_provider import ( + jq_to_dbbardata, + dbbardata_to_jq, + _jq_to_bs_code, + _build_qfq_factor, + LocalUnifiedProvider, +) + + +# ======================== Task 0: 代码转换 ======================== +class TestCodeFormat: + def test_jq_to_dbbardata_sh(self): + assert jq_to_dbbardata("600519.XSHG") == ("600519", "SSE") + + def test_jq_to_dbbardata_sz(self): + assert jq_to_dbbardata("000001.XSHE") == ("000001", "SZSE") + + def test_jq_to_dbbardata_pure_digit_sh(self): + # 6 开头 → SSE + assert jq_to_dbbardata("600519") == ("600519", "SSE") + + def test_jq_to_dbbardata_pure_digit_sz(self): + # 0/3 开头 → SZSE + assert jq_to_dbbardata("000001") == ("000001", "SZSE") + assert jq_to_dbbardata("300001") == ("300001", "SZSE") + + def test_dbbardata_to_jq_sh(self): + assert dbbardata_to_jq("600519", "SSE") == "600519.XSHG" + + def test_dbbardata_to_jq_sz(self): + assert dbbardata_to_jq("000001", "SZSE") == "000001.XSHE" + + def test_round_trip_jq_to_dbbardata_to_jq(self): + original = "600519.XSHG" + sym, exc = jq_to_dbbardata(original) + assert dbbardata_to_jq(sym, exc) == original + + def test_jq_to_bs_code_sh(self): + # 600519.XSHG → 'sh.600519'(bs_adjust_factor.code 格式) + assert _jq_to_bs_code("600519.XSHG") == "sh.600519" + + def test_jq_to_bs_code_sz(self): + assert _jq_to_bs_code("000001.XSHE") == "sz.000001" + + +# ======================== Task 0: 复权因子 asof ======================== +class TestBuildQfqFactor: + def test_asof_before_all_events_uses_earliest(self, tmp_path): + # 2 除权事件, 最新=1.0 + db = tmp_path / "t.db" + c = sqlite3.connect(str(db)) + c.execute( + "CREATE TABLE bs_adjust_factor(code TEXT, dividOperateDate TEXT, " + "foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL)" + ) + c.executemany( + "INSERT INTO bs_adjust_factor VALUES(?,?,?,?,?)", + [ + ("sh.600519", "2024-06-19", 0.90, 0, 0), + ("sh.600519", "2025-06-19", 1.00, 0, 0), + ], + ) + c.commit() + c.close() + # 2023 早于所有事件 → 用最早 factor=0.90 + dates = pd.to_datetime(["2023-01-01"]) + f = _build_qfq_factor("sh.600519", sqlite3.connect(str(db)), dates) + assert abs(f.iloc[0] - 0.90) < 1e-6 + + def test_asof_between_events(self, tmp_path): + db = tmp_path / "t.db" + c = sqlite3.connect(str(db)) + c.execute( + "CREATE TABLE bs_adjust_factor(code TEXT, dividOperateDate TEXT, " + "foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL)" + ) + c.executemany( + "INSERT INTO bs_adjust_factor VALUES(?,?,?,?,?)", + [ + ("sh.600519", "2024-06-19", 0.90, 0, 0), + ("sh.600519", "2025-06-19", 1.00, 0, 0), + ], + ) + c.commit() + c.close() + # 2024-07 在两事件之间 → ≤ 的最大事件是 2024-06-19, factor=0.90 + dates = pd.to_datetime(["2024-07-01"]) + f = _build_qfq_factor("sh.600519", sqlite3.connect(str(db)), dates) + assert abs(f.iloc[0] - 0.90) < 1e-6 + + def test_asof_after_all_events_uses_latest(self, tmp_path): + db = tmp_path / "t.db" + c = sqlite3.connect(str(db)) + c.execute( + "CREATE TABLE bs_adjust_factor(code TEXT, dividOperateDate TEXT, " + "foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL)" + ) + c.executemany( + "INSERT INTO bs_adjust_factor VALUES(?,?,?,?,?)", + [ + ("sh.600519", "2024-06-19", 0.90, 0, 0), + ("sh.600519", "2025-06-19", 1.00, 0, 0), + ], + ) + c.commit() + c.close() + # 2025-07 晚于所有事件 → 最新 factor=1.00 + dates = pd.to_datetime(["2025-07-01"]) + f = _build_qfq_factor("sh.600519", sqlite3.connect(str(db)), dates) + assert abs(f.iloc[0] - 1.00) < 1e-6 + + def test_no_events_returns_ones(self, tmp_path): + db = tmp_path / "t.db" + c = sqlite3.connect(str(db)) + c.execute( + "CREATE TABLE bs_adjust_factor(code TEXT, dividOperateDate TEXT, " + "foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL)" + ) + c.commit() + c.close() + # 无事件 → 全 1.0 + dates = pd.to_datetime(["2024-01-01", "2024-06-01"]) + f = _build_qfq_factor("sh.600519", sqlite3.connect(str(db)), dates) + assert len(f) == 2 + assert abs(f.iloc[0] - 1.0) < 1e-6 + assert abs(f.iloc[1] - 1.0) < 1e-6