feat: quant_engine 独立量化引擎（v1.2.0 重构 bootstrap）
从 research_results 抽出纯回测核心能力，形成独立成员仓。 ## 模块（10 个核心） | 模块 | 内容 | |---|---| | alpha_factors | 158 alpha 公式 + 24 基础算子（移植自 qlib alpha158）+ JSONB 工具 | | execution | 执行仿真（成本/滑点/T+1/涨跌停/部分成交/价差）+ 多日 NAV + PnL 拆解 | | indicators | 50+ 技术指标（MACD/KDJ/布林/ATR/ADX/等） | | data_adapter | 桥接 qtdb_pro 长表与新模块（rename/long-wide/复权/vwap 代理） | | backtest | weight-based 多日仿真 | | metrics / perf_stats | 绩效指标 | | factor_library | 通用方法（turnover/IC/winsorize/OLS） | | portfolio_decomp / risk | 组合分解 + 风险指标 | | logging | 统一 logger（标准库 + 可选 loguru） | ## 设计原则 - 零重型依赖（numpy/pandas/scipy） - mypy strict 0 errors（12 source files） - 394 tests passed（从 research_results 复制 + 适配） - ruff clean ## 与 research_results 的关系 - research_results 通过 re-export wrapper 保持向后兼容（src.shared.X → quant_engine.X） - 47 proj 的 import 路径暂时不变，后续逐步迁移 - 本仓角色：researchhub_workspace 引擎层 Co-Authored-By: Mavis <noreply@mavis.local>
This commit is contained in:
George Berkshire
@@ -0,0 +1,400 @@
|
||||
"""数据适配层 —— 桥接 qtdb_pro（Tushare 长表）与新量化模块（alpha158 / execution）。
|
||||
|
||||
背景（v1.2.0 数据对齐核查）：
|
||||
- 现有 47 项目统一用**长表** `(stock_code, trade_date, close_price)`（qtdb_pro.hq_* 表）
|
||||
- alpha_factors.py 需要**单股 Series**（close / open / high / low / volume / vwap）
|
||||
- execution.py 需要**宽表**（date × stock_code）prices / volumes
|
||||
- Tushare 字段命名：`ts_code / vol(手) / amount(千元) / pct_chg`，且**无 vwap 字段**
|
||||
|
||||
本模块提供 6 个纯函数，让新模块直接吃 qtdb_pro 真实数据：
|
||||
1. `long_to_wide()` — 长表 → 宽表（date × stock_code）
|
||||
2. `wide_to_long()` — 宽表 → 长表
|
||||
3. `rename_tushare_columns()` — 列名映射（ts_code→stock_code, vol→volume 等）
|
||||
4. `add_vwap_proxy()` — vwap 代理（Tushare 无 vwap 字段）
|
||||
5. `apply_adj_factor()` — 复权（hq_daily × hq_adj_factor 前复权）
|
||||
6. `prepare_stock_series()` — 单股提取（alpha_factors 输入）
|
||||
7. `prepare_execution_inputs()` — execution 输入（prices + volumes 宽表）
|
||||
8. `load_qtdb_daily()` — 便捷加载（qtdb_pro.hq_daily + 可选复权）
|
||||
|
||||
全部纯 pandas/numpy，零新依赖，mypy strict 兼容。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from quant_engine.logging import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
__all__ = [
|
||||
"long_to_wide",
|
||||
"wide_to_long",
|
||||
"rename_tushare_columns",
|
||||
"add_vwap_proxy",
|
||||
"apply_adj_factor",
|
||||
"prepare_stock_series",
|
||||
"prepare_execution_inputs",
|
||||
"load_qtdb_daily",
|
||||
]
|
||||
|
||||
# Tushare hq_daily 字段 → 统一命名
|
||||
TUSHARE_RENAME: dict[str, str] = {
|
||||
"ts_code": "stock_code",
|
||||
"trade_date": "trade_date",
|
||||
"open": "open",
|
||||
"high": "high",
|
||||
"low": "low",
|
||||
"close": "close",
|
||||
"pre_close": "pre_close",
|
||||
"pct_chg": "pct_chg",
|
||||
"vol": "volume", # 手
|
||||
"amount": "amount", # 千元
|
||||
}
|
||||
|
||||
|
||||
def long_to_wide(
|
||||
df: pd.DataFrame,
|
||||
value_col: str = "close",
|
||||
date_col: str = "trade_date",
|
||||
stock_col: str = "stock_code",
|
||||
) -> pd.DataFrame:
|
||||
"""长表 → 宽表（date × stock_code）。
|
||||
|
||||
Args:
|
||||
df: 长表 DataFrame（含 date_col / stock_col / value_col 三列）
|
||||
value_col: 值列名
|
||||
date_col: 日期列名
|
||||
stock_col: 股票代码列名
|
||||
|
||||
Returns:
|
||||
宽表：index=trade_date（升序），columns=stock_code，值=value_col
|
||||
|
||||
Examples:
|
||||
>>> df = pd.DataFrame({
|
||||
... "stock_code": ["A", "A", "B", "B"],
|
||||
... "trade_date": ["2024-01-01", "2024-01-02", "2024-01-01", "2024-01-02"],
|
||||
... "close": [10.0, 11.0, 20.0, 21.0],
|
||||
... })
|
||||
>>> wide = long_to_wide(df)
|
||||
>>> wide.shape
|
||||
(2, 2)
|
||||
"""
|
||||
if df.empty:
|
||||
return pd.DataFrame()
|
||||
if date_col not in df.columns or stock_col not in df.columns or value_col not in df.columns:
|
||||
raise ValueError(
|
||||
f"long_to_wide: 缺少列 date_col={date_col} / stock_col={stock_col} / "
|
||||
f"value_col={value_col}，实际列={list(df.columns)}"
|
||||
)
|
||||
wide = df.pivot(index=date_col, columns=stock_col, values=value_col)
|
||||
wide.index = pd.to_datetime(wide.index)
|
||||
return wide.sort_index()
|
||||
|
||||
|
||||
def wide_to_long(
|
||||
wide: pd.DataFrame,
|
||||
value_name: str = "close",
|
||||
date_col: str = "trade_date",
|
||||
stock_col: str = "stock_code",
|
||||
) -> pd.DataFrame:
|
||||
"""宽表 → 长表（date × stock_code → 长表）。
|
||||
|
||||
Args:
|
||||
wide: 宽表 DataFrame（index=date，columns=stock_code）
|
||||
value_name: 值列名（如 'close' / 'volume'）
|
||||
date_col: 日期列名（输出）
|
||||
stock_col: 股票代码列名（输出）
|
||||
|
||||
Returns:
|
||||
长表 DataFrame（date_col / stock_col / value_name 三列）
|
||||
"""
|
||||
if wide.empty:
|
||||
return pd.DataFrame(columns=[date_col, stock_col, value_name])
|
||||
long_df = wide.reset_index().melt(
|
||||
id_vars=wide.index.name or "index",
|
||||
var_name=stock_col,
|
||||
value_name=value_name,
|
||||
)
|
||||
long_df = long_df.rename(columns={wide.index.name or "index": date_col})
|
||||
return long_df.dropna(subset=[value_name])
|
||||
|
||||
|
||||
def rename_tushare_columns(df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""Tushare 列名 → 统一命名（只重命名存在的列）。
|
||||
|
||||
Args:
|
||||
df: qtdb_pro 查询结果 DataFrame
|
||||
|
||||
Returns:
|
||||
rename 后的 DataFrame
|
||||
|
||||
Examples:
|
||||
>>> df = pd.DataFrame({"ts_code": ["000001.SZ"], "vol": [100.0]})
|
||||
>>> rename_tushare_columns(df).columns.tolist()
|
||||
['stock_code', 'volume']
|
||||
"""
|
||||
rename_map = {k: v for k, v in TUSHARE_RENAME.items() if k in df.columns}
|
||||
return df.rename(columns=rename_map)
|
||||
|
||||
|
||||
def add_vwap_proxy(
|
||||
df: pd.DataFrame,
|
||||
method: str = "amount_vol",
|
||||
) -> pd.DataFrame:
|
||||
"""vwap 代理（Tushare hq_daily 无 vwap 字段）。
|
||||
|
||||
Args:
|
||||
df: 含 open/high/low/close（必），volume/amount（method=amount_vol 时）
|
||||
method:
|
||||
- 'amount_vol': vwap = amount(千元) / volume(手) 换算为元/股
|
||||
vwap = amount*1000 / (volume*100) = amount * 10 / volume
|
||||
（Tushare amount 单位千元，vol 单位手=100股）
|
||||
- 'typical': vwap = (high + low + close) / 3
|
||||
- 'close': vwap = close
|
||||
|
||||
Returns:
|
||||
新增 'vwap' 列的 DataFrame（原列保留）
|
||||
|
||||
Note:
|
||||
amount_vol 更接近真实 VWAP（成交额/成交量）；
|
||||
typical / close 为兜底（缺 amount 或 volume 时）。
|
||||
"""
|
||||
out = df.copy()
|
||||
if method == "amount_vol" and {"amount", "volume"}.issubset(out.columns):
|
||||
out["vwap"] = out["amount"] * 10.0 / out["volume"].replace(0, np.nan)
|
||||
elif method == "typical" and {"high", "low", "close"}.issubset(out.columns):
|
||||
out["vwap"] = (out["high"] + out["low"] + out["close"]) / 3.0
|
||||
elif {"close"}.issubset(out.columns):
|
||||
out["vwap"] = out["close"]
|
||||
else:
|
||||
raise ValueError(f"add_vwap_proxy: 列不足。method={method}，实际列={list(out.columns)}")
|
||||
return out
|
||||
|
||||
|
||||
def apply_adj_factor(
|
||||
daily: pd.DataFrame,
|
||||
adj_factor: pd.DataFrame,
|
||||
adj_col: str = "adj_factor",
|
||||
date_col: str = "trade_date",
|
||||
stock_col: str = "stock_code",
|
||||
mode: str = "qfq",
|
||||
) -> pd.DataFrame:
|
||||
"""复权：前复权（qfq）/ 后复权（hfq）。
|
||||
|
||||
Tushare 复权因子（hq_adj_factor.adj_factor）：
|
||||
- 后复权价 = 原始价 × adj_factor
|
||||
- 前复权价 = 原始价 × adj_factor / 最新 adj_factor（按股票）
|
||||
|
||||
Args:
|
||||
daily: 原始行情（含 open/high/low/close/volume 等列 + date_col/stock_col）
|
||||
adj_factor: 复权因子表（含 stock_col/date_col/adj_col 三列）
|
||||
adj_col: 因子列名
|
||||
date_col: 日期列名
|
||||
stock_col: 股票代码列名
|
||||
mode: 'qfq'（前复权，默认）/ 'hfq'（后复权）
|
||||
|
||||
Returns:
|
||||
复权后的行情 DataFrame（新增 adj_factor 列；复权仅作用于 OHLC）
|
||||
volume / amount 不复权（成交量本身不受价格复权影响）
|
||||
|
||||
Note:
|
||||
- 只对 open/high/low/close/pre_close 复权
|
||||
- 无 adj_factor 的日期保持原值（不 drop）
|
||||
"""
|
||||
if daily.empty:
|
||||
return daily
|
||||
if adj_factor.empty:
|
||||
logger.warning("apply_adj_factor: adj_factor 为空，返回原始行情")
|
||||
return daily
|
||||
|
||||
out = daily.copy()
|
||||
adj = adj_factor[[stock_col, date_col, adj_col]].copy()
|
||||
# 合并
|
||||
merged = out.merge(adj, on=[stock_col, date_col], how="left")
|
||||
# 每股"最新日期"因子（前复权基准：最后交易日 adj_factor）
|
||||
# 先按日期排序，再取每股最后一条
|
||||
merged = merged.sort_values([stock_col, date_col])
|
||||
latest_map = merged.groupby(stock_col)[adj_col].last()
|
||||
latest = merged[stock_col].map(latest_map)
|
||||
if mode == "qfq":
|
||||
ratio = merged[adj_col] / latest
|
||||
elif mode == "hfq":
|
||||
ratio = merged[adj_col]
|
||||
else:
|
||||
raise ValueError(f"apply_adj_factor: mode={mode} 不支持（qfq/hfq）")
|
||||
ratio = ratio.fillna(1.0) # 无因子日期保持原值
|
||||
|
||||
for col in ("open", "high", "low", "close", "pre_close"):
|
||||
if col in merged.columns:
|
||||
merged[col] = merged[col] * ratio
|
||||
return merged
|
||||
|
||||
|
||||
def prepare_stock_series(
|
||||
df: pd.DataFrame,
|
||||
stock_code: str,
|
||||
stock_col: str = "stock_code",
|
||||
date_col: str = "trade_date",
|
||||
) -> dict[str, pd.Series]:
|
||||
"""单股提取 → alpha_factors 输入（close/open/high/low/volume/vwap Series）。
|
||||
|
||||
Args:
|
||||
df: 长表行情（Tushare rename 后）
|
||||
stock_code: 目标股票代码
|
||||
stock_col: 股票代码列名
|
||||
date_col: 日期列名
|
||||
|
||||
Returns:
|
||||
dict[str, pd.Series]：{'close': Series, 'open': Series, ...}
|
||||
index=trade_date（datetime），按日期升序
|
||||
|
||||
Note:
|
||||
- 只返回存在的列（如无 volume 则不含 'volume' key）
|
||||
- 自动补 vwap（若缺）：先尝试 amount/volume，否则 typical price
|
||||
"""
|
||||
if df.empty:
|
||||
return {}
|
||||
sub = df[df[stock_col] == stock_code].sort_values(date_col)
|
||||
if sub.empty:
|
||||
return {}
|
||||
series_map: dict[str, pd.Series] = {}
|
||||
for col in ("open", "high", "low", "close", "pre_close", "volume", "amount"):
|
||||
if col in sub.columns:
|
||||
s = sub.set_index(date_col)[col]
|
||||
s.index = pd.to_datetime(s.index)
|
||||
series_map[col] = s.astype(float)
|
||||
# vwap 代理
|
||||
if "vwap" not in series_map:
|
||||
if {"amount", "volume"}.issubset(series_map):
|
||||
series_map["vwap"] = (
|
||||
series_map["amount"] * 10.0 / series_map["volume"].replace(0, np.nan)
|
||||
)
|
||||
elif {"high", "low", "close"}.issubset(series_map):
|
||||
series_map["vwap"] = (
|
||||
series_map["high"] + series_map["low"] + series_map["close"]
|
||||
) / 3.0
|
||||
elif "close" in series_map:
|
||||
series_map["vwap"] = series_map["close"]
|
||||
return series_map
|
||||
|
||||
|
||||
def prepare_execution_inputs(
|
||||
df: pd.DataFrame,
|
||||
stock_col: str = "stock_code",
|
||||
date_col: str = "trade_date",
|
||||
) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||||
"""长表行情 → execution 输入（prices + volumes 宽表）。
|
||||
|
||||
Args:
|
||||
df: 长表行情（含 close / volume 列，Tushare rename 后）
|
||||
stock_col: 股票代码列名
|
||||
date_col: 日期列名
|
||||
|
||||
Returns:
|
||||
(prices_wide, volumes_wide)：
|
||||
- prices_wide: date × stock_code，值=close
|
||||
- volumes_wide: date × stock_code，值=volume（若无 volume 列则全 1.0）
|
||||
|
||||
Examples:
|
||||
>>> df = pd.DataFrame({
|
||||
... "stock_code": ["A", "A", "B", "B"],
|
||||
... "trade_date": ["2024-01-01", "2024-01-02"] * 2,
|
||||
... "close": [10.0, 11.0, 20.0, 21.0],
|
||||
... "volume": [100.0, 200.0, 300.0, 400.0],
|
||||
... })
|
||||
>>> prices, volumes = prepare_execution_inputs(df)
|
||||
>>> prices.shape
|
||||
(2, 2)
|
||||
"""
|
||||
if df.empty:
|
||||
return pd.DataFrame(), pd.DataFrame()
|
||||
if "close" not in df.columns:
|
||||
raise ValueError(f"prepare_execution_inputs: 缺 close 列，实际列={list(df.columns)}")
|
||||
prices = long_to_wide(df, value_col="close", date_col=date_col, stock_col=stock_col)
|
||||
if "volume" in df.columns:
|
||||
volumes = long_to_wide(df, value_col="volume", date_col=date_col, stock_col=stock_col)
|
||||
else:
|
||||
logger.warning("prepare_execution_inputs: 无 volume 列，volumes 全 1.0")
|
||||
volumes = pd.DataFrame(1.0, index=prices.index, columns=prices.columns)
|
||||
return prices, volumes
|
||||
|
||||
|
||||
def load_qtdb_daily(
|
||||
stock_codes: Sequence[str],
|
||||
begin_date: str,
|
||||
end_date: str | None = None,
|
||||
with_adj: bool = False,
|
||||
vwap_method: str = "amount_vol",
|
||||
) -> pd.DataFrame:
|
||||
"""便捷加载：qtdb_pro.hq_daily 长表行情（Tushare rename + vwap + 可选复权）。
|
||||
|
||||
Args:
|
||||
stock_codes: 股票代码列表
|
||||
begin_date: 起始日期（YYYY-MM-DD）
|
||||
end_date: 结束日期（可选）
|
||||
with_adj: 是否 join hq_adj_factor 前复权
|
||||
vwap_method: vwap 代理方法（amount_vol / typical / close）
|
||||
|
||||
Returns:
|
||||
长表 DataFrame：
|
||||
stock_code / trade_date / open / high / low / close / pre_close /
|
||||
pct_chg / volume / amount / vwap（+ adj_factor 若 with_adj）
|
||||
|
||||
Note:
|
||||
- 需要真实 ClickHouse 连接（.env 配置 CLICKHOUSE_QTDB_*）
|
||||
- 无连接时返回空 DataFrame + 告警
|
||||
"""
|
||||
try:
|
||||
from src.data.clickhouse import qtdb_query
|
||||
except ImportError:
|
||||
logger.warning("load_qtdb_daily: src.data.clickhouse 不可用，返回空表")
|
||||
return pd.DataFrame()
|
||||
|
||||
where = "ts_code IN %(codes)s AND trade_date >= %(begin)s"
|
||||
params: dict[str, Any] = {"codes": tuple(stock_codes), "begin": begin_date}
|
||||
if end_date:
|
||||
where += " AND trade_date <= %(end)s"
|
||||
params["end"] = end_date
|
||||
|
||||
sql = (
|
||||
"SELECT ts_code, trade_date, open, high, low, close, pre_close, "
|
||||
"pct_chg, vol, amount "
|
||||
f"FROM qtdb_pro.hq_daily WHERE {where} ORDER BY stock_code, trade_date"
|
||||
)
|
||||
try:
|
||||
result = qtdb_query(sql, params)
|
||||
if hasattr(result, "to_df"):
|
||||
df = result.to_df()
|
||||
else:
|
||||
df = pd.DataFrame(result)
|
||||
except Exception as e:
|
||||
logger.warning("load_qtdb_daily: 查询失败（返回空表）: %s", e)
|
||||
return pd.DataFrame()
|
||||
|
||||
if df.empty:
|
||||
return df
|
||||
|
||||
df = rename_tushare_columns(df)
|
||||
df = add_vwap_proxy(df, method=vwap_method)
|
||||
|
||||
if with_adj:
|
||||
try:
|
||||
adj_sql = (
|
||||
"SELECT ts_code, trade_date, adj_factor "
|
||||
f"FROM qtdb_pro.hq_adj_factor WHERE {where} ORDER BY stock_code, trade_date"
|
||||
)
|
||||
adj_result = qtdb_query(adj_sql, params)
|
||||
adj_df = (
|
||||
adj_result.to_df() if hasattr(adj_result, "to_df") else pd.DataFrame(adj_result)
|
||||
)
|
||||
if not adj_df.empty:
|
||||
adj_df = adj_df.rename(columns={"ts_code": "stock_code"})
|
||||
df = apply_adj_factor(df, adj_df, mode="qfq")
|
||||
except Exception as e:
|
||||
logger.warning("load_qtdb_daily: 复权失败（返回未复权）: %s", e)
|
||||
return df
|
||||
Reference in New Issue
Block a user