"""半自动：人工定方向/目标并授权 → 机器盯开盯平 → 平完停等人工。""" from __future__ import annotations import json from dataclasses import dataclass from typing import Any from ..config import get_settings from ..sim.ledger import Ledger PHASE_WAIT_HUMAN = "wait_human" REASON_POINTS = "semi_target_points" REASON_PERP_NET = "semi_perp_exit" TRADE_LOCK_KEY = "semi_trade_lock" OTM_LEV_FLOOR = 180.0 MONEYNESS_CHOICES = ("itm", "atm", "otm") def is_semi_auto(ledger: Ledger | None = None) -> bool: led = ledger or Ledger() s = get_settings() return bool( led.get_setting_bool("semi_auto_enabled", s.semi_auto_enabled) ) def is_armed(ledger: Ledger | None = None) -> bool: led = ledger or Ledger() return bool(led.get_setting_bool("semi_armed", False)) def set_armed(db: Any, armed: bool) -> None: db.set_setting("semi_armed", "true" if armed else "false") def clear_armed_if_flat(db: Any, *, has_open_position: bool) -> None: """进程重启等：无仓时清授权，避免陈旧 armed 自动开仓。""" if not has_open_position: set_armed(db, False) def read_trade_lock(ledger: Ledger | None = None) -> dict[str, Any] | None: led = ledger or Ledger() raw = led.get_setting_str(TRADE_LOCK_KEY, "") or "" raw = str(raw).strip() if not raw: return None try: data = json.loads(raw) except Exception: return None if not isinstance(data, dict) or not data.get("group_id"): return None return data def lock_trade_params( db: Any, *, group_id: str, view_side: str, option_move_points: float, perp_exit_unit: float, moneyness: str = "otm", otm_max_offset: float = 25.0, perp_unit: float = 1.0, option_unit: float = 4.0, ) -> None: mny = str(moneyness or "otm").strip().lower() if mny not in MONEYNESS_CHOICES: mny = "otm" payload = { "group_id": str(group_id), "view_side": "short" if view_side == "short" else "long", "option_move_points": float(option_move_points), "perp_exit_unit": float(perp_exit_unit), "moneyness": mny, "otm_max_offset": float(otm_max_offset), "perp_unit": float(perp_unit), "option_unit": float(option_unit), } db.set_setting(TRADE_LOCK_KEY, json.dumps(payload, ensure_ascii=False)) def clear_trade_lock(db: Any) -> None: db.set_setting(TRADE_LOCK_KEY, "") def _norm_moneyness(raw: str | None, default: str = "otm") -> str: m = str(raw or default).strip().lower() return m if m in MONEYNESS_CHOICES else default def effective_min_leverage(moneyness: str, configured: float) -> float: """虚值强制不低于 180；实值/平值用配置值。""" lev = max(1.0, float(configured)) if _norm_moneyness(moneyness) == "otm": return max(lev, OTM_LEV_FLOOR) return lev def read_semi_params( ledger: Ledger | None = None, *, group_id: str | None = None, ) -> dict[str, Any]: led = ledger or Ledger() s = get_settings() view = str( led.get_setting_str("semi_view_side", s.semi_view_side) or s.semi_view_side ).strip().lower() if view not in ("long", "short"): view = "long" move = float( led.get_setting_float( "semi_option_move_points", s.semi_option_move_points ) or s.semi_option_move_points ) exit_unit = float( led.get_setting_float("semi_perp_exit_unit", s.semi_perp_exit_unit) or s.semi_perp_exit_unit ) moneyness = _norm_moneyness( led.get_setting_str("semi_moneyness", s.semi_moneyness), s.semi_moneyness, ) otm_off = float( led.get_setting_float("semi_otm_max_offset", s.semi_otm_max_offset) or s.semi_otm_max_offset ) perp_u = float( led.get_setting_float("semi_perp_unit", s.semi_perp_unit) or s.semi_perp_unit ) opt_u = float( led.get_setting_float("semi_option_unit", s.semi_option_unit) or s.semi_option_unit ) if perp_u <= 0: perp_u = float(s.semi_perp_unit) if opt_u <= 0: opt_u = float(s.semi_option_unit) min_lev_cfg = float( led.get_setting_float( "semi_min_option_leverage", s.semi_min_option_leverage ) or s.semi_min_option_leverage ) # 持仓中优先用开仓时锁定的出场/选约参数 lock = read_trade_lock(led) if lock and group_id and str(lock.get("group_id")) == str(group_id): view = str(lock.get("view_side") or view).strip().lower() if view not in ("long", "short"): view = "long" try: move = float(lock.get("option_move_points", move)) exit_unit = float(lock.get("perp_exit_unit", exit_unit)) otm_off = float(lock.get("otm_max_offset", otm_off)) perp_u = float(lock.get("perp_unit", perp_u)) opt_u = float(lock.get("option_unit", opt_u)) except (TypeError, ValueError): pass moneyness = _norm_moneyness(str(lock.get("moneyness") or moneyness), moneyness) return { "enabled": is_semi_auto(led), "armed": is_armed(led), "view_side": view, # 看法多 → Call+永续空；看法空 → Put+永续多 "option_side": "call" if view == "long" else "put", "perp_side": "short" if view == "long" else "long", "option_move_points": move, "perp_exit_unit": exit_unit, "moneyness": moneyness, "otm_max_offset": max(0.0, otm_off), "perp_unit": perp_u, "option_unit": opt_u, "min_option_hours": float( led.get_setting_float( "semi_min_option_hours", s.semi_min_option_hours ) or s.semi_min_option_hours ), "min_option_leverage": effective_min_leverage(moneyness, min_lev_cfg), "min_option_leverage_cfg": min_lev_cfg, "trade_locked": bool( lock and group_id and str(lock.get("group_id")) == str(group_id) ), } @dataclass(slots=True) class SemiExitDecision: should_close: bool reason: str = "" detail: str = "" target_index: float | None = None net_target: float | None = None def check_semi_exits( *, net_pnl: float, strike: float | None, index_px: float | None, view_side: str, option_move_points: float, perp_exit_unit: float, risk_k: float = 1.0, entry_index: float | None = None, ) -> SemiExitDecision: """ 顺方向：指数到达「行权价 ± 波动点」且组合净利>0 → 全平。 多/Call：目标 = K + N；空/Put：目标 = K − N（N 为设置的波动点，不是现价±N）。 逆方向兑现（永续锁定净利）：组合净利 ≥ 净利基数×k → 全平。 """ view = (view_side or "long").strip().lower() if view not in ("long", "short"): view = "long" move = max(0.0, float(option_move_points)) k = float(risk_k) if risk_k and risk_k > 0 else 1.0 net_tgt = max(0.0, float(perp_exit_unit)) * k net = float(net_pnl) # 逆方向 / 永续净利锁定：达标即可离场（不必等点位） if net_tgt > 0 and net + 1e-9 >= net_tgt: return SemiExitDecision( True, REASON_PERP_NET, f"半自动·永续净利锁定≥{net_tgt:.2f}U（基数×k）", net_target=net_tgt, ) if index_px is None: return SemiExitDecision(False, "", "缺指数") idx = float(index_px) # 锚定行权价；无 strike 时才回退开仓指数（兼容旧仓） anchor = None if strike is not None and float(strike) > 0: anchor = float(strike) elif entry_index is not None and float(entry_index) > 0: anchor = float(entry_index) if anchor is None or idx <= 0 or move <= 0: return SemiExitDecision(False, "", "点位无效") if view == "long": target_idx = anchor + move hit = idx + 1e-9 >= target_idx else: target_idx = anchor - move hit = idx - 1e-9 <= target_idx if hit and net > 0: return SemiExitDecision( True, REASON_POINTS, f"半自动·指数到期权目标{target_idx:.2f}（K{anchor:g}±{move:g}）且净利>0", target_index=target_idx, net_target=0.0, ) if hit and net <= 0: return SemiExitDecision( False, "", f"已到期权目标{target_idx:.2f}但组合净利≤0（{net:.2f}），继续持有", target_index=target_idx, ) return SemiExitDecision( False, "", f"未到期权目标（K{anchor:g}→{target_idx:.2f}）", target_index=target_idx, net_target=net_tgt, )