121 lines
4.1 KiB
Python
121 lines
4.1 KiB
Python
"""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/<year>.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
|