kis_bot/kis_trader/utils/perm_price_alert.py

"""
영구구독 목표가 도달 → Mattermost 알람.

- permanent_subscriptions.alert_price / alert_side / alert_armed
- 시세: KR=WSManager(LS 우선)·US=해외 WS RAM
- 채널: KIS_PERM_SUB_MM_CHANNEL (기본 alias=permanent)
"""
from __future__ import annotations

import logging
import time
from datetime import datetime
from typing import Any, Callable, Dict, Optional

logger = logging.getLogger("perm_price_alert")


def _px_from_dict(d: Optional[Dict]) -> float:
    if not d:
        return 0.0
    for k in ("stck_prpr", "last", "price", "close"):
        try:
            v = float(str(d.get(k) or "").replace(",", "").strip())
            if v > 0:
                return v
        except (TypeError, ValueError):
            continue
    return 0.0


def resolve_live_price(
    code: str,
    market_type: str,
    *,
    ws_manager: Any = None,
    overseas_ws: Any = None,
) -> float:
    """실시간(또는 지연) 현재가. 없으면 0."""
    c = str(code or "").strip().upper()
    if not c:
        return 0.0
    mt = str(market_type or "KR").strip().upper()
    if mt == "US":
        if overseas_ws is not None and hasattr(overseas_ws, "get_price"):
            try:
                return _px_from_dict(overseas_ws.get_price(c, max_age_sec=None))
            except Exception:
                return 0.0
        return 0.0
    # KR: LS → 통합 inquire 경로
    if ws_manager is None:
        return 0.0
    try:
        ls = getattr(ws_manager, "_get_ls_ws", lambda: None)()
        if ls is not None and hasattr(ls, "get_price"):
            px = _px_from_dict(ls.get_price(c, max_age_sec=None))
            if px > 0:
                return px
    except Exception:
        pass
    try:
        if hasattr(ws_manager, "get_price"):
            return _px_from_dict(ws_manager.get_price(c, max_age_sec=None))
    except Exception:
        pass
    try:
        # 폴백: 벤더 체인
        for v in ("ls", "kiwoom", "kis"):
            if hasattr(ws_manager, "_vendor_price"):
                px = _px_from_dict(ws_manager._vendor_price(v, c, None))
                if px > 0:
                    return px
    except Exception:
        pass
    return 0.0


def _condition_met(px: float, target: float, side: str) -> bool:
    if px <= 0 or target <= 0:
        return False
    s = (side or "gte").strip().lower()
    if s in ("lte", "below", "down", "<="):
        return px <= target
    # 기본: 목표가 이상 (도달/돌파)
    return px >= target


def tick_perm_price_alerts(
    db: Any,
    *,
    ws_manager: Any = None,
    overseas_ws: Any = None,
    send_mm: Optional[Callable[[str, str], bool]] = None,
) -> int:
    """
    armed 목표가 검사 1회. 도달 시 MM 발송 후 armed=0.
    반환: 발송 시도 건수.
    """
    from kis_trader.utils.env import get_env_bool, get_env_from_db

    if not get_env_bool("PERM_ALERT_ENABLED", True):
        return 0
    try:
        import permanent_subs as ps
    except Exception as e:
        logger.debug("perm_price_alert import: %s", e)
        return 0
    try:
        rows = ps.list_armed_alerts(db)
    except Exception as e:
        logger.debug("list_armed_alerts: %s", e)
        return 0
    if not rows:
        return 0
    ch = (
        get_env_from_db("KIS_PERM_SUB_MM_CHANNEL", "permanent") or "permanent"
    ).strip() or "permanent"
    n = 0
    for r in rows:
        code = str(r.get("code") or "").strip().upper()
        mt = str(r.get("market_type") or "KR").strip().upper()
        try:
            target = float(r.get("alert_price") or 0)
        except (TypeError, ValueError):
            target = 0.0
        side = str(r.get("alert_side") or "gte").strip().lower()
        if not code or target <= 0:
            continue
        px = resolve_live_price(
            code, mt, ws_manager=ws_manager, overseas_ws=overseas_ws,
        )
        if not _condition_met(px, target, side):
            continue
        side_kr = "이하" if side in ("lte", "below", "down", "<=") else "이상"
        name = ""
        try:
            if mt == "US":
                name = str(r.get("symbol") or code).strip().upper()
            else:
                from kis_trader.utils.stock_name import resolve_stock_display_name
                name = resolve_stock_display_name(db, code, fallback="") or ""
                if name == code:
                    name = ""
        except Exception:
            name = ""
        title = f"**{code}**"
        if name and name != code:
            title += f" {name}"
        if mt == "US":
            body = (
                f"📡 영구구독 목표가\n"
                f"{title} ({mt})\n"
                f"현재가 **${px:.4f}** → 지정가 **${target:.4f}** ({side_kr}) 도달"
            )
        else:
            body = (
                f"📡 영구구독 목표가\n"
                f"{title} ({mt})\n"
                f"현재가 **{px:,.0f}원** → 지정가 **{target:,.0f}원** ({side_kr}) 도달"
            )
        note = str(r.get("note") or "").strip()
        if note:
            body += f"\n메모: {note}"
        ok = False
        if send_mm:
            try:
                ok = bool(send_mm(body, ch))
            except Exception as e:
                logger.warning("perm alert MM 실패 %s: %s", code, e)
        else:
            try:
                from kis_trader.utils.logger import msg_mm
                ok = bool(msg_mm(body, channel_alias=ch, jitter=False))
            except Exception as e:
                logger.warning("perm alert MM 실패 %s: %s", code, e)
        try:
            ps.mark_alert_fired(db, code)
        except Exception as e:
            logger.warning("mark_alert_fired %s: %s", code, e)
        n += 1
        logger.info(
            "📡 목표가알람 %s px=%s target=%s side=%s mm=%s",
            code, px, target, side, ok,
        )
        time.sleep(0.05)
    return n