feat(api): 轻量 FastAPI 5 路由（无 JWT/WS/前端）

config/backtest.yaml

@@ -0,0 +1,9 @@
+# Sanguo VeighNa Backtest Configuration
+backtest:
+  max_workers: 2
+  db_path: /volume1/stock/sanguo_vnpy/data/backtest_results.db
+  file_dir: /volume1/stock/sanguo_vnpy/data/backtest_files
+
+api:
+  host: 0.0.0.0
+  port: 8000

sanguo_api/__init__.py

@@ -0,0 +1,16 @@
+"""
+Sanguo Quant API Module
+FastAPI-based REST API for backtesting services
+"""
+
+from .app import create_app
+from .schemas import CtaBacktestRequest, OptimizeRequest, FactorAnalysisRequest
+
+__version__ = "0.1.0"
+
+__all__ = [
+    "create_app",
+    "CtaBacktestRequest",
+    "OptimizeRequest",
+    "FactorAnalysisRequest"
+]

sanguo_api/app.py

@@ -0,0 +1,20 @@
+"""
+FastAPI application factory for Sanguo Quant API
+"""
+from fastapi import FastAPI
+from .routes import router, set_orchestrator
+from sanguo_orchestrator.runner import Orchestrator
+
+
+def create_app(db_path: str, file_dir=None) -> FastAPI:
+    """Create FastAPI application with orchestrator"""
+    app = FastAPI(title="Sanguo Quant API")
+
+    # Initialize orchestrator
+    orch = Orchestrator(db_path=db_path, file_dir=file_dir)
+    set_orchestrator(orch)
+
+    # Include routes
+    app.include_router(router, prefix="/api/v1")
+
+    return app

sanguo_api/routes.py

@@ -0,0 +1,72 @@
+"""
+FastAPI routes for Sanguo Quant API
+"""
+from fastapi import APIRouter, HTTPException
+from .schemas import CtaBacktestRequest, OptimizeRequest, FactorAnalysisRequest
+
+
+router = APIRouter()
+_orchestrator = None
+
+
+def set_orchestrator(orch):
+    """Set the global orchestrator instance"""
+    global _orchestrator
+    _orchestrator = orch
+
+
+def get_orchestrator():
+    """Get the global orchestrator instance"""
+    return _orchestrator
+
+
+@router.post("/backtest/cta")
+def submit_cta(req: CtaBacktestRequest):
+    """Submit CTA backtest task"""
+    tid = get_orchestrator().submit_cta(
+        strategy_class=req.strategy,
+        symbol=req.symbol,
+        params=req.params,
+        start=req.start,
+        end=req.end,
+        cfg=None
+    )
+    return {"task_id": tid}
+
+
+@router.post("/backtest/optimize")
+def submit_optimize(req: OptimizeRequest):
+    """Submit optimization task (placeholder)"""
+    # TODO: Implement optimize submission in Phase 3
+    return {"task_id": "pending_impl"}
+
+
+@router.post("/factor/analyze")
+def submit_factor(req: FactorAnalysisRequest):
+    """Submit factor analysis task (placeholder)"""
+    # TODO: Implement factor analysis submission in Phase 3
+    return {"task_id": "pending_impl"}
+
+
+@router.get("/task/{task_id}")
+def get_status(task_id: str):
+    """Get task status"""
+    status = get_orchestrator().get_status(task_id)
+    if status is None:
+        raise HTTPException(status_code=404, detail="task not found")
+    return {
+        "task_id": task_id,
+        "status": status.value if hasattr(status, "value") else str(status)
+    }
+
+
+@router.get("/task/{task_id}/result")
+def get_result(task_id: str):
+    """Get task result"""
+    result = get_orchestrator().get_result(task_id)
+    if result is None:
+        raise HTTPException(status_code=404, detail="result not ready")
+    return {
+        "task_id": task_id,
+        "statistics": result.statistics
+    }

sanguo_api/schemas.py

@@ -0,0 +1,31 @@
+"""
+Pydantic schemas for FastAPI request/response models
+"""
+from pydantic import BaseModel
+
+
+class CtaBacktestRequest(BaseModel):
+    """CTA backtest request schema"""
+    symbol: str
+    strategy: str
+    params: dict = {}
+    start: str
+    end: str
+
+
+class OptimizeRequest(BaseModel):
+    """Optimization request schema"""
+    symbol: str
+    strategy: str
+    grid: dict
+    start: str
+    end: str
+    max_workers: int = 2
+
+
+class FactorAnalysisRequest(BaseModel):
+    """Factor analysis request schema"""
+    symbols: list[str]
+    factor_names: list[str]
+    start: str
+    end: str

tests/api/__init__.py

@@ -0,0 +1 @@
+# API 测试包

tests/api/test_routes.py

@@ -0,0 +1,219 @@
+"""
+FastAPI routes tests using TestClient
+"""
+from unittest.mock import Mock, patch
+import pytest
+from fastapi.testclient import TestClient
+
+
+def test_submit_cta_backtest():
+    """Test POST /api/v1/backtest/cta returns task_id"""
+    # Create app with temporary DB
+    from sanguo_api.app import create_app
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        # Mock get_orchestrator to return mock orchestrator
+        with patch("sanguo_api.routes.get_orchestrator") as mock_get_orch:
+            mock_orch = Mock()
+            mock_orch.submit_cta.return_value = "cta_test_123"
+            mock_get_orch.return_value = mock_orch
+
+            client = TestClient(app)
+            response = client.post(
+                "/api/v1/backtest/cta",
+                json={
+                    "symbol": "600000SH",
+                    "strategy": "DoubleSMA",
+                    "params": {"fast": 5, "slow": 20},
+                    "start": "2024-01-01",
+                    "end": "2024-12-31"
+                }
+            )
+
+            assert response.status_code == 200
+            data = response.json()
+            assert "task_id" in data
+            assert data["task_id"] == "cta_test_123"
+
+
+def test_get_task_status():
+    """Test GET /api/v1/task/{task_id} returns status"""
+    from sanguo_api.app import create_app
+    from sanguo_orchestrator.task import TaskState
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        with patch("sanguo_api.routes.get_orchestrator") as mock_get_orch:
+            mock_orch = Mock()
+            mock_orch.get_status.return_value = TaskState.DONE
+            mock_get_orch.return_value = mock_orch
+
+            client = TestClient(app)
+            response = client.get("/api/v1/task/cta_test_123")
+
+            assert response.status_code == 200
+            data = response.json()
+            assert "task_id" in data
+            assert data["task_id"] == "cta_test_123"
+            assert data["status"] == "done"
+
+
+def test_get_task_status_not_found():
+    """Test GET /api/v1/task/{task_id} returns 404 for unknown task"""
+    from sanguo_api.app import create_app
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        with patch("sanguo_api.routes.get_orchestrator") as mock_get_orch:
+            mock_orch = Mock()
+            mock_orch.get_status.return_value = None
+            mock_get_orch.return_value = mock_orch
+
+            client = TestClient(app)
+            response = client.get("/api/v1/task/unknown_task")
+
+            assert response.status_code == 404
+
+
+def test_invalid_params_returns_422():
+    """Test POST /api/v1/backtest/cta with missing fields returns 422"""
+    from sanguo_api.app import create_app
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        client = TestClient(app)
+        # Missing required field: strategy
+        response = client.post(
+            "/api/v1/backtest/cta",
+            json={
+                "symbol": "600000SH",
+                "params": {"fast": 5, "slow": 20},
+                "start": "2024-01-01",
+                "end": "2024-12-31"
+            }
+        )
+
+        assert response.status_code == 422
+
+
+def test_get_task_result():
+    """Test GET /api/v1/task/{task_id}/result returns statistics"""
+    from sanguo_api.app import create_app
+    from sanguo_orchestrator.task import TaskState
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        with patch("sanguo_api.routes.get_orchestrator") as mock_get_orch:
+            mock_orch = Mock()
+            mock_result = Mock()
+            mock_result.statistics = {"total_trades": 10, "total_return": 0.15}
+            mock_orch.get_result.return_value = mock_result
+            mock_get_orch.return_value = mock_orch
+
+            client = TestClient(app)
+            response = client.get("/api/v1/task/cta_test_123/result")
+
+            assert response.status_code == 200
+            data = response.json()
+            assert "task_id" in data
+            assert data["task_id"] == "cta_test_123"
+            assert "statistics" in data
+            assert data["statistics"]["total_trades"] == 10
+
+
+def test_get_task_result_not_found():
+    """Test GET /api/v1/task/{task_id}/result returns 404 when result not ready"""
+    from sanguo_api.app import create_app
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        with patch("sanguo_api.routes.get_orchestrator") as mock_get_orch:
+            mock_orch = Mock()
+            mock_orch.get_result.return_value = None
+            mock_get_orch.return_value = mock_orch
+
+            client = TestClient(app)
+            response = client.get("/api/v1/task/unknown_task/result")
+
+            assert response.status_code == 404
+
+
+def test_submit_optimize_returns_pending():
+    """Test POST /api/v1/backtest/optimize returns pending placeholder"""
+    from sanguo_api.app import create_app
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        client = TestClient(app)
+        response = client.post(
+            "/api/v1/backtest/optimize",
+            json={
+                "symbol": "600000SH",
+                "strategy": "DoubleSMA",
+                "grid": {"fast": [5, 10], "slow": [20, 30]},
+                "start": "2024-01-01",
+                "end": "2024-12-31",
+                "max_workers": 2
+            }
+        )
+
+        assert response.status_code == 200
+        data = response.json()
+        assert "task_id" in data
+        assert data["task_id"] == "pending_impl"
+
+
+def test_submit_factor_returns_pending():
+    """Test POST /api/v1/factor/analyze returns pending placeholder"""
+    from sanguo_api.app import create_app
+    import tempfile
+    import os
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = os.path.join(tmp, "test.db")
+        app = create_app(db_path=db_path, file_dir=None)
+
+        client = TestClient(app)
+        response = client.post(
+            "/api/v1/factor/analyze",
+            json={
+                "symbols": ["600000SH", "000001SZ"],
+                "factor_names": ["ts_mean_5", "ts_mean_20"],
+                "start": "2024-01-01",
+                "end": "2024-12-31"
+            }
+        )
+
+        assert response.status_code == 200
+        data = response.json()
+        assert "task_id" in data
+        assert data["task_id"] == "pending_impl"