crypto_monitor/lib/trade/entry_model_lib.py

"""趋势户开仓类型：反转·启动 / 顺势·大分歧 / 波段·小分歧；日内户单独 profile。"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Mapping, Optional, Sequence, Tuple

from lib.trade.trade_policy_lib import TradePolicy

PROFILE_TREND_DIV = "trend_div"
PROFILE_INTRADAY = "intraday"

ENTRY_CATEGORY_REVERSAL = "reversal"
ENTRY_CATEGORY_TREND = "trend"
ENTRY_CATEGORY_SWING = "swing"

ENTRY_MODEL_LAUNCH_A = "launch_a"
ENTRY_MODEL_LAUNCH_B = "launch_b"
ENTRY_MODEL_BIG_DIV_A = "big_div_a"
ENTRY_MODEL_BIG_DIV_B = "big_div_b"
ENTRY_MODEL_SMALL_DIV = "small_div"

VALID_ENTRY_MODEL_CODES = frozenset(
    {
        ENTRY_MODEL_LAUNCH_A,
        ENTRY_MODEL_LAUNCH_B,
        ENTRY_MODEL_BIG_DIV_A,
        ENTRY_MODEL_BIG_DIV_B,
        ENTRY_MODEL_SMALL_DIV,
    }
)

ENTRY_CATEGORY_LABELS: dict[str, str] = {
    ENTRY_CATEGORY_REVERSAL: "反转",
    ENTRY_CATEGORY_TREND: "顺势",
    ENTRY_CATEGORY_SWING: "波段",
}

TREND_DIV_ENTRY_REASON_LABELS: Tuple[str, ...] = (
    "启动A",
    "启动B",
    "大分歧A",
    "大分歧B",
    "小分歧",
)

INTRADAY_LEGACY_TREND_ENTRY_REASONS: Tuple[str, ...] = (
    "趋势多头：4h大结构突破前进场，确认条件：三次探顶，5m收敛不创新低",
    "趋势空头：4h大结构突破前进场，确认条件：三次探底，5m收敛不创新高",
    "趋势多头：小分歧低吸入场（左侧），确认条件：二次探底",
    "趋势空头：小分歧高吸入场（左侧），确认条件：二次探顶",
    "波段单：5m顺势突破，确认条件：2根k线+成交量放大+4h同向+日成交量前20",
)

# code, label, category, trade_style, help
_ENTRY_SPECS: Tuple[Tuple[str, str, str, str, str], ...] = (
    (
        ENTRY_MODEL_LAUNCH_A,
        "启动A",
        ENTRY_CATEGORY_REVERSAL,
        "trend",
        "反转链结构内：摸参考极值前小收敛，或 B 失败后 5m N 字试仓（不单列）",
    ),
    (
        ENTRY_MODEL_LAUNCH_B,
        "启动B",
        ENTRY_CATEGORY_REVERSAL,
        "trend",
        "第二次到参考极值附近、无小收敛时的实体突破",
    ),
    (
        ENTRY_MODEL_BIG_DIV_A,
        "大分歧A",
        ENTRY_CATEGORY_TREND,
        "trend",
        "主升已确立：突破前收敛，不创新低企稳（空：不创新高）",
    ),
    (
        ENTRY_MODEL_BIG_DIV_B,
        "大分歧B",
        ENTRY_CATEGORY_TREND,
        "trend",
        "主升已确立：结构实体突破确认后入场",
    ),
    (
        ENTRY_MODEL_SMALL_DIV,
        "小分歧",
        ENTRY_CATEGORY_SWING,
        "swing",
        "主升已确立：二次探底 N 字或 5m 三均线重新多头（空：二次探顶 / 空头均线）",
    ),
)

_CODE_TO_LABEL = {code: label for code, label, _, _, _ in _ENTRY_SPECS}
_CODE_TO_STYLE = {code: style for code, _, _, style, _ in _ENTRY_SPECS}
_CODE_TO_CATEGORY = {code: cat for code, _, cat, _, _ in _ENTRY_SPECS}
_LABEL_TO_CODE = {label: code for code, label, _, _, _ in _ENTRY_SPECS}
_CODE_TO_HELP = {code: help for code, _, _, _, help in _ENTRY_SPECS}

_CATEGORY_ORDER: Tuple[str, ...] = (
    ENTRY_CATEGORY_REVERSAL,
    ENTRY_CATEGORY_TREND,
    ENTRY_CATEGORY_SWING,
)

_INTRADAY_WHITELIST = frozenset({"BTC", "ETH"})


@dataclass(frozen=True)
class EntryModelOption:
    code: str
    label: str
    category: str
    trade_style: str
    help: str


def is_intraday_trading_profile(policy: TradePolicy) -> bool:
    """日内户：启用 BTC/ETH 白名单（env 中 TRADE_SYMBOL_WHITELIST）。"""
    if not policy.symbol_restrict_enabled:
        return False
    if not policy.symbol_whitelist:
        return False
    return all(s in _INTRADAY_WHITELIST for s in policy.symbol_whitelist)


def order_entry_profile(policy: TradePolicy) -> str:
    return PROFILE_INTRADAY if is_intraday_trading_profile(policy) else PROFILE_TREND_DIV


def entry_model_options() -> Tuple[EntryModelOption, ...]:
    return tuple(
        EntryModelOption(code=code, label=label, category=cat, trade_style=style, help=help)
        for code, label, cat, style, help in _ENTRY_SPECS
    )


def entry_model_categories() -> list[dict[str, Any]]:
    """两级 UI：反转 / 顺势 / 波段 → 子选项。"""
    opts = entry_model_options()
    out: list[dict[str, Any]] = []
    for cat_key in _CATEGORY_ORDER:
        children = [
            {
                "code": o.code,
                "label": o.label,
                "trade_style": o.trade_style,
                "help": o.help,
            }
            for o in opts
            if o.category == cat_key
        ]
        if not children:
            continue
        out.append(
            {
                "key": cat_key,
                "label": ENTRY_CATEGORY_LABELS.get(cat_key, cat_key),
                "options": children,
            }
        )
    return out


def entry_model_category(code: Optional[str]) -> str:
    c = normalize_entry_model_code(code)
    return _CODE_TO_CATEGORY.get(c, "")


def normalize_entry_model_code(raw: Optional[str]) -> str:
    v = (raw or "").strip().lower()
    if v in VALID_ENTRY_MODEL_CODES:
        return v
    label = (raw or "").strip()
    if label in _LABEL_TO_CODE:
        return _LABEL_TO_CODE[label]
    return ""


def entry_model_label(code: Optional[str]) -> str:
    c = normalize_entry_model_code(code)
    return _CODE_TO_LABEL.get(c, "")


def trade_style_for_entry_model(code: Optional[str]) -> str:
    c = normalize_entry_model_code(code)
    return _CODE_TO_STYLE.get(c, "trend")


def trade_style_label_zh(trade_style: str) -> str:
    return "波段单" if (trade_style or "").strip().lower() == "swing" else "趋势单"


def trend_manual_entry_reason_count(policy: TradePolicy) -> int:
    if is_intraday_trading_profile(policy):
        return len(INTRADAY_LEGACY_TREND_ENTRY_REASONS)
    return len(TREND_DIV_ENTRY_REASON_LABELS)


def build_trend_div_entry_reason_options(
    strategy_options: Sequence[str],
) -> Tuple[str, ...]:
    return TREND_DIV_ENTRY_REASON_LABELS + tuple(strategy_options)


def build_intraday_entry_reason_options(
    key_options: Sequence[str],
    strategy_options: Sequence[str],
) -> Tuple[str, ...]:
    return INTRADAY_LEGACY_TREND_ENTRY_REASONS + tuple(key_options) + tuple(strategy_options)


def entry_reason_options_for_policy(
    policy: TradePolicy,
    key_options: Sequence[str],
    strategy_options: Sequence[str],
) -> Tuple[str, ...]:
    if is_intraday_trading_profile(policy):
        return build_intraday_entry_reason_options(key_options, strategy_options)
    return build_trend_div_entry_reason_options(strategy_options)


def parse_manual_order_style_fields(
    policy: TradePolicy,
    form: Mapping[str, Any],
    *,
    default_trade_style: str = "trend",
) -> Tuple[str, Optional[str], Optional[str]]:
    """返回 (trade_style, entry_model_code|None, error_message|None)。"""
    if is_intraday_trading_profile(policy):
        trade_style = (form.get("trade_style") or default_trade_style or "trend").strip().lower()
        if trade_style not in ("trend", "swing"):
            trade_style = "trend"
        return trade_style, None, None

    entry_model = normalize_entry_model_code(form.get("entry_model"))
    if not entry_model:
        return "", None, "请选择开仓类型（反转 / 顺势 / 波段）"
    trade_style = trade_style_for_entry_model(entry_model)
    return trade_style, entry_model, None


def resolve_trade_record_entry_reason(
    *,
    entry_reason: Optional[str] = None,
    entry_model: Optional[str] = None,
    key_signal_type: Optional[str] = None,
    monitor_type: Optional[str] = None,
    entry_reason_from_key_signal=None,
    entry_reason_for_monitor_type=None,
) -> str:
    er = (entry_reason or "").strip()
    if er:
        return er
    label = entry_model_label(entry_model)
    if label:
        return label
    kst = (key_signal_type or "").strip()
    if kst and entry_reason_from_key_signal is not None:
        from_key = (entry_reason_from_key_signal(kst) or "").strip()
        if from_key:
            return from_key
    if entry_reason_for_monitor_type is not None:
        from_mt = (entry_reason_for_monitor_type(monitor_type) or "").strip()
        if from_mt:
            return from_mt
    return ""


def enrich_entry_model_display(item: dict) -> dict:
    code = normalize_entry_model_code(item.get("entry_model"))
    if code:
        item["entry_model"] = code
        item["entry_model_label"] = entry_model_label(code)
        cat = entry_model_category(code)
        if cat:
            item["entry_model_category"] = cat
            item["entry_model_category_label"] = ENTRY_CATEGORY_LABELS.get(cat, "")
    else:
        item.setdefault("entry_model_label", "")
    return item


def order_entry_template_context(policy: TradePolicy) -> dict:
    profile = order_entry_profile(policy)
    opts = entry_model_options()
    return {
        "order_entry_profile": profile,
        "intraday_discipline": profile == PROFILE_INTRADAY,
        "entry_model_options": [
            {
                "code": o.code,
                "label": o.label,
                "category": o.category,
                "trade_style": o.trade_style,
                "help": o.help,
            }
            for o in opts
        ],
        "entry_model_categories": entry_model_categories(),
        "entry_model_trade_style_map": {o.code: o.trade_style for o in opts},
        "entry_model_code_to_category": {o.code: o.category for o in opts},
    }


def hub_meta_entry_context(policy: TradePolicy) -> dict:
    """供 /api/hub/meta：中控按 profile 隐藏平仓/委托等。"""
    profile = order_entry_profile(policy)
    return {
        "order_entry_profile": profile,
        "intraday_discipline": profile == PROFILE_INTRADAY,
    }


def migrate_entry_model_columns(conn) -> None:
    for table in ("order_monitors", "trade_records"):
        try:
            conn.execute(f"ALTER TABLE {table} ADD COLUMN entry_model TEXT")
        except Exception:
            pass