feat(quant): QlibEngine v1 — 本地行情落盘 QlibDataset → D.features 读取 → 因子回测
- qlib_adapter/provider.py：SQLite 行情按 qlib 0.9.8 二进制格式落盘（起始索引头 + 逐日 float32、instruments 3 列、小写 instrument、晚上市 offset）
- qlib_adapter/dataset.py：qlib.init 幂等（{'day': uri}）+ D.features 读取 close 面板
- qlib_adapter/engine.py：QlibEngine（QuantEngine）v1 —— Qlib 数据管线回测与 LocalEngine 同记账规则（无未来函数/成本/涨跌停标注），factor_test 复用共享实现；Alpha158+LightGBM 为 TODO
- 真实 20 股验证：qlib 落盘 142 文件→读取→回测（-12.81%，Local 对照 -12.97%，差异为 qlib float32 存储）
- tests/test_qlib_engine.py 5 项（格式/roundtrip/晚上市 offset/回测/因子测试）→ pytest 86 passed / ruff clean
This commit is contained in:
Simon
@@ -0,0 +1,47 @@
|
||||
"""Qlib 数据读取（dataset）：qlib.init 幂等 + 从 provider_uri 读行情面板。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
_init_state: dict = {"uri": None}
|
||||
|
||||
|
||||
def ensure_qlib_init(provider_uri: Path) -> None:
|
||||
"""初始化 qlib（进程内同 uri 幂等；不同 uri 会重新 init）。"""
|
||||
if _init_state["uri"] == str(provider_uri):
|
||||
return
|
||||
import qlib
|
||||
from qlib.config import REG_CN
|
||||
|
||||
qlib.init(provider_uri={"day": str(provider_uri)}, region=REG_CN)
|
||||
_init_state["uri"] = str(provider_uri)
|
||||
|
||||
|
||||
def load_close_panel(
|
||||
provider_uri: Path,
|
||||
symbols: list[str],
|
||||
start: date,
|
||||
end: date,
|
||||
) -> pd.DataFrame:
|
||||
"""经 qlib D.features 读取 $close，返回 date×symbol 面板（index datetime）。"""
|
||||
ensure_qlib_init(provider_uri)
|
||||
from qlib.data import D
|
||||
|
||||
lower = [s.lower() for s in symbols]
|
||||
df = D.features(
|
||||
lower,
|
||||
["$close"],
|
||||
start_time=start.strftime("%Y-%m-%d"),
|
||||
end_time=end.strftime("%Y-%m-%d"),
|
||||
freq="day",
|
||||
)
|
||||
if df.empty:
|
||||
return pd.DataFrame()
|
||||
close = df["$close"].unstack(level="instrument")
|
||||
close.index = pd.to_datetime(close.index)
|
||||
close.columns = [c.upper() for c in close.columns]
|
||||
return close.sort_index()
|
||||
@@ -1,38 +1,80 @@
|
||||
"""Qlib 后端引擎（桥接占位）。
|
||||
"""QlibEngine —— 基于 Qlib 数据管线的研究引擎（QuantEngine 实现）。
|
||||
|
||||
本模块是 Qlib 引擎的桥接边界（AGENT.md §14 / §15：Qlib 只允许在
|
||||
quant/qlib_adapter/ 内被引用）。Qlib（pyqlib）已通过「源码 git 安装」成功
|
||||
部署到本机（Linux aarch64 + CPython 3.12，backend/.venv，uv 管理；
|
||||
backend/pyproject.toml 固定 commit，细节见 docs/ROADMAP.md §2 备注）。
|
||||
v1 能力（本阶段已打通并测试）：
|
||||
1. provider.build_qlib_dataset：把本地行情按 qlib 二进制格式落盘（data/qlib）
|
||||
2. dataset.ensure_qlib_init + D.features：从 QlibDataset 读取行情（真实 qlib 通路）
|
||||
3. 在 Qlib 读取的行情面板上执行 TopK 因子回测 → 标准 BacktestResult
|
||||
（与 LocalEngine 记账规则一致：无未来函数、成本/涨跌停/停牌近似 + unimplemented 标注）
|
||||
|
||||
基于 Qlib 的因子 / 回测工作流（Parquet/本地行情 → QlibDataset → Alpha158 →
|
||||
LightGBM 训练/预测 → 回测，并归一化为 domain.entities.research 输出）属
|
||||
Phase 2 后续实现；当前默认研究引擎仍为 LocalEngine（纯 pandas，见
|
||||
app/quant/engine.py），业务层经 QuantEngine Protocol 注入，切换无需改业务代码。
|
||||
说明：
|
||||
- 因子分（score）在源行情上按 app.quant.factors 计算（注册表口径一致）；
|
||||
qlib 侧负责数据读取供给（未来可由 Qlib Dataset 直接产出特征）。
|
||||
- Alpha158 特征集 + LightGBM 预测信号的模型增强（walk-forward 训练/预测）为下一步
|
||||
TODO，详见 docs/ROADMAP.md §2 与 docs/QLIB_VERIFICATION.md。
|
||||
- 默认研究引擎仍是 LocalEngine（app/quant/engine.py）；切换只需注入本类。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from app.domain.entities.research import BacktestResult, FactorTestReport, ResearchSpec
|
||||
from app.quant.engine import QuantEngine
|
||||
|
||||
_MSG = (
|
||||
"Qlib 引擎工作流尚未实现：pyqlib 现已可在本机安装使用（见 docs/ROADMAP.md §2），"
|
||||
"但 QlibDataset → LightGBM 的实现待 Phase 2 补齐。当前请使用默认 LocalEngine。"
|
||||
from app.quant.local_engine import (
|
||||
TopKBacktestRunner,
|
||||
build_factor_panels,
|
||||
composite_score,
|
||||
run_spec_factor_test,
|
||||
)
|
||||
from app.quant.qlib_adapter.dataset import ensure_qlib_init, load_close_panel
|
||||
from app.quant.qlib_adapter.provider import build_qlib_dataset
|
||||
|
||||
_ENGINE_NOTE = "QlibEngine v1：本地行情→QlibDataset(bin)→D.features 读取→TopK 因子回测"
|
||||
|
||||
|
||||
class QlibEngine(QuantEngine):
|
||||
"""pyqlib 后端占位：抛 NotImplementedError 并给出启用指引。"""
|
||||
"""基于 Qlib 数据管线的引擎；因子评估复用共享实现，回测从 QlibDataset 读取行情。"""
|
||||
|
||||
name = "qlib"
|
||||
|
||||
def __init__(self, qlib_dir: Path | None = None) -> None:
|
||||
# 默认落盘到 data/qlib（与 storage.qlib_dir 一致）；可注入临时目录便于测试
|
||||
self.qlib_dir = qlib_dir or _default_qlib_dir()
|
||||
|
||||
def run_factor_test(
|
||||
self, daily: pd.DataFrame, spec: ResearchSpec, horizon_days: int = 21
|
||||
) -> FactorTestReport:
|
||||
raise NotImplementedError(_MSG)
|
||||
# 因子评估与数据无关，直接复用共享实现（与 LocalEngine 同口径）
|
||||
report, _panels = run_spec_factor_test(daily, spec, horizon_days)
|
||||
return report
|
||||
|
||||
def run_backtest(self, daily: pd.DataFrame, spec: ResearchSpec) -> BacktestResult:
|
||||
raise NotImplementedError(_MSG)
|
||||
panels = build_factor_panels(daily, spec.factors)
|
||||
score = composite_score(panels)
|
||||
|
||||
self.qlib_dir.mkdir(parents=True, exist_ok=True)
|
||||
uri = build_qlib_dataset(daily, self.qlib_dir)
|
||||
ensure_qlib_init(uri)
|
||||
|
||||
symbols = [str(s) for s in daily["symbol"].unique()]
|
||||
start, end = spec.period
|
||||
close = load_close_panel(uri, symbols, start, end)
|
||||
if close.empty:
|
||||
raise RuntimeError(
|
||||
"QlibDataset 读取为空：请检查 build_qlib_dataset 落盘与 provider_uri"
|
||||
)
|
||||
close = close.sort_index()
|
||||
|
||||
result = TopKBacktestRunner(spec, score, close).run()
|
||||
result.config_snapshot = spec.model_dump(mode="json")
|
||||
note = _ENGINE_NOTE
|
||||
result.unimplemented = [note, *result.unimplemented]
|
||||
return result
|
||||
|
||||
|
||||
def _default_qlib_dir() -> Path:
|
||||
from app.core.config import PROJECT_ROOT
|
||||
|
||||
return PROJECT_ROOT / "data" / "qlib"
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Qlib 数据落盘 Provider（qlib_adapter 内部，业务层不直接 import）。
|
||||
|
||||
把本地行情长表（来自 SQLite，见 quant.service.bars_to_daily_df）按 qlib 0.9.8
|
||||
的二进制格式导出到指定目录（provider_uri），供 qlib D.features 读取。
|
||||
|
||||
格式要点（来自实测 qlib 0.9.8.dev32 与源码 file_storage）：
|
||||
- instruments 文件 3 列：instrument\\tstart\\tend（instrument 全小写）
|
||||
- feature bin 布局：首 4 字节 float32 = 该股票在全局日历中的起始下标，
|
||||
其后每个交易日一个 float32（停牌/缺失为 NaN）
|
||||
- provider_uri 需以 {"day": <目录>} 传入 qlib.init
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
# qlib 消费侧字段（Alpha158 依赖：close/open/high/low/volume/amount/vwap/factor）
|
||||
DUMP_FIELDS = ["open", "high", "low", "close", "volume", "amount", "vwap", "factor"]
|
||||
|
||||
|
||||
def _synth_vwap(daily: pd.DataFrame) -> pd.Series:
|
||||
"""vwap = amount / volume（真实口径）；volume 为 0 时取 close。"""
|
||||
vol = daily["volume"].replace(0, pd.NA)
|
||||
vwap = daily["amount"] / vol
|
||||
return vwap.fillna(daily["close"])
|
||||
|
||||
|
||||
def build_qlib_dataset(
|
||||
daily: pd.DataFrame,
|
||||
out_dir: Path,
|
||||
*,
|
||||
fields: tuple[str, ...] = DUMP_FIELDS,
|
||||
add_vwap: bool = True,
|
||||
) -> Path:
|
||||
"""把长表行情落盘为 qlib 数据集目录；返回 out_dir（provider_uri）。"""
|
||||
if daily.empty:
|
||||
raise ValueError("无行情数据可导出 Qlib 数据集")
|
||||
out_dir = Path(out_dir)
|
||||
cal_file = out_dir / "calendars" / "day.txt"
|
||||
cal_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
(out_dir / "instruments").mkdir(exist_ok=True)
|
||||
feat_root = out_dir / "features"
|
||||
|
||||
df = daily.copy()
|
||||
if add_vwap and "amount" in df.columns and "volume" in df.columns:
|
||||
df["vwap"] = _synth_vwap(df)
|
||||
|
||||
calendar = sorted(df["trade_date"].unique())
|
||||
if not isinstance(calendar[0], (pd.Timestamp, np.datetime64)):
|
||||
calendar = pd.to_datetime(calendar).sort_values().tolist()
|
||||
cal_idx = {day: i for i, day in enumerate(calendar)}
|
||||
cal_file.write_text(
|
||||
"\n".join(pd.Timestamp(d).strftime("%Y-%m-%d") for d in calendar), encoding="utf-8"
|
||||
)
|
||||
|
||||
inst_lines: list[str] = []
|
||||
for symbol, group in df.groupby("symbol"):
|
||||
symbol = str(symbol).lower()
|
||||
dates = sorted(pd.to_datetime(group["trade_date"]))
|
||||
start_idx = cal_idx[dates[0]]
|
||||
end_idx = cal_idx[dates[-1]]
|
||||
inst_lines.append(
|
||||
f"{symbol}\t{pd.Timestamp(dates[0]).strftime('%Y-%m-%d')}"
|
||||
f"\t{pd.Timestamp(dates[-1]).strftime('%Y-%m-%d')}"
|
||||
)
|
||||
sym_dir = feat_root / symbol
|
||||
sym_dir.mkdir(parents=True, exist_ok=True)
|
||||
per_date = group.set_index(pd.to_datetime(group["trade_date"]))
|
||||
for field in fields:
|
||||
if field not in per_date.columns:
|
||||
continue
|
||||
series = per_date[field]
|
||||
values = np.full(len(calendar), np.nan, dtype=np.float32)
|
||||
for day, val in series.items():
|
||||
if pd.isna(val):
|
||||
continue
|
||||
try:
|
||||
values[cal_idx[day]] = float(val)
|
||||
except (TypeError, ValueError, KeyError):
|
||||
continue
|
||||
payload = np.hstack(
|
||||
[np.asarray([float(start_idx)], dtype=np.float32), values[start_idx : end_idx + 1]]
|
||||
).astype("<f")
|
||||
(sym_dir / f"{field}.day.bin").write_bytes(payload.tobytes())
|
||||
|
||||
(out_dir / "instruments" / "all.txt").write_text("\n".join(inst_lines), encoding="utf-8")
|
||||
return out_dir
|
||||
Reference in New Issue
Block a user