eth_hedge_sim/backend/app/strategy/semi_auto.py

"""半自动：人工定方向/目标并授权 → 机器盯开盯平 → 平完停等人工。"""

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"


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,
) -> None:
    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),
    }
    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 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
    )
    # 持仓中优先用开仓时锁定的出场参数
    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))
        except (TypeError, ValueError):
            pass
    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,
        "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": float(
            led.get_setting_float(
                "semi_min_option_leverage", s.semi_min_option_leverage
            )
            or s.semi_min_option_leverage
        ),
        "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,
    entry_index: float | None,
    index_px: float | None,
    view_side: str,
    option_move_points: float,
    perp_exit_unit: float,
    risk_k: float = 1.0,
) -> SemiExitDecision:
    """
    顺方向：标的波动达到目标点 且 组合净利>0 → 全平。
    逆方向兑现：组合净利 ≥ 永续出场基数×k → 全平。
    流动性在 close_group 内再验；平仓顺序已是先期权后永续。
    """
    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 entry_index is None or index_px is None:
        return SemiExitDecision(False, "", "缺指数")
    entry = float(entry_index)
    idx = float(index_px)
    if entry <= 0 or idx <= 0 or move <= 0:
        return SemiExitDecision(False, "", "点位无效")

    if view == "long":
        target_idx = entry + move
        hit = idx + 1e-9 >= target_idx
    else:
        target_idx = entry - move
        hit = idx - 1e-9 <= target_idx

    if hit and net > 0:
        return SemiExitDecision(
            True,
            REASON_POINTS,
            f"半自动·标的到{target_idx:.2f}且组合净利>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"未到点位（目标{target_idx:.2f}）",
        target_index=target_idx,
        net_target=net_tgt,
    )