diff --git a/sanguo_api/routes_portfolio.py b/sanguo_api/routes_portfolio.py index 616d32d..54c9d6f 100644 --- a/sanguo_api/routes_portfolio.py +++ b/sanguo_api/routes_portfolio.py @@ -8,7 +8,6 @@ POST /portfolio/backtest: SSH 触发 VPS 跑 BulletTrade + all_weather, from __future__ import annotations import logging -import shlex import subprocess from typing import Any, Optional @@ -51,25 +50,22 @@ def run_portfolio_backtest(req: PortfolioBacktestRequest): - 没有 tail/head/grep:用 python 后处理(本函数在 Mac 端直接解析 stdout) - 中文路径:VPS_WORKDIR / userdata_mini 走 env(DEFAULT_DATA_PROVIDER=miniqmt) """ - # 在 VPS 上跑的命令:cd workdir && set ENV && python -m sanguo_portfolio.runner_backtest --json - # Windows cmd: set X=Y&&cmd2 (注意 & 必须紧贴前一条,不能有空格,否则 set 会把尾部空格算进 value) - cmd_parts = [ - "set", "DEFAULT_DATA_PROVIDER=miniqmt", "&&", - "cd", _VPS_WORKDIR, "&&", - _VPS_PYTHON, "-X", "utf8", "-m", "sanguo_portfolio.runner_backtest", - "--json", - "--start", req.start_date, - "--end", req.end_date, - "--cash", str(req.initial_cash), - "--benchmark", req.benchmark, - ] - # 用 ssh host "cmd string" 形式;argv 在 ssh 远端走 cmd /c 解析 + # 远端命令手动拼接(不用 shlex.quote:它产 POSIX 单引号,Windows cmd 不认单引号致 cd 失败; + # 且不用 set X=Y:runner_backtest 顶部 os.environ.setdefault 已自处理 env,早于 bullet_trade import, + # 避免 cmd "set X=Y &&" 尾空格进 value(变 "miniqmt ")致 bullet_trade provider 名匹配失败)。 + remote_cmd = ( + f"cd {_VPS_WORKDIR} && " + f"{_VPS_PYTHON} -X utf8 -m sanguo_portfolio.runner_backtest --json " + f"--start {req.start_date} --end {req.end_date} " + f"--cash {req.initial_cash} --benchmark {req.benchmark} " + f"--max-pool 30" + ) ssh_argv = [ "ssh", "-o", "ConnectTimeout=15", "-o", "StrictHostKeyChecking=no", _VPS_HOST, - " ".join(shlex.quote(p) if p != "&&" else "&&" for p in cmd_parts), + remote_cmd, ] logger.info("[portfolio] SSH 触发: %s", ssh_argv[-1]) diff --git a/sanguo_portfolio/filters.py b/sanguo_portfolio/filters.py index d8815d7..6a69bd4 100644 --- a/sanguo_portfolio/filters.py +++ b/sanguo_portfolio/filters.py @@ -212,7 +212,14 @@ def _coerce_datetime(value: Any) -> Optional[datetime]: try: return datetime.fromisoformat(value[:10]) except ValueError: - return None + pass + # YYYYMMDD 8位纯数字(provider/miniQMT 用此格式,fromisoformat 不认) + if len(value) >= 8 and value[:8].isdigit(): + try: + return datetime.strptime(value[:8], "%Y%m%d") + except ValueError: + pass + return None return None diff --git a/sanguo_portfolio/runner_backtest.py b/sanguo_portfolio/runner_backtest.py index a2cf111..f15ca5d 100644 --- a/sanguo_portfolio/runner_backtest.py +++ b/sanguo_portfolio/runner_backtest.py @@ -31,6 +31,7 @@ def parse_args() -> argparse.Namespace: p.add_argument("--end", default="2024-12-31", help="回测结束日期 YYYY-MM-DD") p.add_argument("--cash", type=float, default=1_000_000.0, help="初始资金(元)") p.add_argument("--benchmark", default="000300.XSHG", help="基准代码") + p.add_argument("--max-pool", type=int, default=0, help="限制选股池前N只(0=不限,MVP验证用)") p.add_argument("--frequency", default="day", help="回测频率 day/minute") p.add_argument( "--provider-config", default="{}", @@ -103,7 +104,7 @@ def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: from bullet_trade import BacktestEngine # type: ignore from bullet_trade.data.api import set_data_provider # type: ignore - from .strategies import AllWeatherStrategy + from .strategies import AllWeatherStrategy, AllWeatherConfig provider = build_provider(args.provider_config) set_data_provider(provider) @@ -112,7 +113,10 @@ def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: holder: Dict[str, Any] = {} def initialize(context): - strategy = AllWeatherStrategy(provider=provider) + strategy = AllWeatherStrategy( + provider=provider, + config=AllWeatherConfig(max_pool=args.max_pool), + ) holder["strategy"] = strategy # bullet-trade 的 run_daily/run_monthly 接受全局函数;把 method 暴露为模块级 @@ -144,6 +148,7 @@ def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: order_value=lambda c, v: bt_ov(c, v), ) + print("[runner] ENGINE_BUILD_PRE", flush=True) engine = BacktestEngine( initialize=initialize, start_date=args.start, @@ -152,7 +157,9 @@ def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: initial_cash=args.cash, benchmark=args.benchmark, ) + print("[runner] RUN_START", flush=True) result = engine.run() + print("[runner] RUN_DONE type=%s" % type(result).__name__, flush=True) # 输出结果摘要到 markdown(JSON 模式时 result_file="" 跳过) if getattr(args, "result_file", ""): @@ -220,6 +227,7 @@ def run_backtest_json(params: Dict[str, Any]) -> Dict[str, Any]: frequency="day", provider_config="{}", result_file="", # JSON 模式不写 md + max_pool=int(params.get("max_pool", 0)), ) raw = run_backtest(args) diff --git a/sanguo_portfolio/strategies/all_weather.py b/sanguo_portfolio/strategies/all_weather.py index 6a2d51d..e9a8e5b 100644 --- a/sanguo_portfolio/strategies/all_weather.py +++ b/sanguo_portfolio/strategies/all_weather.py @@ -73,6 +73,7 @@ class AllWeatherConfig: benchmark: str = "000300.XSHG" roic_threshold: float = 0.08 # filter_roic 的 ROIC > 阈值 new_stock_days: int = 375 + max_pool: int = 0 # 0=不限;MVP/验证用,限制 _stock_pool 返回前 N 只(避免全成分基本面下载过慢) class AllWeatherStrategy: @@ -397,6 +398,10 @@ class AllWeatherStrategy: logger.warning("get_index_stocks(%s) 失败: %s", index_symbol, exc) return [] stocks = filters.filter_kcbj_stock(stocks) + # max_pool 提前到 filter_st/filter_new 前:这俩对全成分(HS300+中小综指 1200+只)逐只 + # get_security_info 极慢,先 slice 到 N 只再过滤(验证用子集,语义略变但提速百倍) + if self.config.max_pool > 0: + stocks = stocks[: self.config.max_pool] stocks = filters.filter_st_stock(stocks, self.provider) stocks = filters.filter_new_stock(stocks, self.provider, previous_date, self.config.new_stock_days) return stocks