scripts/launch_platform.py

#!/usr/bin/env python3
"""One-command launcher for the unified trading + research platform."""

from __future__ import annotations

import argparse
import os
import signal
import subprocess
import sys
import time
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[1]
WEB_ROOT = PROJECT_ROOT / "web_development"
BACKEND_DIR = WEB_ROOT / "backend"
FRONTEND_DIR = WEB_ROOT / "frontend"


def _popen(cmd: list[str], cwd: Path, env: dict | None = None) -> subprocess.Popen:
    merged = os.environ.copy()
    if env:
        merged.update(env)
    return subprocess.Popen(
        cmd,
        cwd=str(cwd),
        env=merged,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        start_new_session=True,
    )


def start_backend(host: str, port: int, reload: bool) -> subprocess.Popen:
    args = [
        sys.executable, "-m", "uvicorn",
        "app.main:app",
        "--host", host,
        "--port", str(port),
    ]
    if reload:
        args.append("--reload")
    return _popen(args, BACKEND_DIR)


def start_frontend(host: str, port: int) -> subprocess.Popen:
    return _popen(
        ["npm", "run", "dev", "--", "--host", host, "--port", str(port)],
        FRONTEND_DIR,
    )


def enable_qlib_research(host: str, port: int) -> bool:
    import urllib.error
    import urllib.request

    url = f"http://{host}:{port}/api/platform/services/qlib_research/start"
    req = urllib.request.Request(url, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            return resp.status == 200
    except urllib.error.URLError as exc:
        print(f"[warn] Could not auto-start qlib research: {exc}")
        return False


def main() -> int:
    parser = argparse.ArgumentParser(description="Launch ML-Alpha unified platform")
    parser.add_argument("--host", default="0.0.0.0")
    parser.add_argument("--backend-port", type=int, default=8000)
    parser.add_argument("--frontend-port", type=int, default=5173)
    parser.add_argument("--backend-only", action="store_true")
    parser.add_argument("--frontend-only", action="store_true")
    parser.add_argument("--no-reload", action="store_true")
    parser.add_argument("--enable-research", action="store_true", help="POST start qlib_research after backend is up")
    parser.add_argument("--start-all-modules", action="store_true", help="Start market/strategies/research via platform API")
    args = parser.parse_args()

    procs: list[subprocess.Popen] = []

    if not args.frontend_only:
        print(f"[launch] Backend  http://{args.host}:{args.backend_port}/docs")
        procs.append(start_backend(args.host, args.backend_port, reload=not args.no_reload))

    if not args.backend_only:
        print(f"[launch] Frontend http://{args.host}:{args.frontend_port}/")
        procs.append(start_frontend(args.host, args.frontend_port))

    def shutdown(*_):
        print("\n[launch] Shutting down...")
        for p in procs:
            if p.poll() is None:
                try:
                    os.killpg(os.getpgid(p.pid), signal.SIGTERM)
                except ProcessLookupError:
                    pass
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    if args.enable_research or args.start_all_modules:
        time.sleep(3)
        host = "127.0.0.1" if args.host == "0.0.0.0" else args.host
        if args.start_all_modules:
            import urllib.request
            try:
                urllib.request.urlopen(
                    f"http://{host}:{args.backend_port}/api/platform/services/start-all",
                    method="POST",
                )
                print("[launch] All platform modules started")
            except Exception as exc:
                print(f"[warn] start-all failed: {exc}")
        elif args.enable_research:
            if enable_qlib_research(host, args.backend_port):
                print("[launch] Qlib research module enabled")

    print("[launch] Running. Ctrl+C to stop.")
    try:
        while True:
            for p in procs:
                if p.poll() is not None:
                    out = (p.stdout.read() or b"").decode(errors="replace")
                    print(out)
                    print(f"[launch] Process exited with code {p.returncode}")
                    shutdown()
            time.sleep(1)
    except KeyboardInterrupt:
        shutdown()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())