Files
kis_bot/kis_trader/utils/ops_alert.py
Your Name 36a3e2b4a1 feat: Enhance trading system with new permanent subscription features and order book management
Changes:
- Added a new API endpoint for managing permanent subscriptions, allowing users to enable or disable subscriptions dynamically.
- Implemented a function to fill candle data from Kiwoom, ensuring that only relevant data is inserted into the database.
- Introduced a mechanism to handle master subscription states, improving the management of subscription statuses.
- Updated the database schema to include new fields for managing subscription states and order book filtering.

Impact:
- These enhancements improve the flexibility and reliability of the trading system, allowing for better management of subscriptions and order book data, while reducing the risk of data inconsistencies.

히스토리 align 제거 븅신같은 초기설계 아예 제거
진입모드에 구멍메움
호가진입을 켜도 호가가 안들어올때 호가 안보고 그냥 사버림
2026-08-15 23:01:14 +09:00

170 lines
5.3 KiB
Python

"""
운영 치명 알림 — 예외 삼킴/로그만 하던 인프라 장애를 MM(+선택 TG)로 올린다.
- 체결·기동 알림과 분리. 채널 기본 = KIS_SYSTEM_MM_CHANNEL.
- 코드별 쿨다운 + 전역 최소간격으로 스팸 방지.
- 장중 전용 코드는 session gate (주말·장외 오탐 완화).
"""
from __future__ import annotations
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Optional
from .env import get_env_bool, get_env_from_db, get_env_int
from .logger import atomic_load_json, atomic_save_json, get_logger, msg_mm, msg_tg
logger = get_logger("kis_trader.ops_alert")
_ROOT = Path(__file__).resolve().parents[2]
_STATE_PATH = _ROOT / "logs" / "ops_alert_state.json"
_LOCK = threading.Lock()
# 장중 세션에서만 의미 있는 코드 (토큰·PANIC·VI 는 상시)
_SESSION_CODES = frozenset({
"ws_kis_down",
"ws_kiwoom_down",
"ws_ls_down",
"ws_tick_silence",
"universe_zero",
"history_stale",
"kwcond_off",
"order_buy_reject",
"order_sell_reject",
"rate_limit",
})
def _channel() -> str:
ch = (get_env_from_db("OPS_ALERT_MM_CHANNEL", "") or "").strip()
if not ch:
ch = (get_env_from_db("KIS_SYSTEM_MM_CHANNEL", "default") or "default").strip()
return ch or "default"
def _cooldown_sec(code: str) -> int:
per = get_env_int(f"OPS_ALERT_COOLDOWN_{code.upper()}_SEC", 0)
if per > 0:
return max(30, int(per))
return max(30, get_env_int("OPS_ALERT_COOLDOWN_SEC", 300))
def _in_kr_session(now: Optional[datetime] = None) -> bool:
now = now or datetime.now()
if now.weekday() >= 5:
return False
hm = now.hour * 100 + now.minute
start = int(get_env_int("OPS_ALERT_SESSION_START_HM", 900) or 900)
end = int(get_env_int("OPS_ALERT_SESSION_END_HM", 1535) or 1535)
return start <= hm <= end
def _load_state() -> Dict[str, Any]:
st = atomic_load_json(_STATE_PATH, default={})
return st if isinstance(st, dict) else {}
def _save_state(st: Dict[str, Any]) -> None:
try:
_STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
atomic_save_json(_STATE_PATH, st)
except Exception as e:
logger.debug("ops_alert state save: %s", e)
def ops_alert(
code: str,
title: str,
*,
detail: str = "",
level: str = "critical",
force: bool = False,
session_only: Optional[bool] = None,
) -> bool:
"""운영 알림 1건. True=발송됨.
``level``: critical | warn
``session_only``: None 이면 코드 기본(장중 전용 집합), True/False 강제.
"""
if not get_env_bool("OPS_ALERT_ENABLED", True):
return False
code = str(code or "").strip().lower() or "unknown"
title = str(title or "").strip() or code
level = str(level or "critical").strip().lower()
if level not in ("critical", "warn"):
level = "critical"
need_session = (
bool(session_only) if session_only is not None
else (code in _SESSION_CODES)
)
if need_session and not _in_kr_session():
return False
now = time.time()
with _LOCK:
st = _load_state()
by_code = st.setdefault("by_code", {})
if not isinstance(by_code, dict):
by_code = {}
st["by_code"] = by_code
last = float(by_code.get(code) or 0.0)
cd = _cooldown_sec(code)
if not force and (now - last) < cd:
return False
global_gap = float(get_env_int("OPS_ALERT_GLOBAL_MIN_GAP_SEC", 20) or 20)
last_any = float(st.get("last_any_ts") or 0.0)
if not force and (now - last_any) < global_gap:
return False
icon = "🚨" if level == "critical" else "⚠️"
lines = [
f"{icon} **[운영알림/{level.upper()}] {title}**",
f"- 코드: `{code}`",
f"- 시각: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
]
if detail:
clipped = str(detail).strip()
if len(clipped) > 1200:
clipped = clipped[:1200] + ""
lines.append(f"- 상세: {clipped}")
body = "\n".join(lines)
ok = False
try:
ok = bool(msg_mm(body, channel_alias=_channel(), jitter=False))
except Exception as e:
logger.debug("ops_alert MM 실패: %s", e)
if level == "critical" and get_env_bool("OPS_ALERT_TG_ON_CRITICAL", True):
try:
msg_tg(body, jitter=False)
except Exception as e:
logger.debug("ops_alert TG 실패: %s", e)
by_code[code] = now
st["last_any_ts"] = now
st["last_code"] = code
_save_state(st)
logger.warning("[ops_alert] sent code=%s ok_mm=%s title=%s", code, ok, title)
return ok
def note_counter(key: str, *, reset: bool = False) -> int:
"""연속 실패 카운터 (주문거부·유량 등). reset=True 이면 0."""
with _LOCK:
st = _load_state()
counters = st.setdefault("counters", {})
if not isinstance(counters, dict):
counters = {}
st["counters"] = counters
if reset:
counters[key] = 0
_save_state(st)
return 0
n = int(counters.get(key) or 0) + 1
counters[key] = n
_save_state(st)
return n