feat(orchestrator): runner 任务调度器

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

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