claude_dev
8d55e414fa
审计发现包装层系统性失真(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.
84 lines
3.6 KiB
Python
84 lines
3.6 KiB
Python
"""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_ctastrategy；Mac 本机自动 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}（费用未计入）"
|