dekun
9295dcffe3
Align monitor profit-exit with the same book VWAP so card and auto-exit stay consistent. Co-authored-by: Cursor <cursoragent@cursor.com>
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
"""对冲计划出场条件:读取/校验/更新."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
|
|
def _sf(v: Any) -> Optional[float]:
|
|
try:
|
|
if v is None or v == "":
|
|
return None
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def normalize_exit_mode(raw: Any, *, profit_target: Any = None) -> str:
|
|
m = str(raw or "").strip().lower()
|
|
if m in ("profit", "target"):
|
|
return m
|
|
if _sf(profit_target) is not None and float(profit_target) > 0:
|
|
return "profit"
|
|
return "target"
|
|
|
|
|
|
def validate_exit_payload(body: dict[str, Any] | None) -> tuple[bool, str, dict[str, Any]]:
|
|
"""校验改出场请求 → (ok, msg, fields_for_update)."""
|
|
b = body or {}
|
|
mode = normalize_exit_mode(b.get("exit_mode") or b.get("mode"))
|
|
if mode == "profit":
|
|
pt = _sf(b.get("profit_target_usdt"))
|
|
if pt is None or pt <= 0:
|
|
return False, "盈利金额须大于 0", {}
|
|
return True, "", {
|
|
"exit_mode": "profit",
|
|
"profit_target_usdt": float(pt),
|
|
}
|
|
opt_pts = _sf(b.get("option_target_points"))
|
|
perp_pts = _sf(b.get("perp_target_points"))
|
|
if opt_pts is None or opt_pts <= 0:
|
|
return False, "期权目标点数须大于 0", {}
|
|
if perp_pts is None or perp_pts <= 0:
|
|
return False, "永续目标点数须大于 0", {}
|
|
return True, "", {
|
|
"exit_mode": "target",
|
|
"option_target_points": float(opt_pts),
|
|
"perp_target_points": float(perp_pts),
|
|
}
|
|
|
|
|
|
def update_plan_exit(cfg: dict[str, Any], conn: Any, plan_id: int, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""更新进行中计划的出场条件."""
|
|
from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, update_plan
|
|
|
|
plan = get_plan(conn, int(plan_id))
|
|
if not plan:
|
|
return {"ok": False, "msg": "计划不存在"}
|
|
st = str(plan.get("status") or "").lower()
|
|
legs = get_plan_legs(conn, int(plan_id))
|
|
has_open = any(str(x.get("status") or "").lower() == "open" for x in legs)
|
|
if st not in ("opening", "active", "partial", "watching") and not (st == "closed" and has_open):
|
|
return {"ok": False, "msg": f"当前状态 {st or '—'} 不可改出场"}
|
|
|
|
ok, msg, fields = validate_exit_payload(body)
|
|
if not ok:
|
|
return {"ok": False, "msg": msg}
|
|
|
|
update_plan(conn, int(plan_id), **fields)
|
|
plan2 = get_plan(conn, int(plan_id))
|
|
return {
|
|
"ok": True,
|
|
"plan_id": int(plan_id),
|
|
"msg": "出场条件已更新",
|
|
"exit": {
|
|
"mode": fields.get("exit_mode"),
|
|
"profit_target_usdt": plan2.get("profit_target_usdt") if plan2 else fields.get("profit_target_usdt"),
|
|
"option_target_points": plan2.get("option_target_points") if plan2 else fields.get("option_target_points"),
|
|
"perp_target_points": plan2.get("perp_target_points") if plan2 else fields.get("perp_target_points"),
|
|
},
|
|
}
|