fix(backtest): A股适配层—定寸/做空拦截/真实费用/口径统一(Phase1+2)

审计发现包装层系统性失真(2 CRITICAL+7 HIGH),vnpy底座可信但A股场景未适配:
- C1 定寸: engine.size=N(满仓手数),策略volume=1手=N股,开平对称(pos归零)
- C2 做空拦截: SHORT+OPEN拒单,long-only,SHORT+CLOSE平多允许
- H3 A股费用: AShareDailyResult重算(佣金保底5元/印花税卖方/过户费沪市)
- H4 收益口径: simple return从balance算(不再用vnpy log return喂empyrical)
- H5+口径: benchmark ffill对齐不缩样本; sizing_shares_per_lot暴露
- H7 退化检测: 零成交/空数据标degenerate不静默done
- H8 task_id: optimize/factor用uuid4(原id()内存地址)
- 静默except改warning

验证: 容器内真实vnpy DoubleMa 600000 2022-2024, total_return 1e-6→42.3%,
end_balance 100万→142万, SHORT+OPEN成交0笔, N=7800股/手.
22 backtest测试全绿(含集成测试), API健康200.
This commit is contained in:
2026-07-12 23:39:45 +08:00
parent 292de31eaf
commit 8d55e414fa
11 changed files with 578 additions and 85 deletions
+83
View File
@@ -0,0 +1,83 @@
"""A股适配层真实集成测试(容器内跑,Mac 本机无法运行)。
需要 vnpy_ctastrategy + quant_trading.db + A 股 K 线数据。
用法:容器内 `pytest tests/backtest/test_integration_ashare.py -m integration`
验证 Phase 1+2 四项核心断言:
- C1 定寸:成交金额 ≈ 满仓量级(size=N 方案,volume=1 手=N 股)
- C2 做空拦截:无 SHORT+OPEN 成交
- H3 真实费用:end_balance != capital(非空转,费用+盈亏反映在余额)
- 非噪声:|total_return| > 1e-3
"""
import pytest
# 容器内才有 vnpy_ctastrategyMac 本机自动 skip(不中断 pytest 全量跑)
pytest.importorskip("vnpy_ctastrategy")
pytestmark = [pytest.mark.integration]
def test_double_ma_600000_2022_2024():
"""DoubleMa 600000 2022-2024 真实回测验证。
标记 integration → 仅容器内跑(需 vnpy + quant_trading.db + A 股日线数据)。
用 3 年窗口确保 ArrayManager(100) 充分暖机 + 产生足够多 MA 交叉信号
2024H1 窗口太短,仅 111 根 bar 暖机后信号窗口不足,会误判退化)。
"""
from vnpy_ctastrategy.strategies.double_ma_strategy import DoubleMaStrategy
from sanguo_backtest.cta_engine import run_cta_backtest
capital = 1_000_000
position_pct = 0.95
result = run_cta_backtest(
strategy_class=DoubleMaStrategy,
symbol="600000",
params={"fast_window": 5, "slow_window": 10},
start="2022-01-01",
end="2024-12-31",
cfg=None, # cta_engine 内部 load_config
db_path="/tmp/test_integration_ashare.db",
benchmark="hs300",
capital=capital,
position_pct=position_pct,
)
# 基本成功检查
assert result.status in ("done", "degenerate"), f"回测失败: {result.error_msg}"
assert result.status == "done", f"回测退化(不应退化): {result.statistics.get('degenerate_reason')}"
stats = result.statistics
trades = result.trades
# H3: end_balance != capital(非空转——有费用+盈亏)
end_balance = stats.get("end_balance")
assert end_balance is not None, "statistics 缺 end_balance"
assert abs(end_balance - 1_000_000) > 1.0, f"end_balance={end_balance} 与 capital 几乎相同(空转)"
# C2: 做空拦截——无 SHORT+OPEN
if trades is not None and not trades.empty:
short_opens = trades[
(trades["direction"].str.contains("SHORT"))
& (trades["offset"].str.contains("OPEN"))
]
assert len(short_opens) == 0, f"存在 SHORT+OPEN 成交(做空未拦截): {short_opens}"
# C1: 定寸生效——成交金额 ≈ 满仓量级(size=N 方案:volume=1 手,turnover=1*N*price
sizing_shares_per_lot = stats.get("sizing_shares_per_lot", 0)
if trades is not None and not trades.empty and sizing_shares_per_lot > 0:
first_trade = trades.iloc[0]
turnover = first_trade["volume"] * sizing_shares_per_lot * first_trade["price"]
assert turnover > capital * position_pct * 0.5, (
f"首笔成交金额={turnover:.0f} 未达满仓量级 "
f"(capital={capital} pct={position_pct} N={sizing_shares_per_lot})"
)
# 非噪声:|total_return| > 1e-3
total_return = stats.get("total_return")
if total_return is not None:
assert abs(total_return) > 1e-3, f"|total_return|={abs(total_return)} <= 1e-3(噪声)"
# H3: 费用可见——statistics 含 stamp_duty 或 commission > 0
total_commission = stats.get("total_commission", 0)
assert total_commission > 0, f"total_commission={total_commission}(费用未计入)"