diff --git a/sanguo_api/routes_factor.py b/sanguo_api/routes_factor.py index 9ae911c..7549def 100644 --- a/sanguo_api/routes_factor.py +++ b/sanguo_api/routes_factor.py @@ -113,5 +113,6 @@ async def eval_submit(req: FactorBatchEvalRequest): task_id = await orch.submit_batch_eval( factor_names=req.factors, categories=req.categories, start=req.start, end=req.end, symbols=req.symbols or None, label=req.label, + run_id=req.run_id or None, ) return {"task_id": task_id} diff --git a/sanguo_api/schemas.py b/sanguo_api/schemas.py index d979244..ee6265e 100644 --- a/sanguo_api/schemas.py +++ b/sanguo_api/schemas.py @@ -52,3 +52,4 @@ class FactorBatchEvalRequest(BaseModel): start: str end: str label: str = "" + run_id: str = "" # 断点续跑:复用既有批次 diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py index ae1ee1a..8ebb626 100644 --- a/sanguo_orchestrator/runner.py +++ b/sanguo_orchestrator/runner.py @@ -191,7 +191,8 @@ class Orchestrator: return task_id async def submit_batch_eval(self, factor_names: list, categories: list, - start: str, end: str, symbols, label: str) -> str: + start: str, end: str, symbols, label: str, + run_id: str | None = None) -> str: """Submit a factor batch evaluation task asynchronously""" task_id = f"factor_eval_{uuid.uuid4().hex[:8]}" self._record_submit(task_id) @@ -199,7 +200,7 @@ class Orchestrator: eval_db = resolve_eval_db(self.db_path) fut: Future = self.pool.submit_work( task_id, _batch_eval_worker, factor_names, categories, - start, end, symbols, label or "batch", eval_db, + start, end, symbols, label or "batch", eval_db, run_id, ) task = self.pool.get_task(task_id) task.start() @@ -363,7 +364,7 @@ def resolve_eval_db(backtest_db_path: str) -> str: def _batch_eval_worker(factor_names: list, categories: list, start: str, end: str, - symbols, label: str, eval_db: str) -> dict: + symbols, label: str, eval_db: str, run_id: str | None = None) -> dict: """进程池 worker:批量评估(spawn 环境,自行 load config/挂载因子).""" from sanguo_factor.alpha_datasets import mount_all mount_all() @@ -371,7 +372,7 @@ def _batch_eval_worker(factor_names: list, categories: list, start: str, end: st from sanguo_factor.registry import list_factors factor_names = [f["name"] for c in categories for f in list_factors(c)] from sanguo_factor.batch_eval import run_batch_eval - return run_batch_eval(factor_names, start, end, eval_db, label=label, symbols=symbols) + return run_batch_eval(factor_names, start, end, eval_db, label=label, symbols=symbols, run_id=run_id) def _opt_worker(strategy_class, symbol: str, grid: dict, start: str, end: str, cfg, db_path: str, task_id: str) -> any: