File size: 4,514 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | #!/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())
|