crypto_okx/lib/trade/trade_labels_lib.py

"""交易记录 / 下单监控标签工具(原 lib.strategy.strategy_trade_labels 瘦身后)."""
from __future__ import annotations

from typing import Optional

ORDER_TYPE_MANUAL = "下单监控"
ORDER_TYPE_KEY = "关键位监控"

# 历史数据兼容可读;新复盘下拉不再提供策略开仓类型
MONITOR_TYPE_TREND_PULLBACK = "趋势回调"
MONITOR_TYPE_ROLL = "顺势加仓"
ENTRY_REASON_TREND_PULLBACK = "趋势回调"
ENTRY_REASON_ROLL = "顺势加仓"

JOURNAL_ORDER_TYPE_OPTIONS = (
    ORDER_TYPE_MANUAL,
    ORDER_TYPE_KEY,
)

# 旧库兼容(normalize / 读历史);不进 JOURNAL 下拉
STRATEGY_ENTRY_REASON_OPTIONS = (
    ENTRY_REASON_TREND_PULLBACK,
    ENTRY_REASON_ROLL,
)

TREND_HANDOFF_KEY_SIGNAL = ENTRY_REASON_TREND_PULLBACK
TREND_HANDOFF_TRADE_NOTE = "趋势回调计划"


def normalize_journal_order_type(raw: Optional[str]) -> str:
    s = (raw or "").strip()
    if s in JOURNAL_ORDER_TYPE_OPTIONS:
        return s
    # 历史策略类型仍可读,映射到相近展示
    if s in (MONITOR_TYPE_TREND_PULLBACK, MONITOR_TYPE_ROLL):
        return s
    if "关键位" in s:
        return ORDER_TYPE_KEY
    return ""


def order_type_from_monitor_type(
    monitor_type: Optional[str],
    key_signal_type: Optional[str] = None,
) -> str:
    del key_signal_type
    mt = (monitor_type or "").strip()
    if mt == MONITOR_TYPE_TREND_PULLBACK:
        return MONITOR_TYPE_TREND_PULLBACK
    if mt == MONITOR_TYPE_ROLL:
        return MONITOR_TYPE_ROLL
    if mt == ORDER_TYPE_KEY or "关键位" in mt:
        return ORDER_TYPE_KEY
    return ORDER_TYPE_MANUAL


def handoff_trade_miss_reason(miss_reason, row) -> Optional[str]:
    """历史趋势保本移交监控单平仓备注兼容."""
    if trend_plan_id_from_monitor_row(row) is None:
        return miss_reason
    base = (miss_reason or "").strip()
    if TREND_HANDOFF_TRADE_NOTE in base:
        return base or TREND_HANDOFF_TRADE_NOTE
    if base:
        return f"{TREND_HANDOFF_TRADE_NOTE};{base}"
    return TREND_HANDOFF_TRADE_NOTE


def trend_plan_id_from_monitor_row(row) -> Optional[int]:
    if row is None:
        return None
    try:
        keys = row.keys() if hasattr(row, "keys") else []
    except Exception:
        keys = []
    if "trend_plan_id" not in keys or row["trend_plan_id"] in (None, ""):
        return None
    try:
        tid = int(row["trend_plan_id"])
        return tid if tid > 0 else None
    except (TypeError, ValueError):
        return None


def order_had_roll_fills(conn, order_monitor_id) -> bool:
    """策略已移除:始终 False(roll_legs 表可能仍在旧库)."""
    del conn, order_monitor_id
    return False


def _row_monitor_type(row, default_manual: str) -> str:
    if row is None:
        return default_manual
    try:
        keys = row.keys() if hasattr(row, "keys") else []
    except Exception:
        keys = []
    if "monitor_type" in keys:
        mt = (row["monitor_type"] or "").strip()
        if mt:
            return mt
    return default_manual


def _row_key_signal_type(row) -> str:
    if row is None:
        return ""
    try:
        keys = row.keys() if hasattr(row, "keys") else []
    except Exception:
        keys = []
    if "key_signal_type" not in keys:
        return ""
    return (row["key_signal_type"] or "").strip()


def order_monitor_source_type(row, *, default_manual: str = "下单监控") -> str:
    """展示/平仓记录:历史趋势移交单仍标「趋势回调」."""
    if trend_plan_id_from_monitor_row(row) is not None:
        return MONITOR_TYPE_TREND_PULLBACK
    mt = _row_monitor_type(row, default_manual)
    if mt != default_manual:
        return mt
    kst = _row_key_signal_type(row)
    if kst in (
        MONITOR_TYPE_TREND_PULLBACK,
        TREND_HANDOFF_KEY_SIGNAL,
        TREND_HANDOFF_TRADE_NOTE,
        ENTRY_REASON_TREND_PULLBACK,
    ):
        return MONITOR_TYPE_TREND_PULLBACK
    return mt


def apply_order_monitor_source_labels(item: dict, *, default_manual: str = "下单监控") -> dict:
    out = dict(item or {})
    out["monitor_type"] = order_monitor_source_type(out, default_manual=default_manual)
    return out


def trade_record_monitor_type(conn, order_row, *, default_manual: str = "下单监控") -> str:
    del conn
    return order_monitor_source_type(order_row, default_manual=default_manual)


def entry_reason_for_monitor_type(monitor_type: str | None) -> str:
    mt = (monitor_type or "").strip()
    if mt == MONITOR_TYPE_TREND_PULLBACK:
        return ENTRY_REASON_TREND_PULLBACK
    if mt == MONITOR_TYPE_ROLL:
        return ENTRY_REASON_ROLL
    return ""


def order_monitor_excluded_from_position_limit(conn, row) -> bool:
    del conn
    return order_monitor_source_type(row) == MONITOR_TYPE_TREND_PULLBACK


def count_position_limit_active_monitors(conn) -> int:
    try:
        rows = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
    except Exception:
        return 0
    n = 0
    for row in rows:
        if not order_monitor_excluded_from_position_limit(conn, row):
            n += 1
    return n