Files
crypto_okx/lib/hub/hub_bridge_register.py
T

565 lines
23 KiB
Python

"""中控 Bridge:只读快照 + 受限写(买一全平) + SSO 登录入口."""
from __future__ import annotations
from typing import Any
from flask import Flask, jsonify, redirect, request, session
from urllib.parse import quote
from lib.hub.hub_bridge_auth import (
frame_ancestors_header_value,
hub_bridge_token,
hub_bridge_token_ok,
make_embed_session_token,
verify_sso_token,
)
def _safe_next_path(raw: str, default: str = "/embed") -> str:
p = (raw or "").strip() or default
if not p.startswith("/"):
p = "/" + p
if "://" in p or p.startswith("//"):
return default
return p
def _enrich_hub_protocol(snap: dict[str, Any], *, app_module: Any) -> dict[str, Any]:
"""对齐中控设计文档 version=1 字段,同时保留原 snapshot 键."""
out = dict(snap or {})
out["version"] = 1
mode = "live"
try:
from lib.sim.mode_lib import is_sim_mode
get_db = getattr(app_module, "get_db", None)
if callable(get_db) and is_sim_mode(get_db):
mode = "sim"
except Exception:
mode = "live"
label = ""
try:
from lib.sim.mode_lib import exchange_mode_label
get_db = getattr(app_module, "get_db", None)
if callable(get_db):
label = exchange_mode_label(get_db=get_db, exchange_key="okx") or ""
except Exception:
label = ""
if not label:
label = "okx_sim" if mode == "sim" else "okx_live"
positions = out.get("positions") if isinstance(out.get("positions"), list) else []
slim_pos = []
premium_total_usdt = 0.0
has_premium = False
for p in positions:
if not isinstance(p, dict):
continue
preview = p.get("close_preview") if isinstance(p.get("close_preview"), dict) else {}
# 仅在实例已附带 close_preview 时判定残档;缺字段=未同步,勿假装「无买一/残档」
preview_present = bool(preview) and (
"bid_invalid" in preview
or "levels" in preview
or "covered_sheets" in preview
or "liquidity_ok" in preview
)
levels = preview.get("levels") if isinstance(preview.get("levels"), list) else []
top = levels[0] if levels and isinstance(levels[0], dict) else {}
bid_px = top.get("px")
if bid_px is None:
bid_px = preview.get("raw_bid_px")
bid_sz = top.get("available_sheets")
try:
uncovered = int(float(preview.get("uncovered_sheets") or 0)) if preview_present else 0
except (TypeError, ValueError):
uncovered = 0
try:
covered = int(float(preview.get("covered_sheets") or 0)) if preview_present else 0
except (TypeError, ValueError):
covered = 0
bid_invalid = bool(preview.get("bid_invalid")) if preview_present else False
bid_invalid_reason = preview.get("bid_invalid_reason") if preview_present else None
# 残档/无效买一,或买一深度盖不住持仓 → 流动性不足;未同步时不标黄残档
if not preview_present:
liq_ok = None
else:
liq_ok = (not bid_invalid) and uncovered <= 0 and (
preview.get("liquidity_ok") is not False
)
if not levels and not bid_invalid and covered <= 0:
# 已拉盘口但无可用买一档
liq_ok = False
bid_invalid = True
if not bid_invalid_reason:
bid_invalid_reason = "无买一"
pnl_usdt = p.get("display_pnl_usdt")
if pnl_usdt is None:
try:
from lib.options.options_positions_lib import display_pnl_usdt_from_option_row
pnl_usdt = display_pnl_usdt_from_option_row(p)
except Exception:
pnl_usdt = p.get("net_pnl_usdt")
premium_paid = p.get("premium_paid")
premium_usdt = None
try:
from lib.options.options_positions_lib import premium_paid_usdt_from_option_row
premium_usdt = premium_paid_usdt_from_option_row(p)
except Exception:
premium_usdt = None
if premium_usdt is not None:
has_premium = True
premium_total_usdt += float(premium_usdt)
contract_leverage = p.get("contract_leverage")
if contract_leverage is None:
try:
from lib.options.options_pricing_lib import equivalent_contract_leverage
contract_leverage = equivalent_contract_leverage(
index_px=p.get("idx_px") or p.get("index_px"),
eth_amount=p.get("eth_amount"),
total_premium=premium_paid,
margin_mode=p.get("margin_mode"),
)
except Exception:
contract_leverage = None
slim_pos.append(
{
"inst_id": p.get("inst_id"),
"opt_type": p.get("opt_type"),
"sheets": p.get("sheets") or p.get("pos"),
"pnl_usdt": pnl_usdt,
"premium_paid": premium_paid,
"premium_paid_usdt": round(float(premium_usdt), 4) if premium_usdt is not None else None,
"premium_ccy": p.get("premium_ccy"),
"contract_leverage": contract_leverage,
"mark": p.get("mark") or p.get("mark_px"),
"underlying": p.get("underlying"),
"source_label": p.get("source_label"),
"target_index": p.get("target_index"),
"profit_exit_enabled": p.get("profit_exit_enabled"),
"profit_exit_mult": p.get("profit_exit_mult"),
"profit_exit_state": p.get("profit_exit_state"),
"target_monitor_text": p.get("target_monitor_text"),
"bid_px": bid_px,
"bid_sz": bid_sz,
"covered_sheets": covered if preview_present else None,
"uncovered_sheets": uncovered if preview_present else None,
"bid_invalid": bid_invalid,
"bid_invalid_reason": bid_invalid_reason,
"liquidity_ok": liq_ok,
}
)
out["instance"] = {
"label": label,
"mode": mode,
"options_enabled": bool(out.get("enabled")),
"account_label": out.get("account_label") or "",
"margin_mode": out.get("options_margin_mode") or "coin",
"underly": out.get("options_underly") or "ETH",
}
funding_usdt = out.get("funding_usdt")
trading_usdt = out.get("trading_usdt")
funding_usdc = out.get("funding_usdc")
trading_usdc = out.get("trading_usdc")
out["funds"] = {
"funding_usdt": funding_usdt,
"trading_usdt": trading_usdt,
"funding_usdc": funding_usdc,
"trading_usdc": trading_usdc,
}
def _f(v: Any) -> float | None:
if v is None or v == "":
return None
try:
return float(v)
except (TypeError, ValueError):
return None
# 账户可用余额:优先 USDT 资+交;都空则回退 USDC(≈U)
cash_parts = [_f(trading_usdt), _f(funding_usdt)]
if all(x is None for x in cash_parts):
cash_parts = [_f(trading_usdc), _f(funding_usdc)]
cash_known = [x for x in cash_parts if x is not None]
cash_usdt = round(sum(cash_known), 4) if cash_known else None
premium_usdt = round(premium_total_usdt, 4) if has_premium else (0.0 if not slim_pos else None)
# 占比分母 = 当前余额 + 已付权利金(权利金已从余额扣出,需加回才是完整账户资金)
equity_usdt = None
if cash_usdt is not None or premium_usdt is not None:
equity_usdt = round(float(cash_usdt or 0.0) + float(premium_usdt or 0.0), 4)
ratio_pct = None
if premium_usdt is not None and equity_usdt is not None and equity_usdt > 1e-9:
ratio_pct = round(float(premium_usdt) / float(equity_usdt) * 100.0, 2)
out["pnl"] = {
"unrealized_usdt": out.get("upl_total_usdc"),
}
initial_cash_usdt = None
try:
import os
if mode == "sim":
initial_cash_usdt = float(os.getenv("SIM_INITIAL_EQUITY_USDT") or 10000)
else:
for key in (
"LIVE_INITIAL_EQUITY_USDT",
"OPTIONS_INITIAL_EQUITY_USDT",
"HUB_INITIAL_CASH_USDT",
):
raw = (os.getenv(key) or "").strip()
if not raw:
continue
initial_cash_usdt = float(raw)
break
if initial_cash_usdt is not None and initial_cash_usdt <= 0:
initial_cash_usdt = None
elif initial_cash_usdt is not None:
initial_cash_usdt = round(float(initial_cash_usdt), 4)
except (TypeError, ValueError):
initial_cash_usdt = None
out["premium"] = {
"paid_usdt": premium_usdt,
"cash_usdt": cash_usdt,
"equity_usdt": equity_usdt,
"ratio_pct": ratio_pct,
"initial_cash_usdt": initial_cash_usdt,
}
out["positions_brief"] = slim_pos
out["cond_order_locked"] = bool(out.get("cond_order_locked"))
active_cond = out.get("cond_orders_active")
out["cond_orders_active"] = active_cond if isinstance(active_cond, list) else []
try:
from lib.options.options_time_lib import options_app_now_str
out["ts"] = options_app_now_str()
except Exception:
from datetime import datetime
out["ts"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
quotes = {"tickers": [], "updated_at": out.get("ts") or ""}
try:
from lib.market.nav_spot_tickers_lib import build_nav_spot_tickers
ex = getattr(app_module, "exchange", None)
reset_hour = int(getattr(app_module, "TRADING_DAY_RESET_HOUR", 8) or 8)
tickers = build_nav_spot_tickers(ex, reset_hour=reset_hour) if ex is not None else []
quotes = {
"tickers": tickers if isinstance(tickers, list) else [],
"updated_at": out.get("ts") or "",
}
except Exception:
pass
out["quotes"] = quotes
return out
def register_hub_bridge_routes(app: Flask, app_module: Any) -> None:
@app.route("/api/hub/options/snapshot")
def api_hub_options_snapshot():
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
hub_ctx = app.config.get("HUB_CTX") or {}
fn = hub_ctx.get("options_snapshot_fn")
if not callable(fn):
return jsonify(
{"ok": False, "msg": "期权快照未注册", "version": 1, "enabled": False}
), 503
try:
snap = fn() or {}
except Exception as e:
return jsonify({"ok": False, "msg": str(e), "version": 1, "enabled": True}), 500
return jsonify(_enrich_hub_protocol(snap, app_module=app_module))
@app.route("/api/hub/options/stats")
def api_hub_options_stats():
"""中控汇总用:窗口化已实现盈亏统计(不含浮盈进净盈亏主指标以外的展示字段仍附带)."""
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
hub_ctx = app.config.get("HUB_CTX") or {}
cfg = hub_ctx.get("options_cfg")
if not isinstance(cfg, dict):
return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503
from lib.options.options_history_lib import load_options_history
from lib.options.options_margin_mode_lib import is_coin_margin_mode
from lib.options.options_positions_lib import sum_options_net_pnl_usdc
from lib.options.options_stats_lib import (
compute_options_stats_for_window,
history_pnl_to_usdt,
normalize_stats_window,
)
window = normalize_stats_window(request.args.get("window") or "7d")
ex = cfg.get("exchange_options")
if ex is None:
return jsonify({"ok": False, "msg": "交易所未就绪", "version": 1}), 503
try:
raw_live = cfg["fetch_option_positions"](ex)
if raw_live is None:
return jsonify({"ok": False, "msg": "获取期权持仓失败", "version": 1})
history = load_options_history(ex, cfg)
stats = compute_options_stats_for_window(history_pnl_to_usdt(history, ex), window)
open_float = sum_options_net_pnl_usdc(cfg, ex, raw_live)
net_realized = stats.get("net_realized_pnl")
try:
net_f = float(net_realized) if net_realized is not None else 0.0
except (TypeError, ValueError):
net_f = 0.0
total_pnl = None
if open_float is not None:
total_pnl = round(net_f + float(open_float), 4)
elif stats.get("total_closed"):
total_pnl = round(net_f, 4)
pnl_unit = "U" if is_coin_margin_mode() else "USDC"
return jsonify(
{
"ok": True,
"version": 1,
"pnl_unit": pnl_unit,
"open_float_pnl": open_float,
"total_pnl": total_pnl,
**stats,
}
)
except Exception as e:
return jsonify({"ok": False, "msg": str(e) or "统计失败", "version": 1}), 500
@app.route("/api/hub/options/close", methods=["POST"])
def api_hub_options_close():
"""中控下发全平指令:客户端本地买一限价执行;无效买一拒绝."""
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
data = request.get_json(silent=True) or {}
if data.get("market"):
return jsonify({"ok": False, "msg": "已禁用市价平仓,仅支持买一限价", "version": 1})
# 总览约定只全平,忽略/拒绝自定义张数
if data.get("sheets") is not None and str(data.get("sheets")).strip() != "":
return jsonify({"ok": False, "msg": "中控平仓仅支持全平,勿传 sheets", "version": 1})
inst_id = (data.get("inst_id") or "").strip()
if not inst_id:
return jsonify({"ok": False, "msg": "缺少 inst_id", "version": 1})
hub_ctx = app.config.get("HUB_CTX") or {}
fn = hub_ctx.get("options_close_fn")
if not callable(fn):
return jsonify({"ok": False, "msg": "平仓接口未注册", "version": 1}), 503
try:
result = fn(inst_id) or {}
except Exception as e:
return jsonify({"ok": False, "msg": str(e) or "平仓失败", "version": 1}), 500
if not isinstance(result, dict):
return jsonify({"ok": False, "msg": "平仓返回异常", "version": 1}), 500
out = dict(result)
out["version"] = 1
return jsonify(out)
@app.route("/api/hub/options/cond_orders", methods=["GET"])
def api_hub_options_cond_orders_list():
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
hub_ctx = app.config.get("HUB_CTX") or {}
cfg = hub_ctx.get("options_cfg")
if not isinstance(cfg, dict):
return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503
from lib.options.options_cond_order_lib import list_cond_orders
from lib.options.options_db import init_options_tables
conn = cfg["get_db"]()
try:
init_options_tables(conn)
orders = list_cond_orders(conn, active_only=False)
return jsonify({"ok": True, "version": 1, "orders": orders})
finally:
conn.close()
@app.route("/api/hub/options/cond_orders", methods=["POST"])
def api_hub_options_cond_orders_create():
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
hub_ctx = app.config.get("HUB_CTX") or {}
cfg = hub_ctx.get("options_cfg")
if not isinstance(cfg, dict):
return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503
from lib.options.options_cond_order_lib import (
create_cond_order,
machine_has_option_position,
)
from lib.options.options_db import init_options_tables
from lib.options.options_position_limit_lib import option_position_limit_block_msg
ex = cfg.get("exchange_options")
if machine_has_option_position(cfg, ex):
return jsonify({"ok": False, "msg": "本机已有期权仓,拒绝条件单", "version": 1})
limit_msg = option_position_limit_block_msg(
ex,
opening_inst_id="",
fetch_positions=cfg.get("fetch_option_positions"),
)
if limit_msg:
return jsonify({"ok": False, "msg": limit_msg, "version": 1})
data = request.get_json(silent=True) or {}
conn = cfg["get_db"]()
try:
init_options_tables(conn)
out = create_cond_order(
conn,
data,
send_wechat=cfg.get("send_wechat"),
account_label=str(cfg.get("account_label") or ""),
)
out["version"] = 1
return jsonify(out)
finally:
conn.close()
@app.route("/api/hub/options/cond_orders/<hub_order_id>/cancel", methods=["POST"])
def api_hub_options_cond_orders_cancel(hub_order_id: str):
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
hub_ctx = app.config.get("HUB_CTX") or {}
cfg = hub_ctx.get("options_cfg")
if not isinstance(cfg, dict):
return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503
from lib.options.options_cond_order_lib import cancel_cond_order
from lib.options.options_db import init_options_tables
conn = cfg["get_db"]()
try:
init_options_tables(conn)
out = cancel_cond_order(
conn,
hub_order_id,
send_wechat=cfg.get("send_wechat"),
account_label=str(cfg.get("account_label") or ""),
)
out["version"] = 1
return jsonify(out)
finally:
conn.close()
@app.route("/api/hub/options/cond_orders/<hub_order_id>/exit", methods=["POST"])
def api_hub_options_cond_orders_exit(hub_order_id: str):
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权", "version": 1}), 401
hub_ctx = app.config.get("HUB_CTX") or {}
cfg = hub_ctx.get("options_cfg")
if not isinstance(cfg, dict):
return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503
from lib.options.options_cond_order_lib import update_cond_order_exit
from lib.options.options_db import init_options_tables
data = request.get_json(silent=True) or {}
conn = cfg["get_db"]()
try:
init_options_tables(conn)
out = update_cond_order_exit(
conn,
hub_order_id,
target_index=data.get("target_index"),
profit_exit_enabled=data.get("profit_exit_enabled"),
profit_exit_mult=data.get("profit_exit_mult"),
)
out["version"] = 1
return jsonify(out)
finally:
conn.close()
@app.route("/api/hub/version", methods=["GET"])
def hub_version():
"""中控探测客户端当前 git 版本/在线."""
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权"}), 401
from lib.instance.instance_update_lib import read_git_version
ver = read_git_version()
ver["online"] = True
mode = "live"
try:
from lib.sim.mode_lib import exchange_mode_label, is_sim_mode
get_db = getattr(app_module, "get_db", None)
if callable(get_db) and is_sim_mode(get_db):
mode = "sim"
ver["mode"] = mode
if callable(get_db):
ver["mode_label"] = exchange_mode_label(get_db=get_db, exchange_key="okx") or ""
except Exception:
ver["mode"] = mode
return jsonify(ver)
@app.route("/api/hub/update", methods=["POST"])
def hub_update():
"""中控远程快速更新:git pull + pm2 restart(异步)."""
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权"}), 401
from lib.instance.instance_update_lib import schedule_pull_and_restart
out = schedule_pull_and_restart(delay_seconds=1.5)
return jsonify(out), (200 if out.get("ok") else 400)
@app.route("/api/hub/backup/list", methods=["GET"])
def hub_backup_list():
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权"}), 401
from lib.instance.backup_lib import backup_dir, backup_keep_days, list_backups
return jsonify(
{
"ok": True,
"items": list_backups(),
"backup_dir": backup_dir(),
"keep_days": backup_keep_days(),
"auto_hour_bj": 3,
"includes": [".env", "database", "options_journal"],
}
)
@app.route("/api/hub/backup/download/<name>", methods=["GET"])
def hub_backup_download(name: str):
if not hub_bridge_token_ok(request.headers):
return jsonify({"ok": False, "msg": "未授权"}), 401
import os
from flask import send_file
from lib.instance.backup_lib import backup_path
path = backup_path(name)
if not path or not os.path.isfile(path):
return jsonify({"ok": False, "msg": "备份不存在"}), 404
return send_file(
path,
mimetype="application/gzip",
as_attachment=True,
download_name=os.path.basename(path),
)
@app.route("/hub_sso")
def hub_sso_login():
token = (request.args.get("token") or "").strip()
next_path = _safe_next_path(request.args.get("next") or "/embed")
if not verify_sso_token(token):
return (
"<!doctype html><meta charset=utf-8><title>SSO 失败</title>"
"<p>中控登录链无效或已过期,请从中控重新打开.</p>",
403,
)
session["logged_in"] = True
session.permanent = True
session.modified = True
# iframe 跨站时常丢 Cookie:附带可放入 sessionStorage 的 embed_sso
embed_tok = make_embed_session_token(hub_bridge_token(), ttl_sec=604800)
joiner = "&" if "?" in next_path else "?"
return redirect(f"{next_path}{joiner}embed_sso={quote(embed_tok, safe='')}")
@app.after_request
def _hub_frame_ancestors(resp):
ancestors = frame_ancestors_header_value()
if ancestors:
resp.headers["Content-Security-Policy"] = f"frame-ancestors {ancestors}"
resp.headers.pop("X-Frame-Options", None)
return resp