CoinOX/lib/key_monitor/key_auto_order_lib.py

"""关键位程序自动下单开关(三所共用,与 POSITION_SIZING_MODE 联动)."""
from __future__ import annotations

import os
from typing import Any, Optional, Sequence, Tuple

from lib.key_monitor.fib_key_monitor_lib import is_fib_key_monitor_type
from lib.key_monitor.false_breakout_key_monitor_lib import is_false_breakout_key_monitor_type
from lib.key_monitor.key_monitor_full_margin_lib import monitor_type_disallowed_in_full_margin
from lib.key_monitor.key_monitor_lib import KEY_MONITOR_AUTO_TYPES, KEY_MONITOR_RS_TYPES
from lib.key_monitor.trigger_entry_key_monitor_lib import is_trigger_entry_key_monitor_type
from lib.trade.position_sizing_lib import is_full_margin_mode

KEY_ENTRY_REASON_OPTIONS: Tuple[str, ...] = (
    "关键位箱体突破",
    "关键位收敛突破",
    "关键位斐波0.618",
    "关键位斐波0.786",
    "关键位假突破",
    "关键位回调触价开仓",
    "关键位突破触价开仓",
)

KEY_ENTRY_REASON_TRIGGER_OPTIONS: frozenset[str] = frozenset(
    {
        "关键位回调触价开仓",
        "关键位突破触价开仓",
    }
)

KEY_STATS_SEGMENT_KEYS: frozenset[str] = frozenset(
    {
        "key_box",
        "key_conv",
        "key_fib618",
        "key_fib786",
        "key_false_breakout",
        "key_trigger",
    }
)

KEY_STATS_TRIGGER_ONLY: frozenset[str] = frozenset({"key_trigger"})

TREND_MANUAL_ENTRY_REASON_COUNT = 5


def _env_bool(raw: Optional[str], default: bool = False) -> bool:
    if raw is None:
        return default
    return (raw or "").strip().lower() in ("1", "true", "yes", "on")


def load_key_auto_order_enabled(env: Optional[dict] = None) -> bool:
    e = env if env is not None else os.environ
    return _env_bool(e.get("KEY_AUTO_ORDER_ENABLED"), default=False)


def is_key_level_entry_reason(reason: str) -> bool:
    return (reason or "").strip() in KEY_ENTRY_REASON_OPTIONS


def visible_key_entry_reasons(sizing_mode: str, key_auto_enabled: bool) -> Tuple[str, ...]:
    if not key_auto_enabled:
        return ()
    if is_full_margin_mode(sizing_mode):
        return tuple(x for x in KEY_ENTRY_REASON_OPTIONS if x in KEY_ENTRY_REASON_TRIGGER_OPTIONS)
    return KEY_ENTRY_REASON_OPTIONS


def effective_entry_reason_options(
    all_options: Sequence[str],
    sizing_mode: str,
    key_auto_enabled: bool,
    *,
    trend_manual_count: int = TREND_MANUAL_ENTRY_REASON_COUNT,
) -> Tuple[str, ...]:
    """复盘/表单下拉:按开关与计仓模式裁剪关键位开仓类型."""
    opts = list(all_options)
    if len(opts) <= trend_manual_count:
        return tuple(opts)
    key_visible = set(visible_key_entry_reasons(sizing_mode, key_auto_enabled))
    out: list[str] = []
    for i, item in enumerate(opts):
        if i < trend_manual_count:
            out.append(item)
        elif is_key_level_entry_reason(item):
            if item in key_visible:
                out.append(item)
        else:
            out.append(item)
    return tuple(out)


def effective_stats_segment_defs(
    segment_defs: Sequence[Tuple[str, str, Any]],
    sizing_mode: str,
    key_auto_enabled: bool,
) -> Tuple[Tuple[str, str, Any], ...]:
    if not key_auto_enabled:
        hidden = KEY_STATS_SEGMENT_KEYS
    elif is_full_margin_mode(sizing_mode):
        hidden = KEY_STATS_SEGMENT_KEYS - KEY_STATS_TRIGGER_ONLY
    else:
        hidden = frozenset()
    return tuple(x for x in segment_defs if x[0] not in hidden)


def is_key_auto_monitor_type(monitor_type: str) -> bool:
    mt = (monitor_type or "").strip()
    if mt in KEY_MONITOR_AUTO_TYPES:
        return True
    if is_fib_key_monitor_type(mt):
        return True
    if is_false_breakout_key_monitor_type(mt):
        return True
    if is_trigger_entry_key_monitor_type(mt):
        return True
    return False


def is_rs_key_monitor_type(monitor_type: str) -> bool:
    return (monitor_type or "").strip() in KEY_MONITOR_RS_TYPES


def check_monitor_type_add_allowed(
    monitor_type: str,
    sizing_mode: str,
    key_auto_enabled: bool,
) -> Tuple[bool, str]:
    mt = (monitor_type or "").strip()
    if is_rs_key_monitor_type(mt):
        return True, ""
    if not key_auto_enabled:
        return False, (
            "已关闭关键位程序自动单(KEY_AUTO_ORDER_ENABLED=false);"
            "仅可添加「关键支撑阻力」(微信提醒,不下单)."
        )
    if is_full_margin_mode(sizing_mode):
        if monitor_type_disallowed_in_full_margin(mt):
            return False, (
                "全仓杠杆模式下不可添加箱体/收敛突破,斐波或假突破监控;"
                "可使用「回调/突破触价开仓」或关键支撑阻力(仅提醒)."
            )
    if not is_key_auto_monitor_type(mt) and not is_rs_key_monitor_type(mt):
        return False, "监控类型无效"
    return True, ""


def key_auto_order_env_comment_lines() -> Tuple[str, ...]:
    return (
        "# 关键位程序自动下单(与 POSITION_SIZING_MODE 联动,修改后须重启 PM2)",
        "# 默认 false = 关闭所有关键位程序自动单(箱体/收敛/斐波/假突破/触价)",
        "#",
        "#   POSITION_SIZING_MODE=risk(以损定仓)",
        "#     false → 不执行任何关键位自动单;支撑/阻力提醒,人工下单,顺势加仓不受影响",
        "#     true  → 允许关键位全套自动(含触价)",
        "#",
        "#   POSITION_SIZING_MODE=full_margin(全仓杠杆,须无仓切换)",
        "#     false → 不执行触价自动单",
        "#     true  → 仅回调/突破触价可程序自动开仓;箱体/斐波等仍禁止",
        "#",
        "# 顺势加仓,趋势回调不受本开关控制;全仓模式下策略自动仍禁止.",
        "KEY_AUTO_ORDER_ENABLED=false",
    )