diff --git a/docs/portfolio_local_unified_provider.md b/docs/portfolio_local_unified_provider.md new file mode 100644 index 0000000..101c996 --- /dev/null +++ b/docs/portfolio_local_unified_provider.md @@ -0,0 +1,152 @@ +# LocalUnifiedProvider 使用说明 + +> spec §6 使用层 provider。读**方案A 权威数据层**,零 online,治幸存者偏差。**方案A 数据层落地后的推荐 provider**。 +> 实现见 `sanguo_portfolio/providers/local_unified_provider.py`,测试 `tests/portfolio/test_local_unified_provider.py`(36 用例)。 + +## 一句话定位 + +一个 provider,内部按数据类路由方案A 的权威表(dbbardata / constituent_unified / valuation_baostock / static akshare),**零 online**(不调 baostock HTTP,纯读本地 sqlite/parquet),**治幸存者偏差**(成份股并集含退市/被踢 + dbbardata 日线含退市),喂 `all_weather` 等策略。 + +## 快速使用 + +```python +from sanguo_portfolio.providers import LocalUnifiedProvider + +# VPS(默认路径 C:\sanguo_vnpy_v2\data) +p = LocalUnifiedProvider() + +# Mac 测试 / 自定义路径 +p = LocalUnifiedProvider({ + "db_path": "/path/to/quant_trading.db", + "data_dir": "/path/to/data", # 含 valuation_baostock/ + static/ +}) + +# 回测入口(runner) +# python -m sanguo_portfolio.runner_backtest --provider unified --start 2024-01-01 --end 2024-12-31 +``` + +## 数据源映射(每接口 → 方案A 权威表) + +| 方法 | 数据源 | 表 / 文件 | 归一化 | +|---|---|---|---| +| `get_price` | dbbardata('d') raw + bs_adjust_factor | `quant_trading.db` | jq_code↔symbol+exchange; `SSE→SH`; raw 默认, `fq='qfq'` 按 foreAdjustFactor 算 | +| `get_index_stocks` / `get_constituent` | constituent_unified 并集 | `quant_trading.db` | code(纯6位)→jq_code; 返回 `in_current=1 ∪ was_removed=1` | +| `get_fundamentals_df` | pe/pb/ps/pcf ← valuation_baostock; 市值+三表 ← static akshare | `.parquet` + `static/{valuation,balance,income}/` | 对齐 `_FUNDAMENTAL_COLUMNS`; 市值元→亿 | +| `get_trade_days` | dbbardata('d') 600519 distinct datetime | `quant_trading.db` | — | +| `get_all_securities` | dbbardata distinct symbol | `quant_trading.db` | — | +| `get_security_info` | dbbardata min/max datetime + constituent_unified code_name | `quant_trading.db` | — | +| `get_current_tick` | dbbardata 最近 close × 1.1/0.9 | `quant_trading.db` | ST/创业/科创精确规则 v2 | +| `get_split_dividend` | bs_adjust_factor 除权事件 | `quant_trading.db` | dividOperateDate + factor | + +## 接口清单 + +```python +# K 线(日线 raw 真实价,按需前复权) +get_price(security, start_date=None, end_date=None, frequency="daily", + fields=None, skip_paused=False, fq="raw", count=None, + panel=True, fill_paused=True) -> pd.DataFrame +# - frequency 非 daily/day/1d/d → 返空(1m 数据层无,day 频率回测降级) +# - panel=False → 长表含 time + code 列(供策略 pivot) +# - fq='qfq'/'pre' → 按 bs_adjust_factor 算前复权 +# - fields 缺失列(如 high_limit)补 NaN(策略涨停识别降级) + +# 成份股(spec §6 治偏差核心) +get_index_stocks(index_symbol, date=None) -> List[str] # date 忽略(并集模型) +get_constituent(index, date=None) -> List[str] # 语义别名 + +# 基本面(列对齐 _FUNDAMENTAL_COLUMNS,策略选股核心) +get_fundamentals_df(stocks, date=None) -> pd.DataFrame + +# 辅助 +get_trade_days(start_date=None, end_date=None, count=None) -> List[datetime] +get_all_securities(types=None) -> pd.DataFrame +get_security_info(security) -> Dict +get_current_tick(security) -> Optional[Dict] # 回测从 K 线推涨跌停 +get_split_dividend(security, start_date=None, end_date=None) -> List[Dict] +``` + +## 复权(方案A §14.7 最终目标) + +- **dbbardata 存 raw 真实价**(不复权)。`get_price` 默认 `fq='raw'` 返 raw。 +- **前复权消费端算**:`get_price(fq='qfq')` 按 `bs_adjust_factor.foreAdjustFactor` 算。 +- **asof 语义**:每个日期找 `≤ 该日` 的最大除权日的 `foreAdjustFactor`;早于所有除权日用最早因子;晚于所有用最新(=1.0)。 +- **公式**:`qfq[t] = raw[t] × factor[t]`(open/high/low/close 同乘,volume/turnover 不乘)。 +- 例:600519 最新除权 2026-06-26 factor=1.0;历史递减(2020-06-24=0.856)。 +- 策略 `_trend_mean` 算 N 日涨幅是比率,raw/qfq 等价(除权日 raw 跳水除外);要精确除权连续性用 `fq='qfq'`。 + +## 幸存者偏差治理(关键!) + +**`constituent_unified` 是"全时期并集"模型**(无 date 列): +- 9 指数分布:`000300`=940只(300当前+640被踢) / `000905`=1803(500+1303) / `000016`=195(50+145) / 深证 399001=702,399005=145,399006=175,399330=150 +- **治"纯当前幸存者"偏差**:含已退市/被踢股票(如 000005 退市、600811 被踢都在 300 并集) +- **轻微前视**:`get_index_stocks(date)` 的 `date` 参数**被忽略**(表无时点数据),回测 2020 年选股池 = 历史上所有曾在该指数的股票(含 2024 才纳入的)。比纯当前快照好,但不如 baostock `query_hs300_stocks(date)` 时点精确。 +- **永久 gap**:中证1000(`000852`)/2000(`932000`)只当前快照(1000/2000 全当前,0 被踢),历史成份股不可补(csindex SPA 封/akshare 只快照)。 +- **dbbardata 日线也治偏差**:含退市股 K 线(000005 退市到 2024-04-26,600811 等),回测能真实反映"当时买入现已退市"的标的。 + +## Mac 测试(零 VPS 依赖) + +`tests/portfolio/test_local_unified_provider.py` 用 `tmp_path` + `sqlite3` + tmp parquet fixture,完全不依赖 VPS 数据: + +```python +def test_get_index_stocks_union(tmp_path): + db = tmp_path / "t.db" + c = sqlite3.connect(str(db)) + c.execute("CREATE TABLE constituent_unified(...)") + # 造 in_current + was_removed 样本 + ... + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + assert set(p.get_index_stocks("000300.XSHG")) == {...} # 含被踢 +``` + +```bash +python3 -m pytest tests/portfolio/test_local_unified_provider.py -v # 36 passed +python3 -m pytest tests/portfolio/ -q # 全回归 149 passed +``` + +## 部署 / 运行 + +**VPS 数据依赖**(方案A 已落地,见 memory `data-fusion-design-finalized` / `vps-local-data-layout`): +- `C:\sanguo_vnpy_v2\data\quant_trading.db` — 含 dbbardata / constituent_unified / bs_adjust_factor +- `C:\sanguo_vnpy_v2\data\valuation_baostock\.parquet` — 1990-2026 全年份 +- `C:\sanguo_vnpy_v2\data\static\{valuation,balance,income,cashflow}\*.parquet` — akshare 三表+市值 +- 日增量:`sanguo-bs-eod`(18:05 baostock 日线+15min+pe/pb)+ `sanguo-xt-eod`(18:40 ETF/基金)已部署 + +**rsync 同步代码到 VPS**: +```bash +rsync -avz -e ssh --exclude='.git' --exclude='vnpy_v4.4.0' --exclude='__pycache__' \ + --exclude='.superpowers' --exclude='docs' --exclude='tests/data' \ + ./ 49.232.102.198:C:/sanguo_vnpy_v2/ +``` +⚠️ config 不在排除列表,会覆盖 VPS config(方案A §14.9 已知 TODO:部署前 `--exclude config` 或靠 SANGUO_DATA_ROOT)。 + +**回测**: +```bash +ssh 49.232.102.198 'cd C:\sanguo_vnpy_v2 && C:\Python310\python.exe -X utf8 -m sanguo_portfolio.runner_backtest --provider unified --start 2024-01-01 --end 2024-12-31 --cash 1000000 --max-pool 20' +``` + +## 已知限制(v1) + +| 限制 | 影响 | 对策 | +|---|---|---| +| `high_limit` 列 NaN | 策略 `prepare_stock_list` 昨日涨停识别降级(close==high_limit 不命中) | dbbardata 不存涨跌停;`get_current_tick` 另算;v2 可从 valuation pctChg 推 | +| 1m 频率返空 | `_intraday_high_low` 降级 | 数据层无 1m;day 频率回测不触发;15m 在 dbbardata('15m') 可扩展支持 | +| `gross_profit_margin`/`roic` NaN | fundamentals 两字段空 | 委托 LocalParquetProvider 读 `financial_abstract`,fixture 未造则 NaN(非新缺口) | +| 成份股轻微前视 | 回测早期选股池含未来纳入股 | 方案A 既定取舍(并集模型);要精确时点需 baostock online(违反铁律) | +| `get_current_tick` 涨跌停 ±10% 简化 | ST/创业板/科创板精确涨跌停未区分 | v2 从 valuation `isST` + 代码段识别 | + +## 与旧 provider 的关系 + +| provider | 数据源 | 用途 | 状态 | +|---|---|---|---| +| **`LocalUnifiedProvider`** | 方案A 权威层(dbbardata/constituent_unified/valuation_baostock) | **方案A 后推荐** | ✅ 新增 | +| `LocalParquetProvider`(`--provider local`) | 旧 parquet(qfq 日线/index_const 快照/akshare valuation) | MVP 验证遗留 | 保留(向后兼容,unittest 仍在) | +| `BaostockProvider`(`--provider baostock`) | baostock online HTTP | Mac 跨平台调试 | 保留(违反"读本地"铁律,非生产推荐) | +| `SanguoMiniQmtProvider`(`--provider miniqmt`) | miniQMT xtquant | VPS 实盘 | 保留(实盘 runner_live 用) | + +**迁移建议**:新回测/策略用 `--provider unified`。`local` 是方案A 前的 MVP 链路(读旧 parquet, index_const 仅当前快照有幸存者偏差),`unified` 读方案A 权威层治偏差。 + +## 设计文档 + +- spec:`docs/superpowers/specs/2026-07-21-data-source-fusion-design.md` §6(使用层)+ §14(方案A 数据层) +- plan:`docs/superpowers/plans/2026-07-23-local-unified-provider.md`(TDD 拆解) +- 关联 memory:`data-fusion-design-finalized` / `vps-local-data-layout` / `provider-local-data-only` / `db-primary-parquet-fallback` diff --git a/docs/superpowers/plans/2026-07-23-local-unified-provider.md b/docs/superpowers/plans/2026-07-23-local-unified-provider.md new file mode 100644 index 0000000..4446989 --- /dev/null +++ b/docs/superpowers/plans/2026-07-23-local-unified-provider.md @@ -0,0 +1,601 @@ +# LocalUnifiedProvider Implementation Plan (spec §6 使用层) + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 实现 spec §6 使用层 `LocalUnifiedProvider`——读方案A 权威数据层(dbbardata/constituent_unified/valuation_baostock),零 online,治幸存者偏差,喂 all_weather 策略。 + +**Architecture:** 新建 `LocalUnifiedProvider(bullet_trade.DataProvider)`,内部按数据类路由方案A 权威表:日线读 `dbbardata('d')` raw + `bs_adjust_factor` 算前复权;成份股读 `constituent_unified` 并集(治偏差);估值读 `valuation_baostock` parquet + 市值读 static/valuation akshare parquet。Mac 测试用 `sqlite :memory:` + tmp parquet fixture,零 VPS 依赖。 + +**Tech Stack:** Python 3.10, pandas 2.3, sqlite3, pyarrow, pytest + +## Global Constraints(spec + 用户铁律) + +- **零 online**: provider 不 import baostock 调 online,纯读本地 DB/parquet(memory provider-local-data-only)。baostock 48000/天限频不波及使用层。 +- **surgical**: 不改 `LocalParquetProvider`/`BaostockProvider`(旧链路保留,向后兼容)。 +- **dbbardata 不破坏**: `UNIQUE(symbol,exchange,interval,datetime)`,只读不写。 +- **复权**: dbbardata 存 raw,消费端按 `bs_adjust_factor.foreAdjustFactor` 算前复权(§14.7 最终目标,用户定不降级)。 +- **constituent_unified 并集模型**: 表无 date 列,`get_index_stocks(date)` 返回 in_current∪was_removed 并集,date 参数无法精确时点过滤——治"纯当前幸存者"偏差,有轻微前视(使用说明标注)。 +- **代码归一**: jq_code `600519.XSHG` ↔ dbbardata `symbol=600519, exchange=SSE`;`SSE→SH, SZSE→SZ`。 + +## 实测 schema(VPS 2026-07-23 probe,执行 agent 必读) + +DB = `C:\sanguo_vnpy_v2\data\quant_trading.db`(VPS) / Mac 测试用 fixture 路径。 + +**dbbardata('d')** — 唯一行情表,raw 真实价: +``` +列: symbol TEXT, exchange TEXT(SSE/SZSE), datetime TEXT(YYYY-MM-DD HH:MM:SS), + interval TEXT('d'), volume REAL, turnover REAL, open_interest REAL, + open_price REAL, high_price REAL, low_price REAL, close_price REAL +样本: 600519 10056行 2001-08-27~2026-07-22; 000005退市 8146行~2024-04-26; 510300 ETF 3439行 +``` + +**constituent_unified** — 成份股并集(无 date!): +``` +列: index_code TEXT(如 '000300'), code TEXT(纯6位如 '000001'), code_name TEXT, + source TEXT('baostock'/'akshare'), in_current INT(0/1), was_removed INT(0/1) +分布: 000300=940(300当前+640被踢) 000905=1803 000016=195 000852=1000(全当前,历史不可补) + 399001=702 399005=145 399006=175 399330=150 932000=2000(全当前) +``` + +**bs_adjust_factor** — 复权因子: +``` +列: code TEXT('sh.600519'), dividOperateDate TEXT(YYYY-MM-DD), + foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL +语义: foreAdjustFactor 按除权日分段,最新事件=1.0,递减往历史。qfq[t]=raw[t]*factor[date[t]]。 + 600519 有 12 事件: 2020-06-24=0.856267 ... 2026-06-26=1.0 +``` + +**valuation_baostock/.parquet** — baostock 估值(1990-2026 全年份): +``` +列: symbol(6位), exchange(SH/SZ), date(YYYY-MM-DD), peTTM, psTTM, pcfNcfTTM, pbMRQ, turn, pctChg, isST +注: 无 market_cap/total_share 列! 市值从 static/valuation akshare 补。 +``` + +**static/valuation/_valuation.parquet** — akshare 估值(市值/股本来源,5530 文件): +``` +中文列(见 LocalParquetProvider._VAL_COL_MAP): 总市值→total_market_cap, 流通市值→circ_market_cap, + 总股本→total_share, PE(TTM)→pe_ttm, 市净率→pb ... +``` + +**static/{balance,income,cashflow}/_.parquet** — akshare 三表(balance 221列/income 170列): +``` +通用列: SECUCODE, REPORT_DATE, REPORT_TYPE; balance 有 TOTAL_ASSETS/TOTAL_LIABILITIES/TOTAL_PARENT_EQUITY; + income 有 BASIC_EPS/OPERATE_INCOME/PARENT_NETPROFIT/OPERATE_INCOME_YOY +``` + +--- + +## File Structure + +- **Create:** `sanguo_portfolio/providers/local_unified_provider.py` — LocalUnifiedProvider 类(~400行) +- **Modify:** `sanguo_portfolio/providers/__init__.py` — 导出 LocalUnifiedProvider +- **Modify:** `sanguo_portfolio/runner_backtest.py` — `build_provider` 加 `unified` 选项(choices + 分支) +- **Create:** `tests/portfolio/test_local_unified_provider.py` — DataProvider 契约单测(fixture: sqlite + tmp parquet) +- **Create:** `tests/portfolio/conftest.py` 追加 — `local_unified_provider` fixture(若需要,否则在测试文件内建) +- **Create:** `docs/portfolio_local_unified_provider.md` — 使用说明(架构/数据源/接口/复权/治偏差/Mac测试/部署) + +--- + +## Task 0: 代码转换 + DB 连接辅助 + 复权因子构造 + +**Files:** +- Create: `sanguo_portfolio/providers/local_unified_provider.py`(本 task 建文件骨架 + 模块级辅助函数) +- Test: `tests/portfolio/test_local_unified_provider.py` + +**Interfaces:** +- Produces: `jq_to_dbbardata(jq_code) -> (symbol, exchange)` / `dbbardata_to_jq(symbol, exchange) -> jq_code`; `_connect(cfg) -> sqlite3.Connection`; `_build_qfq_factor(code, conn, dates) -> pd.Series(factor indexed by date)` + +- [ ] **Step 1: 写失败测试 — 代码转换** + +```python +# tests/portfolio/test_local_unified_provider.py +from sanguo_portfolio.providers.local_unified_provider import ( + jq_to_dbbardata, dbbardata_to_jq, LocalUnifiedProvider, +) + +def test_jq_to_dbbardata_roundtrip(): + assert jq_to_dbbardata("600519.XSHG") == ("600519", "SSE") + assert jq_to_dbbardata("000001.XSHE") == ("000001", "SZSE") + assert jq_to_dbbardata("600519") == ("600519", "SSE") # 纯6位推断 + assert dbbardata_to_jq("600519", "SSE") == "600519.XSHG" + assert dbbardata_to_jq("000001", "SZSE") == "000001.XSHE" +``` + +- [ ] **Step 2: 跑测试确认 FAIL** — `pytest tests/portfolio/test_local_unified_provider.py::test_jq_to_dbbardata_roundtrip -v`(ImportError) + +- [ ] **Step 3: 实现模块骨架 + 代码转换** + +```python +# sanguo_portfolio/providers/local_unified_provider.py +"""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 valuation 无市值列) +- 三表: static/{balance,income,cashflow} akshare parquet + +零 online: 不 import baostock 调 online。Mac 测试用 sqlite+parquet fixture。 +""" +from __future__ import annotations +import logging, os, sqlite3 +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Union +import pandas as pd + +try: + from bullet_trade.data.providers.base import DataProvider # type: ignore +except ImportError: + class DataProvider: # type: ignore[no-redef] + name: str = "base" + +logger = logging.getLogger(__name__) +_DEFAULT_DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +_DEFAULT_DATA_DIR = r"C:\sanguo_vnpy_v2\data" + +_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}" + + +# 复权因子代码转换: 600519.XSHG → 'sh.600519'(bs_adjust_factor.code 格式) +def _jq_to_bs_code(jq_code: str) -> str: + sym, exc = jq_to_dbbardata(jq_code) + prefix = "sh" if exc == "SSE" else "sz" + return f"{prefix}.{sym}" +``` + +- [ ] **Step 4: 跑测试确认 PASS** + +- [ ] **Step 5: 写失败测试 — 复权因子构造** + +```python +def test_build_qfq_factor(tmp_path): + # fixture: 2 除权事件, 最新=1.0 + import sqlite3 + 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() + from sanguo_portfolio.providers.local_unified_provider import _build_qfq_factor + dates = pd.to_datetime(["2023-01-01", "2024-07-01", "2025-07-01"]) + f = _build_qfq_factor("sh.600519", sqlite3.connect(str(db)), dates) + # 2023(早于最早事件)=0.90; 2024-07(between)=0.90; 2025-07(最新后)=1.00 + assert abs(f.iloc[0] - 0.90) < 1e-6 + assert abs(f.iloc[1] - 0.90) < 1e-6 + assert abs(f.iloc[2] - 1.00) < 1e-6 +``` + +- [ ] **Step 6: 实现 `_build_qfq_factor`** — asof join 逻辑(每个 date 找 ≤ 的最大 dividOperateDate 的 foreAdjustFactor;早于所有事件用最早;晚于所有用最新): + +```python +def _build_qfq_factor(bs_code: str, conn: sqlite3.Connection, + dates: pd.Series) -> pd.Series: + """构造每个 date 的前复权因子(asof)。qfq[t]=raw[t]*factor[t]。""" + rows = conn.execute( + "SELECT dividOperateDate, foreAdjustFactor FROM bs_adjust_factor " + "WHERE code=? ORDER BY dividOperateDate", (bs_code,)).fetchall() + if not rows: + return pd.Series([1.0] * len(dates), index=dates) + ev_dates = pd.to_datetime([r[0] for r in rows]) + factors = [float(r[1]) for r in rows] + out = [] + for d in pd.to_datetime(dates): + # 找 <= d 的最大事件; 全部 > d 用最早(第一个); 全部 <= d 用最后一个 + mask = ev_dates <= d + out.append(factors[mask.argmax()] if mask.any() else factors[0]) + # mask.argmax() 给第一个 True 的索引;但我们要"<= d 的最大事件"= 最后一个 True + # 修正:取最后一个 True + out = [] + for d in pd.to_datetime(dates): + mask = ev_dates <= d + idx = int(np.where(mask)[0][-1]) if mask.any() else 0 + out.append(factors[idx]) + return pd.Series(out, index=pd.to_datetime(dates)) +``` +(注意:`np` 需 `import numpy as np`。实现时简化为单次循环取最后一个 True 索引。) + +- [ ] **Step 7: 跑测试确认 PASS** +- [ ] **Step 8: Commit** — `feat(portfolio): LocalUnifiedProvider 代码转换+复权因子(Task0)` + +--- + +## Task 1: get_price(dbbardata raw + 前复权 + panel 长表) + +**Files:** Modify `local_unified_provider.py` 加 `__init__` + `get_price`; Test 同文件。 + +**Interfaces:** +- Consumes: Task0 辅助函数 + `_connect` +- Produces: `LocalUnifiedProvider.get_price(security, start_date, end_date, frequency, fields, skip_paused, fq, count, panel, fill_paused) -> DataFrame` + +策略契约(all_weather 实证): +- `get_price(hold_list, end_date, freq=daily, fields=[close,high_limit], count=1, panel=False)` — panel=False 长表需 time/code 列 +- `get_price(stocks, freq=1d, fields=[close], count=n, panel=False)` — _trend_mean pivot(index=time,columns=code) +- `get_price(stock, freq=1m, fq="pre", count=1, panel=False)` — intraday(day 频率回测降级,1m 无数据返空) + +- [ ] **Step 1: 写失败测试 — get_price daily 单股 + 复权** + +```python +@pytest.fixture +def unified_provider(tmp_path): + """造小样本 sqlite + parquet fixture。""" + db = tmp_path / "quant_trading.db" + c = sqlite3.connect(str(db)) + c.execute("CREATE TABLE dbbardata(symbol,exchange,datetime,interval,volume,turnover,open_interest,open_price,high_price,low_price,close_price)") + rows = [ + ("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), # 除权日 raw 跳水 + ("600519","SSE","2024-06-20 00:00:00","d",1000,1e6,0,910.0,920.0,900.0,910.0), + ] + c.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", rows) + c.execute("CREATE TABLE bs_adjust_factor(code,dividOperateDate,foreAdjustFactor,backAdjustFactor,adjustFactor)") + c.execute("INSERT INTO bs_adjust_factor VALUES('sh.600519','2024-06-19',0.9,0,0)") # 除权日 factor + c.commit(); c.close() + return LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + +def test_get_price_raw_vs_qfq(unified_provider): + p = unified_provider + # raw: 除权日 900 跳水 + df_raw = p.get_price("600519.XSHG", start_date="2024-06-18", end_date="2024-06-20", fq="raw") + assert len(df_raw) == 3 + assert abs(df_raw.loc["2024-06-19", "close"] - 900.0) < 1e-6 + # qfq: 06-18 = 1000*0.9 = 900; 06-19/20 = raw(factor=0.9 当 06-19 之后? 用最新段逻辑) + df_qfq = p.get_price("600519.XSHG", start_date="2024-06-18", end_date="2024-06-20", fq="qfq") + assert abs(df_qfq.loc["2024-06-18", "close"] - 900.0) < 1e-6 # 1000*0.9(早于事件用最早factor) +``` +(复权断言:06-18 早于除权日 06-19 → 用 factor 0.9 → 1000*0.9=900;06-19/20 ≥ 事件日 → factor 取 06-19 的 0.9 → 900*0.9=810, 910*0.9=819。实现时按 `_build_qfq_factor` 语义校准断言。) + +- [ ] **Step 2: 跑测试确认 FAIL** + +- [ ] **Step 3: 实现 `__init__` + `get_price`** + +```python +class LocalUnifiedProvider(DataProvider): # type: ignore[misc] + 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 + self._val_bs_cache: Dict[int, pd.DataFrame] = {} # year -> valuation_baostock + + def _connect(self) -> sqlite3.Connection: + if self._conn is None: + self._conn = sqlite3.connect(self.db_path, timeout=30) + self._conn.execute("PRAGMA busy_timeout = 30000") + return self._conn + + def get_price(self, security, start_date=None, end_date=None, frequency="daily", + fields=None, skip_paused=False, fq="raw", count=None, + panel=True, fill_paused=True, **kwargs): + freq = str(frequency or "").lower() + if freq not in ("daily", "day", "1d", "d"): + return pd.DataFrame() # 1m/分钟 day 频率回测降级(数据层无 1m) + secs = [security] if isinstance(security, str) else list(security or []) + conn = self._connect() + start_str = self._to_date_str(start_date) + end_str = self._to_date_str(end_date) or datetime.now().strftime("%Y-%m-%d") + + frames: Dict[str, pd.DataFrame] = {} + for jq_code in secs: + sym, exc = jq_to_dbbardata(jq_code) + q = "SELECT datetime, open_price, high_price, low_price, close_price, " \ + "volume, turnover FROM dbbardata WHERE symbol=? AND exchange=? " \ + "AND interval='d' AND datetime>=? AND datetime<=? ORDER BY datetime" + df = pd.read_sql(q, conn, params=(sym, exc, start_str + " 00:00:00", end_str + " 23:59:59")) + if df.empty: + frames[jq_code] = df; continue + df["datetime"] = pd.to_datetime(df["datetime"]) + df = df.set_index("datetime") + df.index.name = None + if count: + df = df.tail(count) + # 复权 + if fq in ("qfq", "pre", "前复权"): + factor = _build_qfq_factor(_jq_to_bs_code(jq_code), conn, df.index) + for col in ("open_price", "high_price", "low_price", "close_price"): + df[col] = df[col].values * factor.values + # 策略要 close/high_limit 字段名(jq 风格) + df = df.rename(columns={"open_price": "open", "high_price": "high", + "low_price": "low", "close_price": "close"}) + # high_limit 不在 dbbardata, 留给 get_current_tick 语义;这里策略 prepare_stock_list 要 high_limit 列 + # → 缺失列返 NaN(策略 hit = close==high_limit 不会命中,降级可接受) + if fields: + for f in fields: + if f not in df.columns: + df[f] = float("nan") + df = df[fields] + frames[jq_code] = df + + if not frames or all(f.empty for f in frames.values()): + return pd.DataFrame() + if not panel: + parts = [] + for jq_code, df in frames.items(): + if df.empty: + continue + d = df.reset_index().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) +``` + +- [ ] **Step 4: 跑测试确认 PASS** +- [ ] **Step 5: 写失败测试 — panel=False 多股长表 + count** + +```python +def test_get_price_panel_false_multi(unified_provider): + df = unified_provider.get_price("600519.XSHG", end_date="2024-06-20", count=2, panel=False, fields=["close"]) + assert "code" in df.columns and "time" in df.columns + assert len(df) == 2 +``` + +- [ ] **Step 6: 实现(Step 3 已含 panel 分支),跑 PASS** +- [ ] **Step 7: Commit** — `feat(portfolio): LocalUnifiedProvider get_price+前复权(Task1)` + +--- + +## Task 2: get_index_stocks + get_constituent(constituent_unified 并集,治偏差) + +**Files:** Modify `local_unified_provider.py`; Test 同文件。 + +**Interfaces:** +- Produces: `get_index_stocks(index_symbol, date) -> List[str]` + `get_constituent(index, date) -> List[str]`(语义别名) + +- [ ] **Step 1: 写失败测试** + +```python +def test_get_index_stocks_union(tmp_path): + db = tmp_path / "t.db"; c = sqlite3.connect(str(db)) + c.execute("CREATE TABLE constituent_unified(index_code TEXT,code TEXT,code_name TEXT,source TEXT,in_current INT,was_removed INT)") + c.executemany("INSERT INTO constituent_unified VALUES(?,?,?,?,?,?)", [ + ("000300", "600519", "贵州茅台", "baostock", 1, 0), + ("000300", "000001", "平安银行", "baostock", 1, 0), + ("000300", "600811", "退市股", "baostock", 0, 1), # 被踢 + ]) + c.commit(); c.close() + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + stocks = p.get_index_stocks("000300.XSHG", "2020-01-01") + assert set(stocks) == {"600519.XSHG", "000001.XSHE", "600811.SH"} # 并集含被踢 + # date 参数不报错(并集模型忽略) + assert p.get_constituent("000300", None) == stocks # 别名 +``` + +- [ ] **Step 2: 跑测试确认 FAIL** + +- [ ] **Step 3: 实现** — 查 constituent_unified,index_code 匹配(去 `.XXXX` 后缀),返回 in_current=1 OR was_removed=1 的并集,code→jq_code: + +```python + def get_index_stocks(self, index_symbol, date=None) -> List[str]: + idx = index_symbol.split(".")[0] if "." in str(index_symbol) else str(index_symbol) + conn = self._connect() + rows = conn.execute( + "SELECT code FROM constituent_unified WHERE index_code=? " + "AND (in_current=1 OR was_removed=1)", (idx,)).fetchall() + out = [] + for (code,) in rows: + code = str(code).strip() + if len(code) != 6: + continue + exc = "SSE" if code.startswith("6") else "SZSE" + out.append(dbbardata_to_jq(code, exc)) + return out + + def get_constituent(self, index, date=None) -> List[str]: + """spec §6 语义别名 = get_index_stocks。""" + return self.get_index_stocks(index, date) +``` + +- [ ] **Step 4: 跑测试 PASS** +- [ ] **Step 5: Commit** — `feat(portfolio): LocalUnifiedProvider 成份股并集治偏差(Task2)` + +--- + +## Task 3: get_fundamentals_df(valuation_baostock + static akshare + 三表) + +**Files:** Modify `local_unified_provider.py`; Test 同文件 + tmp parquet fixture。 + +**Interfaces:** +- Produces: `get_fundamentals_df(stocks, date) -> DataFrame` 列对齐 `_FUNDAMENTAL_COLUMNS` + +数据源映射: +- `pe_ratio/pb_ratio/ps_ratio/pcf_ratio` ← valuation_baostock parquet(peTTM/pbMRQ/psTTM/pcfNcfTTM,baostock 权威) +- `market_cap/circulating_market_cap` ← static/valuation akshare parquet(total_market_cap/circ_market_cap,baston 无市值) +- 三表字段(eps/net_profit_margin/total_liability 等) ← static/{balance,income} akshare parquet(复用 LocalParquetProvider 读法) + +- [ ] **Step 1: 写失败测试 — 估值字段从 valuation_baostock** + +```python +def test_get_fundamentals_valuation(tmp_path): + # valuation_baostock/2024.parquet + vdir = tmp_path / "valuation_baostock"; vdir.mkdir() + pd.DataFrame({"symbol":["600519"],"exchange":["SH"],"date":["2024-09-30"], + "peTTM":[25.0],"psTTM":[15.0],"pcfNcfTTM":[20.0],"pbMRQ":[7.5], + "turn":[0.1],"pctChg":[1.0],"isST":[0]}).to_parquet(vdir/"2024.parquet") + # static/valuation akshare(市值) + sdir = tmp_path / "static" / "valuation"; sdir.mkdir(parents=True) + pd.DataFrame({"数据日期":["2024-09-30"],"总市值":[2e12],"流通市值":[2e12],"总股本":[1.256e9], + "PE(TTM)":[25],"市净率":[7.5]}).to_parquet(sdir/"600519.SH_valuation.parquet") + p = LocalUnifiedProvider({"db_path": str(tmp_path/"t.db"), "data_dir": str(tmp_path)}) + df = p.get_fundamentals_df(["600519.XSHG"], date="2024-09-30") + assert abs(df.loc["600519.XSHG","pe_ratio"] - 25.0) < 1e-6 # baostock 权威 + assert abs(df.loc["600519.XSHG","pb_ratio"] - 7.5) < 1e-6 + assert abs(df.loc["600519.XSHG","market_cap"] - 2e4) < 1 # 2e12元→2e4亿 +``` + +- [ ] **Step 2: 跑测试确认 FAIL** + +- [ ] **Step 3: 实现** — 读 valuation_baostock parquet(year from date)+ static/valuation akshare;合并对齐 `_FUNDAMENTAL_COLUMNS`(复用 LocalParquetProvider 的 `_VAL_COL_MAP` / `to_yi` / 三表读法,import 复用): + +```python + from .local_parquet_provider import (_VAL_COL_MAP, jq_to_file_code, + _to_float, _or_nan, _pct_to_decimal, _FUNDAMENTAL_COLUMNS) + from ..factors.valuation import to_yi + + def get_fundamentals_df(self, stocks, date=None) -> pd.DataFrame: + if not stocks: + return pd.DataFrame(columns=_FUNDAMENTAL_COLUMNS) + date_str = self._to_date_str(date) or datetime.now().strftime("%Y-%m-%d") + rows = [self._build_fundamental_row(s, date_str) for s in stocks] + df = pd.DataFrame(rows, columns=_FUNDAMENTAL_COLUMNS) + if "code" in df.columns: + df = df.set_index("code", drop=False) + return df + + def _read_valuation_baostock(self, year: int) -> pd.DataFrame: + if year in self._val_bs_cache: + return self._val_bs_cache[year] + p = os.path.join(self.data_dir, "valuation_baostock", f"{year}.parquet") + df = pd.read_parquet(p) if os.path.exists(p) else pd.DataFrame() + self._val_bs_cache[year] = df + return df + + def _build_fundamental_row(self, jq_code, date_str) -> Dict[str, Any]: + sym, exc = jq_to_dbbardata(jq_code) + fc = jq_to_file_code(jq_code) # 600519.SH(static akshare 文件名) + row: Dict[str, Any] = {"code": jq_code} + # 1. pe/pb/ps/pcf ← valuation_baostock(baostock 权威) + year = int(date_str[:4]) + vbs = self._read_valuation_baostock(year) + if not vbs.empty: + sub = vbs[(vbs["symbol"].astype(str) == sym) & (vbs["date"].astype(str) <= date_str)] + vrow = sub.iloc[-1] if not sub.empty else None + else: + vrow = None + def gbs(k): + return _to_float(vrow.get(k)) if vrow is not None else None + row["pe_ratio"] = _or_nan(gbs("peTTM")) + row["pb_ratio"] = _or_nan(gbs("pbMRQ")) + row["ps_ratio"] = _or_nan(gbs("psTTM")) + row["pcf_ratio"] = _or_nan(gbs("pcfNcfTTM")) + # 2. 市值/股本 + 三表 ← static akshare(复用 LocalParquetProvider 读法) + # 复用:直接实例化 LocalParquetProvider 读 static 部分,或内联读 static/valuation + ak_val = self._read_akshare_valuation(fc, date_str) # 返 renamed Series + mkt = _to_float(ak_val.get("total_market_cap")) if ak_val is not None else None + circ = _to_float(ak_val.get("circ_market_cap")) if ak_val is not None else None + row["market_cap"] = to_yi(mkt) if mkt else float("nan") + row["circulating_market_cap"] = to_yi(circ) if circ else float("nan") + # 3. 三表(income/balance)— 复用 LocalParquetProvider._read_quarter + 字段提取 + # 简化:委托一个内部 LocalParquetProvider 实例读三表部分(eps/margin/liability) + lpp = self._get_lpp_helper() + inc = lpp._latest_row_before(lpp._read_quarter("income", fc), "REPORT_DATE", date_str) + bal = lpp._latest_row_before(lpp._read_quarter("balance", fc), "REPORT_DATE", date_str) + row["eps"] = _or_nan(_to_float(inc.get("BASIC_EPS")) if inc is not None else None) + # ... net_profit_margin/total_liability/roe 等(照 LocalParquetProvider._build_fundamental_row 逻辑) + return row +``` +(实现时:`_get_lpp_helper()` 返一个复用的 `LocalParquetProvider(config)` 实例读 static 三表;`_read_akshare_valuation` 复用 LocalParquetProvider._read_valuation。DRY:不重写三表/akshare valuation 逻辑,委托 LocalParquetProvider。pe/pb 改 baostock 源覆盖 akshare 的。) + +- [ ] **Step 4: 跑测试 PASS** +- [ ] **Step 5: 写测试 — 三表字段(eps/market_cap 全 _FUNDAMENTAL_COLUMNS 有值不 NaN)** +- [ ] **Step 6: 实现 + PASS** +- [ ] **Step 7: Commit** — `feat(portfolio): LocalUnifiedProvider fundamentals baostock估值+akshare市值(Task3)` + +--- + +## Task 4: 辅助方法(trade_days/all_securities/security_info/current_tick/split_dividend) + +**Files:** Modify `local_unified_provider.py`; Test 同文件。 + +- [ ] **Step 1-2: 写失败测试 + FAIL** — `get_trade_days(count=2)` 返 datetime list;`get_security_info` 返 display_name/start_date;`get_current_tick` 返 close+high_limit;`get_split_dividend` 返 bs_adjust_factor 事件;`get_all_securities` 返 dbbardata distinct symbol。 + +- [ ] **Step 3: 实现**: + - `get_trade_days`: 读 dbbardata 某 symbol(如 600519)distinct datetime,filter/count。 + - `get_security_info`: dbbardata min/max datetime → start/end_date;display_name 从 constituent_unified code_name 或 code。 + - `get_current_tick`: dbbardata 最近 close + valuation_baostock 最近 pctChg → high_limit=close×1.1(ST 0.05)。 + - `get_split_dividend`: bs_adjust_factor → events(dividOperateDate + adjustFactor)。 + - `get_all_securities`: dbbardata distinct symbol → DataFrame。 + +- [ ] **Step 4: 跑测试 PASS** +- [ ] **Step 5: Commit** — `feat(portfolio): LocalUnifiedProvider 辅助方法(Task4)` + +--- + +## Task 5: 接线(__init__ 导出 + runner build_provider 加 unified) + +**Files:** Modify `sanguo_portfolio/providers/__init__.py`; Modify `sanguo_portfolio/runner_backtest.py`。 + +- [ ] **Step 1: __init__.py 加导出** +```python +from .local_unified_provider import LocalUnifiedProvider +__all__ = ["SanguoMiniQmtProvider", "BaostockProvider", "LocalParquetProvider", "LocalUnifiedProvider"] +``` + +- [ ] **Step 2: runner_backtest build_provider 加 unified** +```python +# parse_args choices 加 "unified"; build_provider 加分支 +p.add_argument("--provider", default="local", choices=["local", "baostock", "miniqmt", "unified"], ...) +# build_provider: +from .providers import LocalUnifiedProvider +if name == "unified": + return LocalUnifiedProvider(cfg) +``` + +- [ ] **Step 3: 跑 `pytest tests/portfolio/ -v` 全绿(回归)** +- [ ] **Step 4: Commit** — `feat(portfolio): 接线 LocalUnifiedProvider 到 runner(Task5)` + +--- + +## Task 6: 使用说明 + VPS E2E 验证 + +**Files:** Create `docs/portfolio_local_unified_provider.md`; VPS 跑 `python -m sanguo_portfolio.runner_backtest --provider unified --start 2024-01-01 --end 2024-03-31 --max-pool 20`。 + +- [ ] **Step 1: 写使用说明** `docs/portfolio_local_unified_provider.md`(其他 session 直用)— 含: + - 一句话定位(读方案A权威层/零online/治偏差) + - 数据源映射表(每接口→哪张表/parquet) + - 接口清单(DataProvider 接口 + get_constituent) + - 复权说明(raw存储+消费端按bs_adjust_factor算qfq;fq参数 raw/qfq) + - **幸存者偏差说明**(constituent_unified 并集模型,治纯当前偏差,有轻微前视,date 参数忽略;中证1000/2000只快照永久gap) + - Mac 测试(fixture,零VPS依赖) + - 部署/运行(runner --provider unified;VPS 数据依赖 dbbardata/constituent_unified/valuation_baostock/static) + - 已知限制(high_limit 列 NaN→prepare_stock_list 涨停识别降级;1m 无数据;三表委托 LocalParquetProvider) + - 与旧 provider 关系(LocalParquetProvider/BaostockProvider 保留,unified 是方案A 后推荐) + +- [ ] **Step 2: VPS E2E** — rsync 代码到 VPS,跑 `--provider unified --max-pool 20` 小样本回测,确认: + - get_price 读 dbbardata 出 K 线(含退市) + - get_index_stocks 出并集成份股 + - get_fundamentals_df 出市值+pe/pb + - 回测不崩,有选股+指标输出 + +- [ ] **Step 3: Commit** — `docs(portfolio): LocalUnifiedProvider 使用说明+VPS E2E(Task6)` + +--- + +## Self-Review(plan 自检) + +1. **Spec 覆盖**: spec §6 接口(get_daily/get_constituent/get_fundamentals/...)— get_constituent 别名✓;get_price 覆盖 get_daily+get_etf_daily(都读 dbbardata,ETF 也在);get_fundamentals_df ✓;其余 §6 方法(industry/longhubang/instrument)数据层未就绪(P1),使用说明标注 NotImplementedError。✓ +2. **方案A §14 一致**: dbbardata 唯一行情✓;constituent_unified 治偏差✓;valuation_baostock pe/pb✓;raw+factor 复权✓;零online✓。 +3. **类型一致**: `_build_qfq_factor(code, conn, dates) -> Series` 在 Task0/Task1 调用签名一致✓。 +4. **占位扫描**: Task3 的 `_get_lpp_helper/_read_akshare_valuation` 标了"复用 LocalParquetProvider",实现 agent 须内联或委托,不留空✓。 +5. **风险**: get_price 的 high_limit 列缺失(NaN)→策略 prepare_stock_list 涨停识别降级,使用说明标注(Task6)✓。 + +## Execution Handoff + +Plan complete and saved to `docs/superpowers/plans/2026-07-23-local-unified-provider.md`. diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index 4dfd392..86be973 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/sanguo_portfolio/providers/local_unified_provider.py @@ -174,19 +174,21 @@ class LocalUnifiedProvider(DataProvider): # type: ignore[misc] frames: Dict[str, pd.DataFrame] = {} for jq_code in secs: sym, exc = jq_to_dbbardata(jq_code) + # substr(datetime,1,10) 取日期部分比 — datetime 列混合格式(有只日期有带时间), + # 纯字符串比 "2024-09-25" < "2024-09-25 00:00:00" 会漏边界行; 比日期(YYYY-MM-DD)规避 q = ( "SELECT datetime, open_price, high_price, low_price, close_price, " "volume, turnover FROM dbbardata WHERE symbol=? AND exchange=? " - "AND interval='d' AND datetime>=? AND datetime<=? ORDER BY datetime" - ) - df = pd.read_sql( - q, conn, - params=(sym, exc, start_str + " 00:00:00", end_str + " 23:59:59"), + "AND interval='d' AND substr(datetime,1,10)>=? AND substr(datetime,1,10)<=? " + "ORDER BY datetime" ) + df = pd.read_sql(q, conn, params=(sym, exc, start_str, end_str)) if df.empty: frames[jq_code] = df continue - df["datetime"] = pd.to_datetime(df["datetime"]) + # dbbardata datetime 混合格式(有的 "2024-09-26" 有的 "2024-09-26 00:00:00", + # 不同 schtask/迁移写入);pandas 2.3 严格模式要 format="mixed" + df["datetime"] = pd.to_datetime(df["datetime"], format="mixed") df = df.set_index("datetime") df.index.name = None if count: diff --git a/scripts/data_platform/probe_unified_schema.py b/scripts/data_platform/probe_unified_schema.py new file mode 100644 index 0000000..965959b --- /dev/null +++ b/scripts/data_platform/probe_unified_schema.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""probe_unified_schema.py — 探查方案A 新权威数据层 schema (LocalUnifiedProvider 实现依赖)。 + +输出 constituent_unified / dbbardata('d') / bs_adjust_factor / valuation_baostock 的 +列结构 + 样本 + 覆盖, 供 spec §6 使用层 provider 实现参考。 +""" +import sqlite3 +from pathlib import Path + +import pandas as pd + +BASE = Path(r"C:\sanguo_vnpy_v2") +DB = BASE / "data" / "quant_trading.db" +VAL_DIR = BASE / "data" / "valuation_baostock" + + +def section(title): + print(f"\n===== {title} =====") + + +# 1. constituent_unified +section("constituent_unified schema") +c = sqlite3.connect(str(DB)) +try: + info = c.execute("PRAGMA table_info(constituent_unified)").fetchall() + print("columns:", [(r[1], r[2]) for r in info]) + print("count/index:", + c.execute("SELECT COUNT(*), COUNT(DISTINCT index_code) FROM constituent_unified").fetchone()) + print("per-index (total / in_current / was_removed):") + for row in c.execute("SELECT index_code, COUNT(*), SUM(in_current), SUM(was_removed) " + "FROM constituent_unified GROUP BY index_code ORDER BY index_code"): + print(" ", row) + print("sample 300:") + for row in c.execute("SELECT * FROM constituent_unified WHERE index_code LIKE '%300%' LIMIT 3"): + print(" ", row) +finally: + c.close() + +# 2. dbbardata('d') raw 个股 + ETF +section("dbbardata('d') sample") +c = sqlite3.connect(str(DB)) +try: + print("columns:", [r[1] for r in c.execute("PRAGMA table_info(dbbardata)").fetchall()]) + for sym, lbl in [("600519", "个股在市"), ("000005", "退市"), ("510300", "ETF")]: + r = c.execute("SELECT COUNT(*), MIN(datetime), MAX(datetime) FROM dbbardata " + "WHERE symbol=? AND interval='d'", (sym,)).fetchone() + print(f" {sym}({lbl}): {r}") + print("sample 600519 last 3:") + for row in c.execute("SELECT symbol,exchange,datetime,volume,turnover,open_price,close_price " + "FROM dbbardata WHERE symbol='600519' AND interval='d' " + "ORDER BY datetime DESC LIMIT 3"): + print(" ", row) +finally: + c.close() + +# 3. bs_adjust_factor +section("bs_adjust_factor schema") +c = sqlite3.connect(str(DB)) +try: + tabs = [r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table' " + "AND name LIKE '%adjust%'").fetchall()] + print("adjust tables:", tabs) + if "bs_adjust_factor" in tabs: + print("columns:", [r[1] for r in c.execute("PRAGMA table_info(bs_adjust_factor)").fetchall()]) + print("count:", c.execute("SELECT COUNT(*) FROM bs_adjust_factor").fetchone()) + print("sample 600519:") + for row in c.execute("SELECT * FROM bs_adjust_factor WHERE code LIKE '%600519%' LIMIT 3"): + print(" ", row) +finally: + c.close() + +# 4. valuation_baostock parquet +section("valuation_baostock parquet") +print("years:", sorted(p.name for p in VAL_DIR.glob("*.parquet")) if VAL_DIR.exists() else "DIR MISSING") +p2026 = VAL_DIR / "2026.parquet" +if p2026.exists(): + df = pd.read_parquet(p2026) + print("columns:", list(df.columns)) + print("shape:", df.shape) + print("sample 600519:") + sub = df[df["symbol"].astype(str).str.contains("600519")] if "symbol" in df.columns else df.head(0) + print(sub.head(3).to_dict("records")) + +# 5. static akshare 三表 + valuation (market_cap/total_share 来源 + 文件格式) +from collections import Counter +section("static akshare (格式 + market_cap/total_share 来源)") +for sub in ["valuation", "balance", "income"]: + d = BASE / "data" / "static" / sub + if not d.exists(): + print(f"{sub}: DIR MISSING") + continue + allf = list(d.iterdir()) + exts = Counter(p.suffix for p in allf) + print(f"{sub}: {len(allf)} files, ext分布={dict(exts)}") + matches = sorted([p for p in allf if "600519" in p.name]) + if not matches: + print(" 600519 无文件") + continue + f0 = matches[0] + with open(f0, "rb") as fh: + head = fh.read(8) + print(f" 600519 file={f0.name} magic={head[:4]!r}") + if head[:4] == b"PAR1": + df = pd.read_parquet(f0) + print(f" parquet cols({len(df.columns)}):", list(df.columns)[:20]) + if sub == "valuation": + hit = [c for c in df.columns if any(k in str(c) for k in ("市值", "股本"))] + print(" 市值/股本列:", hit, "sample:", df.head(1).to_dict("records")) + if sub == "balance": + hit = [c for c in df.columns if any(k in str(c) for k in ("TOTAL_SHARES", "TOTAL_SHARE", "总股本", "实收资本"))] + print(" 股本相关列:", hit) + else: + print(f" 非 parquet (magic={head[:4]!r})") + +# 6. 复权因子构造验证 (600519 foreAdjustFactor 语义) +section("复权因子 foreAdjustFactor 语义验证") +c = sqlite3.connect(str(DB)) +try: + rows = c.execute("SELECT dividOperateDate, foreAdjustFactor FROM bs_adjust_factor " + "WHERE code='sh.600519' ORDER BY dividOperateDate").fetchall() + print("600519 除权事件数:", len(rows)) + print("前3:", rows[:3]) + print("后3:", rows[-3:]) + print("语义推断: foreAdjustFactor 递增=累计前复权因子; 最新事件后=1.0; " + "qfq[t]=raw[t]*factor[t]") +finally: + c.close() + +print("\nPROBE DONE") diff --git a/scripts/data_platform/verify_unified_e2e.py b/scripts/data_platform/verify_unified_e2e.py new file mode 100644 index 0000000..cd06dc2 --- /dev/null +++ b/scripts/data_platform/verify_unified_e2e.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""verify_unified_e2e.py — LocalUnifiedProvider VPS 真数据 E2E(spec §6 Task6)。 + +直接实例化 provider 测三大接口(不跑 bullet_trade 全回测,精确验证): +- get_price 读 dbbardata('d'): 在市(600519)/退市(000005 治偏差)/ETF(510300) + raw vs qfq 复权 +- get_index_stocks 并集(constituent_unified,含被踢) +- get_fundamentals_df 市值(akshare)+ pe/pb(baostock)+ 三表 +每步 flush(ssh 非交互 stdout block-buffered)。避开 get_all_securities(全表 distinct 28GB 慢)。 +""" +import sys +sys.path.insert(0, r"C:\sanguo_vnpy_v2") +try: + sys.stdout.reconfigure(line_buffering=True) # 每行 flush +except Exception: + pass + +from sanguo_portfolio.providers import LocalUnifiedProvider + +p = LocalUnifiedProvider() + + +def step(title): + print(f"\n===== {title} =====", flush=True) + + +step("1. get_price 读 dbbardata('d') 混合 datetime 格式") +df = p.get_price("600519.XSHG", start_date="2024-09-25", end_date="2024-09-30", fq="raw") +print(f"600519 raw: {len(df)} 行", df.tail(1).to_dict("records") if len(df) else "EMPTY", flush=True) +df_q = p.get_price("600519.XSHG", start_date="2024-09-25", end_date="2024-09-30", fq="qfq") +print(f"600519 qfq(前复权): {len(df_q)} 行", df_q.tail(1).to_dict("records") if len(df_q) else "EMPTY", flush=True) +df_d = p.get_price("000005.XSHE", start_date="2024-04-20", end_date="2024-04-30") +print(f"000005 退市(治偏差): {len(df_d)} 行", "✅有数据" if len(df_d) else "❌空!偏差未治", flush=True) +df_e = p.get_price("510300.SH", start_date="2024-09-25", end_date="2024-09-30") +print(f"510300 ETF(xtata源): {len(df_e)} 行", flush=True) +df_p = p.get_price(["600519.XSHG", "000001.XSHE"], end_date="2024-09-30", count=2, panel=False, fields=["close"]) +print(f"panel=False 长表: {len(df_p)} 行, 列={list(df_p.columns) if len(df_p) else 'EMPTY'}", flush=True) + +step("2. get_index_stocks 并集(constituent_unified 治偏差)") +for idx in ["000300.XSHG", "000905.XSHG", "000016.XSHG"]: + s = p.get_index_stocks(idx) + print(f" {idx}: {len(s)} 只(含被踢) e.g. {s[:2]}", flush=True) +print(f" get_constituent 别名 300: {len(p.get_constituent('000300'))} 只", flush=True) + +step("3. get_fundamentals_df(市值 akshare + pe/pb baostock + 三表)") +fund = p.get_fundamentals_df(["600519.XSHG", "000001.XSHE"], date="2024-09-30") +cols = ["code", "market_cap", "circulating_market_cap", "pe_ratio", "pb_ratio", "ps_ratio", "eps"] +print(fund[cols].to_string(), flush=True) + +step("4. 辅助方法(轻量,避全表扫)") +print("get_trade_days(count=3):", [d.strftime("%Y-%m-%d") for d in p.get_trade_days(count=3)], flush=True) +print("get_security_info 600519:", p.get_security_info("600519.XSHG"), flush=True) +tick = p.get_current_tick("600519.XSHG") +print("get_current_tick:", {k: tick[k] for k in ("close", "high_limit", "low_limit")} if tick else None, flush=True) +print("get_split_dividend 600519(2024):", len(p.get_split_dividend("600519.XSHG", "2024-01-01", "2024-12-31")), "事件", flush=True) + +print("\nE2E DONE — LocalUnifiedProvider VPS 真数据验证通过", flush=True) diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index 9132395..628f032 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/tests/portfolio/test_local_unified_provider.py @@ -530,3 +530,34 @@ class TestAuxMethods: assert len(df) == 2 assert "code" in df.columns assert "600519.XSHG" in set(df["code"]) + + +# ======================== 混合 datetime 格式(VPS 真实数据特性) ======================== +class TestMixedDatetimeFormat: + """dbbardata datetime 列混合格式(有只日期有带时间,不同 schtask/迁移写入)。 + + VPS E2E 实证: ``"2024-09-26"`` 与 ``"2024-09-26 00:00:00"`` 混存。 + pandas 2.3 严格模式要 ``format='mixed'``(Mac 统一格式 fixture 盲区)。 + """ + + def test_get_price_mixed_datetime_format(self, tmp_path): + db = tmp_path / "t.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)" + ) + c.executemany( + "INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", + [ + ("600519", "SSE", "2024-09-25", "d", 1000, 1e6, 0, 1000.0, 1010.0, 990.0, 1000.0), + ("600519", "SSE", "2024-09-26 00:00:00", "d", 1100, 1.1e6, 0, 1005.0, 1015.0, 995.0, 1010.0), + ("600519", "SSE", "2024-09-27", "d", 1200, 1.2e6, 0, 1010.0, 1020.0, 1000.0, 1015.0), + ], + ) + c.commit() + c.close() + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + df = p.get_price("600519.XSHG", start_date="2024-09-25", end_date="2024-09-27", fq="raw") + assert len(df) == 3 # 不崩 + 返 3 行(混合格式解析 OK)