""" kis_trader/strategies/updown_watchlist.py — UPDOWN(박스권) sticky 관심종목 관리 ================================================================================ [역할] - 박스권(횡보) 후보를 담는 **멤버십 전용 테이블** (`updown_watchlist`). - 파라미터는 여기 저장하지 않는다. (글로벌 env ``UPDOWN_*`` 사용, 종목별 override 는 선택적으로 ``updow_stock_config`` 핀(pin)). 이 테이블은 "누구를 들고 보느냐"만 관리. [핵심 설계 — sticky] - 조건식/관심그룹에서 잠깐 빠져도 **자동 삭제하지 않는다** (sticky). 박스권은 며칠~몇 주 유지되므로, 리프레시 사이 멤버십이 출렁여도 watchlist 는 흔들리면 안 된다. - 진짜 퇴출(졸업/탈락/만료)은 ``DELETE`` 가 아니라 ``status`` 플래그(soft-delete). 재진입 대비로 보관하다가, retention 기간 초과 시에만 청소(purge) → 갈비지 방지. [한도/교체] - active(매수체크 대상) 행은 ``UPDOWN_WATCH_MAX`` (기본 40) 로 상한. - 꽉 찼는데 새 박스 종목이 들어오면 **시간순 교체**: 가장 오래된(신호도 없는) active 를 밀어낸다(evicted). 단 **현재 보유 중인 종목은 절대 밀어내지 않는다**(청산 감시 필요). [상태(status)] - active : 매수체크 대상 (WS 구독 + 트리거) - graduated : 익절/박스상단 돌파로 청산 성공 → 보관(쿨다운 후 재진입 가능) - failed : 박스 하단 이탈 손절 → 보관(재진입 시 재평가) - expired : N거래일 무신호 → 보관 - evicted : 한도 초과로 시간순 밀려남 → 보관 모든 수치는 하드코딩 금지 — ``get_env_int`` 로 DB/Env 에서 로드한다. """ from __future__ import annotations import logging from datetime import datetime as dt from typing import Any, Dict, List, Optional, Set from ..utils.env import get_env_int logger = logging.getLogger("kis_trader.updown_watchlist") # active(매수체크/구독) 로 인정하는 상태 — 나머지는 inactive(보관) ACTIVE_STATUS = "active" INACTIVE_STATUSES = ("graduated", "failed", "expired", "evicted") # 현재 보유 중인 종목은 eviction/expire 에서 보호 (구·신 전략명 모두) _HELD_STRATEGIES = ("UPDOWN", "UPDOW") _UPDOWN_WATCHLIST_DDL = """ CREATE TABLE IF NOT EXISTS updown_watchlist ( id BIGINT AUTO_INCREMENT PRIMARY KEY, code VARCHAR(20) NOT NULL, name VARCHAR(50) NOT NULL DEFAULT '', market_type VARCHAR(8) NOT NULL DEFAULT 'KR', exchange VARCHAR(16) NOT NULL DEFAULT 'KRX', source VARCHAR(16) NOT NULL DEFAULT 'manual', status VARCHAR(16) NOT NULL DEFAULT 'active', box_low DOUBLE NOT NULL DEFAULT 0, box_high DOUBLE NOT NULL DEFAULT 0, box_score DOUBLE NOT NULL DEFAULT 0, added_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, last_seen_at DATETIME NULL, last_signal_at DATETIME NULL, removed_at DATETIME NULL, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq_updown_watch_code (code), KEY idx_updown_watch_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='UPDOWN 박스권 sticky 관심종목(멤버십)' """ def _trade_db_core(db: Any): """TradeDBExt 이면 ``.raw``, 아니면 그대로 (다른 cfg 모듈과 동일 규약).""" return getattr(db, "raw", db) # ---------------------------------------------------------------------- # env 파라미터 (하드코딩 금지) # ---------------------------------------------------------------------- def watch_max(db: Any = None) -> int: """active 상한 — ``UPDOWN_WATCH_MAX`` (기본 30). 30 = KIS 멀티시세(intstock_multprice) **1콜 한도**. watchlist 전체를 REST 1번으로 훑을 수 있게 30 으로 맞춘다(31개 이상이면 배치가 2콜로 늘어 효율 저하). """ return max(1, get_env_int("UPDOWN_WATCH_MAX", 30)) def watch_expire_days(db: Any = None) -> int: """무신호 만료 거래일 — ``UPDOWN_WATCH_EXPIRE_DAYS`` (기본 10). 자본·슬롯 묶임 방지.""" return max(1, get_env_int("UPDOWN_WATCH_EXPIRE_DAYS", 10)) def watch_retention_days(db: Any = None) -> int: """inactive 보관 일수 — ``UPDOWN_WATCH_RETENTION_DAYS`` (기본 30). 초과 시 purge.""" return max(1, get_env_int("UPDOWN_WATCH_RETENTION_DAYS", 30)) # ---------------------------------------------------------------------- # 테이블 준비 # ---------------------------------------------------------------------- def ensure_updown_watchlist_table(db: Any) -> None: """``updown_watchlist`` 테이블 생성 (없으면).""" raw = _trade_db_core(db) try: raw.conn.execute(_UPDOWN_WATCHLIST_DDL.strip()) raw.conn.commit() logger.info("📌 updown_watchlist 테이블 확인/생성") except Exception as e: logger.warning("updown_watchlist 테이블 생성 경고: %s", e) # ---------------------------------------------------------------------- # 보유 종목 (eviction/expire 보호) # ---------------------------------------------------------------------- def _held_codes(db: Any) -> Set[str]: """현재 UPDOWN 으로 보유 중인 종목코드 (active_trades). eviction/expire 제외 대상.""" raw = _trade_db_core(db) try: ph = ", ".join(["%s"] * len(_HELD_STRATEGIES)) rows = raw.conn.execute( f"SELECT DISTINCT code FROM active_trades WHERE strategy IN ({ph})", tuple(_HELD_STRATEGIES), ).fetchall() return {str(r["code"]).strip() for r in (rows or []) if r and r.get("code")} except Exception as e: logger.debug("held_codes 조회 실패(무시): %s", e) return set() # ---------------------------------------------------------------------- # 조회 # ---------------------------------------------------------------------- def list_active_watchlist(db: Any) -> List[Dict[str, Any]]: """active 상태 종목 (WS 구독·트리거 대상). 오래된 순(신호 우선)으로 정렬.""" raw = _trade_db_core(db) try: rows = raw.conn.execute( "SELECT * FROM updown_watchlist WHERE status = %s " "ORDER BY COALESCE(last_signal_at, added_at) ASC, id ASC", (ACTIVE_STATUS,), ).fetchall() return [dict(r) for r in (rows or [])] except Exception as e: logger.warning("list_active_watchlist 실패: %s", e) return [] def list_all_watchlist(db: Any) -> List[Dict[str, Any]]: """전체(active + inactive) — 웹 UI 표시용.""" raw = _trade_db_core(db) try: rows = raw.conn.execute( "SELECT * FROM updown_watchlist " "ORDER BY (status = %s) DESC, updated_at DESC, id DESC", (ACTIVE_STATUS,), ).fetchall() return [dict(r) for r in (rows or [])] except Exception as e: logger.warning("list_all_watchlist 실패: %s", e) return [] def _get_row(db: Any, code: str) -> Optional[Dict[str, Any]]: raw = _trade_db_core(db) try: r = raw.conn.execute( "SELECT * FROM updown_watchlist WHERE code = %s", (str(code).strip(),) ).fetchone() return dict(r) if r else None except Exception: return None def active_count(db: Any) -> int: raw = _trade_db_core(db) try: r = raw.conn.execute( "SELECT COUNT(*) c FROM updown_watchlist WHERE status = %s", (ACTIVE_STATUS,) ).fetchone() return int(r["c"]) if r else 0 except Exception: return 0 # ---------------------------------------------------------------------- # 추가/교체 (sticky 핵심) # ---------------------------------------------------------------------- def _activate_row(raw: Any, code: str, name: str, source: str, box_low: float, box_high: float, box_score: float) -> None: """기존 행을 active 로 되살리거나(reactivate) 박스 정보만 갱신.""" now = dt.now().strftime("%Y-%m-%d %H:%M:%S") raw.conn.execute( "UPDATE updown_watchlist " "SET status=%s, name=%s, source=%s, box_low=%s, box_high=%s, box_score=%s, " " last_seen_at=%s, removed_at=NULL " "WHERE code=%s", (ACTIVE_STATUS, name, source, box_low, box_high, box_score, now, code), ) def _insert_row(raw: Any, code: str, name: str, source: str, box_low: float, box_high: float, box_score: float) -> None: now = dt.now().strftime("%Y-%m-%d %H:%M:%S") raw.conn.execute( "INSERT INTO updown_watchlist " "(code, name, market_type, exchange, source, status, box_low, box_high, box_score, " " added_at, last_seen_at) " "VALUES (%s, %s, 'KR', 'KRX', %s, %s, %s, %s, %s, %s, %s)", (code, name, source, ACTIVE_STATUS, box_low, box_high, box_score, now, now), ) def _evict_oldest(raw: Any, held: Set[str]) -> bool: """한도 초과 시 시간순(가장 오래·무신호) active 1개를 밀어낸다(evicted). 보유종목 제외. Returns True 면 1개 비웠음(빈 슬롯 생김), False 면 밀어낼 게 없음(전부 보유 중 등). """ rows = raw.conn.execute( "SELECT code FROM updown_watchlist WHERE status = %s " "ORDER BY COALESCE(last_signal_at, added_at) ASC, id ASC", (ACTIVE_STATUS,), ).fetchall() for r in (rows or []): c = str(r["code"]).strip() if c in held: continue # 보유 종목은 청산 감시 위해 보호 → 밀어내지 않음 now = dt.now().strftime("%Y-%m-%d %H:%M:%S") raw.conn.execute( "UPDATE updown_watchlist SET status=%s, removed_at=%s WHERE code=%s", ("evicted", now, c), ) logger.info("🔄 updown_watchlist 시간순 교체 → evicted: %s", c) return True return False def upsert_from_scan( db: Any, candidates: List[Dict[str, Any]], *, source: str = "condition", ) -> Dict[str, int]: """박스권 후보(SCAN 결과)를 sticky 로 반영. candidates: [{"code","name","box_low","box_high","box_score"}, ...] - 이미 active → 박스정보/last_seen 갱신 (그대로 유지) - inactive 인데 재등장 → reactivate (재진입, 박스 이력 보존) - 신규 → 빈 슬롯 있으면 추가, 꽉 찼으면 시간순 교체 후 추가 (보유종목만 남아 못 비우면 스킵) ※ 이 함수는 절대 '교체(replace)' 가 아니라 '추가(append)' 다. 후보에서 빠진 종목을 여기서 삭제하지 않는다(sticky). 삭제는 졸업/탈락/만료/purge 만. """ raw = _trade_db_core(db) cap = watch_max(db) held = _held_codes(db) stats = {"added": 0, "reactivated": 0, "updated": 0, "skipped_full": 0} for cand in candidates or []: code = str(cand.get("code") or "").strip() if not code: continue name = str(cand.get("name") or code)[:50] box_low = float(cand.get("box_low") or 0) box_high = float(cand.get("box_high") or 0) box_score = float(cand.get("box_score") or 0) row = _get_row(db, code) if row and row.get("status") == ACTIVE_STATUS: # 이미 active → 박스 정보·last_seen 만 갱신 (sticky 유지) _activate_row(raw, code, name, source, box_low, box_high, box_score) stats["updated"] += 1 continue # 신규 or 재진입 → active 슬롯 확보 필요 cnt = active_count(db) if cnt >= cap: if not _evict_oldest(raw, held): stats["skipped_full"] += 1 continue # 보유종목만 남아 비울 수 없음 if row: _activate_row(raw, code, name, source, box_low, box_high, box_score) stats["reactivated"] += 1 logger.info("♻️ updown_watchlist 재진입: %s (%s)", code, name) else: _insert_row(raw, code, name, source, box_low, box_high, box_score) stats["added"] += 1 logger.info("➕ updown_watchlist 추가: %s (%s) [%s]", code, name, source) raw.conn.commit() return stats # ---------------------------------------------------------------------- # 수동 추가/삭제 (웹 UI) # ---------------------------------------------------------------------- def add_manual(db: Any, code: str, name: str = "", box_low: float = 0.0, box_high: float = 0.0) -> bool: """UI 수동 추가. 한도 초과면 시간순 교체 시도.""" raw = _trade_db_core(db) code = str(code).strip() if not code: return False row = _get_row(db, code) if not (row and row.get("status") == ACTIVE_STATUS): if active_count(db) >= watch_max(db): if not _evict_oldest(raw, _held_codes(db)): logger.warning("⚠️ watchlist 가득 + 전부 보유 중 → 수동추가 보류: %s", code) return False if row: _activate_row(raw, code, str(name or code)[:50], "manual", box_low, box_high, 0.0) else: _insert_row(raw, code, str(name or code)[:50], "manual", box_low, box_high, 0.0) raw.conn.commit() logger.info("➕ updown_watchlist 수동추가: %s (%s)", code, name) return True def remove_manual(db: Any, code: str) -> bool: """UI 수동 삭제 — 명시적 사용자 행동이므로 하드 삭제.""" raw = _trade_db_core(db) code = str(code).strip() if not code: return False raw.conn.execute("DELETE FROM updown_watchlist WHERE code = %s", (code,)) raw.conn.commit() logger.info("🗑 updown_watchlist 수동삭제: %s", code) return True # ---------------------------------------------------------------------- # 상태 전이 (졸업/탈락/신호) — 엔진/전략에서 호출 # ---------------------------------------------------------------------- def mark_status(db: Any, code: str, status: str) -> None: """graduated/failed/expired 등으로 soft-delete (보관). 재진입 대비.""" if status == ACTIVE_STATUS: status = ACTIVE_STATUS raw = _trade_db_core(db) now = dt.now().strftime("%Y-%m-%d %H:%M:%S") try: raw.conn.execute( "UPDATE updown_watchlist SET status=%s, removed_at=%s WHERE code=%s", (status, now, str(code).strip()), ) raw.conn.commit() except Exception as e: logger.debug("mark_status 실패(무시): %s", e) def touch_signal(db: Any, code: str) -> None: """매수신호(진입봉) 발생 시 호출 — 만료 시계 리셋.""" raw = _trade_db_core(db) now = dt.now().strftime("%Y-%m-%d %H:%M:%S") try: raw.conn.execute( "UPDATE updown_watchlist SET last_signal_at=%s WHERE code=%s", (now, str(code).strip()), ) raw.conn.commit() except Exception as e: logger.debug("touch_signal 실패(무시): %s", e) # ---------------------------------------------------------------------- # 만료/청소 (장 마감 배치 등에서 호출) # ---------------------------------------------------------------------- def expire_stale(db: Any) -> int: """active 중 ``UPDOWN_WATCH_EXPIRE_DAYS`` 거래일 이상 무신호 + 미보유 → expired. last_signal_at 없으면 added_at 기준. 보유 중 종목은 보호. Returns 만료 처리 건수. """ raw = _trade_db_core(db) days = watch_expire_days(db) held = _held_codes(db) try: rows = raw.conn.execute( "SELECT code FROM updown_watchlist WHERE status = %s " "AND COALESCE(last_signal_at, added_at) < (NOW() - INTERVAL %s DAY)", (ACTIVE_STATUS, days), ).fetchall() except Exception as e: logger.warning("expire_stale 조회 실패: %s", e) return 0 n = 0 now = dt.now().strftime("%Y-%m-%d %H:%M:%S") for r in (rows or []): c = str(r["code"]).strip() if c in held: continue # 보유 중이면 만료 보류 raw.conn.execute( "UPDATE updown_watchlist SET status='expired', removed_at=%s WHERE code=%s", (now, c), ) n += 1 if n: raw.conn.commit() logger.info("⌛ updown_watchlist 만료(expired) %d건 (>%d거래일 무신호)", n, days) return n def purge_old(db: Any) -> int: """inactive 중 ``UPDOWN_WATCH_RETENTION_DAYS`` 초과한 행 삭제(갈비지 방지).""" raw = _trade_db_core(db) days = watch_retention_days(db) ph = ", ".join(["%s"] * len(INACTIVE_STATUSES)) try: cur = raw.conn.execute( f"DELETE FROM updown_watchlist WHERE status IN ({ph}) " "AND removed_at IS NOT NULL AND removed_at < (NOW() - INTERVAL %s DAY)", (*INACTIVE_STATUSES, days), ) raw.conn.commit() n = int(getattr(cur, "rowcount", 0) or 0) if n: logger.info("🧹 updown_watchlist purge %d건 (보관 %d일 초과)", n, days) return n except Exception as e: logger.warning("purge_old 실패: %s", e) return 0