diff --git a/kis_trader/utils/perm_price_alert.py b/kis_trader/utils/perm_price_alert.py new file mode 100644 index 0000000..4e6a9d2 --- /dev/null +++ b/kis_trader/utils/perm_price_alert.py @@ -0,0 +1,186 @@ +""" +영구구독 목표가 도달 → 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 diff --git a/permanent_subs.py b/permanent_subs.py index ab116d6..40e0fbd 100644 --- a/permanent_subs.py +++ b/permanent_subs.py @@ -18,6 +18,7 @@ permanent_subs.py from __future__ import annotations import logging +from datetime import datetime from typing import Any, Dict, List, Optional logger = logging.getLogger("permanent_subs") @@ -39,11 +40,22 @@ CREATE TABLE IF NOT EXISTS permanent_subscriptions ( tf_min INT NOT NULL DEFAULT 1, enabled TINYINT NOT NULL DEFAULT 1, note VARCHAR(100) NOT NULL DEFAULT '', + alert_price DOUBLE NULL, + alert_side VARCHAR(8) NOT NULL DEFAULT 'gte', + alert_armed TINYINT NOT NULL DEFAULT 1, + alert_fired_at DATETIME NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uq_perm_code (code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='영구구독(국내 WS + 해외 WS) 종목' """ +_PERM_ALERT_COLS = ( + ("alert_price", "DOUBLE NULL COMMENT '목표가 알람 가격(NULL=미설정)'"), + ("alert_side", "VARCHAR(8) NOT NULL DEFAULT 'gte' COMMENT 'gte=이상 lte=이하'"), + ("alert_armed", "TINYINT NOT NULL DEFAULT 1 COMMENT '1=감시중 0=발화후대기'"), + ("alert_fired_at", "DATETIME NULL COMMENT '마지막 알람 시각'"), +) + def _core(db: Any): """TradeDBExt 이면 ``.raw``, 아니면 그대로.""" @@ -53,6 +65,23 @@ def _core(db: Any): def ensure_permanent_subs_table(db: Any) -> None: raw = _core(db) raw.conn.execute(_PERM_DDL.strip()) + # 기존 테이블에 알람 컬럼 보강 + try: + cols = {c["Field"] for c in raw.conn.execute( + "SHOW COLUMNS FROM permanent_subscriptions" + ).fetchall() or []} + except Exception: + cols = set() + for name, ddl in _PERM_ALERT_COLS: + if name in cols: + continue + try: + raw.conn.execute( + f"ALTER TABLE permanent_subscriptions ADD COLUMN `{name}` {ddl}" + ) + logger.info("📌 permanent_subscriptions.%s 컬럼 추가", name) + except Exception as e: + logger.debug("perm alert col %s: %s", name, e) raw.conn.commit() @@ -88,11 +117,15 @@ def upsert_permanent_sub( tf_min: int = 1, enabled: bool = True, note: str = "", + alert_price: Optional[Any] = None, + alert_side: Optional[str] = None, + update_alert: bool = False, ) -> None: """코드 1건 등록/갱신 (code UNIQUE). tf_min 은 저장·표시용 기준봉. WS는 틱→1분 집계가 기본이고 상위봉은 롤업하므로 기본·권장은 1분 (env ``PERM_SUB_BASE_TF_MIN``). + update_alert=True 일 때만 alert_* 를 덮어씀 (일반 저장 시 알람값 유지). """ ensure_permanent_subs_table(db) code = str(code or "").strip().upper() @@ -109,26 +142,148 @@ def upsert_permanent_sub( base_tf = 1 if base_tf < 1: base_tf = 1 - # 요청값과 무관하게 기준봉으로 통일 (1분→롤업) try: _ = int(tf_min) except (TypeError, ValueError): pass tfv = base_tf raw = _core(db) - raw.conn.execute( + if update_alert: + ap = None + if alert_price not in (None, ""): + try: + ap = float(alert_price) + if ap <= 0: + ap = None + except (TypeError, ValueError): + ap = None + aside = str(alert_side or "gte").strip().lower() + if aside not in ("gte", "lte"): + aside = "gte" + # 목표가 변경/설정 시 재무장 + armed = 1 if ap is not None else 0 + raw.conn.execute( + """ + INSERT INTO permanent_subscriptions + (code, market_type, exchange, symbol, tf_min, enabled, note, + alert_price, alert_side, alert_armed, alert_fired_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NULL) + ON DUPLICATE KEY UPDATE + market_type=VALUES(market_type), exchange=VALUES(exchange), + symbol=VALUES(symbol), tf_min=VALUES(tf_min), + enabled=VALUES(enabled), note=VALUES(note), + alert_price=VALUES(alert_price), alert_side=VALUES(alert_side), + alert_armed=VALUES(alert_armed), alert_fired_at=NULL + """, + [code, mt, ex, sym, tfv, 1 if enabled else 0, str(note or "")[:100], + ap, aside, armed], + ) + else: + raw.conn.execute( + """ + INSERT INTO permanent_subscriptions + (code, market_type, exchange, symbol, tf_min, enabled, note) + VALUES (%s, %s, %s, %s, %s, %s, %s) + ON DUPLICATE KEY UPDATE + market_type=VALUES(market_type), exchange=VALUES(exchange), + symbol=VALUES(symbol), tf_min=VALUES(tf_min), + enabled=VALUES(enabled), note=VALUES(note) + """, + [code, mt, ex, sym, tfv, 1 if enabled else 0, str(note or "")[:100]], + ) + raw.conn.commit() + + +def set_alert( + db: Any, + code: str, + alert_price: Optional[Any] = None, + alert_side: str = "gte", + rearm: bool = True, +) -> bool: + """목표가 알람만 갱신. alert_price 비우면 해제.""" + ensure_permanent_subs_table(db) + code = str(code or "").strip().upper() + if not code: + return False + ap = None + if alert_price not in (None, ""): + try: + ap = float(alert_price) + if ap <= 0: + ap = None + except (TypeError, ValueError): + ap = None + aside = str(alert_side or "gte").strip().lower() + if aside not in ("gte", "lte"): + aside = "gte" + armed = 1 if (ap is not None and rearm) else 0 + raw = _core(db) + cur = raw.conn.execute( """ - INSERT INTO permanent_subscriptions - (code, market_type, exchange, symbol, tf_min, enabled, note) - VALUES (%s, %s, %s, %s, %s, %s, %s) - ON DUPLICATE KEY UPDATE - market_type=VALUES(market_type), exchange=VALUES(exchange), - symbol=VALUES(symbol), tf_min=VALUES(tf_min), - enabled=VALUES(enabled), note=VALUES(note) + UPDATE permanent_subscriptions + SET alert_price=%s, alert_side=%s, alert_armed=%s, + alert_fired_at=IF(%s IS NULL, NULL, NULL) + WHERE code=%s """, - [code, mt, ex, sym, tfv, 1 if enabled else 0, str(note or "")[:100]], + [ap, aside, armed, ap, code], ) raw.conn.commit() + try: + return bool(cur.rowcount) + except Exception: + return True + + +def rearm_alert(db: Any, code: str) -> bool: + """발화 후 재감시.""" + ensure_permanent_subs_table(db) + code = str(code or "").strip().upper() + raw = _core(db) + cur = raw.conn.execute( + """ + UPDATE permanent_subscriptions + SET alert_armed=1 + WHERE code=%s AND alert_price IS NOT NULL AND alert_price>0 + """, + [code], + ) + raw.conn.commit() + try: + return bool(cur.rowcount) + except Exception: + return True + + +def mark_alert_fired(db: Any, code: str) -> None: + ensure_permanent_subs_table(db) + raw = _core(db) + raw.conn.execute( + """ + UPDATE permanent_subscriptions + SET alert_armed=0, alert_fired_at=%s + WHERE code=%s + """, + [datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + str(code or "").strip().upper()], + ) + raw.conn.commit() + + +def list_armed_alerts(db: Any) -> List[Dict[str, Any]]: + """감시 중인 목표가 행.""" + ensure_permanent_subs_table(db) + raw = _core(db) + rows = raw.conn.execute( + """ + SELECT code, market_type, exchange, symbol, note, + alert_price, alert_side, alert_armed, alert_fired_at + FROM permanent_subscriptions + WHERE enabled=1 AND alert_armed=1 + AND alert_price IS NOT NULL AND alert_price>0 + """ + ).fetchall() or [] + return [dict(r) for r in rows] def normalize_all_tf_to_base(db: Any) -> int: @@ -152,6 +307,7 @@ def normalize_all_tf_to_base(db: Any) -> int: except Exception: return 0 + def remove_permanent_sub(db: Any, code: str) -> bool: ensure_permanent_subs_table(db) code = str(code or "").strip().upper() @@ -179,7 +335,11 @@ def set_enabled(db: Any, code: str, enabled: bool) -> None: def list_permanent_subs(db: Any, enabled_only: bool = False) -> List[Dict[str, Any]]: ensure_permanent_subs_table(db) raw = _core(db) - sql = "SELECT code, market_type, exchange, symbol, tf_min, enabled, note FROM permanent_subscriptions" + sql = ( + "SELECT code, market_type, exchange, symbol, tf_min, enabled, note, " + "alert_price, alert_side, alert_armed, alert_fired_at " + "FROM permanent_subscriptions" + ) if enabled_only: sql += " WHERE enabled=1" sql += " ORDER BY market_type, code" @@ -189,6 +349,10 @@ def list_permanent_subs(db: Any, enabled_only: bool = False) -> List[Dict[str, A for r in rows: d = dict(r) if not isinstance(r, dict) else dict(r) d["enabled"] = int(d.get("enabled", 1)) + try: + d["alert_armed"] = int(d.get("alert_armed", 1) or 0) + except (TypeError, ValueError): + d["alert_armed"] = 0 out.append(d) return out