claude_dev
b270faf4b9
- BaostockProvider: 读 VPS daily_baostock_full(本地,不调online,守 provider-local-data-only 铁律) - LocalParquetProvider: 读 parquet 兜底,回测117交易日0.4s/月出JSON - all_weather 策略 + runner_backtest 适配 - 数据源融合使用层(单 Provider 内部路由,见 data-fusion spec §6)
572 lines
24 KiB
Python
572 lines
24 KiB
Python
"""LocalParquetProvider: 读 VPS 本地 parquet/csv,零 online 调用。
|
|
|
|
数据布局(VPS ``C:\\sanguo_vnpy_v2\\data\\``,用户多源汇总,见 memory vps-local-data-layout):
|
|
- 日线 K 线: ``qfq/{年}/{code}_daily.parquet`` (date/open/high/low/close/volume)
|
|
- 三大表: ``static/{balance,income,cashflow}/{code}_{type}.parquet``
|
|
akshare 东财大写列,通用列 SECUCODE/**REPORT_DATE**/REPORT_TYPE
|
|
- 每日估值: ``static/valuation/{code}_valuation.parquet`` (中文列 PE(TTM)/市净率/总市值...)
|
|
- 财务摘要: ``static/financial_abstract/{code}_*.parquet`` (宽表 指标×季度)
|
|
- 成分股: ``static/index_const/index_const.parquet`` (⚠️ 仅当前快照→幸存者偏差缺口)
|
|
|
|
实现 bullet_trade ``DataProvider`` 接口; ``get_fundamentals_df`` 字段对齐
|
|
``BaostockProvider._FUNDAMENTAL_COLUMNS``(策略 all_weather 依赖)。
|
|
|
|
优势(vs BaostockProvider 实时调 baostock HTTP):
|
|
- 三表是完整绝对值(balance 221列/income 170列), ``total_liability``/``retained_profit`` 填真值
|
|
(BaostockProvider 比率字段反推受限,多 NaN)
|
|
- valuation PE(TTM)/PB/PS/PCF 是 akshare 服务端现成值,不用 4 季自滚 TTM
|
|
- 零 online: 不踩 baostock 限频/黑名单/休市坑(见 memory provider-local-data-only)
|
|
|
|
已知缺口(V1 标注,不阻塞 MVP):
|
|
- 历史成分股: index_const 仅 2026-07-17 最新一期 → 回测历史有幸存者偏差
|
|
- gross_profit_margin: income 无明确"营业成本"列, V1 NaN, v2 改读 financial_abstract 现成值
|
|
- roic: 需有息负债拆分, V1 NaN
|
|
- 单位口径假设: 市值=元(/1e8转亿)、PE/PB=数值、YOY=百分数(/100转小数); 验证时看数值范围校准
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
import pandas as pd
|
|
|
|
# bullet-trade 可能未装,容错 import DataProvider(照 baostock_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"
|
|
|
|
from ..factors.valuation import to_yi
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# VPS 数据根目录(Windows 路径; Mac 测试时通过 config["data_dir"] 覆盖)
|
|
_DEFAULT_DATA_DIR = r"C:\sanguo_vnpy_v2\data"
|
|
|
|
# jq 代码 ↔ VPS 文件名代码(600519.XSHG ↔ 600519.SH)
|
|
_JQ_TO_FILE_SUFFIX = {"XSHG": "SH", "XSHE": "SZ", "SH": "SH", "SZ": "SZ"}
|
|
_FILE_TO_JQ_SUFFIX = {"SH": "XSHG", "SZ": "XSHE"}
|
|
|
|
|
|
def jq_to_file_code(jq_code: str) -> str:
|
|
"""``600519.XSHG`` → ``600519.SH`` (VPS parquet 文件名格式)。纯数字透传。"""
|
|
if not jq_code or "." not in jq_code:
|
|
return jq_code
|
|
code, suffix = jq_code.split(".", 1)
|
|
file_suffix = _JQ_TO_FILE_SUFFIX.get(suffix.upper())
|
|
return f"{code}.{file_suffix}" if file_suffix else jq_code
|
|
|
|
|
|
def file_to_jq_code(file_code: str) -> str:
|
|
"""``600519.SH`` → ``600519.XSHG``。纯 6 位按 6开头=sh/0,3开头=sz 推断。"""
|
|
if not file_code:
|
|
return file_code
|
|
if "." not in file_code:
|
|
if len(file_code) == 6:
|
|
return f"{file_code}.{'XSHG' if file_code.startswith('6') else 'XSHE'}"
|
|
return file_code
|
|
code, suffix = file_code.split(".", 1)
|
|
jq_suffix = _FILE_TO_JQ_SUFFIX.get(suffix.upper())
|
|
return f"{code}.{jq_suffix}" if jq_suffix else file_code
|
|
|
|
|
|
# jq → VPS K 线文件名(baostock 风格 sh/sz 前缀无点; 与三表 jq 后缀格式不同!)
|
|
_KLINE_PREFIX = {"XSHG": "sh", "XSHE": "sz", "SH": "sh", "SZ": "sz"}
|
|
|
|
|
|
def jq_to_kline_code(jq_code: str) -> str:
|
|
"""``600519.XSHG`` → ``sh600519`` (VPS qfq/raw K线文件名)。纯 6 位按 6开头=sh 推断。"""
|
|
if not jq_code:
|
|
return jq_code
|
|
if "." not in jq_code:
|
|
if len(jq_code) == 6:
|
|
return ("sh" if jq_code.startswith("6") else "sz") + jq_code
|
|
return jq_code
|
|
code, suffix = jq_code.split(".", 1)
|
|
prefix = _KLINE_PREFIX.get(suffix.upper())
|
|
return (prefix + code) if prefix else jq_code
|
|
|
|
|
|
# valuation parquet 中文列 → 英文
|
|
_VAL_COL_MAP = {
|
|
"数据日期": "date", "当日收盘价": "close", "当日涨跌幅": "pct_chg",
|
|
"总市值": "total_market_cap", "流通市值": "circ_market_cap",
|
|
"总股本": "total_share", "流通股本": "circ_share",
|
|
"PE(TTM)": "pe_ttm", "PE(静)": "pe_static",
|
|
"市净率": "pb", "PEG值": "peg", "市现率": "pcf", "市销率": "ps",
|
|
}
|
|
|
|
|
|
def _to_float(v: Any) -> Optional[float]:
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, (int, float)):
|
|
return float(v)
|
|
try:
|
|
s = str(v).strip().replace(",", "").replace("%", "")
|
|
return float(s) if s else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _pct_to_decimal(v: Any) -> float:
|
|
"""百分数(18.5 表示 18.5%) → 小数(0.185)。None/异常 → NaN。akshare YOY 通常百分数。"""
|
|
f = _to_float(v)
|
|
if f is None:
|
|
return float("nan")
|
|
return f / 100.0
|
|
|
|
|
|
def _or_nan(v: Any) -> float:
|
|
f = _to_float(v)
|
|
return f if f is not None else float("nan")
|
|
|
|
|
|
# 策略 all_weather 依赖的 fundamentals 输出列(对齐 BaostockProvider._FUNDAMENTAL_COLUMNS)
|
|
_FUNDAMENTAL_COLUMNS: List[str] = [
|
|
"code", "market_cap", "circulating_market_cap",
|
|
"pe_ratio", "pb_ratio", "ps_ratio", "pcf_ratio",
|
|
"roe", "roa", "eps", "gross_profit_margin", "net_profit_margin",
|
|
"inc_revenue_year_on_year", "inc_operation_profit_year_on_year",
|
|
"inc_total_revenue_year_on_year",
|
|
"total_liability", "total_sheet_owner_equities", "retained_profit",
|
|
"roic",
|
|
]
|
|
|
|
|
|
class LocalParquetProvider(DataProvider): # type: ignore[misc]
|
|
"""读 VPS 本地 parquet 的数据 provider(回测专用,零 online)。
|
|
|
|
所有方法读 ``data_dir`` 下 parquet 文件,不调任何外部 API。
|
|
"""
|
|
|
|
name: str = "sanguo_local_parquet"
|
|
requires_live_data: bool = False
|
|
|
|
def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
|
|
cfg = config or {}
|
|
self.data_dir: str = cfg.get("data_dir", _DEFAULT_DATA_DIR)
|
|
# 缓存:同股多次读只一次 IO
|
|
self._val_cache: Dict[str, pd.DataFrame] = {}
|
|
self._quarter_cache: Dict[tuple, pd.DataFrame] = {}
|
|
self._index_const_cache: Optional[pd.DataFrame] = None
|
|
|
|
# ==================== 路径辅助 ====================
|
|
def _valuation_path(self, file_code: str) -> str:
|
|
return os.path.join(self.data_dir, "static", "valuation", f"{file_code}_valuation.parquet")
|
|
|
|
def _static_path(self, table: str, file_code: str) -> str:
|
|
return os.path.join(self.data_dir, "static", table, f"{file_code}_{table}.parquet")
|
|
|
|
@staticmethod
|
|
def _year_range(start: Optional[pd.Timestamp], end: Optional[pd.Timestamp]) -> range:
|
|
s = start.year if start is not None else 2010
|
|
e = end.year if end is not None else datetime.now().year
|
|
if e < s:
|
|
s, e = e, s
|
|
return range(s, e + 1)
|
|
|
|
# ==================== get_price ====================
|
|
def get_price(
|
|
self,
|
|
security: Union[str, List[str]],
|
|
start_date: Union[str, datetime] = None,
|
|
end_date: Union[str, datetime] = None,
|
|
frequency: str = "day",
|
|
fields: Optional[List[str]] = None,
|
|
skip_paused: bool = True,
|
|
fq: str = "qfq",
|
|
count: Optional[int] = None,
|
|
panel: bool = True,
|
|
fill_paused: bool = True,
|
|
) -> pd.DataFrame:
|
|
"""读本地 qfq/raw 日线 parquet,拼多年 + 过滤日期区间。
|
|
|
|
聚宽/bullet_trade 兼容参数:
|
|
- ``count``: 无 start_date 时取 end_date 前 N 根
|
|
- ``panel``: True=多股 panel(index=date,外层 code); False=长表(time/code/fields)
|
|
bullet_trade ``_trend_mean`` 用 panel=False + pivot(index=time,columns=code)
|
|
- ``fill_paused``: 停牌填充(忽略,直接读原始)
|
|
"""
|
|
codes = [security] if isinstance(security, str) else list(security or [])
|
|
freq_dir = "raw" if fq == "raw" else "qfq"
|
|
if frequency.startswith("min") or frequency in ("1m", "1min"):
|
|
freq_dir = "minute_15" # V1: 分钟线只支持 15min 目录
|
|
|
|
start = pd.Timestamp(start_date) if start_date else None
|
|
end = pd.Timestamp(end_date) if end_date else None
|
|
# count 模式: 无 start_date, 读 end 前 N 根(近 3 年覆盖足够)
|
|
if count and start is None:
|
|
end_for_count = end or pd.Timestamp.now()
|
|
years = range(end_for_count.year - 2, end_for_count.year + 1)
|
|
else:
|
|
years = self._year_range(start, end)
|
|
|
|
frames: Dict[str, pd.DataFrame] = {}
|
|
for jq_code in codes:
|
|
fc = jq_to_kline_code(jq_code)
|
|
parts: List[pd.DataFrame] = []
|
|
for y in years:
|
|
p = os.path.join(self.data_dir, freq_dir, str(y), f"{fc}_daily.parquet")
|
|
if os.path.exists(p):
|
|
try:
|
|
parts.append(pd.read_parquet(p))
|
|
except Exception as exc:
|
|
logger.warning("读 K 线失败 %s/%s: %s", y, fc, exc)
|
|
if not parts:
|
|
continue
|
|
df = pd.concat(parts, ignore_index=True)
|
|
if "date" in df.columns:
|
|
df["date"] = pd.to_datetime(df["date"])
|
|
df = df.sort_values("date")
|
|
if end is not None:
|
|
df = df[df["date"] <= end]
|
|
if start is not None:
|
|
df = df[df["date"] >= start]
|
|
if count:
|
|
df = df.tail(count) # 取最近 count 根
|
|
df = df.set_index("date")
|
|
if fields:
|
|
keep = [c for c in fields if c in df.columns]
|
|
df = df[keep] if keep else df
|
|
frames[jq_code] = df
|
|
|
|
if not frames:
|
|
return pd.DataFrame()
|
|
# panel=False: 长表(time/code/fields), 兼容 bullet_trade pivot
|
|
if not panel:
|
|
long_parts = []
|
|
for jq_code, df in frames.items():
|
|
d = df.reset_index().rename(columns={"date": "time"})
|
|
d.insert(0, "code", jq_code)
|
|
long_parts.append(d)
|
|
return pd.concat(long_parts, ignore_index=True) if long_parts else pd.DataFrame()
|
|
if len(frames) == 1:
|
|
return next(iter(frames.values()))
|
|
try:
|
|
return pd.concat(frames, axis=1)
|
|
except Exception as exc:
|
|
logger.warning("多股 panel concat 失败,返回首只: %s", exc)
|
|
return next(iter(frames.values()))
|
|
|
|
# ==================== 估值/三表 读取 ====================
|
|
def _read_valuation(self, file_code: str) -> pd.DataFrame:
|
|
if file_code in self._val_cache:
|
|
return self._val_cache[file_code]
|
|
p = self._valuation_path(file_code)
|
|
if not os.path.exists(p):
|
|
self._val_cache[file_code] = pd.DataFrame()
|
|
return pd.DataFrame()
|
|
try:
|
|
df = pd.read_parquet(p).rename(columns=_VAL_COL_MAP)
|
|
if "date" in df.columns:
|
|
df["date"] = pd.to_datetime(df["date"], errors="coerce")
|
|
df = df.sort_values("date")
|
|
self._val_cache[file_code] = df
|
|
return df
|
|
except Exception as exc:
|
|
logger.warning("读 valuation 失败 %s: %s", file_code, exc)
|
|
self._val_cache[file_code] = pd.DataFrame()
|
|
return pd.DataFrame()
|
|
|
|
def _read_quarter(self, table: str, file_code: str) -> pd.DataFrame:
|
|
key = (table, file_code)
|
|
if key in self._quarter_cache:
|
|
return self._quarter_cache[key]
|
|
p = self._static_path(table, file_code)
|
|
if not os.path.exists(p):
|
|
self._quarter_cache[key] = pd.DataFrame()
|
|
return pd.DataFrame()
|
|
try:
|
|
df = pd.read_parquet(p)
|
|
if "REPORT_DATE" in df.columns:
|
|
df["REPORT_DATE"] = pd.to_datetime(df["REPORT_DATE"], errors="coerce")
|
|
df = df.sort_values("REPORT_DATE")
|
|
self._quarter_cache[key] = df
|
|
return df
|
|
except Exception as exc:
|
|
logger.warning("读 %s 失败 %s: %s", table, file_code, exc)
|
|
self._quarter_cache[key] = pd.DataFrame()
|
|
return pd.DataFrame()
|
|
|
|
def _read_financial_abstract(self, file_code: str) -> Optional[pd.DataFrame]:
|
|
"""读 financial_abstract 宽表(指标×季度, 列: 选项/指标/20260331/20251231/...)。"""
|
|
p = os.path.join(self.data_dir, "static", "financial_abstract", f"{file_code}_financial_abstract.parquet")
|
|
if not os.path.exists(p):
|
|
return None
|
|
try:
|
|
return pd.read_parquet(p)
|
|
except Exception as exc:
|
|
logger.warning("读 financial_abstract 失败 %s: %s", file_code, exc)
|
|
return None
|
|
|
|
@staticmethod
|
|
def _latest_indicator(fa_df: Optional[pd.DataFrame], indicator_name: str) -> Optional[float]:
|
|
"""从 financial_abstract 宽表取指定指标最新季度值(第一个季度列)。"""
|
|
if fa_df is None or "指标" not in fa_df.columns:
|
|
return None
|
|
rows = fa_df[fa_df["指标"] == indicator_name]
|
|
if rows.empty:
|
|
return None
|
|
quarter_cols = [c for c in fa_df.columns if c not in ("选项", "指标")]
|
|
if not quarter_cols:
|
|
return None
|
|
return _to_float(rows.iloc[0][quarter_cols[0]])
|
|
|
|
@staticmethod
|
|
def _latest_row_before(
|
|
df: pd.DataFrame, date_col: str, date_str: str,
|
|
) -> Optional[pd.Series]:
|
|
"""取 ``date_col <= date_str`` 的最后一行(最新已披露)。"""
|
|
if df is None or df.empty or date_col not in df.columns:
|
|
return None
|
|
ts = pd.Timestamp(date_str)
|
|
sub = df[df[date_col] <= ts]
|
|
return sub.iloc[-1] if not sub.empty else None
|
|
|
|
# ==================== get_fundamentals_df ====================
|
|
def get_fundamentals_df(
|
|
self,
|
|
stocks: List[str],
|
|
date: Optional[Union[str, datetime]] = None,
|
|
) -> pd.DataFrame:
|
|
"""合并多股 fundamentals,列对齐 ``_FUNDAMENTAL_COLUMNS``。
|
|
|
|
数据源(本地 parquet):
|
|
- valuation: market_cap/pe/pb/ps/pcf(akshare 服务端现成值)
|
|
- income: eps(BASIC_EPS)/inc_*_yoy(OPERATE_*_YOY)/net_profit_margin(NP÷营收 算)
|
|
- balance: total_liability/total_sheet_owner_equities/retained_profit(绝对值现成)
|
|
- 算: roe(NP÷权益)/roa(NP÷总资产)
|
|
"""
|
|
if not stocks:
|
|
return pd.DataFrame(columns=_FUNDAMENTAL_COLUMNS)
|
|
|
|
date_str = self._to_date_str(date) or datetime.now().strftime("%Y-%m-%d")
|
|
rows: List[Dict[str, Any]] = [
|
|
self._build_fundamental_row(jq_code, date_str) for jq_code in stocks
|
|
]
|
|
df = pd.DataFrame(rows, columns=_FUNDAMENTAL_COLUMNS)
|
|
if "code" in df.columns:
|
|
df = df.set_index("code", drop=False)
|
|
return df
|
|
|
|
def _build_fundamental_row(self, jq_code: str, date_str: str) -> Dict[str, Any]:
|
|
fc = jq_to_file_code(jq_code)
|
|
row: Dict[str, Any] = {"code": jq_code}
|
|
|
|
val = self._latest_row_before(self._read_valuation(fc), "date", date_str)
|
|
inc = self._latest_row_before(self._read_quarter("income", fc), "REPORT_DATE", date_str)
|
|
bal = self._latest_row_before(self._read_quarter("balance", fc), "REPORT_DATE", date_str)
|
|
|
|
def g(d: Optional[pd.Series], k: str) -> Optional[float]:
|
|
return _to_float(d.get(k)) if d is not None else None
|
|
|
|
# --- 估值字段(akshare valuation: 市值元, PE/PB 数值) ---
|
|
mkt = g(val, "total_market_cap")
|
|
circ = g(val, "circ_market_cap")
|
|
row["market_cap"] = to_yi(mkt) if mkt else float("nan")
|
|
row["circulating_market_cap"] = to_yi(circ) if circ else float("nan")
|
|
row["pe_ratio"] = _or_nan(g(val, "pe_ttm"))
|
|
row["pb_ratio"] = _or_nan(g(val, "pb"))
|
|
row["ps_ratio"] = _or_nan(g(val, "ps"))
|
|
row["pcf_ratio"] = _or_nan(g(val, "pcf"))
|
|
|
|
# --- 利润表字段 ---
|
|
row["eps"] = _or_nan(g(inc, "BASIC_EPS"))
|
|
row["inc_revenue_year_on_year"] = _pct_to_decimal(g(inc, "OPERATE_INCOME_YOY"))
|
|
row["inc_operation_profit_year_on_year"] = _pct_to_decimal(g(inc, "OPERATE_PROFIT_YOY"))
|
|
# inc_total_revenue: akshare income 无 total_revenue_YOY 独立列,用 OPERATE_INCOME_YOY 近似
|
|
row["inc_total_revenue_year_on_year"] = _pct_to_decimal(g(inc, "OPERATE_INCOME_YOY"))
|
|
|
|
net_profit = g(inc, "PARENT_NETPROFIT") or g(inc, "NETPROFIT")
|
|
revenue = g(inc, "OPERATE_INCOME")
|
|
total_assets = g(bal, "TOTAL_ASSETS")
|
|
parent_equity = g(bal, "TOTAL_PARENT_EQUITY")
|
|
|
|
# net_profit_margin = 归母净利润 / 营收(小数)
|
|
row["net_profit_margin"] = (
|
|
net_profit / revenue
|
|
if (net_profit and revenue and revenue != 0)
|
|
else float("nan")
|
|
)
|
|
|
|
# --- 资产负债表(绝对值,元→亿) ---
|
|
total_liab = g(bal, "TOTAL_LIABILITIES")
|
|
row["total_liability"] = to_yi(total_liab) if total_liab else float("nan")
|
|
row["total_sheet_owner_equities"] = to_yi(parent_equity) if parent_equity else float("nan")
|
|
retained = (g(bal, "SURPLUS_RESERVE") or 0) + (g(bal, "UNASSIGN_RPOFIT") or 0)
|
|
row["retained_profit"] = to_yi(retained) if retained else float("nan")
|
|
|
|
# --- 算指标(单期非年化TTM;v2 改 financial_abstract 现成年化值) ---
|
|
row["roe"] = (
|
|
net_profit / parent_equity
|
|
if (net_profit and parent_equity and parent_equity != 0)
|
|
else float("nan")
|
|
)
|
|
row["roa"] = (
|
|
net_profit / total_assets
|
|
if (net_profit and total_assets and total_assets != 0)
|
|
else float("nan")
|
|
)
|
|
|
|
# gross_profit_margin: 从 financial_abstract 读现成"毛利率"(百分数→小数)
|
|
fa = self._read_financial_abstract(fc)
|
|
row["gross_profit_margin"] = _pct_to_decimal(self._latest_indicator(fa, "毛利率"))
|
|
# roic: 需有息负债拆分 → V1 NaN
|
|
# TODO v2: roic = NOPAT / (权益 + 有息负债 - 现金)
|
|
row["roic"] = float("nan")
|
|
return row
|
|
|
|
@staticmethod
|
|
def _to_date_str(value: Optional[Union[str, datetime]]) -> Optional[str]:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, str):
|
|
return value
|
|
return value.strftime("%Y-%m-%d")
|
|
|
|
# ==================== get_security_info ====================
|
|
def get_security_info(self, security: str) -> Dict[str, Any]:
|
|
fc = jq_to_file_code(security)
|
|
val_df = self._read_valuation(fc)
|
|
if val_df.empty:
|
|
return {"code": security, "display_name": security, "name": security}
|
|
last = val_df.iloc[-1]
|
|
return {
|
|
"code": security,
|
|
"display_name": security, # valuation 无名称,用 code
|
|
"name": security,
|
|
"start_date": val_df["date"].min().strftime("%Y-%m-%d") if "date" in val_df.columns else None,
|
|
"end_date": val_df["date"].max().strftime("%Y-%m-%d") if "date" in val_df.columns else None,
|
|
"type": "stock",
|
|
}
|
|
|
|
# ==================== get_trade_days ====================
|
|
def get_trade_days(
|
|
self,
|
|
start_date: Optional[Union[str, datetime]] = None,
|
|
end_date: Optional[Union[str, datetime]] = None,
|
|
count: Optional[int] = None,
|
|
) -> List[datetime]:
|
|
"""从蓝筹 sh600000 K 线 date 列取交易日(锚定,全市场交易日一致)。
|
|
|
|
bullet_trade 引擎调 ``get_trade_days(count=N)`` 取最近 N 天(无 start_date),
|
|
故兼容 count 参数(其他基类方法也可能传 count)。
|
|
"""
|
|
fc = "sh600000"
|
|
start_ts = pd.Timestamp(start_date) if start_date else None
|
|
end_ts = pd.Timestamp(end_date) if end_date else None
|
|
if count and not start_ts:
|
|
now_y = datetime.now().year
|
|
years = range(now_y - 2, now_y + 1) # 近 3 年足够覆盖 count 天
|
|
else:
|
|
years = self._year_range(start_ts, end_ts)
|
|
days: List[datetime] = []
|
|
for y in years:
|
|
p = os.path.join(self.data_dir, "qfq", str(y), f"{fc}_daily.parquet")
|
|
if not os.path.exists(p):
|
|
continue
|
|
try:
|
|
df = pd.read_parquet(p, columns=["date"])
|
|
for d in pd.to_datetime(df["date"]):
|
|
days.append(d.to_pydatetime())
|
|
except Exception as exc:
|
|
logger.warning("读交易日失败 %s: %s", y, exc)
|
|
if not days:
|
|
return []
|
|
days = sorted(set(days))
|
|
if start_ts:
|
|
days = [d for d in days if pd.Timestamp(d) >= start_ts]
|
|
if end_ts:
|
|
days = [d for d in days if pd.Timestamp(d) <= end_ts]
|
|
if count:
|
|
days = days[-count:]
|
|
return days
|
|
|
|
# ==================== get_all_securities ====================
|
|
def get_all_securities(
|
|
self, types: Optional[List[str]] = None,
|
|
) -> pd.DataFrame:
|
|
"""列 ``static/valuation/`` 下所有股票(文件名 → jq code)。"""
|
|
val_dir = os.path.join(self.data_dir, "static", "valuation")
|
|
if not os.path.isdir(val_dir):
|
|
return pd.DataFrame(columns=["code", "display_name"])
|
|
codes: List[str] = []
|
|
for fn in os.listdir(val_dir):
|
|
if fn.endswith("_valuation.parquet"):
|
|
codes.append(file_to_jq_code(fn.replace("_valuation.parquet", "")))
|
|
return pd.DataFrame({"code": codes, "display_name": codes})
|
|
|
|
# ==================== get_index_stocks ====================
|
|
def get_index_stocks(
|
|
self,
|
|
index_symbol: str,
|
|
date: Optional[Union[str, datetime]] = None,
|
|
) -> List[str]:
|
|
"""读 ``index_const.parquet`` 过滤指数成分。
|
|
|
|
⚠️ 缺口:VPS index_const 仅 2026-07-17 最新一期(当前快照),
|
|
回测历史日期会用到"现在还在指数里的股票"→ 幸存者偏差(结果虚高)。
|
|
``date`` 参数目前忽略(无历史数据),待补 csindex 历史成分。
|
|
"""
|
|
ic = self._load_index_const()
|
|
if ic is None or ic.empty:
|
|
logger.warning("index_const.parquet 无数据,get_index_stocks 返回空")
|
|
return []
|
|
idx = index_symbol.split(".")[0] if "." in index_symbol else index_symbol
|
|
col = "指数代码" if "指数代码" in ic.columns else "index_code"
|
|
sub = ic[ic[col].astype(str).str.contains(idx, na=False)]
|
|
code_col = "成分券代码" if "成分券代码" in ic.columns else None
|
|
if code_col is None:
|
|
return []
|
|
return [file_to_jq_code(str(c)) for c in sub[code_col].tolist()]
|
|
|
|
def _load_index_const(self) -> Optional[pd.DataFrame]:
|
|
if self._index_const_cache is not None:
|
|
return self._index_const_cache
|
|
p = os.path.join(self.data_dir, "static", "index_const", "index_const.parquet")
|
|
if not os.path.exists(p):
|
|
self._index_const_cache = None
|
|
return None
|
|
try:
|
|
self._index_const_cache = pd.read_parquet(p)
|
|
return self._index_const_cache
|
|
except Exception as exc:
|
|
logger.warning("读 index_const 失败: %s", exc)
|
|
self._index_const_cache = None
|
|
return None
|
|
|
|
# ==================== get_split_dividend (qfq 已复权,占位) ====================
|
|
def get_split_dividend(
|
|
self,
|
|
security: str,
|
|
start_date: Optional[Union[str, datetime]] = None,
|
|
end_date: Optional[Union[str, datetime]] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""除权除息记录。读 qfq 日线已前复权,回测不依赖此方法 → 占位返空 list。
|
|
|
|
TODO v2:若需 raw→qfq 自算,从 outstanding_share 变化 + 派息记录派生。
|
|
"""
|
|
return []
|
|
|
|
# ==================== get_current_tick (回测不用,占位) ====================
|
|
def get_current_tick(self, security: str) -> Optional[Dict[str, Any]]:
|
|
"""回测不用实时 tick;从 valuation 最新行推算 close + 涨跌停(filter_limitup 用)。"""
|
|
val_df = self._read_valuation(jq_to_file_code(security))
|
|
if val_df.empty:
|
|
return None
|
|
last = val_df.iloc[-1]
|
|
close = _to_float(last.get("close"))
|
|
pct = _to_float(last.get("pct_chg")) or 0.0
|
|
# 涨跌停:主板 ±10%(ST/创业/科创 精确规则 v2 补)
|
|
high_limit = round(close * 1.1, 2) if close else None
|
|
low_limit = round(close * 0.9, 2) if close else None
|
|
return {
|
|
"code": security, "current_price": close, "close": close,
|
|
"high_limit": high_limit, "low_limit": low_limit,
|
|
"change_percent": pct,
|
|
}
|