Simon
7e369d9680
「因子组合」/`POST /api/research/backtests` 走的是 `LocalEngine/TopKBacktestRunner`， 上一版只把理由接进了组合引擎，同一件事在两个引擎上就会有两种说法。这次补齐： - `engine.py` / `qlib_adapter/engine.py`：因子面板**只算一次** （`build_factor_panels_full`）→ 复合分与「理由里引用的因子原始值」同源同张面板； 复合分口径逐字未变（与 `selection.score_panel_for_factors` 相同）。 - `local_engine.py`：调仓日保留完整排名与合格集，各站点写入结构化理由 —— 买入（按名次建仓 / 顺延成交 / 涨停 / 停牌 / 现金不足 / 不足最低佣金）、 卖出（全量换仓 / 跌出 TopN / 不在候选池 / 停牌顺延 / 跌停顺延）； `Trade.entry_reason/exit_reason` 两端齐全；每个交易日记录持仓市值， 结果填 `factor_curves`（持仓市值加权平均的因子原始值，空仓日不落点）。 - 新增 `SELL_REBALANCE_FULL`（「调仓换仓卖出」）：单策略调仓是「先全清再建仓」， 被卖出的股票**可能仍排在 TopN 内**（如 rank=1），这时写「跌出 TopN」就是假解释； 按事实分 code（仍在 TopN 内 → 全量换仓；否则 → 跌出 TopN / 不在候选池）。 - 顺延成交不拿挂单日的旧名次冒充当日名次（rank/total/score=None，因子值/成交价/预算 取成交当日真实值）；「候选池不足」的提示记录保持 reason=None（词表里没有对应语义， 硬套就是编理由）。 验证： - 新增 `tests/test_local_engine_reasons.py` 14 条：理由数字对回面板、涨停比值对回行情与 板块规则、停牌/跌停/现金不足/最低佣金、顺延成交、全量换仓 vs 不在池两个分支、 Trade 两端理由、因子曲线市值加权（手算加权值断言 + 等权平均对不上）、空仓不落点。 后端 524 条全过（510 + 14），ruff clean。 - 强回归：用改前引擎并排跑 9 个场景，`signal_history`(日期/方向/成交/原因文案/价格)、 `trades`、`positions`、`summary`、净值/回撤、`unimplemented` 逐条一致 —— 理由与曲线 是纯新增字段，成交行为零变化。
503 lines
22 KiB
Python
503 lines
22 KiB
Python
"""LocalEngine（单策略回测）买卖理由与因子曲线的常驻回归。
|
||
|
||
用户要求「回测结果里所有买卖点详细说明买卖理由，用数据说话」，因此这里验证的**不是文案
|
||
长什么样**，而是：理由里的每个数字都等于引擎当时算出来的值，能独立地对回算出来 ——
|
||
|
||
- 名次 / 候选数 / 综合分对得上复合分面板；`factors` 原始值对得上因子面板同一格；
|
||
- 涨跌停比值、预算、最低佣金、持有交易日对得上行情与配置；
|
||
- code 按**事实**选：仍在前列的清仓换仓（本引擎每次调仓先全清再建仓）用
|
||
`sell_rebalance_full`，只有确实不在池 / 名次掉出 / 当日无分数才用 `sell_drop_topn`；
|
||
- `Trade.entry_reason / exit_reason` 两端齐全，且与成交价一致（理由不是事后补的）；
|
||
- `factor_curves` = 当日持仓**市值加权平均原始值**（用手算的加权值断言），空仓日不落点。
|
||
|
||
数据全部由本文件确定性合成（无外部依赖、无随机数），场景通过覆盖个别交易日的收盘价
|
||
（精确到「上一有效收盘 × 目标幅度」）来触发涨停 / 跌停 / 停牌。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import date
|
||
|
||
import pandas as pd
|
||
import pytest
|
||
from app.domain.entities.research import (
|
||
CostSpec,
|
||
FactorSpec,
|
||
ResearchSpec,
|
||
SelectionSpec,
|
||
UniverseSpec,
|
||
)
|
||
from app.quant.composite import build_factor_panels_full
|
||
from app.quant.engine import LocalEngine
|
||
from app.quant.selection import score_panel_for_factors
|
||
from app.quant.trade_reasons import (
|
||
BUY_DEFER_FILLED,
|
||
BUY_ENTER,
|
||
BUY_SKIP_HALTED,
|
||
BUY_SKIP_LIMIT_UP,
|
||
BUY_SKIP_MIN_COMMISSION,
|
||
BUY_SKIP_NO_CASH,
|
||
SELL_DEFER_HALTED,
|
||
SELL_DEFER_LIMIT_DOWN,
|
||
SELL_DROP_TOPN,
|
||
SELL_REBALANCE_FULL,
|
||
)
|
||
|
||
# 三只标的的确定性漂移：600000 最强、600002 最弱（动量排序稳定可预期）
|
||
_SYMS = (("600000.SH", 0.004), ("600001.SH", 0.0015), ("600002.SH", -0.002))
|
||
_START = date(2024, 3, 1)
|
||
_END = date(2024, 10, 31)
|
||
_APR_REBAL = date(2024, 4, 1) # 4 月调仓日
|
||
_MAY_REBAL = date(2024, 5, 1) # 5 月调仓日
|
||
_MOMENTUM = [FactorSpec(name="momentum_20")]
|
||
_VOLUME = [FactorSpec(name="volume_ratio_5_60")]
|
||
|
||
|
||
def _daily(*, overrides=None, nan_quotes=None, n=320, base=100.0) -> pd.DataFrame:
|
||
"""确定性合成日线长表：p[j] = p[j-1] * (1 + drift + 0.012·sin((j+i)·0.8))。
|
||
|
||
`overrides={(symbol, date): close}` 制造涨停/跌停，`nan_quotes` 制造停牌（无行情）；
|
||
两者只改当日收盘（成交/撮合与因子都据此计算），保证场景可复现。
|
||
"""
|
||
dates = pd.bdate_range("2024-01-01", periods=n)
|
||
overrides = overrides or {}
|
||
nan_quotes = set(nan_quotes or ())
|
||
rows: list[dict] = []
|
||
for i, (sym, drift) in enumerate(_SYMS):
|
||
price = base
|
||
for j, d in enumerate(dates):
|
||
prev = price
|
||
price = price * (1 + drift + 0.012 * math.sin((j + i) * 0.8))
|
||
px = float(overrides.get((sym, d.date()), price))
|
||
if (sym, d.date()) in nan_quotes:
|
||
px = float("nan")
|
||
volume = float(1_000_000 + j * 1000 + i * 3000)
|
||
rows.append(
|
||
{
|
||
"symbol": sym,
|
||
"trade_date": d.date(),
|
||
"open": prev,
|
||
"high": float("nan") if math.isnan(px) else max(prev, px) * 1.008,
|
||
"low": float("nan") if math.isnan(px) else min(prev, px) * 0.992,
|
||
"close": px,
|
||
"volume": volume,
|
||
"amount": float("nan") if math.isnan(px) else px * volume,
|
||
}
|
||
)
|
||
return pd.DataFrame(rows)
|
||
|
||
|
||
def _spec(**over) -> ResearchSpec:
|
||
base = dict(
|
||
type="backtest",
|
||
universe=UniverseSpec(exclude_st=False, min_listing_days=0),
|
||
factors=list(_MOMENTUM),
|
||
selection=SelectionSpec(top_n=1),
|
||
rebalance="monthly",
|
||
period=(_START, _END),
|
||
costs=CostSpec(),
|
||
)
|
||
base.update(over)
|
||
return ResearchSpec(**base)
|
||
|
||
|
||
def _run(daily: pd.DataFrame, **spec_over):
|
||
return LocalEngine().run_backtest(daily, _spec(**spec_over))
|
||
|
||
|
||
def _by_code(result, code, *, signal=None, filled=None):
|
||
"""按 code 取记录（可再按 BUY/SELL 与是否成交过滤）。"""
|
||
return [
|
||
a
|
||
for a in result.signal_history
|
||
if a.reason is not None
|
||
and a.reason.code == code
|
||
and (signal is None or a.signal == signal)
|
||
and (filled is None or a.filled == filled)
|
||
]
|
||
|
||
|
||
def _panels(daily: pd.DataFrame, spec: ResearchSpec):
|
||
"""因子原始面板 {name: (defn, panel)}（与引擎注入理由的来源同一构建函数）。"""
|
||
return {d.name: (d, p) for d, p, _w in build_factor_panels_full(daily, spec.factors)}
|
||
|
||
|
||
def _close_panel(daily: pd.DataFrame) -> pd.DataFrame:
|
||
close = daily.pivot(index="trade_date", columns="symbol", values="close").sort_index()
|
||
close.index = pd.to_datetime(close.index)
|
||
return close
|
||
|
||
|
||
def _close_before(close: pd.DataFrame, day: date, symbol: str) -> float:
|
||
"""`day` 之前最后一个有效收盘价（用来精确构造涨停/跌停的当日价）。"""
|
||
series = close[symbol].dropna()
|
||
return float(series[series.index < pd.Timestamp(day)].iloc[-1])
|
||
|
||
|
||
def _leader_at(daily: pd.DataFrame, factors, day: date) -> str:
|
||
"""该日复合分第一名（据此构造「涨停/停牌」的标的，避免写死代码）。"""
|
||
score = score_panel_for_factors(daily, factors)
|
||
return str(score.loc[pd.Timestamp(day)].dropna().idxmax())
|
||
|
||
|
||
def _held_on(daily: pd.DataFrame, day: date, **spec_over) -> set[str]:
|
||
"""基线回测在该日的持仓（据此决定把哪只标的的行情改成跌停/停牌）。"""
|
||
result = LocalEngine().run_backtest(daily, _spec(**spec_over))
|
||
return {p.symbol for p in result.positions if p.date == day}
|
||
|
||
|
||
# ---------- 1. 买入理由的数字来源 ----------
|
||
|
||
|
||
def test_buy_reason_numbers_come_from_engine():
|
||
"""买入理由的 rank/total/top_n/score 与 factors 原始值都能对回引擎面板。"""
|
||
daily = _daily()
|
||
spec = _spec()
|
||
result = LocalEngine().run_backtest(daily, spec)
|
||
|
||
buys = [a for a in result.signal_history if a.signal == "BUY" and a.filled]
|
||
assert buys, "主路径应有成交买入"
|
||
rec = buys[0]
|
||
reason = rec.reason
|
||
assert reason is not None and reason.code == BUY_ENTER
|
||
|
||
# 名次 / 候选数 / 综合分 == 复合分面板当日真实排序（独立算一遍）
|
||
score = score_panel_for_factors(daily, spec.factors)
|
||
d = pd.Timestamp(rec.date)
|
||
ranked = score.loc[d].dropna().sort_values(ascending=False)
|
||
assert reason.data["rank"] == ranked.index.get_loc(rec.symbol) + 1
|
||
assert reason.data["rank"] == 1
|
||
assert reason.data["total"] == len(ranked)
|
||
assert reason.data["top_n"] == spec.selection.top_n
|
||
assert reason.data["score"] == pytest.approx(round(float(ranked[rec.symbol]), 6))
|
||
|
||
# 因子原始值 == 因子面板同一格（不是重算、不是估算）
|
||
_defn, panel = _panels(daily, spec)["momentum_20"]
|
||
assert reason.data["factors"]["momentum_20"] == pytest.approx(
|
||
round(float(panel.at[d, rec.symbol]), 6)
|
||
)
|
||
assert f"第 {reason.data['rank']}" in reason.text and "成交价" in reason.text
|
||
|
||
|
||
# ---------- 2. 涨停未买 ----------
|
||
|
||
|
||
def test_buy_skip_limit_up_uses_real_ratio():
|
||
"""涨停未买：data 里的收盘/前收/比值/阈值全部来自当日行情与板块规则。"""
|
||
base_daily = _daily()
|
||
close0 = _close_panel(base_daily)
|
||
leader = _leader_at(base_daily, _MOMENTUM, _APR_REBAL)
|
||
prev = _close_before(close0, _APR_REBAL, leader)
|
||
daily = _daily(overrides={(leader, _APR_REBAL): prev * 1.12})
|
||
|
||
result = LocalEngine().run_backtest(daily, _spec())
|
||
skips = _by_code(result, BUY_SKIP_LIMIT_UP, signal="BUY", filled=False)
|
||
assert skips, "当日涨停应记录 buy_skip_limit_up"
|
||
rec = skips[0]
|
||
assert rec.symbol == leader
|
||
reason = rec.reason
|
||
|
||
close = _close_panel(daily)
|
||
d = pd.Timestamp(_APR_REBAL)
|
||
real_close = float(close.at[d, leader])
|
||
real_prev = float(close.ffill().shift(1).at[d, leader])
|
||
assert reason.data["close"] == pytest.approx(round(real_close, 4))
|
||
assert reason.data["prev_close"] == pytest.approx(round(real_prev, 4))
|
||
assert reason.data["close_prev_ratio"] == pytest.approx(round(real_close / real_prev, 4))
|
||
assert reason.data["close_prev_ratio"] == pytest.approx(1.12)
|
||
assert reason.data["limit_ratio"] == pytest.approx(round(1.0 + (1.099 - 1.0), 4))
|
||
assert reason.data["close_prev_ratio"] >= reason.data["limit_ratio"]
|
||
assert rec.reject_reason == "涨停，无法追买" # 既有文案未被理由改动
|
||
assert "涨停" in reason.text and "无法追买" in reason.text
|
||
|
||
|
||
# ---------- 3. 停牌未买 / 停牌未卖 ----------
|
||
|
||
|
||
def test_buy_skip_halted_keeps_rank_and_reject_text():
|
||
"""停牌未买：标的仍被选中（有真实名次），只是当日无行情无法成交。"""
|
||
base_daily = _daily()
|
||
# 用「只依赖 volume」的因子：close 缺失时该股仍能进候选池，才能走到执行层停牌分支
|
||
leader = _leader_at(base_daily, _VOLUME, _MAY_REBAL)
|
||
daily = _daily(nan_quotes={(leader, _MAY_REBAL)})
|
||
result = LocalEngine().run_backtest(daily, _spec(factors=list(_VOLUME)))
|
||
|
||
skips = _by_code(result, BUY_SKIP_HALTED, signal="BUY", filled=False)
|
||
assert skips, "停牌应记录 buy_skip_halted"
|
||
rec = skips[0]
|
||
assert rec.symbol == leader
|
||
assert rec.reject_reason == "无行情（停牌），无法买入" # 既有文案未变
|
||
assert rec.reason.data["rank"] == 1 # 停牌的是被选中的第一名，不是随便一只
|
||
assert "停牌" in rec.reason.text
|
||
|
||
|
||
def test_sell_defer_halted_keeps_position():
|
||
"""停牌未卖：顺延理由带真实持有交易日，且 reject_reason 保持原文案。"""
|
||
base_daily = _daily()
|
||
held = _held_on(base_daily, _MAY_REBAL)
|
||
assert held, "基线在 5 月调仓日应有持仓"
|
||
daily = _daily(nan_quotes={(s, _MAY_REBAL) for s in held})
|
||
result = _run(daily)
|
||
|
||
defers = _by_code(result, SELL_DEFER_HALTED, signal="SELL", filled=False)
|
||
assert defers, "持仓股无行情应记录 sell_defer_halted"
|
||
rec = defers[0]
|
||
assert rec.symbol in held
|
||
assert rec.reject_reason == "无行情（停牌），保留持仓"
|
||
assert rec.reason.data["hold_days"] > 0
|
||
# 当日没有该股的成交卖出（停牌只是顺延，仓位保留）
|
||
assert not [
|
||
a
|
||
for a in result.signal_history
|
||
if a.symbol == rec.symbol
|
||
and a.date == _MAY_REBAL
|
||
and a.signal == "SELL"
|
||
and a.filled
|
||
]
|
||
|
||
|
||
# ---------- 4. 跌停未卖 ----------
|
||
|
||
|
||
def test_sell_defer_limit_down_uses_real_ratio():
|
||
"""跌停未卖：data 里的收盘/前收/比值/阈值与行情一致（比值 ≤ 阈值）。"""
|
||
base_daily = _daily()
|
||
close0 = _close_panel(base_daily)
|
||
held = _held_on(base_daily, _APR_REBAL)
|
||
assert held
|
||
overrides = {
|
||
(s, _APR_REBAL): _close_before(close0, _APR_REBAL, s) * 0.90 for s in held
|
||
}
|
||
daily = _daily(overrides=overrides)
|
||
result = _run(daily)
|
||
|
||
defers = _by_code(result, SELL_DEFER_LIMIT_DOWN, signal="SELL", filled=False)
|
||
assert defers, "持仓股跌停应记录 sell_defer_limit_down"
|
||
rec = defers[0]
|
||
reason = rec.reason
|
||
close = _close_panel(daily)
|
||
d = pd.Timestamp(_APR_REBAL)
|
||
assert reason.data["close"] == pytest.approx(round(float(close.at[d, rec.symbol]), 4))
|
||
assert reason.data["prev_close"] == pytest.approx(
|
||
round(float(close.ffill().shift(1).at[d, rec.symbol]), 4)
|
||
)
|
||
assert reason.data["close_prev_ratio"] == pytest.approx(0.90)
|
||
assert reason.data["limit_ratio"] == pytest.approx(0.901)
|
||
assert reason.data["close_prev_ratio"] <= reason.data["limit_ratio"]
|
||
assert reason.data["hold_days"] > 0
|
||
assert rec.reject_reason == "跌停无法卖出，保留到下一调仓"
|
||
assert "跌停" in reason.text and "顺延" in reason.text
|
||
|
||
|
||
# ---------- 5. 现金不足 / 不足最低佣金 ----------
|
||
|
||
|
||
def test_buy_skip_no_cash_when_budget_exhausted():
|
||
"""现金分配耗尽后，池内第二只留痕「资金不足」，budget = 当时真实剩余现金。"""
|
||
daily = _daily()
|
||
result = _run(daily, selection=SelectionSpec(top_n=2, hold_top_x=1))
|
||
|
||
skips = _by_code(result, BUY_SKIP_NO_CASH, signal="BUY", filled=False)
|
||
assert skips
|
||
no_cash = [a for a in skips if a.reject_reason == "资金不足（未成交）"]
|
||
assert no_cash, "替补路径下池内被跳过的标的应给出 buy_skip_no_cash"
|
||
assert no_cash[0].reason.data["budget"] == pytest.approx(0.0) # 唯一目标吃光现金
|
||
assert "可用预算" in no_cash[0].reason.text
|
||
|
||
|
||
def test_buy_skip_min_commission_from_budget_and_config():
|
||
"""不足最低佣金：budget = 等权分配额、min_commission = 配置值，两者都来自引擎。"""
|
||
daily = _daily()
|
||
result = LocalEngine().run_backtest(
|
||
daily,
|
||
_spec(costs=CostSpec(min_commission=5.0), initial_capital=4.0),
|
||
)
|
||
skips = _by_code(result, BUY_SKIP_MIN_COMMISSION, signal="BUY", filled=False)
|
||
assert skips
|
||
reason = skips[0].reason
|
||
assert reason.data["budget"] == pytest.approx(4.0) # 4 元全给唯一目标
|
||
assert reason.data["min_commission"] == pytest.approx(5.0)
|
||
assert reason.data["budget"] < reason.data["min_commission"]
|
||
assert skips[0].reject_reason == "预算不足以覆盖最低佣金，未成交"
|
||
assert result.trades == [] # 该场景确实一笔未成
|
||
|
||
|
||
# ---------- 6. 顺延买入成交 ----------
|
||
|
||
|
||
def test_buy_defer_filled_after_limit_up():
|
||
"""顺延买入：挂单当日涨停未买，之后按真实成交日的价格/因子值成交（不编当日名次）。"""
|
||
base_daily = _daily()
|
||
close0 = _close_panel(base_daily)
|
||
leader = _leader_at(base_daily, _MOMENTUM, _APR_REBAL)
|
||
prev = _close_before(close0, _APR_REBAL, leader)
|
||
daily = _daily(overrides={(leader, _APR_REBAL): prev * 1.12})
|
||
spec = _spec(
|
||
selection=SelectionSpec(top_n=1, allow_substitute=False, defer_buy=True)
|
||
)
|
||
result = LocalEngine().run_backtest(daily, spec)
|
||
|
||
pending = _by_code(result, BUY_SKIP_LIMIT_UP, signal="BUY", filled=False)
|
||
filled = _by_code(result, BUY_DEFER_FILLED, signal="BUY", filled=True)
|
||
assert pending and filled, "顺延应有「挂单当日涨停」+「之后成交」两条记录"
|
||
assert pending[0].symbol == filled[0].symbol == leader
|
||
assert pending[0].date == _APR_REBAL
|
||
assert filled[0].date > _APR_REBAL # 只在之后的交易日补成交，不回溯
|
||
|
||
reason = filled[0].reason
|
||
# 成交日不是择股日 → 不拿旧名次冒充当日名次
|
||
assert "rank" not in reason.data and "score" not in reason.data
|
||
# 因子原始值 / 成交价 / 预算取成交当日的真实值
|
||
_defn, panel = _panels(daily, spec)["momentum_20"]
|
||
fd = pd.Timestamp(filled[0].date)
|
||
assert reason.data["factors"]["momentum_20"] == pytest.approx(
|
||
round(float(panel.at[fd, leader]), 6)
|
||
)
|
||
close = _close_panel(daily)
|
||
price_in = float(close.at[fd, leader]) * (1 + spec.costs.slippage_rate)
|
||
assert reason.data["price"] == pytest.approx(round(price_in, 4))
|
||
assert reason.data["budget"] > 0
|
||
# 成交明细里的建仓理由是「顺延成交」，不是笼统的按名次建仓
|
||
trade = next(
|
||
t for t in result.trades if t.symbol == leader and t.entry_date == filled[0].date
|
||
)
|
||
assert trade.entry_reason is not None and trade.entry_reason.code == BUY_DEFER_FILLED
|
||
|
||
|
||
# ---------- 7. 成交卖出：全量换仓 vs 跌出 TopN ----------
|
||
|
||
|
||
def test_sell_rebalance_full_when_still_in_topn():
|
||
"""仍排在 TopN 内却被清仓（本引擎「先全清再建仓」）→ sell_rebalance_full。
|
||
|
||
这是本次新增 code 的关键回归：老实现会把它说成「跌出 TopN」，与 data 里的
|
||
rank=1/top_n=1 自相矛盾 —— 用假解释掩盖真实原因。
|
||
"""
|
||
daily = _daily()
|
||
spec = _spec() # top_n=1：最强的 600000.SH 每月都排第一
|
||
result = LocalEngine().run_backtest(daily, spec)
|
||
|
||
sells = [a for a in result.signal_history if a.signal == "SELL" and a.filled]
|
||
assert sells, "调仓应产生成交卖出"
|
||
rec = sells[0]
|
||
reason = rec.reason
|
||
assert reason.code == SELL_REBALANCE_FULL
|
||
assert reason.data["rank"] == 1
|
||
assert reason.data["top_n"] == 1
|
||
assert reason.data["rank"] <= reason.data["top_n"] # 关键：当时仍在前列
|
||
assert reason.data["hold_days"] > 0
|
||
assert "全量换仓" in reason.text
|
||
# 该股当日确实有分（不是「当日无分数」才落到这个 code）
|
||
score = score_panel_for_factors(daily, spec.factors)
|
||
assert not math.isnan(float(score.at[pd.Timestamp(rec.date), rec.symbol]))
|
||
|
||
# 关键 data（跑 -s 时可读；失败时也在断言里可见）
|
||
print(
|
||
f"[sell_rebalance_full] code={reason.code} rank={reason.data['rank']} "
|
||
f"total={reason.data['total']} top_n={reason.data['top_n']} "
|
||
f"hold_days={reason.data['hold_days']}"
|
||
)
|
||
|
||
|
||
def test_sell_drop_topn_when_filtered_out_of_pool():
|
||
"""被股票池/条件过滤（已不在候选池）才归 sell_drop_topn，与换仓卖出分开。"""
|
||
daily = _daily()
|
||
spec = _spec()
|
||
result = LocalEngine().run_backtest(
|
||
daily,
|
||
spec,
|
||
eligibility_fn=lambda as_of: {"600001.SH"} if as_of >= _APR_REBAL else None,
|
||
)
|
||
drops = _by_code(result, SELL_DROP_TOPN, signal="SELL", filled=True)
|
||
assert drops, "持仓股被条件过滤后应卖出并归 sell_drop_topn"
|
||
reason = drops[0].reason
|
||
assert reason.data["in_pool"] is False
|
||
assert "已不在候选池" in reason.text
|
||
# 对照：换仓卖出的 code 不应出现在同一条记录上
|
||
assert reason.code != SELL_REBALANCE_FULL
|
||
|
||
|
||
def test_trade_carries_both_end_reasons():
|
||
"""成交明细两端齐全，且理由里的价格与 Trade 的成交价一致（理由跟着成交走）。"""
|
||
daily = _daily()
|
||
result = _run(daily)
|
||
assert result.trades
|
||
trade = result.trades[0]
|
||
assert trade.entry_reason is not None and trade.entry_reason.code == BUY_ENTER
|
||
assert trade.exit_reason is not None and trade.exit_reason.code == SELL_REBALANCE_FULL
|
||
assert trade.entry_reason.data["price"] == pytest.approx(round(trade.entry_price, 4))
|
||
assert trade.exit_reason.data["price"] == pytest.approx(round(trade.exit_price, 4))
|
||
|
||
|
||
# ---------- 8. 因子曲线：市值加权原始值 / 空仓日不落点 ----------
|
||
|
||
|
||
def test_factor_curves_are_market_value_weighted():
|
||
"""曲线值 == 当日持仓按市值加权平均的因子原始值（用反推股数的手算值断言）。"""
|
||
daily = _daily()
|
||
spec = _spec(selection=SelectionSpec(top_n=2)) # 两只持仓，权重会随行情漂移
|
||
result = LocalEngine().run_backtest(daily, spec)
|
||
|
||
_defn, panel = _panels(daily, spec)["momentum_20"]
|
||
curve = next(c for c in result.factor_curves if c.name == "momentum_20")
|
||
points = {p.date: p.value for p in curve.points}
|
||
assert points, "有持仓就应有因子曲线点"
|
||
|
||
# 从首个「两只持仓」的调仓日反推股数（引擎给的 weight × 当日权益 ÷ 当日收盘）
|
||
close = _close_panel(daily)
|
||
by_day: dict = {}
|
||
for pos in result.positions:
|
||
by_day.setdefault(pos.date, {})[pos.symbol] = pos.weight
|
||
d0 = next(d for d in sorted(by_day) if len(by_day[d]) == 2)
|
||
equity0 = next(q.value for q in result.equity_curve if q.date == d0)
|
||
qty = {
|
||
s: by_day[d0][s] * equity0 / float(close.at[pd.Timestamp(d0), s])
|
||
for s in by_day[d0]
|
||
}
|
||
|
||
# 取之后第 10 个交易日（仍在同一持仓期内）：权重已随价格漂移，非等权
|
||
idx = list(close.index)
|
||
d1 = idx[idx.index(pd.Timestamp(d0)) + 10]
|
||
market_value = {s: qty[s] * float(close.at[d1, s]) for s in qty}
|
||
values = {s: float(panel.at[d1, s]) for s in market_value}
|
||
manual = sum(values[s] * market_value[s] for s in values) / sum(market_value.values())
|
||
equal = sum(values.values()) / len(values)
|
||
|
||
assert abs(market_value["600000.SH"] - market_value["600001.SH"]) > 1.0 # 确实漂移了
|
||
assert abs(equal - manual) > 1e-5 # 等权平均对不上 → 能区分「市值加权」
|
||
assert points[d1.date()] == pytest.approx(round(manual, 6), abs=1e-6)
|
||
# 曲线上是因子的**原始值**（未 z-score、未按方向取负）：量级与动量本身一致
|
||
assert all(abs(v) < 5 for v in points.values())
|
||
|
||
|
||
def test_factor_curves_skip_days_without_holdings():
|
||
"""空仓日不落点（不插值、不用 0 填充）：涨停买不进且不替补的整月没有曲线点。"""
|
||
base_daily = _daily()
|
||
close0 = _close_panel(base_daily)
|
||
leader = _leader_at(base_daily, _MOMENTUM, _APR_REBAL)
|
||
prev = _close_before(close0, _APR_REBAL, leader)
|
||
daily = _daily(overrides={(leader, _APR_REBAL): prev * 1.12})
|
||
spec = _spec(
|
||
selection=SelectionSpec(top_n=1, allow_substitute=False, defer_buy=False)
|
||
)
|
||
result = LocalEngine().run_backtest(daily, spec)
|
||
|
||
curve = next(c for c in result.factor_curves if c.name == "momentum_20")
|
||
point_dates = {p.date for p in curve.points}
|
||
april = {d.date() for d in pd.bdate_range("2024-04-01", "2024-04-30")}
|
||
assert not (point_dates & april), "4 月空仓（涨停未买且不替补），不应有任何曲线点"
|
||
assert date(2024, 3, 1) in point_dates # 3 月建仓后有持仓 → 有点
|
||
period_days = {d.date() for d in pd.bdate_range(_START, _END)}
|
||
assert len(point_dates) < len(period_days) # 有缺口 = 没按交易日补齐
|
||
|
||
|
||
def test_factor_curves_empty_when_no_fills():
|
||
"""一笔都没成交（预算不足最低佣金）→ 面板非空但曲线 0 个点，而不是一堆 0 值。"""
|
||
daily = _daily()
|
||
result = LocalEngine().run_backtest(
|
||
daily, _spec(costs=CostSpec(min_commission=5.0), initial_capital=4.0)
|
||
)
|
||
assert result.trades == []
|
||
assert result.factor_curves, "因子曲线按策略因子输出（即便没成交）"
|
||
assert all(c.points == [] for c in result.factor_curves) |