"""中控台持仓:一键保本(成交价 ± 偏移%)。""" from typing import Any, Callable, Optional, Tuple def calc_manual_breakeven_stop(direction: str, entry_price: float, offset_pct: float) -> Optional[float]: try: e = float(entry_price) pct = float(offset_pct) except (TypeError, ValueError): return None if e <= 0 or pct < 0: return None direction = (direction or "long").strip().lower() if direction == "short": return e * (1.0 - pct / 100.0) return e * (1.0 + pct / 100.0) def apply_order_manual_breakeven( row: Any, offset_pct: float, *, calc_stop_fn: Callable[..., Optional[float]], round_price_fn: Callable[[str, float], Any], resolve_ex_sym_fn: Callable[[Any], str], get_position_fn: Callable[[str, str], Any], replace_tpsl_fn: Callable[[Any, float, float], None], entry_price_key: str = "trigger_price", ) -> Tuple[bool, Optional[str], Optional[float]]: if (row["status"] or "").strip() != "active": return False, "持仓已结束", None entry = float(row[entry_price_key] or 0) if entry <= 0: return False, "缺少有效成交价", None direction = (row["direction"] or "long").lower() take_profit = float(row["take_profit"] or 0) if take_profit <= 0: return False, "缺少有效止盈价,无法更新交易所委托", None ex_sym = resolve_ex_sym_fn(row) pos = get_position_fn(ex_sym, direction) if pos is None or float(pos) <= 0: return False, "交易所当前无该方向持仓", None new_sl_raw = calc_stop_fn(direction, entry, offset_pct) if new_sl_raw is None: new_sl_raw = calc_manual_breakeven_stop(direction, entry, offset_pct) if new_sl_raw is None: return False, "保本价计算失败", None new_sl = round_price_fn(ex_sym, new_sl_raw) if new_sl is None: return False, "保本价经交易所精度舍入后无效", None new_sl = float(new_sl) cur_sl = float(row["stop_loss"] or 0) if direction == "long": if cur_sl > 0 and new_sl <= cur_sl: return False, f"新止损 {new_sl} 未高于当前止损 {cur_sl}(多仓需上移)", None else: if cur_sl > 0 and new_sl >= cur_sl: return False, f"新止损 {new_sl} 未低于当前止损 {cur_sl}(空仓需下移)", None try: replace_tpsl_fn(row, new_sl, take_profit) except Exception as e: return False, str(e), None return True, None, new_sl