dekun
cf3e2ee1c9
盈亏比固定用开仓 initial_stop_loss 计算，人工改委托后不变化；轮询交易所止损触发价相对成交价判定已保本，四所实例与中控统一显示绿色标识。 Co-authored-by: Cursor <cursoragent@cursor.com>
112 lines
3.2 KiB
Python
112 lines
3.2 KiB
Python
"""实时持仓展示：开仓快照盈亏比、交易所止损是否已保本。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Callable, Optional
|
|
|
|
|
|
def _positive_float(value: Any) -> Optional[float]:
|
|
try:
|
|
v = float(value)
|
|
return v if v > 0 else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def snapshot_stop_loss(initial_stop_loss: Any, stop_loss: Any) -> Optional[float]:
|
|
"""展示盈亏比时优先用开仓时止损快照。"""
|
|
sl = _positive_float(initial_stop_loss)
|
|
if sl is not None:
|
|
return sl
|
|
return _positive_float(stop_loss)
|
|
|
|
|
|
def snapshot_rr(
|
|
calc_rr_ratio_fn: Callable[..., Optional[float]],
|
|
direction: str,
|
|
trigger_price: Any,
|
|
initial_stop_loss: Any,
|
|
stop_loss: Any,
|
|
take_profit: Any,
|
|
) -> Optional[float]:
|
|
entry = _positive_float(trigger_price)
|
|
sl = snapshot_stop_loss(initial_stop_loss, stop_loss)
|
|
tp = _positive_float(take_profit)
|
|
if entry is None or sl is None or tp is None:
|
|
return None
|
|
return calc_rr_ratio_fn(direction or "long", entry, sl, tp)
|
|
|
|
|
|
def tpsl_slot_trigger_price(slot: Any) -> Optional[float]:
|
|
if not isinstance(slot, dict):
|
|
return None
|
|
for key in ("trigger_price", "trigger_display"):
|
|
v = _positive_float(slot.get(key))
|
|
if v is not None:
|
|
return v
|
|
return None
|
|
|
|
|
|
def is_sl_breakeven_secured(direction: str, entry_price: Any, exchange_sl_price: Any) -> bool:
|
|
"""
|
|
交易所当前止损相对开仓成交价是否已保本。
|
|
做多：止损 >= 成交价；做空：止损 <= 成交价。
|
|
"""
|
|
entry = _positive_float(entry_price)
|
|
sl = _positive_float(exchange_sl_price)
|
|
if entry is None or sl is None:
|
|
return False
|
|
d = (direction or "long").strip().lower()
|
|
if d == "short":
|
|
return sl <= entry
|
|
return sl >= entry
|
|
|
|
|
|
def sl_breakeven_from_exchange_tpsl(
|
|
direction: str,
|
|
entry_price: Any,
|
|
exchange_tpsl: Any,
|
|
) -> bool:
|
|
if not isinstance(exchange_tpsl, dict):
|
|
return False
|
|
sl_px = tpsl_slot_trigger_price(exchange_tpsl.get("sl"))
|
|
if sl_px is None:
|
|
return False
|
|
return is_sl_breakeven_secured(direction, entry_price, sl_px)
|
|
|
|
|
|
def enrich_order_display_fields(item: dict[str, Any], calc_rr_ratio_fn: Callable[..., Optional[float]]) -> dict[str, Any]:
|
|
item["rr_ratio"] = snapshot_rr(
|
|
calc_rr_ratio_fn,
|
|
item.get("direction") or "long",
|
|
item.get("trigger_price"),
|
|
item.get("initial_stop_loss"),
|
|
item.get("stop_loss"),
|
|
item.get("take_profit"),
|
|
)
|
|
return item
|
|
|
|
|
|
def apply_order_price_display_fields(
|
|
payload: dict[str, Any],
|
|
*,
|
|
direction: str,
|
|
entry_price: Any,
|
|
initial_stop_loss: Any,
|
|
stop_loss: Any,
|
|
take_profit: Any,
|
|
calc_rr_ratio_fn: Callable[..., Optional[float]],
|
|
exchange_tpsl: Any = None,
|
|
) -> dict[str, Any]:
|
|
payload["rr_ratio"] = snapshot_rr(
|
|
calc_rr_ratio_fn,
|
|
direction,
|
|
entry_price,
|
|
initial_stop_loss,
|
|
stop_loss,
|
|
take_profit,
|
|
)
|
|
payload["sl_breakeven_secured"] = sl_breakeven_from_exchange_tpsl(
|
|
direction, entry_price, exchange_tpsl
|
|
)
|
|
return payload
|