From 330b687e837491bd0dfca82c3ef713d7c0051815 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Mon, 6 Jul 2026 11:25:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(orchestrator):=20runner=20=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E8=B0=83=E5=BA=A6=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_orchestrator/runner.py | 69 +++++++++++++++++ tests/orchestrator/test_runner.py | 125 ++++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+) create mode 100644 sanguo_orchestrator/runner.py create mode 100644 tests/orchestrator/test_runner.py diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py new file mode 100644 index 0000000..b32988b --- /dev/null +++ b/sanguo_orchestrator/runner.py @@ -0,0 +1,69 @@ +""" +Orchestrator for task coordination and execution +Manages backtesting tasks with lazy imports +""" +from .pool import TaskPool +from .task import TaskState + + +class Orchestrator: + """Task coordinator for backtesting operations""" + + def __init__(self, db_path: str, file_dir=None, max_workers: int = 2): + """Initialize orchestrator with database path and worker limits""" + self.db_path = db_path + self.file_dir = file_dir + self.pool = TaskPool(max_workers=max_workers) + self._pending = {} + + def submit_cta(self, strategy_class, symbol: str, params: dict, + start: str, end: str, cfg) -> str: + """Submit a CTA backtesting task""" + task_id = f"cta_{symbol}_{id(params)}" + self.pool.submit(task_id, "cta") + self._pending = dict( + strategy_class=strategy_class, + symbol=symbol, + params=params, + start=start, + end=end, + cfg=cfg + ) + return task_id + + def _run_sync(self, task_id: str): + """Execute a task synchronously (lazy import)""" + # Lazy import to avoid vnpy dependency issues + from sanguo_backtest.cta_engine import run_cta_backtest + + task = self.pool.get_task(task_id) + task.start() + + try: + result = run_cta_backtest( + self._pending["strategy_class"], + self._pending["symbol"], + self._pending["params"], + self._pending["start"], + self._pending["end"], + self._pending["cfg"], + self.db_path + ) + task.complete(result_id=id(result)) + except Exception as e: + task.fail(f"{type(e).__name__}: {e}") + + return task + + def get_status(self, task_id: str) -> TaskState | None: + """Get task status by ID""" + return self.pool.get_status(task_id) + + def get_result(self, task_id: str): + """Get task result by ID (lazy import)""" + task = self.pool.get_task(task_id) + if task and task.status == TaskState.DONE and task.result_id: + # Lazy import to avoid vnpy dependency issues + from sanguo_backtest.result_store import load_result + return load_result(task.result_id, self.db_path) + return None \ No newline at end of file diff --git a/tests/orchestrator/test_runner.py b/tests/orchestrator/test_runner.py new file mode 100644 index 0000000..9278f14 --- /dev/null +++ b/tests/orchestrator/test_runner.py @@ -0,0 +1,125 @@ +""" +Tests for sanguo_orchestrator.runner module +Tests Orchestrator task coordination +""" +import pytest +from unittest.mock import Mock, patch +from sanguo_orchestrator.runner import Orchestrator +from sanguo_orchestrator.task import TaskState + + +class TestOrchestrator: + """Test Orchestrator initialization and task management""" + + def test_orchestrator_initialization(self): + """Test Orchestrator initializes correctly""" + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + assert orchestrator.db_path == "test.db" + assert orchestrator.pool.max_workers == 2 + assert orchestrator._pending == {} + + @patch('sanguo_orchestrator.runner.TaskPool') + def test_submit_cta_creates_task(self, mock_pool_class): + """Test submit_cta() creates task and stores spec""" + mock_pool = Mock() + mock_pool_class.return_value = mock_pool + mock_pool.submit.return_value = Mock(task_id="test_1") + + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + strategy_class = Mock + symbol = "AAPL" + params = {"param1": "value1"} + start = "2024-01-01" + end = "2024-12-31" + cfg = Mock() + + task_id = orchestrator.submit_cta(strategy_class, symbol, params, start, end, cfg) + + # Verify task was submitted to pool + mock_pool.submit.assert_called_once() + call_args = mock_pool.submit.call_args + assert call_args[0][0] == task_id # task_id + assert call_args[0][1] == "cta" # task_type + + # Verify pending spec was stored + assert orchestrator._pending["strategy_class"] == strategy_class + assert orchestrator._pending["symbol"] == symbol + assert orchestrator._pending["params"] == params + assert orchestrator._pending["start"] == start + assert orchestrator._pending["end"] == end + assert orchestrator._pending["cfg"] == cfg + + assert task_id.startswith("cta_AAPL_") + + @patch('sanguo_orchestrator.runner.TaskPool') + def test_get_status(self, mock_pool_class): + """Test get_status() delegates to pool""" + mock_pool = Mock() + mock_pool_class.return_value = mock_pool + mock_pool.get_status.return_value = TaskState.PENDING + + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + status = orchestrator.get_status("test_1") + + mock_pool.get_status.assert_called_once_with("test_1") + assert status == TaskState.PENDING + + @patch('sanguo_orchestrator.runner.TaskPool') + def test_get_status_nonexistent(self, mock_pool_class): + """Test get_status() returns None for non-existent task""" + mock_pool = Mock() + mock_pool_class.return_value = mock_pool + mock_pool.get_status.return_value = None + + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + status = orchestrator.get_status("nonexistent") + + assert status is None + + @patch('sanguo_backtest.result_store.load_result') + @patch('sanguo_orchestrator.runner.TaskPool') + def test_get_result_done_task(self, mock_pool_class, mock_load_result): + """Test get_result() returns result for DONE task""" + mock_pool = Mock() + mock_pool_class.return_value = mock_pool + + mock_task = Mock() + mock_task.status = TaskState.DONE + mock_task.result_id = 12345 + mock_pool.get_task.return_value = mock_task + + mock_result = Mock() + mock_load_result.return_value = mock_result + + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + result = orchestrator.get_result("test_1") + + mock_load_result.assert_called_once_with(12345, "test.db") + assert result == mock_result + + @patch('sanguo_orchestrator.runner.TaskPool') + def test_get_result_pending_task(self, mock_pool_class): + """Test get_result() returns None for PENDING task""" + mock_pool = Mock() + mock_pool_class.return_value = mock_pool + + mock_task = Mock() + mock_task.status = TaskState.PENDING + mock_pool.get_task.return_value = mock_task + + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + result = orchestrator.get_result("test_1") + + assert result is None + + @patch('sanguo_orchestrator.runner.TaskPool') + def test_get_result_nonexistent_task(self, mock_pool_class): + """Test get_result() returns None for non-existent task""" + mock_pool = Mock() + mock_pool_class.return_value = mock_pool + mock_pool.get_task.return_value = None + + orchestrator = Orchestrator(db_path="test.db", max_workers=2) + result = orchestrator.get_result("nonexistent") + + assert result is None \ No newline at end of file