eth_hedge_sim/backend/app/strategy/risk_sizing.py

"""以损定仓：按可承受最大亏损反推标准组倍数 k（永续1 / 期权2 / 出场15）。"""

from __future__ import annotations

import logging
import math
from dataclasses import dataclass
from typing import Any

from ..config import get_settings
from ..models.db import Database, get_db
from ..sim.ledger import Ledger

logger = logging.getLogger(__name__)

# 标准组基准默认（k=1）；可由设置 risk_*_unit 覆盖
BASE_PERP_ETH = 1.0
BASE_OPTION_ETH = 2.0
BASE_EXIT_USDT = 15.0
MIN_K = 0.1
FEE_LEG_COUNT = 3  # 永续开/平 + 期权一次


def _round2(x: float | None) -> float | None:
    if x is None or not math.isfinite(float(x)):
        return None
    return round(float(x) + 0.0, 2)


def read_risk_units(ledger: Ledger) -> tuple[float, float, float]:
    """永续名义单位 / 期权名义单位 / 出场基数(k=1)。"""
    perp_u = ledger.get_setting_float("risk_perp_unit", BASE_PERP_ETH)
    opt_u = ledger.get_setting_float("risk_option_unit", BASE_OPTION_ETH)
    exit_u = ledger.get_setting_float("risk_exit_unit", BASE_EXIT_USDT)
    if perp_u <= 0:
        perp_u = BASE_PERP_ETH
    if opt_u <= 0:
        opt_u = BASE_OPTION_ETH
    if exit_u <= 0:
        exit_u = BASE_EXIT_USDT
    return float(perp_u), float(opt_u), float(exit_u)


@dataclass(frozen=True, slots=True)
class RiskSizingResult:
    ok: bool
    detail: str
    k: float | None = None
    budget: float | None = None
    capital_base: float | None = None
    premium_est: float | None = None
    fee_est: float | None = None
    max_loss: float | None = None
    perp_qty_eth: float | None = None
    option_qty_eth: float | None = None
    net_profit_target: float | None = None
    index_px: float | None = None
    option_ask: float | None = None  # 定仓用权利金（可能为选约杠杆隐含）
    actual_option_ask: float | None = None  # 盘口真实卖一
    leverage_basis: str | None = None  # actual | selection
    perp_unit: float | None = None
    option_unit: float | None = None
    exit_unit: float | None = None


def normalize_risk_leverage_basis(raw: str | None, default: str = "selection") -> str:
    v = (raw or default or "selection").strip().lower()
    if v in ("selection", "min_option_leverage", "select", "选约", "选约杠杆"):
        return "selection"
    if v in ("actual", "market", "ask", "实际", "实际杠杆"):
        return "actual"
    return "selection" if default == "selection" else "actual"


def resolve_sizing_option_ask(
    *,
    index_px: float,
    option_ask: float,
    leverage_basis: str,
    min_option_leverage: float,
) -> tuple[float, str]:
    """
    返回 (定仓用卖一, 口径 actual|selection)。
    selection：隐含卖一 = 指数 / 选约杠杆；actual：用盘口卖一。
    """
    basis = normalize_risk_leverage_basis(leverage_basis, "selection")
    if basis == "selection":
        lev = float(min_option_leverage)
        if lev > 1e-12 and math.isfinite(lev) and index_px > 0:
            return float(index_px) / lev, "selection"
        # 选约杠杆无效时退回实际卖一，避免拒单
        return float(option_ask), "actual"
    return float(option_ask), "actual"


def is_risk_based(ledger: Ledger | None = None) -> bool:
    led = ledger or Ledger()
    mode = (led.get_setting_str("sizing_mode", "manual") or "manual").strip().lower()
    return mode == "risk_based"


def floor_k_1dp(k_raw: float) -> float:
    """一位小数向下取整，保证不超预算。"""
    if k_raw <= 0 or not math.isfinite(k_raw):
        return 0.0
    return math.floor(k_raw * 10.0 + 1e-12) / 10.0


def unit_cost(
    *,
    index_px: float,
    option_ask: float,
    fee_rate: float,
    option_unit: float = BASE_OPTION_ETH,
) -> float:
    """k=1 时估算最大亏损 = 权利金(option_unit ETH) + 手续费粗估。"""
    premium_unit = float(option_ask) * float(option_unit)
    fee_unit = float(index_px) * float(fee_rate) * FEE_LEG_COUNT
    return premium_unit + fee_unit


def compute_k(
    *,
    budget: float,
    index_px: float,
    option_ask: float,
    fee_rate: float,
    perp_unit: float = BASE_PERP_ETH,
    option_unit: float = BASE_OPTION_ETH,
    exit_unit: float = BASE_EXIT_USDT,
) -> RiskSizingResult:
    if budget is None or budget <= 0 or not math.isfinite(budget):
        return RiskSizingResult(ok=False, detail="以损定仓预算无效（须 > 0）")
    if index_px is None or index_px <= 0 or not math.isfinite(index_px):
        return RiskSizingResult(ok=False, detail="以损定仓缺少有效指数价")
    if option_ask is None or option_ask <= 0 or not math.isfinite(option_ask):
        return RiskSizingResult(ok=False, detail="以损定仓缺少有效期权卖一")
    if option_unit <= 0 or perp_unit <= 0 or exit_unit <= 0:
        return RiskSizingResult(ok=False, detail="以损定仓比例/出场基数须 > 0")

    cost1 = unit_cost(
        index_px=index_px,
        option_ask=option_ask,
        fee_rate=fee_rate,
        option_unit=option_unit,
    )
    if cost1 <= 1e-12:
        return RiskSizingResult(ok=False, detail="以损定仓单位成本无效")

    k_raw = float(budget) / cost1
    k = floor_k_1dp(k_raw)
    if k < MIN_K - 1e-12:
        return RiskSizingResult(
            ok=False,
            detail=(
                f"以损定仓算出 k={k_raw:.4f}，向下取整后 < {MIN_K}，"
                f"预算 {budget:.2f}U 不足以开最小仓（单位成本≈{cost1:.2f}U）"
            ),
            budget=_round2(float(budget)),
            k=k,
            index_px=float(index_px),
            option_ask=float(option_ask),
            perp_unit=float(perp_unit),
            option_unit=float(option_unit),
            exit_unit=float(exit_unit),
        )

    # 若浮点导致仍略超，再降一档
    while k >= MIN_K - 1e-12:
        prem = float(option_ask) * float(option_unit) * k
        fee = float(index_px) * float(fee_rate) * FEE_LEG_COUNT * k
        mx = prem + fee
        if mx <= float(budget) + 1e-6:
            return RiskSizingResult(
                ok=True,
                detail="ok",
                k=k,
                budget=_round2(float(budget)),
                premium_est=_round2(prem),
                fee_est=_round2(fee),
                max_loss=_round2(mx),
                perp_qty_eth=round(float(perp_unit) * k, 4),
                option_qty_eth=round(float(option_unit) * k, 4),
                net_profit_target=_round2(float(exit_unit) * k),
                index_px=float(index_px),
                option_ask=float(option_ask),
                perp_unit=float(perp_unit),
                option_unit=float(option_unit),
                exit_unit=float(exit_unit),
            )
        k = round(k - 0.1, 1)

    return RiskSizingResult(
        ok=False,
        detail=f"以损定仓无法在预算 {budget:.2f}U 内找到合规 k",
        budget=_round2(float(budget)),
        index_px=float(index_px),
        option_ask=float(option_ask),
        perp_unit=float(perp_unit),
        option_unit=float(option_unit),
        exit_unit=float(exit_unit),
    )


def resolve_capital_base(db: Database | None = None) -> tuple[float | None, str]:
    """返回 (本金USDT口径, 说明)。"""
    database = db or get_db()
    ledger = Ledger(database)
    source = (
        ledger.get_setting_str("risk_capital_source", "trading_account") or "trading_account"
    ).strip().lower()
    if source in ("manual", "manual_capital", "fixed"):
        cap = ledger.get_setting_float("risk_manual_capital_usdt", 0.0)
        if cap <= 0:
            return None, "单独本金未设置或 ≤ 0"
        return float(cap), "manual"

    # trading_account：交易账户 USDT + USDC（1:1 折算，与资金条交易账户一致）
    usdt, usdc = _trading_balances(database)
    if usdt is None and usdc is None:
        try:
            from ..exchange.runtime import load_runtime_settings

            ex = str(load_runtime_settings().exchange or "").strip().lower()
            if ex in ("binance", "bn") and not get_settings().is_sim:
                return (
                    None,
                    "币安实盘暂未接入交易账户余额，请改用「单独本金」或「亏损值」",
                )
        except Exception:
            pass
        return None, "无法读取交易账户资金"
    total = float(usdt or 0.0) + float(usdc or 0.0)
    if total <= 1e-9:
        return None, "交易账户总资金为 0"
    return total, "trading_account"


def resolve_budget(db: Database | None = None) -> tuple[float | None, str, float | None]:
    """返回 (budget, detail, capital_base)。"""
    database = db or get_db()
    ledger = Ledger(database)
    loss_mode = (
        ledger.get_setting_str("risk_loss_mode", "percent") or "percent"
    ).strip().lower()
    if loss_mode in ("absolute", "usdt", "value", "亏损值"):
        bud = ledger.get_setting_float("risk_loss_usdt", 0.0)
        if bud <= 0:
            return None, "亏损值未设置或 ≤ 0", None
        return float(bud), "absolute", None

    capital, src = resolve_capital_base(database)
    if capital is None:
        return None, src, None
    pct = ledger.get_setting_float("risk_loss_pct", 1.0)
    if pct <= 0:
        return None, "亏损幅度须 > 0", capital
    mg = resolve_martingale(database, ledger=ledger, base_pct=float(pct))
    effective = float(mg["effective_pct"])
    detail = f"percent@{src}"
    if int(mg.get("doubles") or 0) > 0:
        detail += (
            f"|mg×{int(2 ** int(mg['doubles']))}"
            f"(连亏{int(mg.get('loss_days') or 0)}天)"
        )
    return float(capital) * (effective / 100.0), detail, capital


MARTINGALE_MAX_BASE_PCT = 3.0


def _is_expiry_close_reason(reason: str | None) -> bool:
    r = str(reason or "").strip().lower()
    return r in ("expiry", "到期", "到期结算", "到期结算全平")


def _martingale_day_pnl_contrib(realized_pnl: float, close_reason: str | None) -> float:
    """
    倍投连亏日口径：到期结算无论实际盈亏（含小盈利）一律按亏损计入；
    其它平仓按真实 realized_pnl。
    """
    if _is_expiry_close_reason(close_reason):
        return -1.0
    return float(realized_pnl or 0.0)


def consecutive_loss_days(db: Database | None = None) -> int:
    """
    按上海日历「平仓日」汇总倍投口径盈亏，从最近有平仓的一天往前数连续亏损天数。
    某日合计 < 0 计为亏损日；到期结算组无论盈亏均按亏损计入。
    无平仓的日历日不计入、不打断（按有成交日序列）。
    """
    from collections import defaultdict
    from datetime import datetime, timezone
    from zoneinfo import ZoneInfo

    database = db or get_db()
    rows = database.fetchall(
        """SELECT realized_pnl, close_at_ms, close_reason FROM groups
           WHERE status='closed' AND close_at_ms IS NOT NULL
           ORDER BY close_at_ms ASC"""
    )
    if not rows:
        return 0
    sh = ZoneInfo("Asia/Shanghai")
    day_pnl: dict[str, float] = defaultdict(float)
    for r in rows:
        try:
            ms = int(r["close_at_ms"] or 0)
        except (TypeError, ValueError):
            continue
        if ms <= 0:
            continue
        day = (
            datetime.fromtimestamp(ms / 1000.0, tz=timezone.utc)
            .astimezone(sh)
            .strftime("%Y-%m-%d")
        )
        day_pnl[day] += _martingale_day_pnl_contrib(
            float(r["realized_pnl"] or 0),
            r["close_reason"],
        )
    if not day_pnl:
        return 0
    streak = 0
    for d in reversed(sorted(day_pnl.keys())):
        if float(day_pnl[d]) < 0:
            streak += 1
        else:
            break
    return streak


def resolve_martingale(
    db: Database | None = None,
    *,
    ledger: Ledger | None = None,
    base_pct: float | None = None,
) -> dict[str, Any]:
    """
    倍投状态：仅以损定仓 + 亏损幅度% + 开关开启 + 基础幅度≤3% 时生效。
    doubles: 已翻倍次数（0=用基础幅度）；effective_pct = base * 2^doubles。
    """
    database = db or get_db()
    led = ledger or Ledger(database)
    enabled = led.get_setting_bool("martingale_enabled", False)
    pct = (
        float(base_pct)
        if base_pct is not None
        else float(led.get_setting_float("risk_loss_pct", 1.0))
    )
    start_after = int(
        round(led.get_setting_float("martingale_start_after_loss_days", 2.0))
    )
    max_doubles = int(round(led.get_setting_float("martingale_max_doubles", 3.0)))
    start_after = max(1, min(30, start_after))
    max_doubles = max(1, min(10, max_doubles))
    loss_days = consecutive_loss_days(database)
    out: dict[str, Any] = {
        "enabled": bool(enabled),
        "eligible": False,
        "blocked": "",
        "base_pct": round(pct, 4),
        "effective_pct": round(pct, 4),
        "doubles": 0,
        "loss_days": int(loss_days),
        "start_after_loss_days": start_after,
        "max_doubles": max_doubles,
    }
    if not enabled:
        out["blocked"] = "off"
        return out
    if not is_risk_based(led):
        out["blocked"] = "not_risk_based"
        return out
    loss_mode = (
        led.get_setting_str("risk_loss_mode", "percent") or "percent"
    ).strip().lower()
    if loss_mode not in ("percent", "pct", "%", "幅度"):
        out["blocked"] = "not_percent_mode"
        return out
    if pct > MARTINGALE_MAX_BASE_PCT + 1e-12:
        out["blocked"] = f"base_pct>{MARTINGALE_MAX_BASE_PCT:g}"
        return out
    out["eligible"] = True
    doubles = 0
    if loss_days >= start_after:
        doubles = min(int(loss_days - start_after + 1), max_doubles)
    out["doubles"] = doubles
    out["effective_pct"] = round(float(pct) * (2**doubles), 6)
    return out


def _trading_balances(db: Database) -> tuple[float | None, float | None]:
    s = get_settings()
    if s.is_sim:
        from ..sim.funds_wallets import SimFundsWallets

        w = SimFundsWallets(db)
        v = w.view()
        return float(v["trading_usdt"]), float(v["trading_usdc"])
    try:
        from ..exchange.runtime import load_runtime_settings

        ex = str(load_runtime_settings().exchange or "").strip().lower()
        if ex in ("binance", "bn"):
            return None, None
        from ..live.okx_funds import OkxFundsClient

        client = OkxFundsClient()
        try:
            bal = client.fetch_balances()
            tu = bal.get("trading_usdt")
            tc = bal.get("trading_usdc")
            return (
                float(tu) if tu is not None else None,
                float(tc) if tc is not None else None,
            )
        finally:
            client.close()
    except Exception as e:
        logger.warning("risk_sizing trading balance failed: %s", e)
        return None, None


def compute_risk_sizing(
    *,
    index_px: float,
    option_ask: float,
    db: Database | None = None,
    perp_unit: float | None = None,
    option_unit: float | None = None,
    exit_unit: float | None = None,
    leverage_basis: str | None = None,
) -> RiskSizingResult:
    database = db or get_db()
    ledger = Ledger(database)
    s = get_settings()
    fee_rate = ledger.get_setting_float("fee_rate", s.fee_rate)
    perp_u, opt_u, exit_u = read_risk_units(ledger)
    if perp_unit is not None and float(perp_unit) > 0:
        perp_u = float(perp_unit)
    if option_unit is not None and float(option_unit) > 0:
        opt_u = float(option_unit)
    if exit_unit is not None and float(exit_unit) > 0:
        exit_u = float(exit_unit)
    basis_raw = (
        leverage_basis
        if leverage_basis is not None
        else (
            ledger.get_setting_str("risk_leverage_basis", s.risk_leverage_basis)
            or s.risk_leverage_basis
        )
    )
    min_opt_lev = ledger.get_setting_float(
        "min_option_leverage", s.min_option_leverage
    )
    sizing_ask, basis = resolve_sizing_option_ask(
        index_px=float(index_px),
        option_ask=float(option_ask),
        leverage_basis=str(basis_raw),
        min_option_leverage=float(min_opt_lev),
    )
    budget, bud_detail, capital = resolve_budget(database)
    if budget is None:
        return RiskSizingResult(
            ok=False,
            detail=f"以损定仓预算失败: {bud_detail}",
            leverage_basis=basis,
            actual_option_ask=_round2(float(option_ask)),
            option_ask=_round2(float(sizing_ask)),
        )
    r = compute_k(
        budget=budget,
        index_px=index_px,
        option_ask=sizing_ask,
        fee_rate=fee_rate,
        perp_unit=perp_u,
        option_unit=opt_u,
        exit_unit=exit_u,
    )
    if not r.ok:
        return RiskSizingResult(
            ok=False,
            detail=r.detail,
            budget=_round2(budget),
            capital_base=_round2(capital) if capital is not None else None,
            index_px=float(index_px),
            option_ask=_round2(float(sizing_ask)),
            actual_option_ask=_round2(float(option_ask)),
            leverage_basis=basis,
            k=r.k,
            perp_unit=perp_u,
            option_unit=opt_u,
            exit_unit=exit_u,
        )
    return RiskSizingResult(
        ok=True,
        detail=r.detail,
        k=r.k,
        budget=_round2(budget),
        capital_base=_round2(capital) if capital is not None else None,
        premium_est=r.premium_est,
        fee_est=r.fee_est,
        max_loss=r.max_loss,
        perp_qty_eth=r.perp_qty_eth,
        option_qty_eth=r.option_qty_eth,
        net_profit_target=r.net_profit_target,
        index_px=r.index_px,
        option_ask=_round2(float(sizing_ask)),
        actual_option_ask=_round2(float(option_ask)),
        leverage_basis=basis,
        perp_unit=perp_u,
        option_unit=opt_u,
        exit_unit=exit_u,
    )


@dataclass(frozen=True, slots=True)
class OoSizingResult:
    ok: bool
    detail: str
    budget: float | None = None
    spend: float | None = None
    qty_eth: float | None = None  # 兼容：Call 数量
    call_qty_eth: float | None = None
    put_qty_eth: float | None = None
    call_ask: float | None = None
    put_ask: float | None = None
    call_premium: float | None = None
    put_premium: float | None = None
    max_loss: float | None = None
    net_profit_target: float | None = None
    capital_base: float | None = None
    cushion: float | None = None
    reward_ratio: float | None = None
    leg_budget: float | None = None  # 单腿权利金预算（B/2×cushion）


def compute_oo_sizing(
    *,
    budget: float,
    call_ask: float,
    put_ask: float,
    fee_rate: float = 0.0005,
    index_px: float = 0.0,
    cushion: float = 0.92,
    reward_ratio: float = 2.0,
) -> OoSizingResult:
    """
    期期：总预算 B 平分给 Call/Put（各约 B/2，再乘 cushion 预留）；
    两腿按各自卖一独立定仓 qty=floor_1dp(腿预算/ask)，数量可以不同；
    出场目标 = B × reward_ratio（按全额预算）。
    """
    if budget is None or budget <= 0 or not math.isfinite(budget):
        return OoSizingResult(ok=False, detail="期期预算无效")
    if call_ask <= 0 or put_ask <= 0:
        return OoSizingResult(ok=False, detail="期期缺少有效卖一")
    cush = min(1.0, max(0.5, float(cushion)))
    ratio = max(0.5, float(reward_ratio))
    # 各腿：总预算一半 × 预留
    leg_raw = float(budget) / 2.0
    leg_budget = leg_raw * cush
    # 单腿开仓费粗估（从该腿预算里扣）
    fee_one = 0.0
    if index_px and index_px > 0 and fee_rate > 0:
        fee_one = float(index_px) * float(fee_rate)
    leg_spend = max(0.0, leg_budget - fee_one)
    if leg_spend <= 1e-9:
        return OoSizingResult(ok=False, detail="期期单腿预留后可用权利金不足")

    def _leg_qty(ask: float) -> tuple[float, float]:
        q = floor_k_1dp(leg_spend / float(ask))
        while q >= 0.1 - 1e-12:
            prem = float(ask) * q
            if prem <= leg_spend + 1e-6:
                return round(q, 1), prem
            q = round(q - 0.1, 1)
        return 0.0, 0.0

    q_call, cp = _leg_qty(float(call_ask))
    q_put, pp = _leg_qty(float(put_ask))
    if q_call < 0.1 - 1e-12 or q_put < 0.1 - 1e-12:
        return OoSizingResult(
            ok=False,
            detail=(
                f"期期定仓失败：Call可{q_call} Put可{q_put}（各腿预算约"
                f"{leg_budget:.2f}U），总预算 {budget:.2f}U 不足"
            ),
            budget=_round2(float(budget)),
            leg_budget=_round2(leg_budget),
        )
    spend = leg_budget * 2.0
    return OoSizingResult(
        ok=True,
        detail="ok",
        budget=_round2(float(budget)),
        spend=_round2(spend),
        qty_eth=round(q_call, 1),
        call_qty_eth=round(q_call, 1),
        put_qty_eth=round(q_put, 1),
        call_ask=_round2(float(call_ask)),
        put_ask=_round2(float(put_ask)),
        call_premium=_round2(cp),
        put_premium=_round2(pp),
        max_loss=_round2(cp + pp + fee_one * 2.0),
        net_profit_target=_round2(float(budget) * ratio),
        cushion=cush,
        reward_ratio=ratio,
        leg_budget=_round2(leg_budget),
    )


def apply_oo_sizing_to_ledger(
    *,
    call_ask: float,
    put_ask: float,
    index_px: float,
    db: Database | None = None,
) -> OoSizingResult:
    database = db or get_db()
    ledger = Ledger(database)
    s = get_settings()
    pos = database.fetchone("SELECT status FROM positions WHERE id=1")
    if pos is not None:
        st = str(pos["status"] or "flat")
        if st in ("open", "half_open", "option_closed_perp_pending", "opening"):
            return OoSizingResult(
                ok=False,
                detail="持仓中已锁定本组成交目标与名义，平仓后再自动计算",
            )
    budget, detail, capital = resolve_budget(database)
    if budget is None:
        return OoSizingResult(ok=False, detail=f"期期预算失败: {detail}")
    fee_rate = ledger.get_setting_float("fee_rate", s.fee_rate)
    cushion = ledger.get_setting_float("oo_budget_cushion", s.oo_budget_cushion)
    ratio = ledger.get_setting_float("oo_reward_ratio", s.oo_reward_ratio)
    r = compute_oo_sizing(
        budget=float(budget),
        call_ask=float(call_ask),
        put_ask=float(put_ask),
        fee_rate=fee_rate,
        index_px=float(index_px),
        cushion=cushion,
        reward_ratio=ratio,
    )
    if not r.ok:
        return r
    call_q = float(r.call_qty_eth or r.qty_eth or 0)
    put_q = float(r.put_qty_eth or r.qty_eth or 0)
    database.set_setting("exit_mode", "fixed_usdt")
    database.set_setting("perp_qty_eth", "0")
    database.set_setting("option_qty_eth", str(call_q))
    database.set_setting("oo_put_qty_eth", str(put_q))
    database.set_setting("net_profit_target", str(r.net_profit_target))
    database.set_setting("risk_last_k", str(call_q))
    database.set_setting(
        "risk_last_max_loss",
        f"{r.max_loss:.2f}" if r.max_loss is not None else "",
    )
    logger.info(
        "oo_sizing applied call_qty=%.1f put_qty=%.1f call_ask=%.4f put_ask=%.4f "
        "exit=%.2f max_loss=%.2f budget=%.2f leg=%.2f",
        call_q,
        put_q,
        r.call_ask or 0,
        r.put_ask or 0,
        r.net_profit_target or 0,
        r.max_loss or 0,
        r.budget or 0,
        r.leg_budget or 0,
    )
    # attach capital for callers
    return OoSizingResult(
        ok=True,
        detail=r.detail,
        budget=r.budget,
        spend=r.spend,
        qty_eth=call_q,
        call_qty_eth=call_q,
        put_qty_eth=put_q,
        call_ask=r.call_ask,
        put_ask=r.put_ask,
        call_premium=r.call_premium,
        put_premium=r.put_premium,
        max_loss=r.max_loss,
        net_profit_target=r.net_profit_target,
        capital_base=_round2(capital) if capital is not None else None,
        cushion=r.cushion,
        reward_ratio=r.reward_ratio,
        leg_budget=r.leg_budget,
    )


def apply_risk_sizing_to_ledger(
    *,
    index_px: float,
    option_ask: float,
    db: Database | None = None,
    perp_unit: float | None = None,
    option_unit: float | None = None,
    exit_unit: float | None = None,
    leverage_basis: str | None = None,
) -> RiskSizingResult:
    """计算并写入 perp/option/exit；非以损定仓模式直接 ok 跳过。持仓中拒绝改写。"""
    database = db or get_db()
    ledger = Ledger(database)
    if not is_risk_based(ledger):
        return RiskSizingResult(ok=True, detail="manual_sizing_skip")

    # 有活跃仓：本组成场参数已锁定，禁止重算覆盖
    pos = database.fetchone("SELECT status, group_id FROM positions WHERE id=1")
    if pos is not None:
        st = str(pos["status"] or "flat")
        if st in ("open", "half_open", "option_closed_perp_pending", "opening"):
            return RiskSizingResult(
                ok=False,
                detail="持仓中已锁定本组成交目标与名义，平仓后再自动计算",
            )

    r = compute_risk_sizing(
        index_px=index_px,
        option_ask=option_ask,
        db=database,
        perp_unit=perp_unit,
        option_unit=option_unit,
        exit_unit=exit_unit,
        leverage_basis=leverage_basis,
    )
    if not r.ok:
        return r

    exit_mode = str(
        ledger.get_setting_str("exit_mode", "fixed_usdt") or "fixed_usdt"
    ).strip().lower()
    s = get_settings()
    prem_mult = float(
        ledger.get_setting_float("premium_exit_multiple", s.premium_exit_multiple)
        or s.premium_exit_multiple
    )
    database.set_setting("perp_qty_eth", str(r.perp_qty_eth))
    database.set_setting("option_qty_eth", str(r.option_qty_eth))
    database.set_setting("risk_last_k", str(r.k))
    database.set_setting(
        "risk_last_max_loss",
        f"{r.max_loss:.2f}" if r.max_loss is not None else "",
    )
    if exit_mode == "premium_multiple":
        database.set_setting("exit_mode", "premium_multiple")
        # 预估展示用：估权利金×倍数；开仓后以真实 initial_premium 锁定
        est = float(r.premium_est or 0) * max(0.0, prem_mult)
        if est > 0:
            database.set_setting("net_profit_target", f"{est:.4f}")
        exit_log = f"prem×{prem_mult:g}≈{est:.2f}"
    else:
        database.set_setting("exit_mode", "fixed_usdt")
        database.set_setting("net_profit_target", str(r.net_profit_target))
        exit_log = f"{r.net_profit_target or 0:.4f}"
    logger.info(
        "risk_sizing applied k=%.1f basis=%s sizing_ask=%.4f actual_ask=%.4f "
        "perp=%.4f opt=%.4f exit=%s max_loss=%.4f budget=%.4f",
        r.k or 0,
        r.leverage_basis or "?",
        r.option_ask or 0,
        r.actual_option_ask or 0,
        r.perp_qty_eth or 0,
        r.option_qty_eth or 0,
        exit_log,
        r.max_loss or 0,
        r.budget or 0,
    )
    return r


def _hedge_mode(ledger: Ledger | None = None) -> str:
    led = ledger or Ledger()
    s = get_settings()
    raw = str(led.get_setting_str("hedge_mode", s.hedge_mode) or s.hedge_mode).strip().lower()
    return raw if raw in ("perp_option", "option_option") else "perp_option"


def preview_risk_sizing(db: Database | None = None) -> dict[str, Any]:
    """设置页/计划页预览：用当前盘口粗估。"""
    database = db or get_db()
    ledger = Ledger(database)
    out: dict[str, Any] = {
        "sizing_mode": ledger.get_setting_str("sizing_mode", "manual") or "manual",
        "risk_based": is_risk_based(ledger),
        "hedge_mode": _hedge_mode(ledger),
    }
    if not is_risk_based(ledger):
        out["ok"] = True
        out["detail"] = "当前为手动仓位"
        return out

    if out["hedge_mode"] == "option_option":
        return _preview_oo_sizing(database, ledger, out)

    try:
        from .open_capacity import _index_and_option_ask

        idx, ask = _index_and_option_ask()
    except Exception:
        idx, ask = None, None
    if idx is None or ask is None:
        out["ok"] = False
        out["detail"] = "暂无指数或期权卖一，无法预览"
        return out

    # 半自动：用本单配比/净利基数，并以盘口卖一定仓（与开仓一致）
    semi_perp = semi_opt = semi_exit = None
    semi_on = False
    try:
        from .semi_auto import is_semi_auto, read_semi_params

        if is_semi_auto(ledger):
            sp = read_semi_params(ledger)
            semi_on = True
            semi_perp = float(sp["perp_unit"])
            semi_opt = float(sp["option_unit"])
            semi_exit = float(sp["perp_exit_unit"])
    except Exception:
        logger.debug("preview semi units skipped", exc_info=True)

    r = compute_risk_sizing(
        index_px=float(idx),
        option_ask=float(ask),
        db=database,
        perp_unit=semi_perp,
        option_unit=semi_opt,
        exit_unit=semi_exit,
        leverage_basis="actual" if semi_on else None,
    )
    perp_u, opt_u, exit_u = read_risk_units(ledger)
    if semi_perp is not None:
        perp_u = float(semi_perp)
    if semi_opt is not None:
        opt_u = float(semi_opt)
    if semi_exit is not None:
        exit_u = float(semi_exit)
    mg = resolve_martingale(database, ledger=ledger)
    s = get_settings()
    exit_mode = str(
        ledger.get_setting_str("exit_mode", s.exit_mode) or s.exit_mode
    ).strip().lower()
    prem_mult = float(
        ledger.get_setting_float("premium_exit_multiple", s.premium_exit_multiple)
        or s.premium_exit_multiple
    )
    exit_target = r.net_profit_target
    exit_label = "基数×k"
    if semi_on:
        exit_label = "半自动净利基数×k"
        exit_mode = "fixed_usdt"
    elif exit_mode == "premium_multiple":
        exit_label = f"权利金×{prem_mult:g}"
        if r.ok and r.premium_est is not None:
            exit_target = round(float(r.premium_est) * max(0.0, prem_mult), 2)
    out.update(
        {
            "ok": r.ok,
            "detail": r.detail,
            "k": r.k,
            "budget": r.budget,
            "capital_base": r.capital_base,
            "premium_est": r.premium_est,
            "fee_est": r.fee_est,
            "max_loss": r.max_loss,
            "perp_qty_eth": r.perp_qty_eth,
            "option_qty_eth": r.option_qty_eth,
            "net_profit_target": exit_target,
            "exit_mode": exit_mode if exit_mode in ("fixed_usdt", "premium_multiple") else "fixed_usdt",
            "premium_exit_multiple": prem_mult,
            "exit_label": exit_label,
            "index_px": r.index_px,
            "option_ask": r.option_ask,
            "actual_option_ask": r.actual_option_ask,
            "leverage_basis": r.leverage_basis,
            "perp_unit": perp_u,
            "option_unit": opt_u,
            "exit_unit": exit_u,
            "semi_units": semi_on,
            "martingale": mg,
            "risk_effective_loss_pct": mg.get("effective_pct"),
        }
    )
    return out


def _preview_oo_sizing(
    database: Database, ledger: Ledger, out: dict[str, Any]
) -> dict[str, Any]:
    """期期预览：出场目标 = 预算 × 盈亏比；有卖一时再估单腿 qty。"""
    s = get_settings()
    budget, detail, capital = resolve_budget(database)
    mg = resolve_martingale(database, ledger=ledger)
    ratio = float(
        ledger.get_setting_float("oo_reward_ratio", s.oo_reward_ratio)
        or s.oo_reward_ratio
    )
    cush = float(
        ledger.get_setting_float("oo_budget_cushion", s.oo_budget_cushion)
        or s.oo_budget_cushion
    )
    out["martingale"] = mg
    out["risk_effective_loss_pct"] = mg.get("effective_pct")
    out["reward_ratio"] = ratio
    out["cushion"] = cush
    if budget is None:
        out["ok"] = False
        out["detail"] = f"期期预算失败: {detail}"
        return out
    exit_target = _round2(float(budget) * max(0.5, ratio))
    out.update(
        {
            "budget": _round2(float(budget)),
            "capital_base": _round2(float(capital)) if capital is not None else None,
            "net_profit_target": exit_target,
            "k": None,
            "perp_qty_eth": 0.0,
        }
    )
    call_ask = put_ask = idx = None
    try:
        from .session import get_session

        snap = get_session().snapshot()
        idx = snap.index_px
        if snap.call and snap.call.ask and float(snap.call.ask) > 0:
            call_ask = float(snap.call.ask)
        if snap.put and snap.put.ask and float(snap.put.ask) > 0:
            put_ask = float(snap.put.ask)
        if idx is None and snap.perp and snap.perp.mark_px:
            idx = float(snap.perp.mark_px)
    except Exception:
        pass
    if call_ask is None or put_ask is None:
        try:
            from .open_capacity import _index_and_option_ask

            i2, a2 = _index_and_option_ask()
            if idx is None:
                idx = i2
            # 回退：单腿 ATM 卖一不够准确，但至少能估数量量级
            if call_ask is None and a2 is not None and float(a2) > 0:
                call_ask = float(a2)
            if put_ask is None and a2 is not None and float(a2) > 0:
                put_ask = float(a2)
        except Exception:
            pass
    if call_ask is None or put_ask is None or call_ask <= 0 or put_ask <= 0:
        out["ok"] = True
        out["detail"] = "已估出场目标；虚值双腿卖一未齐，数量待开仓时再算"
        out["option_qty_eth"] = None
        out["sizing_ok"] = False
        return out
    fee_rate = ledger.get_setting_float("fee_rate", s.fee_rate)
    r = compute_oo_sizing(
        budget=float(budget),
        call_ask=float(call_ask),
        put_ask=float(put_ask),
        fee_rate=fee_rate,
        index_px=float(idx or 0),
        cushion=cush,
        reward_ratio=ratio,
    )
    # 出场始终按全额预算×盈亏比；数量估失败仍返回 ok 以便 Plan 展示目标
    out.update(
        {
            "ok": True,
            "sizing_ok": bool(r.ok),
            "detail": "ok" if r.ok else str(r.detail or "期期数量未估出"),
            "option_qty_eth": r.qty_eth if r.ok else None,
            "call_qty_eth": r.call_qty_eth if r.ok else None,
            "put_qty_eth": r.put_qty_eth if r.ok else None,
            "leg_budget": r.leg_budget if r.ok else None,
            "call_ask": r.call_ask,
            "put_ask": r.put_ask,
            "call_premium": r.call_premium if r.ok else None,
            "put_premium": r.put_premium if r.ok else None,
            "max_loss": r.max_loss if r.ok else None,
            "net_profit_target": exit_target,
            "index_px": _round2(float(idx)) if idx is not None else None,
        }
    )
    return out