claude_dev
cb220619ef
修复 cta_engine 在真数据上的多个 bug（Phase 2 未在真数据验证）： - interval "1d" -> "d"（vnpy Interval.DAILY.value） - capital 0 -> 1_000_000（0 致首笔交易即爆仓，统计全 0） - statistics 改用 calculate_statistics(df)（旧代码误用 calculate_result 拿 DataFrame） - statistics JSON-safe（vnpy 可能含 Timestamp） - test_cta_engine mock 匹配新流程（calculate_statistics 返回统计字典） 验证：diag_cta.py 真实回测 DoubleMaStrategy on 600000 (2024H1, 111 天) → 真实统计 total_return -0.017% / sharpe -1.03 / max_drawdown -2.17 / 1 trade 容器 79 tests passed。
52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""Diagnostic: real CTA backtest of DoubleMaStrategy on 600000 (A-share daily).
|
|
Guarded entry for spawn. Throwaway."""
|
|
import sys
|
|
import os
|
|
import traceback
|
|
|
|
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0"))
|
|
_REPO = os.path.dirname(_VNPY_SRC)
|
|
for _p in (_REPO, _VNPY_SRC):
|
|
if _p not in sys.path:
|
|
sys.path.insert(0, _p)
|
|
|
|
|
|
def main():
|
|
# Configure vnpy DB to the real quant_trading.db BEFORE engine.load_data()
|
|
from vnpy.trader.setting import SETTINGS
|
|
from sanguo_data.config import load_config
|
|
cfg = load_config("/app/config/data_platform.yaml")
|
|
SETTINGS["database.name"] = "sqlite"
|
|
SETTINGS["database.database"] = cfg.data_paths["vnpy_db"]
|
|
print("DB:", SETTINGS["database.database"])
|
|
|
|
from vnpy_ctastrategy.strategies.double_ma_strategy import DoubleMaStrategy
|
|
print("DoubleMaStrategy.parameters:", getattr(DoubleMaStrategy, "parameters", "?"))
|
|
|
|
from sanguo_backtest.cta_engine import run_cta_backtest
|
|
# Classic double-MA params (vnpy example defaults); fixed_size=1
|
|
params = {"fast_window": 10, "slow_window": 20, "fixed_size": 1}
|
|
|
|
print(f"run_cta_backtest DoubleMaStrategy on 600000, 2024-01-01..2024-06-30, params={params}")
|
|
try:
|
|
result = run_cta_backtest(
|
|
DoubleMaStrategy, "600000", params,
|
|
"2024-01-01", "2024-06-30", cfg, "/tmp/cta_results.db",
|
|
)
|
|
print("=== status:", result.status)
|
|
print("=== statistics ===")
|
|
if result.statistics:
|
|
for k, v in result.statistics.items():
|
|
print(f" {k}: {v}")
|
|
else:
|
|
print(" (empty)")
|
|
if result.status == "failed":
|
|
print("=== error ===")
|
|
print(result.error_msg)
|
|
except Exception:
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|