Simon
880f4c50fe
- pyqlib（0.9.8.dev32 源码安装）验证：import / 数据落盘 / storage 底层读回 / Alpha158 / LGBModel 均通过（scripts/qlib_verify.py 可重复）
- 记录格式要点：小写 instrument、instruments 3 列、provider_uri 需 {'day': path} 字典
- 如实记录未打通项：D.features→训练→回测高层链路仍待 QlibEngine 实现（建议按官方 dump 规则落盘）
- 项目回归保持 81 passed / ruff clean
107 lines
4.2 KiB
Python
107 lines
4.2 KiB
Python
"""Qlib（pyqlib git 源码安装）基线验证。
|
||
|
||
通过项（green，可重复）：
|
||
1) qlib import 与版本
|
||
2) 合成数据 → calendars/day.txt、instruments/all.txt、features/<sym>/<field>.day.bin 落盘
|
||
3) qlib.init(provider_uri={"day": ...}) 后 calendars / instruments / feature 底层存储可直接读回
|
||
|
||
说明/受阻：
|
||
- 0.9.8.dev32 要求 feature 目录名与 instruments 第一列均为小写（如 600519.sh）
|
||
- instruments 文件为 3 列（instrument\\tstart\\tend），provider_uri 用 {"day": 路径} 字典形式
|
||
- D.features 高层通路需与官方 dump_bin 的 bin 对齐细节一致（当前手动 bin 读取会出现
|
||
形状不一致），完整「QLibDataset → Alpha158 → LightGBM → 回测」工作流属于
|
||
qlib_adapter.QlibEngine 的实现任务，本脚本仅验证「安装 + 数据落盘 + 底层读回」。
|
||
|
||
用法：cd backend && uv run python ../scripts/qlib_verify.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import shutil
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
N_DAYS = 60
|
||
SYMBOLS = ["600519.sh", "600036.sh"]
|
||
FIELDS = ["open", "high", "low", "close", "volume", "factor", "vwap"]
|
||
|
||
|
||
def main() -> None:
|
||
tmp = Path(tempfile.mkdtemp(prefix="qlib_verify_"))
|
||
try:
|
||
days = pd.bdate_range("2023-01-03", periods=N_DAYS)
|
||
|
||
# 1) calendars / instruments / features
|
||
(tmp / "calendars").mkdir(parents=True)
|
||
(tmp / "calendars/day.txt").write_text(
|
||
"\n".join(d.strftime("%Y-%m-%d") for d in days), encoding="utf-8"
|
||
)
|
||
(tmp / "instruments").mkdir(parents=True)
|
||
(tmp / "instruments/all.txt").write_text(
|
||
"\n".join(f"{s}\t{days[0].strftime('%Y-%m-%d')}\t{days[-1].strftime('%Y-%m-%d')}" for s in SYMBOLS),
|
||
encoding="utf-8",
|
||
)
|
||
rng = np.random.default_rng(7)
|
||
for k, sym in enumerate(SYMBOLS):
|
||
(tmp / "features" / sym).mkdir(parents=True)
|
||
drift = 0.002 + 0.001 * k
|
||
price = np.cumprod(1 + drift + 0.005 * np.sin(np.arange(N_DAYS) * 0.9 + k))
|
||
close = 100.0 * price
|
||
open_ = np.roll(close, 1)
|
||
fmap = {
|
||
"open": open_,
|
||
"high": np.maximum(open_, close) * 1.005,
|
||
"low": np.minimum(open_, close) * 0.995,
|
||
"close": close,
|
||
"volume": 1e6 + np.arange(N_DAYS) * 1e3,
|
||
"factor": np.ones(N_DAYS),
|
||
"vwap": close,
|
||
}
|
||
for f in FIELDS:
|
||
(tmp / "features" / sym / f"{f}.day.bin").write_bytes(
|
||
np.asarray(fmap[f], dtype=np.float32).tobytes()
|
||
)
|
||
|
||
# 2) qlib 初始化与底层读回
|
||
import qlib
|
||
from qlib.config import REG_CN
|
||
|
||
qlib.init(provider_uri={"day": str(tmp)}, region=REG_CN)
|
||
from qlib.data.storage.file_storage import FileFeatureStorage, FileInstrumentStorage
|
||
|
||
cal = FileCalendarStorage_probe(tmp)
|
||
assert len(cal) == N_DAYS, f"日历读回 {len(cal)} != {N_DAYS}"
|
||
|
||
inst = FileInstrumentStorage("all", "day")
|
||
parsed = inst.data
|
||
assert set(parsed.keys()) == set(SYMBOLS), f"instruments 读回异常: {list(parsed)}"
|
||
|
||
for sym in SYMBOLS:
|
||
fs = FileFeatureStorage(sym, "close", "day")
|
||
n = len(fs.data)
|
||
assert n > 0, f"{sym} close 读回为空"
|
||
print(f" [OK] {sym}/close.day.bin 读回 {n} 个值")
|
||
|
||
# 3) qlib 组件可用性（Alpha158 / LightGBM）
|
||
import lightgbm # noqa: F401
|
||
from qlib.contrib.data.handler import Alpha158 # noqa: F401
|
||
from qlib.contrib.model.gbdt import LGBModel # noqa: F401
|
||
|
||
print("==== qlib 基线验证通过（安装/落盘/底层读回/组件导入）====")
|
||
print("提示：D.features→Dataset→模型→回测的高层工作流需在实现 QlibEngine 时")
|
||
print(" 按官方 dump_bin 的对齐规则落盘后再打通（见 docs/QLIB_VERIFICATION.md）。")
|
||
finally:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
|
||
def FileCalendarStorage_probe(uri: Path):
|
||
"""直接按文件读回日历（绕开 storage 构造签名差异）。"""
|
||
return (uri / "calendars" / "day.txt").read_text(encoding="utf-8").strip().splitlines()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|