dekun
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
EXIT_MODE_FIXED = "fixed_usdt"
|
|
EXIT_MODE_PREMIUM = "premium_multiple"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ExitDecision:
|
|
should_close: bool
|
|
reason: str = ""
|
|
target: float = 0.0
|
|
|
|
|
|
def resolve_exit_target(
|
|
*,
|
|
exit_mode: str,
|
|
net_profit_target: float,
|
|
premium_exit_multiple: float,
|
|
initial_premium: float,
|
|
) -> tuple[float, str]:
|
|
"""返回 (出场目标金额 USDT, 模式标记)。"""
|
|
mode = (exit_mode or EXIT_MODE_FIXED).strip().lower()
|
|
if mode == EXIT_MODE_PREMIUM:
|
|
mult = max(0.0, float(premium_exit_multiple))
|
|
return float(initial_premium) * mult, EXIT_MODE_PREMIUM
|
|
return float(net_profit_target), EXIT_MODE_FIXED
|
|
|
|
|
|
def check_expiry_close(
|
|
*,
|
|
expiry_ms: int | None,
|
|
now_ms: int | None = None,
|
|
) -> ExitDecision:
|
|
"""期权到期时刻（含）→ 强制全平。"""
|
|
if expiry_ms is None:
|
|
return ExitDecision(False, "", 0.0)
|
|
now = int(time.time() * 1000) if now_ms is None else int(now_ms)
|
|
if now >= int(expiry_ms):
|
|
return ExitDecision(True, "expiry", 0.0)
|
|
return ExitDecision(False, "", 0.0)
|
|
|
|
|
|
def check_exits(
|
|
*,
|
|
net_pnl: float,
|
|
exit_mode: str,
|
|
net_profit_target: float,
|
|
premium_exit_multiple: float,
|
|
initial_premium: float,
|
|
) -> ExitDecision:
|
|
"""净盈利（预估全平后）≥ 所选模式目标则全平。"""
|
|
target, mode = resolve_exit_target(
|
|
exit_mode=exit_mode,
|
|
net_profit_target=net_profit_target,
|
|
premium_exit_multiple=premium_exit_multiple,
|
|
initial_premium=initial_premium,
|
|
)
|
|
if target > 0 and net_pnl + 1e-9 >= target:
|
|
reason = "premium_multiple" if mode == EXIT_MODE_PREMIUM else "fixed_usdt"
|
|
return ExitDecision(True, reason, target)
|
|
return ExitDecision(False, "", target)
|