"""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()