web_development/backend/app/services/platform_manager.py

"""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()