claude_dev
f416a17b6d
策略02 _pick_stocks 对 5128 只按 market_cap+eps 排序, 旧实现逐只读 4 表(valuation+ income+balance+financial_abstract)+算 calc_roic, 首仓全 cache miss 卡死。 优化(就地, 策略代码零改动, fields=None 向后兼容): - fields= 参数: 只读请求字段依赖的源表(策略02 只要 market_cap+eps → 跳 balance/ financial_abstract/roic, 省一半 parquet 读) - ThreadPool 并发逐只(>64 只; 本地文件 I/O 非 baostock 网络, 不触不并发铁律) - _build_fundamental_row 加 need= 守卫读取(lpp + unified 两层) Mac TDD 4 测试(子集/短路/回归/并发保序)全绿, 124 回归通过。 注: 策略02 要拿满提速需在其 get_fundamentals_df 调用加 fields=[market_cap,eps](策略层, 归策略session)
821 lines
36 KiB
Python
821 lines
36 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
|
|
|
|
@staticmethod
|
|
def _latest_published_annual(
|
|
df: pd.DataFrame, date_str: str,
|
|
) -> Optional[pd.Series]:
|
|
"""最新**已披露年报**: ``NOTICE_DATE <= date_str`` 且 ``REPORT_TYPE`` 含"年"。
|
|
|
|
修复前视偏差: 旧逻辑按 ``REPORT_DATE``(报告期)取最新, 会用尚未披露的年报
|
|
(如 3/31 取 REPORT_DATE=上年 12/31 但 NOTICE_DATE=当年 4 月的年报 → 未来信息)。
|
|
年报口径保证 roe/roa 跨股可比(非季报累计); 最多滞后~1年, 月频策略可接受。
|
|
无 NOTICE_DATE 列 / 无已披露年报 → 回退 ``_latest_row_before(REPORT_DATE)`` 兜底。
|
|
"""
|
|
if df is None or df.empty:
|
|
return None
|
|
if "NOTICE_DATE" not in df.columns:
|
|
return LocalParquetProvider._latest_row_before(df, "REPORT_DATE", date_str)
|
|
ts = pd.Timestamp(date_str)
|
|
d = df.assign(_notice=pd.to_datetime(df["NOTICE_DATE"], errors="coerce"))
|
|
sub = d[d["_notice"] <= ts]
|
|
if "REPORT_TYPE" in sub.columns:
|
|
sub = sub[sub["REPORT_TYPE"].astype(str).str.contains("年", na=False)]
|
|
sub = sub.sort_values("_notice")
|
|
if not sub.empty:
|
|
return sub.iloc[-1]
|
|
return LocalParquetProvider._latest_row_before(df, "REPORT_DATE", date_str)
|
|
|
|
# ==================== 多期已披露财报(供 ValueSelectionStrategy) ====================
|
|
@staticmethod
|
|
def _filter_published(df: pd.DataFrame, date_str: str) -> pd.DataFrame:
|
|
"""NOTICE_DATE <= date_str 已披露行, 按 NOTICE_DATE 升序。
|
|
|
|
⚠️ NOTICE_DATE 全有(2026-07-28全扫确认);REPORT_DATE兜底保留但几乎不触发。
|
|
"""
|
|
if df is None or df.empty:
|
|
return pd.DataFrame()
|
|
if "NOTICE_DATE" not in df.columns:
|
|
if "REPORT_DATE" not in df.columns:
|
|
return pd.DataFrame()
|
|
d = df.assign(_notice=pd.to_datetime(df["REPORT_DATE"], errors="coerce"))
|
|
else:
|
|
d = df.assign(_notice=pd.to_datetime(df["NOTICE_DATE"], errors="coerce"))
|
|
ts = pd.Timestamp(date_str)
|
|
sub = d[d["_notice"] <= ts].sort_values("_notice")
|
|
return sub
|
|
|
|
@staticmethod
|
|
def _latest_n_published(
|
|
df: pd.DataFrame, date_str: str, n: int,
|
|
) -> List[pd.Series]:
|
|
"""近 n 个已披露报告期(任意季报),按 NOTICE_DATE 降序(最新在前)。"""
|
|
sub = LocalParquetProvider._filter_published(df, date_str)
|
|
if sub.empty:
|
|
return []
|
|
take = min(n, len(sub))
|
|
return [sub.iloc[-(i + 1)] for i in range(take)]
|
|
|
|
@staticmethod
|
|
def _latest_n_annual(
|
|
df: pd.DataFrame, date_str: str, n: int,
|
|
) -> List[pd.Series]:
|
|
"""近 n 个已披露年报(REPORT_TYPE 含"年"),按 NOTICE_DATE 降序。"""
|
|
sub = LocalParquetProvider._filter_published(df, date_str)
|
|
if sub.empty:
|
|
return []
|
|
if "REPORT_TYPE" in sub.columns:
|
|
sub = sub[sub["REPORT_TYPE"].astype(str).str.contains("年", na=False)]
|
|
if sub.empty:
|
|
return []
|
|
take = min(n, len(sub))
|
|
return [sub.iloc[-(i + 1)] for i in range(take)]
|
|
|
|
def get_value_metrics(
|
|
self,
|
|
stock: str,
|
|
date: Union[str, datetime],
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""单股多期价值精选指标(供 ``ValueSelectionStrategy`` 调用)。
|
|
|
|
数据源(全本地 parquet, 零 online):
|
|
- valuation: 流通市值(akshare 服务端现成值, 单位元→亿元)
|
|
- balance: TOTAL_CURRENT_ASSETS / TOTAL_CURRENT_LIAB(算流动比率) /
|
|
TOTAL_PARENT_EQUITY(算 ROE)
|
|
- income: BASIC_EPS / OPERATE_INCOME_YOY / PARENT_NETPROFIT(算 ROE)
|
|
- cashflow: NETCASH_OPERATE - NETCASH_INVEST(算 FCF, 年报口径)
|
|
|
|
聚宽→东财字段映射(完整表见 ``docs/research/joinquant_strategies/01_value_selection/notes.md``):
|
|
|
|
| 聚宽字段 | 聚宽表 | 东财表 | 东财字段 |
|
|
|---------|--------|--------|---------|
|
|
| circulating_market_cap | valuation | valuation | circ_market_cap |
|
|
| total_current_assets | balance | balance | TOTAL_CURRENT_ASSETS |
|
|
| total_current_liability | balance | balance | TOTAL_CURRENT_LIAB |
|
|
| roe | indicator | income/balance | PARENT_NETPROFIT / TOTAL_PARENT_EQUITY |
|
|
| net_operate_cash_flow | cash_flow | cashflow | NETCASH_OPERATE |
|
|
| net_invest_cash_flow | cash_flow | cashflow | NETCASH_INVEST |
|
|
| inc_revenue_year_on_year | indicator | income | OPERATE_INCOME_YOY |
|
|
| net_profit_growth | indicator | income | PARENT_NETPROFIT_YOY (fallback NETPROFIT_YOY) |
|
|
|
|
前视偏差修复: 所有财报按 ``NOTICE_DATE(公告日) <= date`` 过滤(原聚宽用 REPORT_DATE
|
|
会有前视, 见 notes.md「移植记录」)。
|
|
|
|
Args:
|
|
stock: jq 风格代码 "600519.XSHG"
|
|
date: 取数日期 YYYY-MM-DD
|
|
|
|
Returns:
|
|
None(三表全空 / 完全没数据) 或 dict 含:
|
|
|
|
- circulating_market_cap: float (亿元)
|
|
- current_ratio: float (近一季流动比率, NaN if 缺)
|
|
- roe_series: List[float] (近 4 季 ROE, 小数 0.15=15%, 最新在前)
|
|
- fcf_series: List[float] (近 5 年 FCF, 元, 最新在前)
|
|
- revenue_yoy_series: List[float] (近 4 季营收同比, 百分数 18.5=18.5%)
|
|
- netprofit_yoy_series: List[float] (近 4 季归母净利润同比, 百分数 18.5=18.5%)
|
|
|
|
数据缺口(已知):
|
|
- 北交所920xxx三表空(akshare不覆盖)→返回None;沪深95%+健康(2026-07-28全扫复核,原"1/3"系误报已撤回)
|
|
- NOTICE_DATE 全有(全扫确认);兜底按REPORT_DATE逻辑保留以防万一
|
|
"""
|
|
import math
|
|
fc = jq_to_file_code(stock)
|
|
date_str = self._to_date_str(date) or datetime.now().strftime("%Y-%m-%d")
|
|
|
|
# 流通市值(近一日已披露)
|
|
val = self._latest_row_before(self._read_valuation(fc), "date", date_str)
|
|
circ_cap = float("nan")
|
|
if val is not None:
|
|
cv = _to_float(val.get("circ_market_cap"))
|
|
if cv:
|
|
circ_cap = to_yi(cv)
|
|
|
|
# 三表
|
|
balance_df = self._read_quarter("balance", fc)
|
|
income_df = self._read_quarter("income", fc)
|
|
cashflow_df = self._read_quarter("cashflow", fc)
|
|
|
|
# 三表全空 → 跳过(北交所920xxx空, akshare不覆盖)
|
|
if balance_df.empty and income_df.empty and cashflow_df.empty:
|
|
return None
|
|
|
|
# 近一季流动比率
|
|
cur_ratio = float("nan")
|
|
if not balance_df.empty:
|
|
bal_rows = self._latest_n_published(balance_df, date_str, 1)
|
|
if bal_rows:
|
|
b = bal_rows[0]
|
|
ca = _to_float(b.get("TOTAL_CURRENT_ASSETS"))
|
|
cl = _to_float(b.get("TOTAL_CURRENT_LIAB"))
|
|
if ca is not None and cl and cl != 0:
|
|
cur_ratio = ca / cl
|
|
|
|
# 近 4 季 ROE(PARENT_NETPROFIT / TOTAL_PARENT_EQUITY, 按报告期对齐)
|
|
roe_series: List[float] = []
|
|
if not income_df.empty and not balance_df.empty:
|
|
inc_rows = self._latest_n_published(income_df, date_str, 4)
|
|
bal_rows = self._latest_n_published(balance_df, date_str, 4)
|
|
for inc_row in inc_rows:
|
|
rdate = inc_row.get("REPORT_DATE")
|
|
if rdate is None:
|
|
continue
|
|
# 按报告期对齐: 找同 REPORT_DATE 的 balance 行
|
|
bal_match = next(
|
|
(b for b in bal_rows if b.get("REPORT_DATE") == rdate), None,
|
|
)
|
|
if bal_match is None:
|
|
continue
|
|
np_ = _to_float(inc_row.get("PARENT_NETPROFIT"))
|
|
eq = _to_float(bal_match.get("TOTAL_PARENT_EQUITY"))
|
|
if np_ is not None and eq and eq != 0:
|
|
roe_series.append(np_ / eq)
|
|
|
|
# 近 4 季营收同比(OPERATE_INCOME_YOY, 百分数) + 净利润同比(PARENT_NETPROFIT_YOY, 百分数)
|
|
yoy_series: List[float] = []
|
|
netprofit_yoy_series: List[float] = []
|
|
if not income_df.empty:
|
|
inc_rows = self._latest_n_published(income_df, date_str, 4)
|
|
for inc_row in inc_rows:
|
|
yoy = _to_float(inc_row.get("OPERATE_INCOME_YOY"))
|
|
if yoy is not None:
|
|
yoy_series.append(yoy)
|
|
# 归母净利润同比优先, 缺则用净利润同比 fallback
|
|
np_yoy = _to_float(inc_row.get("PARENT_NETPROFIT_YOY"))
|
|
if np_yoy is None:
|
|
np_yoy = _to_float(inc_row.get("NETPROFIT_YOY"))
|
|
if np_yoy is not None:
|
|
netprofit_yoy_series.append(np_yoy)
|
|
|
|
# 近 5 年 FCF(NETCASH_OPERATE - NETCASH_INVEST, 年报口径)
|
|
fcf_series: List[float] = []
|
|
if not cashflow_df.empty:
|
|
cf_rows = self._latest_n_annual(cashflow_df, date_str, 5)
|
|
for cf_row in cf_rows:
|
|
op = _to_float(cf_row.get("NETCASH_OPERATE"))
|
|
inv = _to_float(cf_row.get("NETCASH_INVEST"))
|
|
if op is not None and inv is not None:
|
|
fcf_series.append(op - inv)
|
|
|
|
# 完全没数据 → 跳过(北交所三表空等边缘情况)
|
|
if (math.isnan(circ_cap) and math.isnan(cur_ratio)
|
|
and not roe_series and not fcf_series
|
|
and not yoy_series and not netprofit_yoy_series):
|
|
return None
|
|
|
|
return {
|
|
"circulating_market_cap": circ_cap,
|
|
"current_ratio": cur_ratio,
|
|
"roe_series": roe_series,
|
|
"fcf_series": fcf_series,
|
|
"revenue_yoy_series": yoy_series,
|
|
"netprofit_yoy_series": netprofit_yoy_series,
|
|
}
|
|
|
|
# ==================== 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,
|
|
need: Optional[Dict[str, bool]] = None,
|
|
) -> Dict[str, Any]:
|
|
fc = jq_to_file_code(jq_code)
|
|
row: Dict[str, Any] = {"code": jq_code}
|
|
n = need or {}
|
|
|
|
def want(key: str) -> bool:
|
|
return n.get(key, True) # 未传 need({})→全读(向后兼容)
|
|
|
|
val = (
|
|
self._latest_row_before(self._read_valuation(fc), "date", date_str)
|
|
if want("akshare_val") else None
|
|
)
|
|
# income/balance 取最新**已披露年报**(NOTICE_DATE<=date, REPORT_TYPE 含"年"):
|
|
# 修复旧按 REPORT_DATE 过滤的前视偏差(用了未披露年报) + 年报口径跨股可比
|
|
inc = (
|
|
self._latest_published_annual(self._read_quarter("income", fc), date_str)
|
|
if want("income") else None
|
|
)
|
|
bal = (
|
|
self._latest_published_annual(self._read_quarter("balance", fc), date_str)
|
|
if want("balance") else None
|
|
)
|
|
|
|
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) if want("fa") else None
|
|
row["gross_profit_margin"] = _pct_to_decimal(self._latest_indicator(fa, "毛利率"))
|
|
# roic = NOPAT / (归母权益 + 有息负债 - 货币资金)
|
|
# actual_tax_rate akshare 无现成指标, 传 None 让 calc_roic 用 inc_tax/total_profit 兜底
|
|
# _num: g() 的 _to_float 对 NaN 返 float('nan')(truthy), 需 v==v 排除 NaN 才能正确 or 0/条件
|
|
from ..factors.roic import calc_roic
|
|
def _num(d, k):
|
|
v = g(d, k)
|
|
return v if (v is not None and v == v) else None
|
|
oper_profit = _num(inc, "OPERATE_PROFIT")
|
|
inc_tax = _num(inc, "INCOME_TAX")
|
|
profit_before_tax = _num(inc, "TOTAL_PROFIT")
|
|
short_loan = _num(bal, "SHORT_LOAN") or 0
|
|
long_loan = _num(bal, "LONG_LOAN") or 0
|
|
bond_pay = (_num(bal, "BOND_PAYABLE") or 0) + (_num(bal, "SHORT_BOND_PAYABLE") or 0)
|
|
interest_bearing_debt = short_loan + long_loan + bond_pay
|
|
cash_equiv = _num(bal, "MONETARYFUNDS")
|
|
_parent = _num(bal, "TOTAL_PARENT_EQUITY")
|
|
if oper_profit is not None and _parent and cash_equiv is not None:
|
|
row["roic"] = calc_roic(
|
|
oper_profit, None, _parent, interest_bearing_debt, cash_equiv,
|
|
inc_tax=inc_tax, profit_before_tax=profit_before_tax,
|
|
)
|
|
else:
|
|
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,
|
|
}
|