"""Platform module launcher — start/stop functional services on demand.""" from __future__ import annotations import asyncio import logging import os import signal import subprocess import sys from dataclasses import dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger(__name__) WORKSPACE_ROOT = Path(__file__).resolve().parents[4] WEB_ROOT = WORKSPACE_ROOT / "web_development" FRONTEND_DIR = WEB_ROOT / "frontend" @dataclass class ServiceSpec: id: str name: str description: str kind: str # embedded | process port: int | None = None url: str | None = None tags: list[str] = field(default_factory=list) @dataclass class ServiceStatus: id: str name: str description: str kind: str status: str # running | stopped | error | unavailable port: int | None = None url: str | None = None pid: int | None = None message: str = "" tags: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "name": self.name, "description": self.description, "kind": self.kind, "status": self.status, "port": self.port, "url": self.url, "pid": self.pid, "message": self.message, "tags": self.tags, } SERVICE_CATALOG: list[ServiceSpec] = [ ServiceSpec( id="trading_api", name="期货交易 API", description="FastAPI 后端:行情、下单、持仓、期货策略回测", kind="embedded", port=8000, url="http://localhost:8000/docs", tags=["futures", "core"], ), ServiceSpec( id="market_data", name="行情推送", description="模拟/实时行情 WebSocket 与 K 线生成", kind="embedded", tags=["futures", "market"], ), ServiceSpec( id="futures_strategies", name="期货量化策略", description="MA/Bollinger/DualThrust 策略引擎", kind="embedded", tags=["futures", "strategy"], ), ServiceSpec( id="qlib_research", name="Qlib 因子研究", description="因子公式库、算子构建、单因子 IC/回测、A 股策略", kind="embedded", tags=["research", "qlib", "stock"], ), ServiceSpec( id="frontend_ui", name="Web 可视化前端", description="Vue3 仪表盘、K 线、交易、研究面板", kind="process", port=5173, url="http://localhost:5173", tags=["ui"], ), ] class PlatformManager: def __init__(self): self._processes: dict[str, subprocess.Popen] = {} self._qlib_research_enabled = False self._market_data_running = False self._futures_strategies_enabled = True def _spec(self, service_id: str) -> ServiceSpec: for s in SERVICE_CATALOG: if s.id == service_id: return s raise KeyError(f"Unknown service: {service_id}") def list_services(self) -> list[ServiceStatus]: return [self.get_status(s.id) for s in SERVICE_CATALOG] def get_status(self, service_id: str) -> ServiceStatus: spec = self._spec(service_id) if service_id == "trading_api": return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="running", port=spec.port, url=spec.url, message="当前进程即 Trading API", tags=spec.tags, ) if service_id == "market_data": from app.services.market_data import market_data_service running = market_data_service._running return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="running" if running else "stopped", message="行情 WebSocket 推送" if running else "已停止", tags=spec.tags, ) if service_id == "futures_strategies": from app.services.strategy_engine import strategy_engine n = len(strategy_engine._tasks) return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="running" if self._futures_strategies_enabled else "stopped", message=f"策略引擎就绪,{n} 个策略运行中", tags=spec.tags, ) if service_id == "qlib_research": return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="running" if self._qlib_research_enabled else "stopped", url="/api/research/health", message="Qlib 因子/策略研究模块" if self._qlib_research_enabled else "未启动(按需加载 qlib)", tags=spec.tags, ) if service_id == "frontend_ui": proc = self._processes.get("frontend_ui") if proc and proc.poll() is None: return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="running", port=spec.port, url=spec.url, pid=proc.pid, message="Vue dev server", tags=spec.tags, ) return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="stopped", port=spec.port, url=spec.url, message="前端未启动", tags=spec.tags, ) return ServiceStatus( id=spec.id, name=spec.name, description=spec.description, kind=spec.kind, status="unavailable", message="Unknown service", tags=spec.tags, ) async def start(self, service_id: str) -> ServiceStatus: spec = self._spec(service_id) if service_id == "trading_api": return self.get_status(service_id) if service_id == "market_data": from app.services.market_data import market_data_service if not market_data_service._running: await market_data_service.start() self._market_data_running = True return self.get_status(service_id) if service_id == "futures_strategies": self._futures_strategies_enabled = True return self.get_status(service_id) if service_id == "qlib_research": if str(WORKSPACE_ROOT) not in sys.path: sys.path.insert(0, str(WORKSPACE_ROOT)) os.environ.setdefault("MLFLOW_ALLOW_FILE_STORE", "true") from data_pipeline.init_qlib import init_qlib loop = asyncio.get_event_loop() await loop.run_in_executor(None, init_qlib) self._qlib_research_enabled = True logger.info("Qlib research module enabled") return self.get_status(service_id) if service_id == "frontend_ui": proc = self._processes.get("frontend_ui") if proc and proc.poll() is None: return self.get_status(service_id) if not FRONTEND_DIR.exists(): raise FileNotFoundError(f"Frontend not found: {FRONTEND_DIR}") proc = subprocess.Popen( ["npm", "run", "dev", "--", "--host", "0.0.0.0", "--port", "5173"], cwd=str(FRONTEND_DIR), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True, ) self._processes["frontend_ui"] = proc await asyncio.sleep(2) return self.get_status(service_id) raise ValueError(f"Cannot start service: {service_id}") async def stop(self, service_id: str) -> ServiceStatus: if service_id == "market_data": from app.services.market_data import market_data_service await market_data_service.stop() self._market_data_running = False return self.get_status(service_id) if service_id == "futures_strategies": from app.services.strategy_engine import strategy_engine for sid in list(strategy_engine._tasks.keys()): strategy_engine.stop_strategy(sid) self._futures_strategies_enabled = False return self.get_status(service_id) if service_id == "qlib_research": self._qlib_research_enabled = False return self.get_status(service_id) if service_id == "frontend_ui": proc = self._processes.get("frontend_ui") if proc and proc.poll() is None: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) proc.wait(timeout=5) self._processes.pop("frontend_ui", None) return self.get_status(service_id) if service_id == "trading_api": return ServiceStatus( **self.get_status(service_id).to_dict(), message="Trading API 由 uvicorn 管理,请停止 uvicorn 进程", ) raise ValueError(f"Cannot stop service: {service_id}") async def start_all(self) -> list[ServiceStatus]: results = [] for spec in SERVICE_CATALOG: if spec.id == "trading_api": results.append(self.get_status(spec.id)) continue try: results.append(await self.start(spec.id)) except Exception as exc: st = self.get_status(spec.id) st.status = "error" st.message = str(exc) results.append(st) return results @property def qlib_research_enabled(self) -> bool: return self._qlib_research_enabled platform_manager = PlatformManager()