Simon
b8f67f99ae
- 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
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
"""QlibEngine 数据管线测试：落盘格式、D.features 读回、端到端回测（合成数据、临时 qlib 目录）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
from app.domain.entities.research import ResearchSpec, UniverseSpec
|
|
from app.quant.qlib_adapter.dataset import load_close_panel
|
|
from app.quant.qlib_adapter.engine import QlibEngine
|
|
from app.quant.qlib_adapter.provider import build_qlib_dataset
|
|
|
|
from conftest_quant import synthetic_daily
|
|
|
|
|
|
def _daily5(n: int = 220) -> 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="<f")[0] == pytest.approx(0.0)
|
|
assert len(raw) // 4 - 1 == 200 # 数据长度 = 日历交易日数
|
|
|
|
def test_bin_with_late_listing(self, tmp_path: Path) -> 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="<f")[0] == pytest.approx(50.0)
|
|
|
|
|
|
class TestQlibDatasetRead:
|
|
def test_load_close_panel_roundtrip(self, tmp_path: Path) -> 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 为正
|