#!/usr/bin/env python3 """ KIS Long Alert - 늘림목 알림 전용 (매매 없음, 하루 2회 Mattermost 브리핑) - kis_long_ver1.py에서 알림(리포트) 로직만 분리 - 13:00 오전 브리핑, 15:35 장마감 최종 브리핑 (환경변수로 시각 변경 가능) """ import json import logging import time from pathlib import Path from datetime import datetime as dt from database import TradeDB # kis_long_ver1과 동일한 설정·클래스 사용 (알림만 필요) from kis_long_ver1 import ( db, get_env_from_db, get_env_float, get_env_bool, SCRIPT_DIR, MM_SERVER_URL, MM_BOT_TOKEN, MM_CONFIG_FILE, MM_CHANNEL_LONG, MattermostBot, KISClientWithOrder, ) logging.basicConfig( format="[%(asctime)s] %(message)s", datefmt="%H:%M:%S", level=logging.INFO, ) logger = logging.getLogger("KISLongAlert") # 알림 발송 시각 (하루 2회). DB/환경: ALERT_TIME_1="13:00", ALERT_TIME_2="15:35" def _parse_time(s: str): """'13:00' -> (13, 0)""" try: h, m = s.strip().split(":") return int(h), int(m) except Exception: return None def _get_alert_times(): t1 = get_env_from_db("ALERT_TIME_1", "13:00").strip() t2 = get_env_from_db("ALERT_TIME_2", "15:35").strip() return _parse_time(t1) or (13, 0), _parse_time(t2) or (15, 35) # 당일 시작 자산 저장 (첫 알림 시점에 설정, 같은 날 재사용) STATE_FILE = SCRIPT_DIR / "kis_long_alert_state.json" def _load_state(): try: if STATE_FILE.exists(): with open(STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: logger.debug(f"상태 파일 로드 실패: {e}") return {} def _save_state(state): try: with open(STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) except Exception as e: logger.warning(f"상태 파일 저장 실패: {e}") class LongAlertBot: """늘림목 알림 전용 - KIS 자산 조회 + Mattermost 2회 전송 (ver1과 동일한 보유 기준: DB LONG + watchlist 병합)""" def __init__(self): self.db = db self.client = KISClientWithOrder() self.mm = MattermostBot() self.mm_channel = MM_CHANNEL_LONG self.current_cash = 0 self.current_total_asset = 0 self.start_day_asset = 0 self.total_deposit = get_env_float("TOTAL_DEPOSIT", 0) self.today_date = dt.now().strftime("%Y-%m-%d") self.holdings = {} # ver1과 동일: DB에서 늘림목(LONG) 보유만 로드 for code, trade in self.db.get_active_trades(strategy_prefix="LONG").items(): qty = trade.get("current_qty", 0) or 0 if qty > 0: self.holdings[code] = {"total_qty": qty, "name": trade.get("name", code)} # watchlist 로드 (새로 발굴·반영한 종목도 알림에 포함되도록) self.watchlist_path = SCRIPT_DIR / "long_term_watchlist.json" self.watchlist = self._load_watchlist() self._update_assets() self._maybe_set_start_day_asset() def _load_watchlist(self): if not self.watchlist_path.exists(): return [] try: with open(self.watchlist_path, "r", encoding="utf-8") as f: return json.load(f).get("items", []) except Exception as e: logger.debug(f"watchlist 로드 실패: {e}") return [] def _merge_watchlist_holdings_from_balance(self, balance): """ver1과 동일: watchlist LONG/dca인 계좌 보유를 holdings에 합침 (inquire-balance → output1=종목 리스트)""" positions_list = balance.get("output1") or [] if isinstance(positions_list, dict): positions_list = [positions_list] if not isinstance(positions_list, list): positions_list = [] watchlist_dca_codes = { (item.get("code") or "").strip() for item in self.watchlist if (item.get("strategy") == "LONG" or item.get("use") == "dca") } watchlist_dca_codes.discard("") for row in positions_list: code = (row.get("pdno") or "").strip() if not code or code in self.holdings or code not in watchlist_dca_codes: continue try: qty = int(float(row.get("hldg_qty") or row.get("ord_psbl_qty") or 0)) if qty <= 0: continue name = (row.get("prdt_name") or row.get("prdt_name_eng") or code).strip() self.holdings[code] = {"total_qty": qty, "name": name or code} except Exception: pass def _update_assets(self): """자산 정보 업데이트 (KIS API)""" try: balance = self.client.get_account_evaluation() if not balance: return # ver1과 동일: watchlist LONG 반영 (단타와 동일 inquire-balance → output2=예수금, output1=종목 리스트) self._merge_watchlist_holdings_from_balance(balance) out2 = balance.get("output2") if isinstance(out2, list) and out2: out2 = out2[0] out2 = out2 if isinstance(out2, dict) else {} ord_psbl = out2.get("ord_psbl_cash") or out2.get("dnca_tot_amt") self.current_cash = float(str(ord_psbl or 0).replace(",", "").strip()) if ord_psbl is not None else 0 output1_list = balance.get("output1") or [] if isinstance(output1_list, dict): output1_list = [output1_list] holdings_value = 0 for code in self.holdings: evlu = None for item in output1_list: if (item.get("pdno") or "").strip() == code: evlu = float(item.get("evlu_amt", 0) or 0) break if evlu is not None: holdings_value += evlu else: price_data = self.client.inquire_price(code) if price_data: holdings_value += abs(float(price_data.get("stck_prpr", 0) or 0)) * self.holdings[code]["total_qty"] self.current_total_asset = self.current_cash + holdings_value except Exception as e: logger.error(f"자산 정보 업데이트 실패: {e}") def _maybe_set_start_day_asset(self): """오늘 날짜 기준으로 start_day_asset 설정 (당일 첫 실행 시)""" state = _load_state() saved_date = state.get("date", "") if saved_date == self.today_date and state.get("start_day_asset"): self.start_day_asset = float(state["start_day_asset"]) return self.start_day_asset = self.current_total_asset _save_state({"date": self.today_date, "start_day_asset": self.current_total_asset}) def send_mm(self, msg): try: self.mm.send(self.mm_channel, msg) except Exception as e: logger.error(f"❌ MM 전송 에러: {e}") def send_report_1(self): """1회차: 오전 브리핑 (13:00)""" self._update_assets() self._maybe_set_start_day_asset() day_pnl = self.current_total_asset - self.start_day_asset day_pnl_pct = (day_pnl / self.start_day_asset * 100) if self.start_day_asset > 0 else 0 msg = f"""📊 **[늘림목 오전 브리핑 - {dt.now().strftime('%H:%M')}]** - 당일 시작: {self.start_day_asset:,.0f}원 - 현재 자산: {self.current_total_asset:,.0f}원 - 당일 손익: {day_pnl:+,.0f}원 ({day_pnl_pct:+.2f}%) - 보유 종목: {len(self.holdings)}개""" self.send_mm(msg) logger.info("📊 오전 브리핑 전송 완료") def send_report_2(self): """2회차: 장마감 최종 브리핑 (15:35)""" self._update_assets() self._maybe_set_start_day_asset() day_pnl = self.current_total_asset - self.start_day_asset day_pnl_pct = (day_pnl / self.start_day_asset * 100) if self.start_day_asset > 0 else 0 cumulative_pnl = self.current_total_asset - self.total_deposit cumulative_pnl_pct = (cumulative_pnl / self.total_deposit * 100) if self.total_deposit > 0 else 0 today_ymd = dt.now().strftime("%Y%m%d") today_trades = self.db.get_trades_by_date(today_ymd) msg = f"""🏁 **[늘림목 장마감 브리핑 - {dt.now().strftime('%H:%M')}]** ━━━━━━━━━━━━━━━━━━━━ 📅 **당일 손익** - 시작: {self.start_day_asset:,.0f}원 - 종료: {self.current_total_asset:,.0f}원 - 손익: {day_pnl:+,.0f}원 ({day_pnl_pct:+.2f}%) 💰 **누적 손익 (총 입금액 대비)** - 총 입금: {self.total_deposit:,.0f}원 - 현재 자산: {self.current_total_asset:,.0f}원 - 누적 손익: {cumulative_pnl:+,.0f}원 ({cumulative_pnl_pct:+.2f}%) 📊 **거래 현황** - 오늘 매매: {len(today_trades)}건 - 보유 종목: {len(self.holdings)}개 - 예수금: {self.current_cash:,.0f}원 ━━━━━━━━━━━━━━━━━━━━""" self.send_mm(msg) logger.info("🏁 장마감 브리핑 전송 완료") def main(): (h1, m1), (h2, m2) = _get_alert_times() logger.info(f"늘림목 알림 봇 시작 (하루 2회: {h1:02d}:{m1:02d}, {h2:02d}:{m2:02d})") sent_1 = False sent_2 = False last_date = "" while True: try: now = dt.now() today = now.strftime("%Y-%m-%d") if today != last_date: last_date = today sent_1 = False sent_2 = False if now.hour == h1 and now.minute == m1 and not sent_1: bot = LongAlertBot() bot.send_report_1() sent_1 = True elif now.hour == h2 and now.minute == m2 and not sent_2: bot = LongAlertBot() bot.send_report_2() sent_2 = True except KeyboardInterrupt: logger.info("종료") break except Exception as e: logger.error(f"알림 루프 에러: {e}") time.sleep(60) if __name__ == "__main__": main()