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 제거 븅신같은 초기설계 아예 제거 진입모드에 구멍메움 호가진입을 켜도 호가가 안들어올때 호가 안보고 그냥 사버림
170 lines
5.3 KiB
Python
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
|