simon/qlib
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backend): Phase 2 研究引擎 — ResearchSpec / 因子 / 评估 / 低频回测 / 引擎抽象

Browse Source 
- domain：ResearchSpec（universe/factors/selection/rebalance/costs 校验）+ 标准化 BacktestResult / FactorTestReport
- 因子引擎：注册表 + 元数据，内置 9 个行情因子（momentum/volatility/量比/乖离/反转），支持自定义注册；只用行情字段规避未来函数
- 评估：横截面 IC / RankIC（rank+pearson 免 scipy）/ ICIR / 分层收益
- 回测：TopK 等权低频，无未来函数记账（t 收盘成交、自 t+1 计收益），成本/涨跌停/停牌约束，未建模项显式写入 unimplemented（AGENT §24）
- 引擎抽象 QuantEngine + LocalEngine（pandas 默认实现）；qlib_adapter 桥接占位 —— pyqlib 无 aarch64+cp312 wheel（ROADMAP 已备注）
- 真实链路冒烟：600519 2024 月度动量回测闭环产出标准结果
- 测试 60 passed / ruff clean
This commit is contained in:
Simon
2026-09-06 17:08:00 +08:00
parent 2da234220a
commit e9f59d3cf8
12 changed files with 1388 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/domain/entities/research.py
+169
View File
@@ -0,0 +1,169 @@
"""研究领域对象：Research Specification、标准化研究结果。
原则（AGENT.md §16/§21/§24、ARCHITECTURE §14）：
- 前端 / Agent / 后端统一经 Research Specification 描述任务，禁止直接拼引擎配置
- 回测结果一律标准化为 BacktestResult；未建模的成本/市场约束显式列在
unimplemented，禁止默认「无成本 / 永远可成交」假设
"""
from __future__ import annotations
from datetime import date
from pydantic import BaseModel, Field, field_validator, model_validator
# ---------- Research Specification ----------
class UniverseSpec(BaseModel):
"""股票池口径。MVP：市场 + 过滤条件；指数成分等 Phase 3 扩展。"""
market: str = Field(default="CN_A", description="CN_A / CN_B / ...")
exclude_st: bool = True
exclude_suspended: bool = True
min_listing_days: int = Field(default=250, ge=0, description="上市至少 N 个自然日")
class FactorSpec(BaseModel):
"""引用一个已注册因子并给定权重。"""
name: str
weight: float = Field(default=1.0, gt=0)
class SelectionSpec(BaseModel):
"""选股方式。MVP：按加权因子得分取 Top N 等权。"""
top_n: int = Field(default=30, ge=1, le=1000)
class CostSpec(BaseModel):
"""交易成本模型（单边比例）。
buy = commission + slippage；sell = commission + stamp_tax + slippage。
"""
commission_rate: float = Field(default=0.0003, ge=0, le=0.01)
stamp_tax_rate: float = Field(default=0.0005, ge=0, le=0.01)
slippage_rate: float = Field(default=0.001, ge=0, le=0.05)
benchmark: str = Field(default="000300.SH", description="对照基准指数代码")
class ResearchSpec(BaseModel):
"""一次研究的完整描述。type 决定执行路径。"""
type: str = Field(default="backtest", pattern="^(factor_test|backtest)$")
universe: UniverseSpec = UniverseSpec()
factors: list[FactorSpec] = Field(min_length=1)
selection: SelectionSpec = SelectionSpec()
rebalance: str = Field(default="monthly", pattern="^(weekly|monthly)$")
period: tuple[date, date]
costs: CostSpec = CostSpec()
initial_capital: float = Field(default=1_000_000.0, gt=0)
@field_validator("period")
@classmethod
def _period_ordered(cls, period: tuple[date, date]) -> tuple[date, date]:
if period[0] >= period[1]:
raise ValueError("period 必须满足 start < end")
return period
@model_validator(mode="after")
def _no_duplicate_factors(self) -> ResearchSpec:
names = [f.name for f in self.factors]
if len(set(names)) != len(names):
raise ValueError("factors 存在重复因子名")
return self
# ---------- 回测结果 ----------
class CurvePoint(BaseModel):
date: date
value: float
class MonthlyReturn(BaseModel):
year: int
month: int
return_pct: float # 百分数，如 3.2 表示 +3.2%
class YearlyReturn(BaseModel):
year: int
return_pct: float
class BacktestSummary(BaseModel):
start: date
end: date
initial_capital: float
final_equity: float
total_return_pct: float
annual_return_pct: float
sharpe: float
max_drawdown_pct: float
volatility_pct: float
win_rate_pct: float
total_trades: int
avg_turnover_pct: float
benchmark_return_pct: float | None = None
class Trade(BaseModel):
entry_date: date
exit_date: date
symbol: str
entry_price: float
exit_price: float
return_pct: float
class Position(BaseModel):
date: date
symbol: str
weight: float
class BacktestResult(BaseModel):
"""标准化回测结果（ARCHITECTURE §14）。前端只依赖该结构。"""
summary: BacktestSummary
equity_curve: list[CurvePoint]
drawdown: list[CurvePoint]
monthly_returns: list[MonthlyReturn]
yearly_returns: list[YearlyReturn]
positions: list[Position]
trades: list[Trade]
turnover_pct: float
unimplemented: list[str] = Field(
default_factory=list,
description="本结果中未建模的约束（AGENT §24：必须显式标注，禁止假装支持）",
)
config_snapshot: dict = Field(default_factory=dict, description="复现用完整配置快照")
# ---------- 因子测试结果 ----------
class QuantileReturn(BaseModel):
"""分层收益：按因子值升序分 N 层后各层等权组合的区间收益。"""
quantile: int
return_pct: float
class FactorTestReport(BaseModel):
factor_name: str
ic_mean: float
icir: float
rank_ic_mean: float
positive_ratio_pct: float
quantile_returns: list[QuantileReturn]
spread_quantile: int | None = Field(
default=None, description="分层价差 = 最高层收益 - 最低层收益（若多头/空头语义适用）"
)
sample_days: int
unimplemented: list[str] = Field(default_factory=list)
config_snapshot: dict = Field(default_factory=dict)
backend/app/quant/engine.py
+48
View File
@@ -0,0 +1,48 @@
"""研究引擎抽象与默认实现（业务层依赖本接口，可替换引擎）。
切换引擎（如未来在支持平台启用 Qlib）只需注入不同实现 —— 业务代码不变。
"""
from __future__ import annotations
from typing import Protocol
import pandas as pd
from app.domain.entities.research import BacktestResult, FactorTestReport, ResearchSpec
from app.quant.local_engine import (
TopKBacktestRunner,
build_factor_panels,
composite_score,
run_spec_factor_test,
)
class QuantEngine(Protocol):
"""研究引擎端口：因子面板构建 / 因子测试 / 回测。"""
name: str
def run_factor_test(
self, daily: pd.DataFrame, spec: ResearchSpec, horizon_days: int = 21
) -> FactorTestReport: ...
def run_backtest(self, daily: pd.DataFrame, spec: ResearchSpec) -> BacktestResult: ...
class LocalEngine:
"""默认引擎：纯 pandas 实现（无 Qlib 依赖），见 local_engine.py 的纪律说明。"""
name = "local"
def run_factor_test(
self, daily: pd.DataFrame, spec: ResearchSpec, horizon_days: int = 21
) -> FactorTestReport:
report, _panels = run_spec_factor_test(daily, spec, horizon_days)
return report
def run_backtest(self, daily: pd.DataFrame, spec: ResearchSpec) -> BacktestResult:
panels = build_factor_panels(daily, spec.factors)
score = composite_score(panels)
close = daily.pivot(index="trade_date", columns="symbol", values="close").sort_index()
return TopKBacktestRunner(spec, score, close).run()
backend/app/quant/evaluation.py
+102
View File
@@ -0,0 +1,102 @@
"""因子评估：截面 IC / RankIC / ICIR / 分层收益（AGENT.md §23）。
输入均为 面板（index=trade_date, columns=symbol）：
- factor：因子值
- forward：未来 horizon 期收益（每行是「当日可见、未来实现」的收益，用于横截面相关）
任何消费侧必须保证 factor 行 t 只用 <= t 的信息，forward 是 t 之后的实现 ——
两者错位即未来函数，由数据构造方负责（本模块只做统计）。
"""
from __future__ import annotations
import math
import pandas as pd
from app.domain.entities.research import FactorTestReport, QuantileReturn
MIN_CROSS_SECTION = 5 # 少于该样本数的日期跳过（避免噪声 IC）
def cross_sectional_ic(
factor: pd.DataFrame, forward: pd.DataFrame, method: str = "pearson"
) -> pd.Series:
"""逐日横截面相关（pearson=IC；spearman=RankIC 用 rank+pearson 等价，免 scipy）。"""
rows: dict[pd.Timestamp, float] = {}
idx = factor.index.intersection(forward.index)
for dt in idx:
f = factor.loc[dt].dropna()
r = forward.loc[dt].reindex(f.index)
pair = pd.concat([f, r], axis=1).dropna()
if len(pair) < MIN_CROSS_SECTION:
continue
a, b = pair.iloc[:, 0], pair.iloc[:, 1]
if method == "spearman":
a, b = a.rank(), b.rank()
ic = a.corr(b)
if math.isfinite(ic):
rows[dt] = float(ic)
return pd.Series(rows, dtype=float).sort_index()
def _icir(series: pd.Series) -> float:
if len(series) < 2:
return 0.0
std = float(series.std(ddof=1))
if std == 0 or math.isnan(std):
return 0.0
return float(series.mean() / std * math.sqrt(len(series)))
def quantile_returns(factor: pd.DataFrame, forward: pd.DataFrame, quantiles: int = 5) -> pd.Series:
"""逐日按因子值升序分层，返回各层平均未来收益（跨日再平均）。"""
acc = {q: [] for q in range(quantiles)}
idx = factor.index.intersection(forward.index)
for dt in idx:
f = factor.loc[dt].dropna()
r = forward.loc[dt].reindex(f.index)
pair = pd.concat([f, r], axis=1).dropna()
if len(pair) < quantiles * 2:
continue
try:
labels = pd.qcut(pair.iloc[:, 0], quantiles, labels=False, duplicates="drop")
except ValueError:
continue
grouped = pair.iloc[:, 1].groupby(labels).mean()
for q, val in grouped.items():
acc[int(q)].append(float(val))
means = {q: (sum(v) / len(v) if v else float("nan")) for q, v in acc.items()}
return pd.Series(means)
def run_factor_test(
factor: pd.DataFrame,
forward: pd.DataFrame,
*,
factor_name: str = "",
quantiles: int = 5,
) -> FactorTestReport:
ic = cross_sectional_ic(factor, forward, "pearson")
rank_ic = cross_sectional_ic(factor, forward, "spearman")
q_ret = quantile_returns(factor, forward, quantiles)
spread: int | None = None
valid = [q for q in range(quantiles) if q in q_ret.index and not math.isnan(q_ret[q])]
if len(valid) >= 2 and q_ret[valid[-1]] > q_ret[valid[0]]:
spread = int(valid[-1]) # 高分层 > 低分层时报告层号
report = FactorTestReport(
factor_name=factor_name or "factor",
ic_mean=float(ic.mean()) if len(ic) else 0.0,
icir=_icir(ic),
rank_ic_mean=float(rank_ic.mean()) if len(rank_ic) else 0.0,
positive_ratio_pct=float((ic > 0).mean() * 100) if len(ic) else 0.0,
quantile_returns=[
QuantileReturn(quantile=int(q), return_pct=round(float(v) * 100, 4))
for q, v in sorted(q_ret.items())
],
spread_quantile=spread,
sample_days=len(ic),
)
return report
backend/app/quant/factors.py
+184
View File
@@ -0,0 +1,184 @@
"""因子引擎：因子注册表、元数据与计算（Phase 2，低频选股因子）。
数据形态：行情长表 DataFrame（列 symbol/trade_date/close/high/low/volume/amount），
因子计算返回 面板 DataFrame（index=trade_date，columns=symbol）。
所有内置因子只用行情字段（无财务），天然规避未来函数；财务因子接入时必须以
announce_date 控制可见性（见 domain.entities.market.FinancialIndicator）。
"""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import pandas as pd
@dataclass(frozen=True)
class FactorDef:
"""因子元数据（AGENT.md §22 要求逐项明确）。"""
name: str
description: str
formula: str
frequency: str = "daily"
lookback: int = 20
direction: str = "higher_is_better" # | lower_is_better
requires: tuple[str, ...] = ("close",)
FactorFn = Callable[[dict[str, pd.DataFrame]], pd.DataFrame]
class FactorError(ValueError):
pass
_REGISTRY: dict[str, tuple[FactorDef, FactorFn]] = {}
def register(defn: FactorDef) -> Callable[[FactorFn], FactorFn]:
"""装饰器：注册自定义因子。"""
def deco(fn: FactorFn) -> FactorFn:
if defn.name in _REGISTRY:
raise FactorError(f"因子 {defn.name} 已注册")
_REGISTRY[defn.name] = (defn, fn)
return fn
return deco
def get_factor(name: str) -> tuple[FactorDef, FactorFn]:
if name not in _REGISTRY:
raise FactorError(f"未知因子：{name}（可用：{', '.join(sorted(_REGISTRY))}）")
return _REGISTRY[name]
def list_factors() -> list[FactorDef]:
return [d for d, _fn in sorted(_REGISTRY.values(), key=lambda x: x[0].name)]
def compute_factor(name: str, daily: pd.DataFrame) -> tuple[FactorDef, pd.DataFrame]:
"""计算因子：从行情长表提取所需字段的面板后调用因子函数。"""
defn, fn = get_factor(name)
fields: dict[str, pd.DataFrame] = {}
for col in defn.requires:
panel = daily.pivot(index="trade_date", columns="symbol", values=col).sort_index()
panel.index = pd.to_datetime(panel.index)
fields[col] = panel
return defn, fn(fields)
# ---------- 内置因子 ----------
def _rolling_return(prices: pd.DataFrame, lookback: int) -> pd.DataFrame:
return prices / prices.shift(lookback) - 1.0
def _rolling_vol(prices: pd.DataFrame, lookback: int) -> pd.DataFrame:
return prices.pct_change().rolling(lookback).std()
@register(
FactorDef("momentum_20", "过去 20 个交易日收益率", "close / close.shift(20) - 1", lookback=20)
)
def _momentum_20(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return _rolling_return(fields["close"], 20)
@register(
FactorDef("momentum_60", "过去 60 个交易日收益率", "close / close.shift(60) - 1", lookback=60)
)
def _momentum_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return _rolling_return(fields["close"], 60)
@register(
FactorDef(
"momentum_120", "过去 120 个交易日收益率", "close / close.shift(120) - 1", lookback=120
)
)
def _momentum_120(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return _rolling_return(fields["close"], 120)
@register(
FactorDef(
"volatility_20",
"过去 20 个交易日收益率波动率",
"std(pct_change, 20)",
lookback=20,
direction="lower_is_better",
)
)
def _volatility_20(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return _rolling_vol(fields["close"], 20)
@register(
FactorDef(
"volatility_60",
"过去 60 个交易日收益率波动率",
"std(pct_change, 60)",
lookback=60,
direction="lower_is_better",
)
)
def _volatility_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return _rolling_vol(fields["close"], 60)
@register(
FactorDef(
"close_to_high_60",
"收盘价相对 60 日最高价的接近程度",
"close / rolling_max(high, 60)",
lookback=60,
requires=("close", "high"),
)
)
def _close_to_high_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
high = fields["high"]
return fields["close"] / high.rolling(60).max()
@register(
FactorDef(
"volume_ratio_5_60",
"量比：5 日均量 / 60 日均量",
"mean(volume, 5) / mean(volume, 60)",
lookback=60,
requires=("volume",),
)
)
def _volume_ratio_5_60(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
vol = fields["volume"]
return vol.rolling(5).mean() / vol.rolling(60).mean()
@register(
FactorDef(
"ma_bias_20",
"20 日均线乖离率",
"(close - ma(close, 20)) / ma(close, 20)",
lookback=20,
)
)
def _ma_bias_20(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
close = fields["close"]
ma = close.rolling(20).mean()
return (close - ma) / ma
@register(
FactorDef(
"reversal_5",
"短期反转：过去 5 日收益率取负（越低越接近超跌）",
"-1 * (close / close.shift(5) - 1)",
lookback=5,
)
)
def _reversal_5(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return -1.0 * _rolling_return(fields["close"], 5)
backend/app/quant/local_engine.py
+338
View File
@@ -0,0 +1,338 @@
"""LocalEngine —— 默认研究引擎（纯 pandas，AGENT.md §40 简单可替换优先）。
无未来函数纪律：
- 调仓日 t 的选股只使用 <=t 的因子值与收盘价
- 成交发生在 t 收盘（价格 = close[t] ± 滑点）；t 当日组合收益用 t-1 收盘持仓结算，
调仓在 t 收盘生效、自 t+1 起计收益 —— 不存在「当日买入当日计收益」的未来函数
- 涨跌停 / 停牌约束按可达信息近似建模，未建模部分显式写入结果 unimplemented
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from datetime import date
import pandas as pd
from app.domain.entities.research import (
BacktestResult,
BacktestSummary,
CurvePoint,
FactorTestReport,
MonthlyReturn,
Position,
ResearchSpec,
Trade,
YearlyReturn,
)
from app.quant.evaluation import run_factor_test
from app.quant.factors import FactorDef, compute_factor
TRADING_DAYS = 252
_DEFAULT_UNIMPLEMENTED = [
"涨跌停按收盘价相对上一有效收盘近似判定（未建模开盘一字 / 集合竞价路径）",
"成交假设发生在调仓日收盘（未建模盘中价格路径与流动性冲击）",
]
def _limit_up_ratio(symbol: str) -> float:
"""按板块近似涨跌停幅度。"""
code = symbol[:3]
if code in {"300", "301", "688"}:
return 1.199
if code.startswith(("8", "4", "92")):
return 1.299
return 1.099
def cross_sectional_zscore(panel: pd.DataFrame) -> pd.DataFrame:
"""截面 z-score。
候选不足 2 只（如单股票池）时退化为 0：无比较基准，但保留为可候选值；
全列缺失才为 NaN（该日不可选股）。
"""
def _row_z(row: pd.Series) -> pd.Series:
valid = row.dropna()
if len(valid) == 0:
return pd.Series(float("nan"), index=row.index)
if len(valid) == 1:
return pd.Series(0.0, index=row.index)
mu, sd = valid.mean(), valid.std()
if sd == 0 or math.isnan(sd):
return pd.Series(0.0, index=row.index)
return (row - mu) / sd
return panel.apply(_row_z, axis=1)
def composite_score(
panels: list[tuple[str, pd.DataFrame, float, str]],
) -> pd.DataFrame:
"""按 (name, panel, weight, direction) 计算加权复合 zscore。
direction="lower_is_better" 的因子取负号后相加（统一为「得分高者优先」）。
"""
total = None
for _name, panel, weight, direction in panels:
z = cross_sectional_zscore(panel)
if direction == "lower_is_better":
z = -z
contribution = z * weight
total = contribution if total is None else total.add(contribution, fill_value=0)
assert total is not None
return total
def rebalance_dates(index: pd.Index, rebalance: str, start: date) -> list[pd.Timestamp]:
"""按频率取首个交易日（>= start）。"""
periods = index.to_period("M" if rebalance == "monthly" else "W")
seen: dict = {}
order: list[pd.Timestamp] = []
for ts, per in zip(index, periods, strict=True):
if per not in seen:
seen[per] = ts
order.append(ts)
return [ts for ts in order if ts.date() >= start]
@dataclass
class EngineResult:
equity: pd.Series # index=date -> equity
trades: list[Trade]
positions: list[Position]
rebalance_notional: list[float]
class TopKBacktestRunner:
"""TopK 等权、固定调仓频率的低频回测。"""
def __init__(self, spec: ResearchSpec, score: pd.DataFrame, close: pd.DataFrame) -> None:
self.spec = spec
close = close.copy()
close.index = pd.to_datetime(close.index)
self.close = close.sort_index()
self.score = score.reindex(self.close.index).sort_index()
self.costs = spec.costs
# 上一有效收盘（用于涨跌停与收益结算，处理停牌日）
self.prev_close = self.close.ffill().shift(1)
def run(self) -> BacktestResult:
end_date = self.spec.period[1]
dates = [d for d in self.close.index if self.spec.period[0] <= d.date() <= end_date]
rebal = {
d
for d in rebalance_dates(self.close.index, self.spec.rebalance, self.spec.period[0])
if d.date() <= end_date
}
cash = float(self.spec.initial_capital)
shares: dict[str, float] = {}
entry_date: dict[str, date] = {}
entry_price: dict[str, float] = {}
equity_rows: dict[pd.Timestamp, float] = {}
trades: list[Trade] = []
positions: list[Position] = []
notional: list[float] = []
def _value(d: pd.Timestamp) -> float:
total = cash
for s, qty in shares.items():
if qty <= 0:
continue
px = self.close.at[d, s] if d in self.close.index else None
if px is None or (isinstance(px, float) and math.isnan(px)):
continue # 无行情日不计该仓（停牌近似，见 unimplemented）
total += float(qty * px)
return total
for d in dates:
if d in rebal:
cash = self._rebalance(
d, cash, shares, entry_date, entry_price, trades, positions, notional
)
equity_rows[d] = _value(d)
equity = pd.Series(equity_rows).sort_index()
return self._to_result(equity, trades, positions, notional)
# ---- 调仓（t 收盘执行，自 t+1 生效） ----
def _rebalance(self, d, cash, shares, entry_date, entry_price, trades, positions, notional):
close_d = self.close.loc[d]
prev_d = self.prev_close.loc[d]
sold_notional = 0.0
# 1) 卖出：跌停或无价（停牌）持仓保留，其余卖出
for s in [s for s in shares if shares[s] > 0]:
c, p = close_d[s], prev_d[s]
if _nan(c):
continue # 停牌无价：保留
if not _nan(p) and p > 0 and c / p <= 1.0 - (_limit_up_ratio(s) - 1.0):
continue # 跌停无法卖出：保留到下一调仓
qty = shares[s]
proceeds = qty * float(c) * (1 - self.costs.slippage_rate)
fee = proceeds * (self.costs.commission_rate + self.costs.stamp_tax_rate)
cash += proceeds - fee
sold_notional += proceeds
trades.append(
Trade(
entry_date=entry_date[s],
exit_date=d.date(),
symbol=s,
entry_price=entry_price[s],
exit_price=float(c),
return_pct=(float(c) / entry_price[s] - 1.0) * 100,
)
)
shares[s] = 0.0
entry_date.pop(s, None)
entry_price.pop(s, None)
# 2) 买入：取得分最高且可买的 TopN（涨停 / 无价剔除）
score_d = self.score.loc[d].dropna()
top = score_d.sort_values(ascending=False).index.tolist()
targets: list[str] = []
for s in top:
if len(targets) >= self.spec.selection.top_n:
break
c, p = close_d[s], prev_d[s]
if _nan(c) or _nan(p) or p <= 0:
continue
if c / p >= _limit_up_ratio(s):
continue # 涨停不可追买
targets.append(s)
if targets:
budget = cash / len(targets)
for s in targets:
c = float(close_d[s])
price_in = c * (1 + self.costs.slippage_rate)
invest = budget * (1 - self.costs.commission_rate)
shares[s] = invest / price_in
entry_date[s] = d.date()
entry_price[s] = price_in
notional.append(budget)
cash -= budget * len(targets)
# 3) 记录调仓后仓位
total = cash + sum(
float(self.close.at[d, s] * qty)
for s, qty in shares.items()
if qty > 0 and not _nan(self.close.at[d, s])
)
if total > 0:
for s, qty in shares.items():
if qty > 0 and not _nan(self.close.at[d, s]):
positions.append(
Position(
date=d.date(), symbol=s, weight=float(qty * self.close.at[d, s] / total)
)
)
return cash
# ---- 指标 ----
def _to_result(self, equity, trades, positions, notional) -> BacktestResult:
start, end = equity.index[0].date(), equity.index[-1].date()
init = float(self.spec.initial_capital)
final = float(equity.iloc[-1])
rets = equity.pct_change().dropna()
n = len(rets)
total_ret = (final / init - 1.0) * 100 if init else 0.0
annual = (
((final / init) ** (TRADING_DAYS / max(n, 1)) - 1.0) * 100
if final > 0 and init > 0
else -100.0
)
mean_r, std_r = (float(rets.mean()), float(rets.std(ddof=1))) if n else (0.0, 0.0)
sharpe = mean_r / std_r * math.sqrt(TRADING_DAYS) if std_r and mean_r else 0.0
vol = std_r * math.sqrt(TRADING_DAYS) * 100
dd = (equity / equity.cummax() - 1.0).min() * 100
wins = [t for t in trades if t.return_pct > 0]
win_rate = len(wins) / len(trades) * 100 if trades else 0.0
avg_turn = (sum(notional) / len(notional) / ((init + final) / 2)) * 100 if notional else 0.0
eq_pts = [CurvePoint(date=d.date(), value=round(float(v), 2)) for d, v in equity.items()]
dd_series = (equity / equity.cummax() - 1.0) * 100
drawdown = [
CurvePoint(date=d.date(), value=round(float(v), 3)) for d, v in dd_series.items()
]
monthly: list[MonthlyReturn] = []
yearly: list[YearlyReturn] = []
if len(equity) > 1:
m = equity.resample("ME").last().pct_change().dropna()
monthly = [
MonthlyReturn(
year=int(d.year), month=int(d.month), return_pct=round(float(v) * 100, 3)
)
for d, v in m.items()
]
y = equity.resample("YE").last().pct_change().dropna()
yearly = [
YearlyReturn(year=int(d.year), return_pct=round(float(v) * 100, 3))
for d, v in y.items()
]
summary = BacktestSummary(
start=start,
end=end,
initial_capital=round(init, 2),
final_equity=round(final, 2),
total_return_pct=round(total_ret, 3),
annual_return_pct=round(annual, 3),
sharpe=round(sharpe, 3),
max_drawdown_pct=round(float(dd), 3),
volatility_pct=round(vol, 3),
win_rate_pct=round(win_rate, 2),
total_trades=len(trades),
avg_turnover_pct=round(avg_turn, 2),
)
return BacktestResult(
summary=summary,
equity_curve=eq_pts,
drawdown=drawdown,
monthly_returns=monthly,
yearly_returns=yearly,
positions=positions,
trades=trades,
turnover_pct=round(sum(notional) / max(init, 1) * 100, 2),
unimplemented=list(_DEFAULT_UNIMPLEMENTED),
config_snapshot=self.spec.model_dump(mode="json"),
)
def build_factor_panels(
daily: pd.DataFrame, factor_specs
) -> list[tuple[str, pd.DataFrame, float, str]]:
"""按 spec.factors 计算面板与权重（因子不存在即报错）。"""
panels: list[tuple[str, pd.DataFrame, float, str]] = []
for fs in factor_specs:
defn: FactorDef
defn, panel = compute_factor(fs.name, daily)
panels.append((fs.name, panel, fs.weight, defn.direction))
return panels
def run_spec_factor_test(
daily: pd.DataFrame,
spec: ResearchSpec,
horizon_days: int = 21,
) -> tuple[FactorTestReport, dict[str, pd.DataFrame]]:
"""单因子测试：因子面板 + 未来 horizon 收益 → FactorTestReport。"""
assert spec.type == "factor_test"
factor_name = spec.factors[0].name
panels = build_factor_panels(daily, spec.factors)
panel = panels[0][1]
close = daily.pivot(index="trade_date", columns="symbol", values="close").sort_index()
forward = close.shift(-horizon_days) / close - 1.0
report = run_factor_test(panel, forward, factor_name=factor_name)
return report, {factor_name: panel}
def _nan(v) -> bool:
try:
return bool(math.isnan(float(v)))
except (TypeError, ValueError):
return False
backend/app/quant/qlib_adapter/engine.py
+39
View File
@@ -0,0 +1,39 @@
"""Qlib 后端引擎（桥接占位）。
当前开发平台为 Linux aarch64 + CPython 3.12：pyqlib 官方仅提供 x86_64 /
macOS / Windows 且最高 Python 3.8 的 wheel（实测 uv 解析不可满足），因此
Qlib 实现在此平台无法安装运行。
受支持平台（如 x86_64 Linux + Python 3.11）启用方式：
cd backend && uv pip install pyqlib==0.9.7
将本模块替换为真实实现：Parquet/本地行情 → QlibDataset → Alpha158 →
LightGBM 训练/预测 → 回测，并归一化为 domain.entities.research 输出。
业务层经 app.quant.engine.QuantEngine Protocol 注入引擎，切换无需改业务代码。
"""
from __future__ import annotations
import pandas as pd
from app.domain.entities.research import BacktestResult, FactorTestReport, ResearchSpec
from app.quant.engine import QuantEngine
_MSG = (
"Qlib 引擎需要 pyqlib，当前平台（Linux aarch64 + Python 3.12）无可用 wheel。"
"请改用 LocalEngine 或在受支持平台安装 pyqlib 后实现（见本模块 docstring）。"
)
class QlibEngine(QuantEngine):
"""pyqlib 后端占位：抛 NotImplementedError 并给出启用指引。"""
name = "qlib"
def run_factor_test(
self, daily: pd.DataFrame, spec: ResearchSpec, horizon_days: int = 21
) -> FactorTestReport:
raise NotImplementedError(_MSG)
def run_backtest(self, daily: pd.DataFrame, spec: ResearchSpec) -> BacktestResult:
raise NotImplementedError(_MSG)
backend/app/quant/service.py
+93
View File
@@ -0,0 +1,93 @@
"""研究服务：把 Research Specification 编排为数据获取 + 引擎执行。
本层是业务入口：API / Agent 只能调用这里的用例（AGENT.md §16/§17），
禁止直接拼接引擎配置。数据一律经 Repository 获取（防未来函数由查询层保证）。
"""
from __future__ import annotations
from datetime import date, timedelta
import pandas as pd
from app.domain.entities.market import Stock
from app.domain.entities.research import (
BacktestResult,
FactorTestReport,
ResearchSpec,
UniverseSpec,
)
from app.domain.repositories.market import (
DailyBarRepository,
StockRepository,
)
from app.quant.engine import QuantEngine
def filter_stocks(stocks: list[Stock], universe: UniverseSpec, as_of: date) -> list[Stock]:
"""按股票池口径过滤（名称含 ST 判定 —— 名称快照为当日口径，属历史可追溯数据）。"""
out: list[Stock] = []
for s in stocks:
if s.delist_date is not None and s.delist_date < as_of:
continue
if universe.exclude_st and s.name and "ST" in s.name.upper():
continue
if (
universe.min_listing_days
and s.list_date
and (as_of - s.list_date).days < universe.min_listing_days
):
continue
out.append(s)
return out
def bars_to_daily_df(bars) -> pd.DataFrame:
"""DailyBar 列表 → 引擎长表 DataFrame（symbol/trade_date/ohlc/volume/amount）。
领域实体中的 Decimal 在此转 float，供 pandas 数值运算（保持 DataFrame 全数值列）。
"""
df = pd.DataFrame([b.model_dump() for b in bars])
if not df.empty:
for col in ("open", "high", "low", "close", "volume", "amount"):
if col in df.columns:
df[col] = df[col].astype(float)
return df
class ResearchService:
"""研究用例入口（因子测试 / 回测）。依赖注入 Repository 与引擎。"""
def __init__(
self,
stock_repo: StockRepository,
daily_repo: DailyBarRepository,
engine: QuantEngine,
) -> None:
self._stock_repo = stock_repo
self._daily_repo = daily_repo
self._engine = engine
def run_factor_test(self, spec: ResearchSpec, horizon_days: int = 21) -> FactorTestReport:
if spec.type != "factor_test":
raise ValueError("factor_test 用例需要 spec.type=factor_test")
daily = self._load_daily(spec)
return self._engine.run_factor_test(daily, spec, horizon_days=horizon_days)
def run_backtest(self, spec: ResearchSpec) -> BacktestResult:
if spec.type != "backtest":
raise ValueError("backtest 用例需要 spec.type=backtest")
daily = self._load_daily(spec)
return self._engine.run_backtest(daily, spec)
# ---- 数据装配 ----
def _load_daily(self, spec: ResearchSpec) -> pd.DataFrame:
start, end = spec.period
# 回测前预留因子 warmup（lookback≤120 交易日，取 300 自然日余量）
data_start = start - timedelta(days=300)
stocks = filter_stocks(self._stock_repo.list(), spec.universe, as_of=start)
bars: list = []
for s in stocks:
bars.extend(self._daily_repo.get_range(s.symbol, data_start, end))
return bars_to_daily_df(bars)
backend/tests/conftest_quant.py
+56
View File
@@ -0,0 +1,56 @@
"""共享测试装置：合成确定性 A 股行情（含涨跌趋势与噪声，无外部依赖）。"""
from __future__ import annotations
from decimal import Decimal
import numpy as np
import pandas as pd
def synthetic_daily(drifts: dict[str, float], n: int = 320, base: float = 100.0) -> pd.DataFrame:
"""生成多股票日线长表。
每股价格：p[j] = p[j-1] * (1 + drift + 0.012 * sin((j + i) * 0.8)) —— 确定性、
趋势 + 微幅周期噪声；含 high/low/volume/amount 供各因子使用。
"""
dates = pd.bdate_range("2024-01-01", periods=n)
rows: list[dict] = []
for i, (sym, drift) in enumerate(drifts.items()):
price = float(base)
for j, d in enumerate(dates):
ret = drift + 0.012 * np.sin((j + i) * 0.8)
prev = price
price = price * (1 + ret)
rows.append(
{
"symbol": sym,
"trade_date": d.date(),
"open": float(prev),
"high": float(price * 1.008),
"low": float(min(prev, price) * 0.992),
"close": float(price),
"volume": float(1_000_000 + j * 1000 + i * 3000),
"amount": float(price * (1_000_000 + j * 1000 + i * 3000)),
}
)
return pd.DataFrame(rows)
def bars_dataframe_to_daily_bars(daily: pd.DataFrame) -> list:
"""测试辅助：DataFrame → domain DailyBar 实体（供 service/仓储层路径测试）。"""
from app.domain.entities.market import DailyBar
return [
DailyBar(
symbol=r.symbol,
trade_date=r.trade_date,
open=Decimal(str(r.open)),
high=Decimal(str(r.high)),
low=Decimal(str(r.low)),
close=Decimal(str(r.close)),
volume=Decimal(str(r.volume)),
amount=Decimal(str(r.amount)),
)
for r in daily.itertuples()
]
backend/tests/test_quant_engine.py
+138
View File
@@ -0,0 +1,138 @@
"""LocalEngine 回测测试：主路径、成本、涨跌停/不可买约束、无未来函数构造。"""
from __future__ import annotations
from datetime import date
import pandas as pd
from app.domain.entities.market import Stock
from app.domain.entities.research import (
CostSpec,
FactorSpec,
ResearchSpec,
SelectionSpec,
UniverseSpec,
)
from app.quant.engine import LocalEngine
from app.quant.service import filter_stocks
from conftest_quant import synthetic_daily
def _spec(
top_n: int = 1,
start: date = date(2024, 3, 1),
end: date = date(2024, 10, 31),
rebalance: str = "monthly",
costs: CostSpec | None = None,
) -> ResearchSpec:
return ResearchSpec(
type="backtest",
universe=UniverseSpec(exclude_st=False, min_listing_days=0),
factors=[FactorSpec(name="momentum_20")],
selection=SelectionSpec(top_n=top_n),
rebalance=rebalance,
period=(start, end),
costs=costs or CostSpec(),
)
class TestBacktestMain:
def test_uptrend_wins_and_profits(self) -> None:
daily = synthetic_daily({"600000.SH": 0.004, "600001.SH": -0.002}, n=320)
res = LocalEngine().run_backtest(daily, _spec(top_n=1))
assert res.summary.total_return_pct > 0
assert res.summary.final_equity > res.summary.initial_capital
assert res.summary.total_trades >= 1
assert res.summary.annual_return_pct > 0
assert res.equity_curve[0].date == date(2024, 3, 1)
assert res.equity_curve[-1].date == date(2024, 10, 31)
assert res.monthly_returns
assert res.trades
# AGENT §24：未建模约束必须显式标注
assert any("涨跌停" in item for item in res.unimplemented)
assert res.config_snapshot["selection"]["top_n"] == 1
def test_costs_reduce_returns(self) -> None:
daily = synthetic_daily({"600000.SH": 0.004, "600001.SH": -0.002}, n=320)
free = CostSpec(commission_rate=0.0, stamp_tax_rate=0.0, slippage_rate=0.0)
with_cost = LocalEngine().run_backtest(daily, _spec(top_n=2, costs=CostSpec()))
without = LocalEngine().run_backtest(daily, _spec(top_n=2, costs=free))
# 有成本时收益不应高于无成本
assert with_cost.summary.total_return_pct <= without.summary.total_return_pct + 1e-6
def test_positions_and_drawdown_wellformed(self) -> None:
daily = synthetic_daily({"600000.SH": 0.004, "600001.SH": -0.001}, n=260)
res = LocalEngine().run_backtest(daily, _spec(top_n=2))
assert res.positions
weights = [p.weight for p in res.positions]
assert all(0 < w <= 1 for w in weights)
assert all(p.value <= 0 for p in res.drawdown)
assert res.summary.max_drawdown_pct <= 0
def _limit_up_scenario_daily() -> pd.DataFrame:
"""Y 在 2024-07-01 相对前一交易日跳涨 10.5%（主板涨停不可追），且动量高于 X。"""
dates = pd.bdate_range("2024-06-03", periods=46)
rows = []
for j, d in enumerate(dates):
x = 100.0 * 1.001**j
y = 100.0 * 1.003**j
# 2024-07-01 是第 21 个工作日（6/28 周五 → 7/1 周一）
if d.date() == date(2024, 7, 1):
y = y / 1.003 * 1.105 # 相对前一日 +10.5%，形成涨停
rows.append(
{
"symbol": "600001.SH",
"trade_date": d.date(),
"close": x,
"open": x,
"high": x * 1.01,
"low": x * 0.99,
"volume": 1e6,
"amount": 1e8,
}
)
rows.append(
{
"symbol": "600002.SH",
"trade_date": d.date(),
"close": y,
"open": y,
"high": y * 1.01,
"low": y * 0.99,
"volume": 1e6,
"amount": 1e8,
}
)
return pd.DataFrame(rows)
class TestLimitUpConstraint:
def test_limit_up_symbol_not_bought(self) -> None:
daily = _limit_up_scenario_daily()
res = LocalEngine().run_backtest(
daily, _spec(top_n=1, start=date(2024, 6, 3), end=date(2024, 8, 2))
)
# 7/1 调仓：Y 动量更高但涨停不可买 → 当日只能买入 X（8 月 Y 恢复可买，不断言之后）
at_0701 = [p for p in res.positions if p.date == date(2024, 7, 1)]
assert at_0701, "7/1 调仓后应记录持仓"
assert {p.symbol for p in at_0701} == {"600001.SH"}
class TestServiceFilter:
def _stock(self, symbol: str, name: str, list_date: date, delist: date | None = None) -> Stock:
return Stock(symbol=symbol, name=name, list_date=list_date, delist_date=delist)
def test_exclude_st_and_new_and_delisted(self) -> None:
stocks = [
self._stock("600001.SH", "*ST 某某", date(2000, 1, 1)),
self._stock("600002.SH", "正常公司", date(2024, 6, 1)), # 上市不足 250 天
self._stock("600003.SH", "正常公司", date(2010, 1, 1), delist=date(2023, 6, 1)),
self._stock("600004.SH", "正常公司", date(2010, 1, 1)),
]
kept = filter_stocks(
stocks, UniverseSpec(exclude_st=True, min_listing_days=250), as_of=date(2024, 8, 1)
)
assert [s.symbol for s in kept] == ["600004.SH"]
backend/tests/test_quant_factors.py
+89
View File
@@ -0,0 +1,89 @@
"""因子计算与注册表测试（合成数据、确定性断言方向与相对排序）。"""
from __future__ import annotations
import pandas as pd
from app.quant.factors import (
FactorDef,
FactorError,
compute_factor,
get_factor,
list_factors,
register,
)
from conftest_quant import synthetic_daily
def test_registry_builtins_present() -> None:
names = {f.name for f in list_factors()}
assert {"momentum_20", "momentum_60", "volatility_20", "ma_bias_20"} <= names
def test_unknown_factor_raises() -> None:
try:
compute_factor("not_a_factor", synthetic_daily({"A": 0.0}))
except FactorError:
return
raise AssertionError("应抛 FactorError")
def test_momentum_ordering_matches_drift() -> None:
daily = synthetic_daily({"AAA": 0.002, "BBB": 0.0, "CCC": -0.002}, n=160)
_defn, panel = compute_factor("momentum_20", daily)
tail = panel.iloc[-1]
assert tail["AAA"] > tail["BBB"] > tail["CCC"]
assert tail["AAA"] > 0 # 上涨股 20 日动量为正
def test_volatility_ranks_noise() -> None:
# 手写：SMOOTH 每日 +0.2%；WILD 在 ±5% 间摆动 → WILD 的 20 日波动率应显著更高
from datetime import date, timedelta
dates = [date(2024, 1, 1) + timedelta(days=i) for i in range(120)]
rows = []
smooth, wild = 100.0, 100.0
for j, d in enumerate(dates):
smooth *= 1.002
wild *= 1.05 if j % 2 == 0 else 0.95
for sym, px in (("SMOOTH", smooth), ("WILD", wild)):
rows.append(
{
"symbol": sym,
"trade_date": d,
"close": px,
"high": px,
"low": px,
"volume": 1e6,
"amount": 1e8,
}
)
daily = pd.DataFrame(rows)
_defn, panel = compute_factor("volatility_20", daily)
assert float(panel["WILD"].iloc[-1]) > float(panel["SMOOTH"].iloc[-1]) * 5
def test_custom_factor_registration() -> None:
@register(FactorDef("test_double_close", "close*2 测试因子", "close * 2", lookback=1))
def _fn(fields: dict[str, pd.DataFrame]) -> pd.DataFrame:
return fields["close"] * 2
try:
daily = synthetic_daily({"A": 0.001}, n=40)
defn, panel = compute_factor("test_double_close", daily)
assert defn.direction == "higher_is_better"
assert float(panel.iloc[-1, 0]) > 200.0
finally:
# 清理注册表，避免污染其他测试
from app.quant import factors as _factors
_factors._REGISTRY.pop("test_double_close", None) # noqa: SLF001
def test_factor_def_metadata_present() -> None:
defn, _fn = get_factor("momentum_60")
assert defn.description
assert defn.formula
assert defn.lookback == 60
assert defn.direction in {"higher_is_better", "lower_is_better"}
backend/tests/test_research_eval.py
+125
View File
@@ -0,0 +1,125 @@
"""ResearchSpec 校验与因子评估（IC/RankIC/分层）测试。"""
from __future__ import annotations
from datetime import date
import numpy as np
import pandas as pd
import pytest
from app.domain.entities.research import (
CostSpec,
FactorSpec,
ResearchSpec,
SelectionSpec,
UniverseSpec,
)
from app.quant.evaluation import run_factor_test
from app.quant.factors import compute_factor
from app.quant.local_engine import composite_score, cross_sectional_zscore, rebalance_dates
from pydantic import ValidationError
from conftest_quant import synthetic_daily
def _spec(start: date = date(2024, 3, 1), end: date = date(2024, 12, 31), **kw) -> ResearchSpec:
base = dict(
type="backtest",
universe=UniverseSpec(exclude_st=False, min_listing_days=0),
factors=[FactorSpec(name="momentum_20")],
selection=SelectionSpec(top_n=10),
rebalance="monthly",
period=(start, end),
)
base.update(kw)
return ResearchSpec(**base)
class TestSpecValidation:
def test_inverted_period_rejected(self) -> None:
with pytest.raises(ValidationError, match="start < end"):
_spec(start=date(2024, 12, 1), end=date(2024, 1, 1))
def test_duplicate_factors_rejected(self) -> None:
with pytest.raises(ValidationError, match="重复"):
_spec(factors=[FactorSpec(name="momentum_20"), FactorSpec(name="momentum_20")])
def test_nonpositive_weight_rejected(self) -> None:
with pytest.raises(ValidationError):
_spec(factors=[FactorSpec(name="momentum_20", weight=0)])
def test_cost_bounds(self) -> None:
with pytest.raises(ValidationError):
CostSpec(commission_rate=0.5) # 超过 1%
with pytest.raises(ValidationError):
CostSpec(slippage_rate=-0.01)
class TestEvaluation:
def _panels(self):
# 强趋势 + 弱噪声，确保截面排序稳定（drift 差异远大于噪声）
drifts = {f"S{i:02d}": v for i, v in enumerate(np.linspace(0.006, -0.006, 12), start=1)}
daily = synthetic_daily(drifts, n=260)
_d, f20 = compute_factor("momentum_20", daily)
_d, f60 = compute_factor("momentum_60", daily)
close = daily.pivot(index="trade_date", columns="symbol", values="close").sort_index()
close.index = pd.to_datetime(close.index)
return daily, f20, f60, close
def test_cross_sectional_zscore_standardized(self) -> None:
_daily, f20, _f60, _close = self._panels()
z = cross_sectional_zscore(f20).dropna(how="all")
row = z.iloc[60].dropna()
assert abs(float(row.mean())) < 1e-9
assert abs(float(row.std()) - 1.0) < 1e-6
def test_composite_score_respects_direction(self) -> None:
_daily, f20, _f60, _close = self._panels()
pos = composite_score([("m", f20, 1.0, "higher_is_better")])
neg = composite_score([("m", f20, 1.0, "lower_is_better")])
row_date = f20.dropna(how="all").iloc[100].name
sym = f20.loc[row_date].dropna().index[0]
assert float(pos.loc[row_date, sym]) == pytest.approx(-float(neg.loc[row_date, sym]))
def test_momentum_ic_positive_on_trend_data(self) -> None:
daily, f20, _f60, close = self._panels()
forward = close.shift(-21) / close - 1.0
report = run_factor_test(f20, forward, factor_name="momentum_20")
assert report.sample_days > 10
assert report.ic_mean > 0
assert report.rank_ic_mean > 0
assert report.positive_ratio_pct > 50
def test_quantile_monotonic_on_trend(self) -> None:
daily, f20, _f60, close = self._panels()
forward = close.shift(-21) / close - 1.0
report = run_factor_test(f20, forward, factor_name="momentum_20", quantiles=5)
qs = {q.quantile: q.return_pct for q in report.quantile_returns}
assert qs[4] > qs[0] # 高动量层未来收益高于低动量层
assert report.spread_quantile is not None
def test_rebalance_dates_monthly_first(self) -> None:
idx = pd.bdate_range("2024-03-01", "2024-05-31")
out = rebalance_dates(idx, "monthly", date(2024, 3, 1))
assert [d.strftime("%Y-%m-%d") for d in out][:3] == [
"2024-03-01",
"2024-04-01",
"2024-05-01",
]
def test_rebalance_dates_respects_start(self) -> None:
idx = pd.bdate_range("2024-03-01", "2024-05-31")
out = rebalance_dates(idx, "monthly", date(2024, 4, 10))
assert out and out[0] >= pd.Timestamp("2024-04-10")
class TestSingleStockDegradation:
def test_zscore_single_stock_keeps_candidate(self) -> None:
from app.quant.local_engine import cross_sectional_zscore
daily = synthetic_daily({"ONLY": 0.001}, n=80)
_d, panel = compute_factor("momentum_20", daily)
z = cross_sectional_zscore(panel)
valid = z.dropna(how="all")
assert not valid.empty
assert (valid == 0.0).all().all() # 单股退化为 0，而非 NaN
docs/ROADMAP.md
+7 -2
View File
@@ -66,9 +66,14 @@
--- ---
## 2. Phase 2 —— Qlib 研究引擎（M2） ## 2. Phase 2 —— 研究引擎（M2）
> 目标：不碰 Qlib 源码，用 Research Specification 驱动完整「因子→模型→回测」。 > 实施备注（2026）：pyqlib 官方 wheel 仅支持 x86_64 / macOS / Windows 且最高 Python 3.8 构建；
> 本项目开发机为 Linux aarch64 + Python 3.12，无法安装 Qlib（已实测 pip 解析不可满足）。
> 依据 AGENT.md §40「更简单、可替换优先」与 Qlib 经 Adapter 隔离的约束，M2 交付**引擎接口 + 默认自研轻量引擎**（pandas 实现因子/评估/低频回测，完整可测）；
> `quant/qlib_adapter/` 保留桥接边界，在支持平台安装 pyqlib 后填充 Qlib Dataset / LightGBM 工作流实现，业务层不感知切换。
> 目标：用 Research Specification 驱动「因子 → 评估 → 低频选股回测」闭环，产出标准化结果。
### 2.1 Qlib Adapter（`quant/qlib_adapter/`） ### 2.1 Qlib Adapter（`quant/qlib_adapter/`）