diff --git a/.env.control.example b/.env.control.example index d18f436..17f6ceb 100644 --- a/.env.control.example +++ b/.env.control.example @@ -6,6 +6,9 @@ CONTROL_AUTH_SECRET=change-me-control-secret-please CONTROL_AUTH_TOKEN_VERSION=1 CONTROL_TOKEN_TTL_SEC=604800 CONTROL_POLL_INTERVAL_SEC=8 +# 监控区 SSE 推送间隔（秒）与单机探测超时 +CONTROL_SSE_INTERVAL_SEC=1 +CONTROL_SSE_PROBE_TIMEOUT_SEC=2.5 CONTROL_HTTP_TIMEOUT_SEC=12 # 局域网免登录：1=开启（仅私网 IP），0=关闭 CONTROL_LAN_AUTH_BYPASS=0 diff --git a/control/backend/app/api/auth_routes.py b/control/backend/app/api/auth_routes.py index f0d3199..8bcbaff 100644 --- a/control/backend/app/api/auth_routes.py +++ b/control/backend/app/api/auth_routes.py @@ -92,6 +92,7 @@ async def me( return { "username": username, "poll_interval_sec": settings.control_poll_interval_sec, + "sse_interval_sec": settings.control_sse_interval_sec, "show_default_hint": settings.is_default_credentials, "lan_bypass_enabled": settings.lan_auth_bypass, } diff --git a/control/backend/app/api/nodes.py b/control/backend/app/api/nodes.py index 75208e0..e9bfd10 100644 --- a/control/backend/app/api/nodes.py +++ b/control/backend/app/api/nodes.py @@ -1,9 +1,13 @@ from __future__ import annotations +import asyncio +import json import secrets -from typing import Annotated, Any +import time +from typing import Annotated, Any, AsyncIterator -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from ..auth import require_control_user @@ -47,6 +51,50 @@ def _raise_node_error(code: int, data: Any) -> None: ) +def _health_looks_like_fleet(health: Any) -> bool: + return isinstance(health, dict) and "strategy" in health and "mode" in health + + +async def _collect_one_status( + node: dict[str, Any], + *, + timeout: float | None = None, +) -> dict[str, Any]: + probe = await probe_health(node, timeout=timeout) + item: dict[str, Any] = {**_public_node(node), **probe} + if probe.get("online") and node.get("token_sealed"): + health = probe.get("health") + # probe_health 在 /health 失败时可能已用 fleet/status 回填到 health + if _health_looks_like_fleet(health): + code, data = 200, health + else: + code, data = await call_node( + node, "GET", "/api/fleet/status", timeout=timeout + ) + if code == 200 and isinstance(data, dict): + item["fleet"] = data + item["fleet_ok"] = True + else: + item["fleet_ok"] = False + item["fleet_error"] = _http_detail(data) if data else f"HTTP {code}" + elif not node.get("token_sealed"): + item["fleet_ok"] = False + item["fleet_error"] = "未生成 Token" + return item + + +async def collect_all_status(*, timeout: float | None = None) -> list[dict[str, Any]]: + db = get_control_db() + nodes = db.list_nodes() + if not nodes: + return [] + return list( + await asyncio.gather( + *[_collect_one_status(n, timeout=timeout) for n in nodes] + ) + ) + + class NodeCreate(BaseModel): name: str = Field(min_length=1, max_length=64) base_url: str = Field(min_length=8, max_length=256) @@ -79,26 +127,65 @@ async def create_node( @router.get("/status/all") async def status_all(_user: Annotated[str, Depends(require_control_user)]) -> dict: - db = get_control_db() - items = [] - for node in db.list_nodes(): - probe = await probe_health(node) - item = {**_public_node(node), **probe} - if probe.get("online") and node.get("token_sealed"): - code, data = await call_node(node, "GET", "/api/fleet/status") - if code == 200 and isinstance(data, dict): - item["fleet"] = data - item["fleet_ok"] = True - else: - item["fleet_ok"] = False - item["fleet_error"] = _http_detail(data) if data else f"HTTP {code}" - elif not node.get("token_sealed"): - item["fleet_ok"] = False - item["fleet_error"] = "未生成 Token" - items.append(item) + items = await collect_all_status() return {"nodes": items} +@router.get("/status/stream") +async def status_stream( + request: Request, + _user: Annotated[str, Depends(require_control_user)], +) -> StreamingResponse: + """SSE：中控约每秒并行拉取策略机状态并推送到浏览器。""" + settings = get_control_settings() + interval = max(0.5, float(settings.control_sse_interval_sec)) + probe_timeout = max(0.5, float(settings.control_sse_probe_timeout_sec)) + + async def event_gen() -> AsyncIterator[str]: + last_payload = "" + last_heartbeat = 0.0 + yield f": connected interval={interval}\n\n" + while True: + if await request.is_disconnected(): + break + started = time.monotonic() + try: + items = await collect_all_status(timeout=probe_timeout) + payload = json.dumps( + {"nodes": items, "ts_ms": int(time.time() * 1000)}, + ensure_ascii=False, + separators=(",", ":"), + default=str, + ) + now = time.monotonic() + if payload != last_payload: + last_payload = payload + yield f"event: nodes\ndata: {payload}\n\n" + last_heartbeat = now + elif now - last_heartbeat >= 5.0: + yield f": heartbeat {int(time.time())}\n\n" + last_heartbeat = now + except Exception as e: + err = json.dumps( + {"detail": str(e)}, + ensure_ascii=False, + separators=(",", ":"), + ) + yield f"event: error\ndata: {err}\n\n" + elapsed = time.monotonic() - started + await asyncio.sleep(max(0.05, interval - elapsed)) + + return StreamingResponse( + event_gen(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + @router.post("/update-batch") async def update_batch( body: dict, @@ -179,13 +266,7 @@ async def node_status( node = db.get_node(node_id) if not node: raise HTTPException(status_code=404, detail="节点不存在") - probe = await probe_health(node) - out = {**_public_node(node), **probe} - if probe.get("online") and node.get("token_sealed"): - code, data = await call_node(node, "GET", "/api/fleet/status") - if code == 200 and isinstance(data, dict): - out["fleet"] = data - return out + return await _collect_one_status(node) @router.post("/{node_id}/start") diff --git a/control/backend/app/config.py b/control/backend/app/config.py index e004f47..5053c5e 100644 --- a/control/backend/app/config.py +++ b/control/backend/app/config.py @@ -31,6 +31,8 @@ class ControlSettings(BaseSettings): control_token_ttl_sec: int = 7 * 24 * 3600 control_db_path: str = "" control_poll_interval_sec: int = 8 + control_sse_interval_sec: float = 1.0 + control_sse_probe_timeout_sec: float = 2.5 control_http_timeout_sec: float = 12.0 control_port: int = 5160 # "1"/"0"：局域网客户端免密登录 diff --git a/control/backend/app/envfile.py b/control/backend/app/envfile.py index d702ea3..88e19e8 100644 --- a/control/backend/app/envfile.py +++ b/control/backend/app/envfile.py @@ -71,6 +71,8 @@ _DEPLOY_DEFAULTS: dict[str, str] = { "CONTROL_AUTH_SECRET": "change-me-control-secret-please", "CONTROL_TOKEN_TTL_SEC": "604800", "CONTROL_POLL_INTERVAL_SEC": "8", + "CONTROL_SSE_INTERVAL_SEC": "1", + "CONTROL_SSE_PROBE_TIMEOUT_SEC": "2.5", "CONTROL_HTTP_TIMEOUT_SEC": "12", "CONTROL_AUTH_TOKEN_VERSION": "1", "CONTROL_LAN_AUTH_BYPASS": "0", diff --git a/control/backend/app/main.py b/control/backend/app/main.py index f802793..9c506fd 100644 --- a/control/backend/app/main.py +++ b/control/backend/app/main.py @@ -49,7 +49,7 @@ app.include_router(api_router) @app.get("/health") async def health() -> dict: s = get_control_settings() - return {"ok": True, "app": "control", "poll_interval_sec": s.control_poll_interval_sec} + return {"ok": True, "app": "control", "sse_interval_sec": s.control_sse_interval_sec} _DIST = resolve_frontend_dist() diff --git a/control/backend/app/proxy.py b/control/backend/app/proxy.py index 92447b6..0fad4ac 100644 --- a/control/backend/app/proxy.py +++ b/control/backend/app/proxy.py @@ -32,6 +32,7 @@ async def call_node( *, require_token: bool = True, json_body: dict | None = None, + timeout: float | None = None, ) -> tuple[int, Any]: settings = get_control_settings() base = str(node["base_url"]).rstrip("/") @@ -43,9 +44,9 @@ async def call_node( headers["X-Fleet-Token"] = tok headers["Authorization"] = f"Fleet {tok}" url = f"{base}{path}" - timeout = settings.control_http_timeout_sec + to = float(timeout if timeout is not None else settings.control_http_timeout_sec) try: - async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client: + async with httpx.AsyncClient(timeout=to, follow_redirects=False) as client: res = await client.request(method, url, headers=headers, json=json_body) # 少数反代会 301/302 补尾斜杠；手动跟一次并保留 Token 头 if res.status_code in (301, 302, 307, 308) and res.headers.get("location"): @@ -64,12 +65,20 @@ async def call_node( return 502, {"detail": f"连接失败: {e}"} -async def probe_health(node: dict[str, Any]) -> dict[str, Any]: - code, data = await call_node(node, "GET", "/health", require_token=False) +async def probe_health( + node: dict[str, Any], + *, + timeout: float | None = None, +) -> dict[str, Any]: + code, data = await call_node( + node, "GET", "/health", require_token=False, timeout=timeout + ) if code == 200 and isinstance(data, dict): return {"online": True, "health": data, "error": None} # fallback fleet status if health blocked - code2, data2 = await call_node(node, "GET", "/api/fleet/status", require_token=True) + code2, data2 = await call_node( + node, "GET", "/api/fleet/status", require_token=True, timeout=timeout + ) if code2 == 200 and isinstance(data2, dict): return {"online": True, "health": data2, "error": None} detail = "" diff --git a/control/frontend/src/pages/Monitor.tsx b/control/frontend/src/pages/Monitor.tsx index 4919cfd..b404bb6 100644 --- a/control/frontend/src/pages/Monitor.tsx +++ b/control/frontend/src/pages/Monitor.tsx @@ -1,5 +1,5 @@ import { useCallback, useEffect, useState } from "react"; -import { apiFetch, type NodeCard } from "../api"; +import { apiFetch, clearSession, getToken, type NodeCard } from "../api"; function pickStrategy(n: NodeCard) { const fleet = (n.fleet || {}) as Record; @@ -55,7 +55,8 @@ export default function MonitorPage() { const [err, setErr] = useState(""); const [busy, setBusy] = useState>({}); const [selected, setSelected] = useState>(new Set()); - const [pollSec, setPollSec] = useState(8); + const [sseSec, setSseSec] = useState(1); + const [liveState, setLiveState] = useState<"connecting" | "live" | "retry">("connecting"); const [detailId, setDetailId] = useState(null); const refresh = useCallback(async () => { @@ -69,15 +70,109 @@ export default function MonitorPage() { }, []); useEffect(() => { - apiFetch<{ poll_interval_sec?: number }>("/api/auth/me") + apiFetch<{ sse_interval_sec?: number }>("/api/auth/me") .then((m) => { - if (m.poll_interval_sec) setPollSec(m.poll_interval_sec); + if (m.sse_interval_sec && Number(m.sse_interval_sec) > 0) { + setSseSec(Number(m.sse_interval_sec)); + } }) .catch(() => undefined); + }, []); + + useEffect(() => { + let cancelled = false; + let retryMs = 1000; + let timer: number | undefined; + const ac = new AbortController(); + void refresh(); - const id = window.setInterval(() => void refresh(), pollSec * 1000); - return () => window.clearInterval(id); - }, [refresh, pollSec]); + + async function runStream() { + while (!cancelled) { + setLiveState(retryMs > 1000 ? "retry" : "connecting"); + try { + const token = getToken(); + const headers = new Headers({ Accept: "text/event-stream" }); + if (token) headers.set("Authorization", `Bearer ${token}`); + const res = await fetch("/api/nodes/status/stream", { + headers, + signal: ac.signal, + }); + if (res.status === 401) { + clearSession(); + setErr("登录已失效，请重新登录"); + setLiveState("retry"); + break; + } + if (!res.ok || !res.body) { + throw new Error(`SSE HTTP ${res.status}`); + } + setLiveState("live"); + setErr(""); + retryMs = 1000; + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + let eventName = "message"; + while (!cancelled) { + const { done, value } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + const parts = buf.split("\n"); + buf = parts.pop() || ""; + for (const rawLine of parts) { + const line = rawLine.replace(/\r$/, ""); + if (line.startsWith("event:")) { + eventName = line.slice(6).trim() || "message"; + continue; + } + if (line.startsWith("data:")) { + const data = line.slice(5).trim(); + if (!data) continue; + try { + const parsed = JSON.parse(data) as { + nodes?: NodeCard[]; + detail?: string; + }; + if (eventName === "error") { + setErr(parsed.detail || "SSE 错误"); + } else if (parsed.nodes) { + setNodes(parsed.nodes); + setErr(""); + } + } catch { + /* ignore malformed chunk */ + } + eventName = "message"; + continue; + } + if (line === "") { + eventName = "message"; + } + } + } + } catch (ex) { + if (cancelled || (ex instanceof DOMException && ex.name === "AbortError")) { + break; + } + setLiveState("retry"); + setErr(ex instanceof Error ? ex.message : String(ex)); + } + if (cancelled) break; + await new Promise((resolve) => { + timer = window.setTimeout(() => resolve(), retryMs); + }); + retryMs = Math.min(retryMs * 2, 15000); + } + } + + void runStream(); + return () => { + cancelled = true; + ac.abort(); + if (timer != null) window.clearTimeout(timer); + }; + }, [refresh]); async function act(id: number, action: "start" | "pause" | "update" | "login") { if (action === "update") { @@ -158,6 +253,18 @@ export default function MonitorPage() {

监控区

+ + {liveState === "live" + ? `实时 · ${sseSec}s` + : liveState === "retry" + ? "重连中…" + : "连接中…"} + diff --git a/control/frontend/src/styles.css b/control/frontend/src/styles.css index 6285986..9cae66d 100644 --- a/control/frontend/src/styles.css +++ b/control/frontend/src/styles.css @@ -177,6 +177,29 @@ input { .toolbar-actions { display: flex; gap: 8px; + align-items: center; + flex-wrap: wrap; +} + +.live-pill { + font-size: 0.82rem; + color: var(--muted); + border: 1px solid var(--line); + border-radius: 999px; + padding: 4px 10px; + white-space: nowrap; +} + +.live-pill.ok { + color: var(--pnl-pos); + border-color: rgba(46, 229, 154, 0.45); + font-weight: 600; +} + +.live-pill.bad { + color: var(--pnl-neg); + border-color: rgba(255, 92, 92, 0.45); + font-weight: 600; } .card-grid { diff --git a/docs/中控Fleet说明.md b/docs/中控Fleet说明.md index dee69a1..e12e731 100644 --- a/docs/中控Fleet说明.md +++ b/docs/中控Fleet说明.md @@ -81,7 +81,9 @@ bash /opt/eth_hedge_sim/control/deploy/update.sh | `CONTROL_AUTH_USERNAME` / `PASSWORD` | 中控登录 | | `CONTROL_AUTH_SECRET` | Token 加密与签名密钥（部署自动填默认值，有值不覆盖） | | `CONTROL_LAN_AUTH_BYPASS` | `1`=局域网免登录，`0`=关闭 | -| `CONTROL_POLL_INTERVAL_SEC` | 监控轮询间隔 | +| `CONTROL_SSE_INTERVAL_SEC` | 监控 SSE 推送间隔（默认 1 秒） | +| `CONTROL_SSE_PROBE_TIMEOUT_SEC` | 单次探测策略机超时（默认 2.5 秒） | +| `CONTROL_POLL_INTERVAL_SEC` | 兼容保留（旧前端轮询；现以 SSE 为主） | **勿提交** `.env.control` 到 git。 @@ -112,7 +114,9 @@ bash /opt/eth_hedge_sim/control/deploy/update.sh ## 5. 监控区 -- **卡片**：在线/离线、SIM/LIVE、阶段、轮次、行情、Token 状态。 +- **实时推送**：浏览器通过 SSE（`GET /api/nodes/status/stream`）接收状态；中控约每秒**并行**拉取各策略机 `/api/fleet/status`，有变化才推送；断线自动重连。工具栏显示「实时 · 1s」。 +- **刷新**：仍可手动走 `/api/nodes/status/all` 拉一次。 +- **卡片**：在线/离线、SIM/LIVE、阶段、持仓、轮次、行情、Token 状态。 - **运行中**：卡片绿色；底部按钮显示「运行中」且不可点启动。 - **点击卡片**：放大弹层 — 策略详情 + 持仓腿表；**净浮盈 / 浮盈** 正绿负红加粗。 - **登录策略机**：免密新标签打开策略页。