eth_hedge_sim/control/backend/app/api/nodes.py

from __future__ import annotations

import asyncio
import json
import secrets
import time
from typing import Annotated, Any, AsyncIterator

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from ..auth import require_control_user
from ..config import get_control_settings
from ..crypto import seal
from ..db import get_control_db
from ..proxy import call_node, probe_health

router = APIRouter(prefix="/api/nodes", tags=["nodes"])


def _public_node(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": row["id"],
        "name": row["name"],
        "base_url": row["base_url"],
        "token_configured": bool((row.get("token_sealed") or "").strip()),
        "created_at_ms": row["created_at_ms"],
        "updated_at_ms": row["updated_at_ms"],
    }


def _http_detail(data: Any) -> str:
    if isinstance(data, dict):
        d = data.get("detail", data)
        return d if isinstance(d, str) else str(d)
    return str(data)


def _raise_node_error(code: int, data: Any) -> None:
    """策略机错误不得用 401 回传，否则中控前端会误清登录态。"""
    detail = _http_detail(data)
    if code in (401, 403):
        raise HTTPException(
            status_code=502,
            detail=f"策略机鉴权失败（{detail}）。请在中控重新生成 Token，并到策略机「系统设置→登录账户」保存同一 Token。",
        )
    raise HTTPException(
        status_code=code if 400 <= code < 600 else 502,
        detail=detail,
    )


async def _collect_one_status(
    node: dict[str, Any],
    *,
    timeout: float | None = None,
) -> dict[str, Any]:
    """有 Token 时只打 fleet/status（含在线+持仓），避免 /health 再多一跳。"""
    base = _public_node(node)
    if not node.get("token_sealed"):
        probe = await probe_health(node, timeout=timeout)
        return {
            **base,
            **probe,
            "fleet_ok": False,
            "fleet_error": "未生成 Token",
        }

    code, data = await call_node(
        node, "GET", "/api/fleet/status", timeout=timeout
    )
    if code == 200 and isinstance(data, dict):
        return {
            **base,
            "online": True,
            "health": None,
            "error": None,
            "from_fleet": True,
            "fleet": data,
            "fleet_ok": True,
        }

    # fleet 失败时再探 /health，区分离线 vs Token 错误
    probe = await probe_health(node, timeout=timeout)
    item: dict[str, Any] = {
        **base,
        **probe,
        "fleet_ok": False,
        "fleet_error": _http_detail(data) if data else f"HTTP {code}",
    }
    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)


class NodeUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=64)
    base_url: str | None = Field(default=None, min_length=8, max_length=256)


@router.get("/")
async def list_nodes(_user: Annotated[str, Depends(require_control_user)]) -> dict:
    db = get_control_db()
    nodes = [_public_node(n) for n in db.list_nodes()]
    return {"nodes": nodes}


@router.post("/")
async def create_node(
    body: NodeCreate,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    try:
        row = db.create_node(body.name, body.base_url)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"创建失败: {e}") from e
    return _public_node(row)


@router.get("/status/all")
async def status_all(_user: Annotated[str, Depends(require_control_user)]) -> dict:
    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,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    ids = body.get("ids") or []
    if not isinstance(ids, list) or not ids:
        raise HTTPException(status_code=400, detail="ids 不能为空")
    db = get_control_db()
    results = []
    for nid in ids:
        node = db.get_node(int(nid))
        if not node:
            results.append({"id": nid, "ok": False, "detail": "不存在"})
            continue
        code, data = await call_node(node, "POST", "/api/fleet/update")
        results.append(
            {
                "id": nid,
                "ok": code < 400,
                "status": code,
                "result": data,
            }
        )
    return {"results": results}


@router.post("/start-batch")
async def start_batch(
    body: dict,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """并行启动多台策略机（Fleet start）。"""
    ids = body.get("ids") or []
    if not isinstance(ids, list) or not ids:
        raise HTTPException(status_code=400, detail="ids 不能为空")
    db = get_control_db()

    async def _one(nid: int) -> dict:
        node = db.get_node(int(nid))
        if not node:
            return {"id": nid, "ok": False, "detail": "不存在", "name": str(nid)}
        name = str(node.get("name") or nid)
        if not node.get("token_sealed"):
            return {"id": nid, "ok": False, "detail": "未生成 Token", "name": name}
        try:
            code, data = await call_node(node, "POST", "/api/fleet/start", timeout=20.0)
        except Exception as ex:
            return {"id": nid, "ok": False, "detail": str(ex), "name": name}
        detail = ""
        if code >= 400:
            detail = _http_detail(data) if data else f"HTTP {code}"
        return {
            "id": nid,
            "name": name,
            "ok": code < 400,
            "status": code,
            "detail": detail,
            "result": data,
        }

    items = list(await asyncio.gather(*[_one(int(x)) for x in ids]))
    ok_n = sum(1 for x in items if x.get("ok"))
    return {"ok": ok_n == len(items), "started": ok_n, "total": len(items), "results": items}


@router.patch("/{node_id}")
async def update_node(
    node_id: int,
    body: NodeUpdate,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    row = db.update_node(node_id, name=body.name, base_url=body.base_url)
    if not row:
        raise HTTPException(status_code=404, detail="节点不存在")
    return _public_node(row)


@router.delete("/{node_id}")
async def delete_node(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    if not db.delete_node(node_id):
        raise HTTPException(status_code=404, detail="节点不存在")
    return {"ok": True}


@router.post("/{node_id}/generate-token")
async def generate_token(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """生成新 Token，加密存中控；明文仅返回一次，需粘贴到策略机设置。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    plain = secrets.token_urlsafe(32)
    sealed = seal(plain, get_control_settings().control_auth_secret)
    db.update_node(node_id, token_sealed=sealed)
    return {
        "ok": True,
        "token": plain,
        "msg": "请立即复制并到策略机「系统设置 → 登录账户 → 中控 API Token」保存",
    }


@router.get("/{node_id}/status")
async def node_status(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    return await _collect_one_status(node)


def _max_single_loss_from_curve(curve: list[Any]) -> float:
    pnls = [float(x.get("realized_pnl") or 0) for x in curve if isinstance(x, dict)]
    if not pnls:
        return 0.0
    worst = min(pnls)
    return float(worst) if worst < 0 else 0.0


def _loss_streak_from_curve(curve: list[Any]) -> int:
    streak = 0
    for item in reversed(curve):
        if not isinstance(item, dict):
            continue
        if float(item.get("realized_pnl") or 0) < 0:
            streak += 1
        else:
            break
    return streak


async def _collect_one_stats(node: dict[str, Any]) -> dict[str, Any]:
    base = {
        "id": node["id"],
        "name": node["name"],
        "ok": False,
        "error": None,
        "initial_funds": None,
        "latest_funds": None,
        "groups": None,
        "fees_perp": None,
        "fees_option": None,
        "total_fees": None,
        "max_single_loss": None,
        "loss_streak": None,
        "total_pnl": None,
    }
    if not node.get("token_sealed"):
        base["error"] = "未生成 Token"
        return base
    try:
        code, data = await call_node(node, "GET", "/api/fleet/stats", timeout=15.0)
    except Exception as ex:
        base["error"] = str(ex)
        return base
    if code >= 400 or not isinstance(data, dict):
        base["error"] = _http_detail(data) if data else f"HTTP {code}"
        return base

    curve = data.get("equity_curve") if isinstance(data.get("equity_curve"), list) else []
    max_loss = data.get("max_single_loss")
    if max_loss is None:
        max_loss = _max_single_loss_from_curve(curve)
    loss_streak = data.get("loss_streak")
    if loss_streak is None:
        loss_streak = _loss_streak_from_curve(curve)

    latest_funds = data.get("latest_funds")
    if latest_funds is None:
        # 旧版策略机 stats 无资金字段：回落 status.latest_funds
        try:
            sc, sd = await call_node(node, "GET", "/api/fleet/status", timeout=8.0)
            if sc < 400 and isinstance(sd, dict) and sd.get("latest_funds") is not None:
                latest_funds = sd.get("latest_funds")
        except Exception:
            pass
    if latest_funds is None:
        latest_funds = 0.0

    initial_funds = data.get("initial_funds")
    if initial_funds is None:
        # 旧版策略机：用最新资金 − 已实现盈亏近似初始资金
        try:
            initial_funds = float(latest_funds) - float(data.get("total_pnl") or 0)
        except Exception:
            initial_funds = None

    base.update(
        {
            "ok": True,
            "initial_funds": initial_funds,
            "latest_funds": latest_funds,
            "groups": data.get("groups"),
            "fees_perp": data.get("fees_perp"),
            "fees_option": data.get("fees_option"),
            "total_fees": data.get("total_fees"),
            "max_single_loss": max_loss,
            "loss_streak": loss_streak,
            "total_pnl": data.get("total_pnl"),
            "mode": data.get("mode"),
        }
    )
    return base


@router.get("/stats/all")
async def stats_all(_user: Annotated[str, Depends(require_control_user)]) -> dict:
    """并行拉取各策略机统计，供监控区「数据统计」表。"""
    db = get_control_db()
    nodes = db.list_nodes()
    if not nodes:
        return {"nodes": []}
    items = list(await asyncio.gather(*[_collect_one_stats(n) for n in nodes]))
    return {"nodes": items}


@router.get("/{node_id}/stats")
async def node_stats(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """代理策略机整体统计（点击详情时按需拉取，不进 SSE）。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    if not node.get("token_sealed"):
        raise HTTPException(status_code=400, detail="未生成 Token")
    code, data = await call_node(node, "GET", "/api/fleet/stats", timeout=15.0)
    if code >= 400:
        _raise_node_error(code, data)
    if not isinstance(data, dict):
        raise HTTPException(status_code=502, detail="策略机统计返回异常")
    return data


class ResidualCloseBody(BaseModel):
    group_id: str = Field(min_length=1, max_length=128)


@router.post("/{node_id}/residual/close")
async def node_residual_close(
    node_id: int,
    body: ResidualCloseBody,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """代理策略机手动平单条残留（只验流动性）。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    if not node.get("token_sealed"):
        raise HTTPException(status_code=400, detail="未生成 Token")
    code, data = await call_node(
        node,
        "POST",
        "/api/fleet/residual/close",
        json_body={"group_id": body.group_id},
        timeout=30.0,
    )
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/start")
async def node_start(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/start")
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/pause")
async def node_pause(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/pause")
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/update")
async def node_update(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/update")
    if code >= 400:
        _raise_node_error(code, data)
    return {"ok": True, "result": data}


@router.post("/{node_id}/login-url")
async def node_login_url(
    node_id: int,
    _user: Annotated[str, Depends(require_control_user)],
) -> dict:
    """用 Fleet Token 向策略机签发一次性登录票，返回可打开的 URL。"""
    db = get_control_db()
    node = db.get_node(node_id)
    if not node:
        raise HTTPException(status_code=404, detail="节点不存在")
    code, data = await call_node(node, "POST", "/api/fleet/issue-login")
    if code >= 400:
        _raise_node_error(code, data)
    path = ""
    if isinstance(data, dict):
        path = str(data.get("login_path") or "")
    if not path:
        raise HTTPException(status_code=502, detail="策略机未返回 login_path")
    base = str(node["base_url"]).rstrip("/")
    return {
        "ok": True,
        "url": f"{base}{path}",
        "expires_in": data.get("expires_in") if isinstance(data, dict) else None,
    }