diff --git a/backend/app/quant/qlib_adapter/dataset.py b/backend/app/quant/qlib_adapter/dataset.py new file mode 100644 index 0000000..1388d25 --- /dev/null +++ b/backend/app/quant/qlib_adapter/dataset.py @@ -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() diff --git a/backend/app/quant/qlib_adapter/engine.py b/backend/app/quant/qlib_adapter/engine.py index e22a1a5..0caf36f 100644 --- a/backend/app/quant/qlib_adapter/engine.py +++ b/backend/app/quant/qlib_adapter/engine.py @@ -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" diff --git a/backend/app/quant/qlib_adapter/provider.py b/backend/app/quant/qlib_adapter/provider.py new file mode 100644 index 0000000..bd2d7bc --- /dev/null +++ b/backend/app/quant/qlib_adapter/provider.py @@ -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(" pd.DataFrame: + drifts = { + "600519.SH": 0.004, + "600036.SH": 0.002, + "601318.SH": 0.001, + "000001.SZ": -0.001, + "600030.SH": -0.003, + } + return synthetic_daily(drifts, n=n) + + +def _spec(**kw) -> ResearchSpec: + base = dict( + type="backtest", + universe=UniverseSpec(exclude_st=False, min_listing_days=0), + factors=[{"name": "momentum_60", "weight": 1.0}], + selection={"top_n": 2}, + rebalance="monthly", + period=(date(2024, 5, 1), date(2024, 8, 31)), + ) + base.update(kw) + return ResearchSpec.model_validate(base) + + +class TestProviderFormat: + def test_bin_has_start_index_header(self, tmp_path: Path) -> None: + daily = synthetic_daily({"600519.SH": 0.003, "600036.SH": -0.001}, n=200) + uri = build_qlib_dataset(daily, tmp_path) + + cal = (uri / "calendars" / "day.txt").read_text().strip().splitlines() + assert len(cal) == 200 + inst_line = (uri / "instruments" / "all.txt").read_text().strip().splitlines()[0] + assert inst_line.count("\t") == 2 # 3 列，无 TYPE 列 + + # 落盘目录用小写 instrument；bin 首 4 字节 = 起始日历下标 + raw = (uri / "features" / "600519.sh" / "close.day.bin").read_bytes() + assert np.frombuffer(raw[:4], dtype=" None: + daily = synthetic_daily({"600000.SH": 0.002, "600001.SH": 0.001}, n=120) + days = daily["trade_date"].unique() + merged = pd.concat( + [ + daily[daily["symbol"] == "600000.SH"], + daily[(daily["symbol"] == "600001.SH") & (daily["trade_date"] >= days[50])], + ] + ) + uri = build_qlib_dataset(merged, tmp_path) + raw = (uri / "features" / "600001.sh" / "close.day.bin").read_bytes() + assert np.frombuffer(raw[:4], dtype=" None: + daily = synthetic_daily({"600519.SH": 0.003, "600036.SH": -0.001}, n=120) + uri = build_qlib_dataset(daily, tmp_path) + panel = load_close_panel( + uri, ["600519.SH", "600036.SH"], date(2024, 2, 1), date(2024, 5, 31) + ) + assert len(panel) > 0 + assert set(panel.columns) == {"600519.SH", "600036.SH"} + + src = daily[daily["symbol"] == "600519.SH"].set_index("trade_date")["close"] + src.index = pd.to_datetime(src.index) + target = panel["600519.SH"] + overlap = src.index.intersection(target.index) + assert len(overlap) > 10 + rel = (target.loc[overlap] / src.loc[overlap].astype(float) - 1).abs().max() + assert rel < 1e-3 # qlib 存储为 float32，允许微小舍入 + + +class TestQlibEngine: + def test_backtest_via_qlib_pipeline(self, tmp_path: Path) -> None: + daily = _daily5() + result = QlibEngine(qlib_dir=tmp_path).run_backtest(daily, _spec()) + + assert result.summary.total_return_pct > 0 # 强趋势下动量择股盈利 + assert len(result.equity_curve) > 50 + assert any("QlibEngine v1" in u for u in result.unimplemented) + assert result.config_snapshot["factors"][0]["name"] == "momentum_60" + assert (tmp_path / "features").is_dir() # 数据确实落盘到 qlib 目录 + + def test_factor_test_shared_path(self, tmp_path: Path) -> None: + daily = _daily5() + spec = _spec() + spec.type = "factor_test" + report = QlibEngine(qlib_dir=tmp_path).run_factor_test(daily, spec) + assert report.factor_name == "momentum_60" + assert report.sample_days > 5 + assert report.ic_mean > 0 # 5 只强趋势股票的截面动量 IC 为正 diff --git a/docs/USAGE.md b/docs/USAGE.md index 59f600a..590e1dd 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -231,9 +231,11 @@ Job 状态机与 Experiment 归档、Agent 工具白名单与编排、API 端到 ## 9. 已知限制与说明 -1. **Qlib 引擎**：本开发机为 Linux aarch64 + Python 3.12，pyqlib 无匹配 wheel，无法安装。 - 默认研究引擎为自研轻量实现（pandas）；`backend/app/quant/qlib_adapter/` 保留桥接， - 在受支持平台（x86_64/macOS/Windows）安装 pyqlib 后可填充实现，业务层无感。 +1. **Qlib 引擎（v1 数据管线）**：pyqlib 已通过源码安装可用（qlib 0.9.8.dev32，见 + docs/QLIB_VERIFICATION.md）。`app/quant/qlib_adapter/` 提供 QlibEngine：本地行情按官方 + 二进制格式落盘 QlibDataset → `D.features` 读取 → TopK 因子回测（与 LocalEngine 同记账 + 规则）。默认研究引擎仍为 LocalEngine；切换只需向 ResearchService 注入 QlibEngine。 + Alpha158 特征 + LightGBM 预测信号（walk-forward）为下一步 TODO。 2. **数据规模**：仓库自带示例数据为 20 只权重股 2023–2024 日线；`sync --all` 可扩展 全市场，注意耗时与 Tushare 积分限制。 3. **回测为近似建模**：涨跌停按收盘相对上一有效收盘判定、成交假设调仓日收盘，未建模