dekun/eth_hedge_sim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add SSE second-level status push for control monitor.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-30 13:32:28 +08:00
parent 7136adcaa9
commit b3ba0fd431
10 changed files with 273 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.control.example
+3
View File
@@ -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
control/backend/app/api/auth_routes.py
+1
View File
@@ -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,
}
control/backend/app/api/nodes.py
+107 -26
View File
@@ -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")
control/backend/app/config.py
+2
View File
@@ -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"：局域网客户端免密登录
control/backend/app/envfile.py
+2
View File
@@ -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",
control/backend/app/main.py
+1 -1
View File
@@ -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()
control/backend/app/proxy.py
+14 -5
View File
@@ -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 = ""
control/frontend/src/pages/Monitor.tsx
+114 -7
View File
@@ -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<string, unknown>;
@@ -55,7 +55,8 @@ export default function MonitorPage() {
const [err, setErr] = useState("");
const [busy, setBusy] = useState<Record<number, string>>({});
const [selected, setSelected] = useState<Set<number>>(new Set());
const [pollSec, setPollSec] = useState(8);
const [sseSec, setSseSec] = useState(1);
const [liveState, setLiveState] = useState<"connecting" | "live" | "retry">("connecting");
const [detailId, setDetailId] = useState<number | null>(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<void>((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() {
<div className="toolbar">
<h2>监控区</h2>
<div className="toolbar-actions">
<span
className={`live-pill ${
liveState === "live" ? "ok" : liveState === "retry" ? "bad" : ""
}`}
title="中控 SSE 秒级推送"
>
{liveState === "live"
? `实时 · ${sseSec}s`
: liveState === "retry"
? "重连中…"
: "连接中…"}
</span>
<button type="button" className="btn ghost" onClick={() => void refresh()}>
刷新
</button>
control/frontend/src/styles.css
+23
View File
@@ -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 {
docs/中控Fleet说明.md
+6 -2
View File
@@ -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 状态。
- **运行中**：卡片绿色；底部按钮显示「运行中」且不可点启动。
- **点击卡片**：放大弹层 — 策略详情 + 持仓腿表；**净浮盈 / 浮盈** 正绿负红加粗。
- **登录策略机**：免密新标签打开策略页。