From e9f59d3cf870419d474e357c3d42b5d16819443a Mon Sep 17 00:00:00 2001 From: Simon Date: Sun, 6 Sep 2026 17:08:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(backend):=20Phase=202=20=E7=A0=94=E7=A9=B6?= =?UTF-8?q?=E5=BC=95=E6=93=8E=20=E2=80=94=20ResearchSpec=20/=20=E5=9B=A0?= =?UTF-8?q?=E5=AD=90=20/=20=E8=AF=84=E4=BC=B0=20/=20=E4=BD=8E=E9=A2=91?= =?UTF-8?q?=E5=9B=9E=E6=B5=8B=20/=20=E5=BC=95=E6=93=8E=E6=8A=BD=E8=B1=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/app/domain/entities/research.py | 169 ++++++++++++ backend/app/quant/engine.py | 48 ++++ backend/app/quant/evaluation.py | 102 +++++++ backend/app/quant/factors.py | 184 ++++++++++++ backend/app/quant/local_engine.py | 338 +++++++++++++++++++++++ backend/app/quant/qlib_adapter/engine.py | 39 +++ backend/app/quant/service.py | 93 +++++++ backend/tests/conftest_quant.py | 56 ++++ backend/tests/test_quant_engine.py | 138 +++++++++ backend/tests/test_quant_factors.py | 89 ++++++ backend/tests/test_research_eval.py | 125 +++++++++ docs/ROADMAP.md | 9 +- 12 files changed, 1388 insertions(+), 2 deletions(-) create mode 100644 backend/app/domain/entities/research.py create mode 100644 backend/app/quant/engine.py create mode 100644 backend/app/quant/evaluation.py create mode 100644 backend/app/quant/factors.py create mode 100644 backend/app/quant/local_engine.py create mode 100644 backend/app/quant/qlib_adapter/engine.py create mode 100644 backend/app/quant/service.py create mode 100644 backend/tests/conftest_quant.py create mode 100644 backend/tests/test_quant_engine.py create mode 100644 backend/tests/test_quant_factors.py create mode 100644 backend/tests/test_research_eval.py diff --git a/backend/app/domain/entities/research.py b/backend/app/domain/entities/research.py new file mode 100644 index 0000000..df69c37 --- /dev/null +++ b/backend/app/domain/entities/research.py @@ -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) diff --git a/backend/app/quant/engine.py b/backend/app/quant/engine.py new file mode 100644 index 0000000..62f454b --- /dev/null +++ b/backend/app/quant/engine.py @@ -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() diff --git a/backend/app/quant/evaluation.py b/backend/app/quant/evaluation.py new file mode 100644 index 0000000..e2bdc87 --- /dev/null +++ b/backend/app/quant/evaluation.py @@ -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 diff --git a/backend/app/quant/factors.py b/backend/app/quant/factors.py new file mode 100644 index 0000000..1a046be --- /dev/null +++ b/backend/app/quant/factors.py @@ -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) diff --git a/backend/app/quant/local_engine.py b/backend/app/quant/local_engine.py new file mode 100644 index 0000000..f466a4b --- /dev/null +++ b/backend/app/quant/local_engine.py @@ -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 diff --git a/backend/app/quant/qlib_adapter/engine.py b/backend/app/quant/qlib_adapter/engine.py new file mode 100644 index 0000000..100bce5 --- /dev/null +++ b/backend/app/quant/qlib_adapter/engine.py @@ -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) diff --git a/backend/app/quant/service.py b/backend/app/quant/service.py new file mode 100644 index 0000000..aedafe1 --- /dev/null +++ b/backend/app/quant/service.py @@ -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) diff --git a/backend/tests/conftest_quant.py b/backend/tests/conftest_quant.py new file mode 100644 index 0000000..b4dfbe9 --- /dev/null +++ b/backend/tests/conftest_quant.py @@ -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() + ] diff --git a/backend/tests/test_quant_engine.py b/backend/tests/test_quant_engine.py new file mode 100644 index 0000000..ca118a8 --- /dev/null +++ b/backend/tests/test_quant_engine.py @@ -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"] diff --git a/backend/tests/test_quant_factors.py b/backend/tests/test_quant_factors.py new file mode 100644 index 0000000..fa537aa --- /dev/null +++ b/backend/tests/test_quant_factors.py @@ -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"} diff --git a/backend/tests/test_research_eval.py b/backend/tests/test_research_eval.py new file mode 100644 index 0000000..e175090 --- /dev/null +++ b/backend/tests/test_research_eval.py @@ -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 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 7c52925..7c85e8e 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -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/`）