CoinOX/lib/key_monitor/fib_key_monitor_lib.py

"""斐波关键位监控:纯计算与类型判断(Gate / Binance 主站共用)."""

from lib.key_monitor.key_monitor_lib import KEY_MONITOR_AUTO_TYPES

FIB_KEY_MONITOR_TYPES = frozenset({"斐波回调0.618", "斐波回调0.786"})
KEY_MONITOR_TRADE_TYPE = "关键位监控"

FIB_RATIO_BY_TYPE = {
    "斐波回调0.618": 0.618,
    "斐波回调0.786": 0.786,
}


def is_fib_key_monitor_type(monitor_type):
    return (monitor_type or "").strip() in FIB_KEY_MONITOR_TYPES


def fib_ratio_from_type(monitor_type):
    return FIB_RATIO_BY_TYPE.get((monitor_type or "").strip())


def calc_fib_plan(direction, upper, lower, ratio):
    """
    上沿 H,下沿 L(H > L).
    做多:自 H 向下回撤 ratio,E = H - ratio*(H-L);SL=L,TP=H.
    做空:自 L 向上反弹 ratio,E = L + ratio*(H-L);SL=H,TP=L.
    返回 (entry, stop_loss, take_profit) 或 None.
    """
    try:
        h = float(upper)
        l = float(lower)
        r = float(ratio)
    except (TypeError, ValueError):
        return None
    if h <= l or r <= 0 or r >= 1:
        return None
    span = h - l
    direction = (direction or "long").strip().lower()
    if direction == "short":
        entry = l + r * span
        return entry, h, l
    entry = h - r * span
    return entry, l, h


def stored_key_signal_type(monitor_type):
    """写入 order_monitors / trade_records 的 key_signal_type(箱体/收敛/斐波/假突破/触价开仓)."""
    mt = (monitor_type or "").strip()
    if mt in FIB_KEY_MONITOR_TYPES:
        return mt
    if mt in ("假突破", "回调触价开仓", "突破触价开仓", "触价开仓"):
        return mt if mt != "触价开仓" else "回调触价开仓"
    if mt in KEY_MONITOR_AUTO_TYPES:
        return mt
    return None


KEY_ENTRY_REASON_BY_SIGNAL = {
    "箱体突破": "关键位箱体突破",
    "收敛突破": "关键位收敛突破",
    "斐波回调0.618": "关键位斐波0.618",
    "斐波回调0.786": "关键位斐波0.786",
    "假突破": "关键位假突破",
    "回调触价开仓": "关键位回调触价开仓",
    "突破触价开仓": "关键位突破触价开仓",
    "触价开仓": "关键位触价开仓",
    "趋势回调": "趋势回调",
}


def entry_reason_from_key_signal(key_signal_type):
    return KEY_ENTRY_REASON_BY_SIGNAL.get((key_signal_type or "").strip())


def key_signal_type_for_trade_record(key_signal_type, box_auto_types):
    """平仓写入 trade_records 时保留箱体/收敛/斐波/假突破来源."""
    kst = (key_signal_type or "").strip()
    if kst in FIB_KEY_MONITOR_TYPES:
        return kst
    if kst in ("假突破", "回调触价开仓", "突破触价开仓", "触价开仓"):
        return kst if kst != "触价开仓" else "回调触价开仓"
    if box_auto_types and kst in box_auto_types:
        return kst
    return None


def backfill_missing_key_signal_types(conn, *, monitor_type: str = KEY_MONITOR_TRADE_TYPE) -> int:
    """补全历史 trade_records / order_monitors 中缺失的箱体/收敛 key_signal_type."""
    mt = (monitor_type or KEY_MONITOR_TRADE_TYPE).strip()
    updated = 0
    for signal in KEY_MONITOR_AUTO_TYPES:
        entry_reason = KEY_ENTRY_REASON_BY_SIGNAL.get(signal)
        if entry_reason:
            cur = conn.execute(
                """UPDATE trade_records SET key_signal_type=?
                   WHERE monitor_type=? AND (key_signal_type IS NULL OR TRIM(key_signal_type)='')
                   AND TRIM(COALESCE(entry_reason, ''))=?""",
                (signal, mt, entry_reason),
            )
            updated += int(cur.rowcount or 0)
    rows = conn.execute(
        """SELECT id, symbol, opened_at FROM trade_records
           WHERE monitor_type=? AND (key_signal_type IS NULL OR TRIM(key_signal_type)='')""",
        (mt,),
    ).fetchall()
    for row in rows:
        # init_db 连接未设 row_factory,结果为 tuple
        rid, sym, opened_at = row[0], row[1], row[2]
        opened = (opened_at or "").strip()
        for signal in KEY_MONITOR_AUTO_TYPES:
            hist = conn.execute(
                """SELECT monitor_type FROM key_monitor_history
                   WHERE symbol=? AND monitor_type=? AND close_reason='auto_opened'
                   AND (?='' OR closed_at <= ?)
                   ORDER BY closed_at DESC LIMIT 1""",
                (sym, signal, opened, opened),
            ).fetchone()
            if not hist:
                continue
            conn.execute(
                "UPDATE trade_records SET key_signal_type=? WHERE id=?",
                (signal, rid),
            )
            updated += 1
            break
    return updated


def fib_invalidate_by_mark(direction, mark_price, upper, lower):
    """先触达止盈侧(标记价)则失效.多:mark>=H;空:mark<=L."""
    try:
        m = float(mark_price)
        h = float(upper)
        l = float(lower)
    except (TypeError, ValueError):
        return False
    direction = (direction or "long").strip().lower()
    if direction == "short":
        return m <= l
    return m >= h