"""AI 전략 서비스 (ai_strategy_service.py) 장 전/장중/장 후 AI 분석을 관리하는 통합 서비스. MCP 서버의 BATCH_STOCK_ANALYSIS, NEWS_SENTIMENT, DAILY_REVIEW를 호출. [호출 타이밍] 08:30 pre_market_analysis() -- Sonnet 배치 분석 -> Redis 캐시 08:50 select_daytrading_candidates() -- Haiku 데이트레이딩 종목 선정 10분 주기 monitor_news() -- Haiku 뉴스 감시 -> 악재 시 Sonnet 긴급 분석 15:40 post_market_review() -- Sonnet 당일 복기 -> Telegram 발송 아키텍처 준수: - DB 쓰기 금지 (read-only SELECT만 수행, CLAUDE.md 3.1 예외 조항 동일) - 결과는 Redis 캐시 또는 반환값으로만 전달 - MCP 호출은 mcp.py의 httpx 싱글톤 클라이언트 패턴 재사용 - JSON 직렬화: _dumps() 사용 (CLAUDE.md 4.2) - import는 함수 내부에서 수행 (순환 import 방지) """ import hashlib import logging from datetime import date, datetime, timedelta from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) # ── Redis 캐시 키 & TTL 상수 ──────────────────────────────────────────────── CACHE_KEY_PRE_MARKET = "ai:pre_market:latest" CACHE_KEY_DT_CANDIDATES = "ai:daytrading_candidates" CACHE_KEY_NEWS_HASH_PREFIX = "ai:news_hash:" TTL_PRE_MARKET = 8 * 3600 # 8시간 TTL_DT_CANDIDATES = 6 * 3600 # 6시간 TTL_NEWS_HASH = 24 * 3600 # 24시간 # ── 배치 사이즈 상수 ──────────────────────────────────────────────────────── _PRE_MARKET_TOP_N = 50 # 장 전 분석 대상 종목 수 _BATCH_SIZE = 10 # MCP 배치 요청 단위 _NEWS_PER_STOCK = 3 # 종목당 크롤링 뉴스 건수 _DT_CANDIDATE_TOP_N = 20 # 데이트레이딩 후보 종목 수 _DT_CANDIDATE_BATCH = 10 # 데이트레이딩 MCP 배치 단위 # ============================================================================= # A. 장 전 배치 분석 # ============================================================================= async def pre_market_analysis() -> dict: """장 전 AI 배치 분석 (08:30 호출). 1. DB에서 stock_analysis 최근 분석 상위 50종목 조회 (score 내림차순) 2. 각 종목의 뉴스 제목 3건 조회 3. 10종목씩 MCP BATCH_STOCK_ANALYSIS 요청 (5회) 4. Redis "ai:pre_market:latest" 캐시 (TTL 8시간) 5. 적합 종목 리스트 반환 Returns: {"total_analyzed": int, "suitable_stocks": [...], "timestamp": str} """ from app.core.config import settings from app.core.database import AsyncSessionLocal from app.core.redis import setex_json from app.core.json_helpers import _dumps result: Dict[str, Any] = { "total_analyzed": 0, "suitable_stocks": [], "errors": [], "timestamp": datetime.now().isoformat(), } # ── 1) DB에서 최근 분석 상위 50종목 조회 ── top_stocks: List[Dict[str, Any]] = [] try: from sqlalchemy import select, desc from app.models.models import StockAnalysis async with AsyncSessionLocal() as db: stmt = ( select(StockAnalysis) .where(StockAnalysis.score.isnot(None)) .order_by(desc(StockAnalysis.score), desc(StockAnalysis.analysis_date)) .limit(_PRE_MARKET_TOP_N) ) rows = await db.execute(stmt) analyses = rows.scalars().all() for a in analyses: top_stocks.append({ "stock_code": a.stock_code, "stock_name": a.stock_name, "score": a.score, "rsi": float(a.rsi) if a.rsi else None, "current_price": a.current_price, "analysis_date": str(a.analysis_date), "signals": a.signals, }) logger.info("[AI-Pre] DB에서 %d개 종목 조회 완료", len(top_stocks)) except Exception as e: logger.error("[AI-Pre] DB 조회 실패: %s", e) result["errors"].append(f"db_query_failed: {e}") return result if not top_stocks: logger.warning("[AI-Pre] 분석 대상 종목 없음") return result # ── 2) 종목별 뉴스 제목 3건 조회 ── for stock in top_stocks: stock["recent_news"] = await _fetch_stock_news_titles( stock["stock_code"], limit=_NEWS_PER_STOCK ) # ── 3) 10종목씩 묶어서 MCP BATCH_STOCK_ANALYSIS 요청 ── all_results: List[Dict[str, Any]] = [] batches = [ top_stocks[i : i + _BATCH_SIZE] for i in range(0, len(top_stocks), _BATCH_SIZE) ] for batch_idx, batch in enumerate(batches): batch_result = await _call_mcp_batch_analysis( batch, context="pre_market" ) if batch_result: all_results.extend(batch_result) else: result["errors"].append(f"batch_{batch_idx}_failed") # ── 4) 적합 종목 필터링 ── suitable: List[Dict[str, Any]] = [] for item in all_results: ai_score = item.get("ai_score", 0) recommendation = str(item.get("recommendation", "")).upper() if ai_score >= settings.MCP_ANALYSIS_MIN_SCORE or recommendation in ( "BUY", "STRONG_BUY", ): suitable.append(item) result["total_analyzed"] = len(all_results) result["suitable_stocks"] = suitable logger.info( "[AI-Pre] 분석 완료: %d개 분석, %d개 적합", len(all_results), len(suitable), ) # ── 5) Redis 캐시 ── try: await setex_json(CACHE_KEY_PRE_MARKET, TTL_PRE_MARKET, result) logger.info("[AI-Pre] Redis 캐시 저장 완료 (TTL=%ds)", TTL_PRE_MARKET) except Exception as e: logger.warning("[AI-Pre] Redis 캐시 저장 실패: %s", e) return result # ============================================================================= # B. 데이트레이딩 종목 선정 # ============================================================================= async def select_daytrading_candidates() -> list[dict]: """데이트레이딩 후보 종목 선정 (08:50 호출). 1. 네이버 시가총액 상위 종목 조회 2. 가격 필터 (5,000원 이상) 적용 3. 상위 20종목을 MCP BATCH_STOCK_ANALYSIS로 전송 (2회 배치) 4. Redis "ai:daytrading_candidates" 캐시 (TTL 6시간) 5. daytrading watchlist에 자동 등록 Returns: [{"stock_code", "stock_name", "ai_score", ...}, ...] """ from app.core.config import settings from app.core.redis import setex_json from app.services.naver_finance import get_naver_stock_list candidates: List[Dict[str, Any]] = [] # ── 1) 네이버 시가총액 상위 종목 조회 ── try: stocks = await get_naver_stock_list( max_pages_per_market=5, min_price=settings.MIN_PRICE_KRW, limit=100, ) logger.info("[AI-DT] 네이버 종목 %d개 수집", len(stocks)) except Exception as e: logger.error("[AI-DT] 네이버 종목 조회 실패: %s", e) return candidates if not stocks: logger.warning("[AI-DT] 조회된 종목 없음") return candidates # ── 2) 상위 N개 선정 ── stock_list = [ {"stock_code": code, "stock_name": name} for code, name in stocks[:_DT_CANDIDATE_TOP_N] ] # ── 3) 종목별 뉴스 수집 + MCP 배치 분석 (데이트레이딩 관점) ── for stock in stock_list: stock["recent_news"] = await _fetch_stock_news_titles( stock["stock_code"], limit=_NEWS_PER_STOCK ) all_results: List[Dict[str, Any]] = [] batches = [ stock_list[i : i + _DT_CANDIDATE_BATCH] for i in range(0, len(stock_list), _DT_CANDIDATE_BATCH) ] for batch_idx, batch in enumerate(batches): batch_result = await _call_mcp_batch_analysis( batch, context="daytrading" ) if batch_result: all_results.extend(batch_result) # ── 4) 적합 종목 필터링 (데이트레이딩 기준) ── for item in all_results: ai_score = item.get("ai_score", 0) dt_suitable = item.get("daytrading_suitable", False) recommendation = str(item.get("recommendation", "")).upper() if dt_suitable or ai_score >= settings.MCP_ANALYSIS_MIN_SCORE or recommendation in ( "BUY", "STRONG_BUY", ): candidates.append(item) logger.info( "[AI-DT] 분석 완료: %d개 분석, %d개 후보", len(all_results), len(candidates), ) # ── 5) Redis 캐시 ── try: await setex_json(CACHE_KEY_DT_CANDIDATES, TTL_DT_CANDIDATES, candidates) except Exception as e: logger.warning("[AI-DT] Redis 캐시 저장 실패: %s", e) # ── 6) daytrading watchlist에 자동 등록 ── if candidates: try: from app.services.daytrading.entry_service import set_watchlist watchlist_items = [ {"code": c["stock_code"], "name": c.get("stock_name", c["stock_code"])} for c in candidates ] set_watchlist(watchlist_items) logger.info("[AI-DT] daytrading watchlist %d종목 등록", len(watchlist_items)) except Exception as e: logger.warning("[AI-DT] watchlist 등록 실패: %s", e) return candidates # ============================================================================= # C. 장중 뉴스 모니터링 # ============================================================================= async def monitor_news() -> list[dict]: """장중 뉴스 모니터링 (10분 주기 호출). 1. 보유 종목 + 데이트레이딩 포지션 조회 2. 종목별 최신 뉴스 3건 크롤링 3. 이전 뉴스 해시 비교 -> 새 뉴스만 MCP NEWS_SENTIMENT 요청 4. NEGATIVE + HIGH severity -> Telegram 알림 5. 뉴스 해시 Redis 업데이트 Returns: [{"stock_code", "stock_name", "sentiment", "severity", "titles": [...], "action": str}, ...] -- sentiment="NEGATIVE" & severity="HIGH"인 종목만 포함 """ from app.core.database import AsyncSessionLocal from app.core.redis import get_json, setex_json alerts: List[Dict[str, Any]] = [] # ── 1) 보유 종목 + 데이트레이딩 포지션 조회 ── monitored_stocks: Dict[str, str] = {} # code -> name try: from sqlalchemy import select from app.models.models import Portfolio async with AsyncSessionLocal() as db: stmt = select(Portfolio).where(Portfolio.quantity > 0) rows = await db.execute(stmt) for p in rows.scalars().all(): monitored_stocks[p.stock_code] = p.stock_name or p.stock_code except Exception as e: logger.error("[AI-News] 포트폴리오 조회 실패: %s", e) # 데이트레이딩 포지션 추가 try: from app.services.daytrading.position_manager import day_trading_state for code, pos in day_trading_state.positions.items(): if code not in monitored_stocks: monitored_stocks[code] = pos.get("stock_name", code) except Exception as e: logger.warning("[AI-News] 데이트레이딩 포지션 조회 실패: %s", e) if not monitored_stocks: logger.debug("[AI-News] 모니터링 대상 종목 없음") return alerts logger.info("[AI-News] %d개 종목 뉴스 모니터링 시작", len(monitored_stocks)) # ── 2) 종목별 뉴스 크롤링 + 해시 비교 ── for stock_code, stock_name in monitored_stocks.items(): try: news_titles = await _fetch_stock_news_titles( stock_code, limit=_NEWS_PER_STOCK ) if not news_titles: continue # 현재 뉴스 해시 계산 current_hash = _compute_news_hash(news_titles) # 이전 해시와 비교 cache_key = f"{CACHE_KEY_NEWS_HASH_PREFIX}{stock_code}" prev_hash = await get_json(cache_key) if prev_hash == current_hash: # 뉴스 변경 없음 -- skip continue # ── 3) 새 뉴스 감지 -> MCP NEWS_SENTIMENT 요청 ── sentiment_result = await _call_mcp_news_sentiment( stock_code, stock_name, news_titles ) # ── 4) 뉴스 해시 업데이트 ── try: await setex_json(cache_key, TTL_NEWS_HASH, current_hash) except Exception as e: logger.warning("[AI-News] 해시 캐시 저장 실패 %s: %s", stock_code, e) if not sentiment_result: continue sentiment = str(sentiment_result.get("sentiment", "")).upper() severity = str(sentiment_result.get("severity", "")).upper() # ── 5) 악재 알림 ── if sentiment == "NEGATIVE" and severity == "HIGH": alert_item = { "stock_code": stock_code, "stock_name": stock_name, "sentiment": sentiment, "severity": severity, "titles": news_titles, "analysis": sentiment_result.get("analysis", ""), "action": sentiment_result.get("recommended_action", "review"), } alerts.append(alert_item) # Telegram 알림 await _send_news_alert_telegram(alert_item) except Exception as e: logger.warning("[AI-News] 종목 %s 뉴스 처리 오류: %s", stock_code, e) if alerts: logger.warning("[AI-News] %d개 악재 알림 발생", len(alerts)) else: logger.info("[AI-News] 뉴스 모니터링 완료 (악재 없음)") return alerts # ============================================================================= # D. 장 후 리뷰 # ============================================================================= async def post_market_review() -> dict: """장 후 AI 리뷰 (15:40 호출). 1. DB에서 당일 TradeHistory (SELL) 조회 2. 승/패, PnL, 전략별 통계 집계 3. 시장 요약 (코스피 변동률, 시장 국면) 4. MCP DAILY_REVIEW 요청 5. Telegram 리뷰 결과 발송 6. 결과 반환 Returns: {"date", "summary", "trades", "pnl", "ai_review", ...} """ from app.core.config import settings from app.core.database import AsyncSessionLocal today = date.today() result: Dict[str, Any] = { "date": str(today), "trades": {"total": 0, "wins": 0, "losses": 0}, "pnl": {"total": 0.0, "by_strategy": {}}, "market_summary": {}, "ai_review": None, "timestamp": datetime.now().isoformat(), } # ── 1) 당일 매도 기록 조회 ── sell_trades: List[Dict[str, Any]] = [] try: from sqlalchemy import select, and_, func, cast, Date from app.models.models import TradeHistory async with AsyncSessionLocal() as db: stmt = ( select(TradeHistory) .where( and_( TradeHistory.order_type == "SELL", cast(TradeHistory.created_at, Date) == today, ) ) .order_by(TradeHistory.created_at) ) rows = await db.execute(stmt) trades = rows.scalars().all() for t in trades: sell_trades.append({ "stock_code": t.stock_code, "stock_name": t.stock_name or t.stock_code, "quantity": t.executed_quantity or t.order_quantity, "price": float(t.executed_price) if t.executed_price else 0, "profit_loss": float(t.net_profit_loss) if t.net_profit_loss else 0, "profit_loss_rate": float(t.profit_loss_rate) if t.profit_loss_rate else 0, "strategy_type": t.strategy_type or "swing", "exit_reason": t.exit_reason or "", }) logger.info("[AI-Review] 당일 매도 %d건 조회", len(sell_trades)) except Exception as e: logger.error("[AI-Review] 매도 이력 조회 실패: %s", e) # ── 2) 통계 집계 ── total_pnl = 0.0 wins = 0 losses = 0 by_strategy: Dict[str, Dict[str, Any]] = {} for t in sell_trades: pnl = t["profit_loss"] total_pnl += pnl if pnl >= 0: wins += 1 else: losses += 1 st = t["strategy_type"] if st not in by_strategy: by_strategy[st] = {"count": 0, "pnl": 0.0, "wins": 0, "losses": 0} by_strategy[st]["count"] += 1 by_strategy[st]["pnl"] += pnl if pnl >= 0: by_strategy[st]["wins"] += 1 else: by_strategy[st]["losses"] += 1 result["trades"] = { "total": len(sell_trades), "wins": wins, "losses": losses, "win_rate": round(wins / len(sell_trades) * 100, 1) if sell_trades else 0.0, } result["pnl"] = { "total": round(total_pnl, 2), "by_strategy": { k: {**v, "pnl": round(v["pnl"], 2)} for k, v in by_strategy.items() }, } # ── 3) 시장 요약 ── try: from app.services.market_analyzer import check_market_regime regime_info = await check_market_regime() result["market_summary"] = { "regime": regime_info.get("regime", "UNKNOWN"), "kospi_change_pct": regime_info.get("kospi_change_pct"), "description": regime_info.get("description", ""), } except Exception as e: logger.warning("[AI-Review] 시장 국면 조회 실패: %s", e) result["market_summary"] = {"regime": "UNKNOWN"} # ── 4) MCP DAILY_REVIEW 요청 ── try: review_data = { "date": str(today), "trades": sell_trades, "statistics": result["trades"], "pnl": result["pnl"], "market_summary": result["market_summary"], } ai_review = await _call_mcp_daily_review(review_data) result["ai_review"] = ai_review except Exception as e: logger.error("[AI-Review] MCP DAILY_REVIEW 호출 실패: %s", e) # ── 5) Telegram 발송 ── await _send_review_telegram(result) logger.info( "[AI-Review] 리뷰 완료: %d건 매매, PnL %.0f원, 승률 %.1f%%", result["trades"]["total"], total_pnl, result["trades"]["win_rate"], ) return result # ============================================================================= # 내부 헬퍼 함수들 # ============================================================================= async def _fetch_stock_news_titles(stock_code: str, limit: int = 3) -> list[str]: """종목별 뉴스 제목을 네이버에서 크롤링하여 반환. news_sentiment.collect_stock_news()를 재활용하되 제목만 추출. Args: stock_code: 종목 코드 (예: "005930") limit: 최대 뉴스 건수 Returns: ["뉴스 제목 1", "뉴스 제목 2", ...] """ try: from app.services.news_sentiment import collect_stock_news articles = await collect_stock_news(stock_code, limit=limit) return [a.get("title", "") for a in articles if a.get("title")] except Exception as e: logger.debug("[AI] 뉴스 크롤링 실패 %s: %s", stock_code, e) return [] def _compute_news_hash(titles: list[str]) -> str: """뉴스 제목 리스트의 해시를 계산하여 변경 감지에 사용. Args: titles: 뉴스 제목 리스트 Returns: SHA-256 해시 문자열 (hex, 16자) """ combined = "|".join(sorted(titles)) return hashlib.sha256(combined.encode("utf-8")).hexdigest()[:16] async def _get_mcp_client(): """MCP httpx 싱글톤 클라이언트를 반환. 초기화 전이면 None. mcp.py의 get_http_client() 패턴과 동일하지만, 이 서비스에서는 RuntimeError 대신 None을 반환하여 호출자가 graceful하게 처리할 수 있도록 함. """ try: from app.api.endpoints.mcp import get_http_client return get_http_client() except RuntimeError: logger.warning("[AI] MCP httpx client 미초기화") return None except Exception as e: logger.warning("[AI] MCP client 획득 실패: %s", e) return None async def _call_mcp_batch_analysis( stocks: list[dict], context: str = "pre_market", ) -> Optional[list[dict]]: """MCP 서버에 BATCH_STOCK_ANALYSIS 요청. Args: stocks: 종목 리스트 [{"stock_code", "stock_name", "score", "recent_news", ...}] context: "pre_market" 또는 "daytrading" (프롬프트 관점 결정) Returns: [{"stock_code", "stock_name", "ai_score", "recommendation", ...}] 또는 None """ from app.core.config import settings client = await _get_mcp_client() if client is None: return None # MCP 요청 페이로드 구성 batch_payload = [] for s in stocks: item = { "stock_code": s.get("stock_code", ""), "stock_name": s.get("stock_name", ""), "recent_news": s.get("recent_news", []), } # pre_market 에서는 기존 분석 점수/지표 포함 if context == "pre_market": item["current_score"] = s.get("score") item["rsi"] = s.get("rsi") item["current_price"] = s.get("current_price") item["signals"] = s.get("signals") batch_payload.append(item) request_data = { "analysis_type": "BATCH_STOCK_ANALYSIS", "data": { "stocks": batch_payload, "context": context, }, } # 데이트레이딩 관점 추가 if context == "daytrading": request_data["data"]["evaluation_focus"] = "daytrading_suitability" request_data["data"]["criteria"] = ( "volatility, volume, intraday_range, liquidity, " "ORB/VWAP/RSI_BB signal potential" ) try: resp = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json=request_data, timeout=60.0, ) if resp.status_code == 200: data = resp.json() # MCP 응답 구조: {"result": {...}} 또는 {"result": [{...}, ...]} payload = data.get("result", data) if isinstance(payload, list): return payload elif isinstance(payload, dict): # 단일 결과가 dict로 올 경우 stocks 키 확인 return payload.get("stocks", payload.get("analyses", [payload])) return None else: logger.warning( "[AI] MCP BATCH_STOCK_ANALYSIS HTTP %d: %s", resp.status_code, resp.text[:200], ) return None except Exception as e: logger.error("[AI] MCP BATCH_STOCK_ANALYSIS 호출 오류: %s", e) return None async def _call_mcp_news_sentiment( stock_code: str, stock_name: str, news_titles: list[str], ) -> Optional[dict]: """MCP 서버에 NEWS_SENTIMENT 요청. Args: stock_code: 종목 코드 stock_name: 종목명 news_titles: 뉴스 제목 리스트 Returns: {"sentiment": "POSITIVE"|"NEGATIVE"|"NEUTRAL", "severity": "HIGH"|"MEDIUM"|"LOW", "analysis": str, "recommended_action": str} 또는 None """ from app.core.config import settings client = await _get_mcp_client() if client is None: return None request_data = { "analysis_type": "NEWS_SENTIMENT", "data": { "stock_code": stock_code, "stock_name": stock_name, "news_titles": news_titles, }, } try: resp = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json=request_data, timeout=30.0, ) if resp.status_code == 200: data = resp.json() return data.get("result", data) else: logger.warning( "[AI] MCP NEWS_SENTIMENT HTTP %d (%s): %s", resp.status_code, stock_code, resp.text[:200], ) return None except Exception as e: logger.error("[AI] MCP NEWS_SENTIMENT 호출 오류 (%s): %s", stock_code, e) return None async def _call_mcp_daily_review(review_data: dict) -> Optional[dict]: """MCP 서버에 DAILY_REVIEW 요청. Args: review_data: 당일 매매 요약 데이터 Returns: {"review_text": str, "lessons": [...], "suggestions": [...], "grade": str} 또는 None """ from app.core.config import settings client = await _get_mcp_client() if client is None: return None request_data = { "analysis_type": "DAILY_REVIEW", "data": review_data, } try: resp = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json=request_data, timeout=60.0, ) if resp.status_code == 200: data = resp.json() return data.get("result", data) else: logger.warning( "[AI] MCP DAILY_REVIEW HTTP %d: %s", resp.status_code, resp.text[:200], ) return None except Exception as e: logger.error("[AI] MCP DAILY_REVIEW 호출 오류: %s", e) return None async def _send_news_alert_telegram(alert: dict) -> None: """악재 뉴스 알림을 Telegram으로 발송. Args: alert: monitor_news()가 생성한 알림 항목 """ try: from app.services.telegram_notifier import telegram_notifier, _escape_markdown_v2 esc = _escape_markdown_v2 stock_code = alert.get("stock_code", "") stock_name = alert.get("stock_name", stock_code) titles = alert.get("titles", []) analysis = alert.get("analysis", "") lines = [ f"\U0001F6A8 *{esc('악재 뉴스 감지')}*", f"", f"\U0001F4C8 {esc(stock_name)} \\({esc(stock_code)}\\)", f"\U0001F4F0 {esc('뉴스')}:", ] for i, title in enumerate(titles[:3], 1): lines.append(f" {esc(str(i))}\\. {esc(title)}") if analysis: lines.append(f"") lines.append(f"\U0001F4DD {esc('분석')}: {esc(str(analysis)[:200])}") action = alert.get("action", "review") lines.append(f"") lines.append(f"\U000026A0 {esc('권장 조치')}: {esc(action)}") text = "\n".join(lines) await telegram_notifier.send_message(text) except Exception as e: logger.warning("[AI-News] Telegram 알림 발송 실패: %s", e) async def _send_review_telegram(review: dict) -> None: """장 후 리뷰를 Telegram으로 발송. Args: review: post_market_review()가 생성한 결과 """ try: from app.services.telegram_notifier import telegram_notifier, _escape_markdown_v2 esc = _escape_markdown_v2 trades_info = review.get("trades", {}) pnl_info = review.get("pnl", {}) total_pnl = pnl_info.get("total", 0) total_count = trades_info.get("total", 0) wins = trades_info.get("wins", 0) losses = trades_info.get("losses", 0) win_rate = trades_info.get("win_rate", 0.0) pnl_sign = "+" if total_pnl >= 0 else "" pnl_emoji = "\U0001F4C8" if total_pnl >= 0 else "\U0001F4C9" lines = [ f"\U0001F4CA *{esc('AI 일일 리뷰')}*", f"\U0001F4C5 {esc(review.get('date', ''))}", f"", f"{pnl_emoji} {esc('당일 PnL')}: {esc(f'{pnl_sign}{total_pnl:,.0f}')}원", f"\U0001F3AF {esc('매매')}: {esc(str(total_count))}건 " f"\\({esc(f'W:{wins}')} / {esc(f'L:{losses}')}\\)", f"\U0001F4AF {esc('승률')}: {esc(f'{win_rate:.1f}%')}", ] # 전략별 PnL by_strategy = pnl_info.get("by_strategy", {}) if by_strategy: lines.append(f"") lines.append(f"\U0001F4CB {esc('전략별')}:") for strategy, stats in by_strategy.items(): st_pnl = stats.get("pnl", 0) st_count = stats.get("count", 0) st_sign = "+" if st_pnl >= 0 else "" lines.append( f" {esc(strategy)}: {esc(f'{st_sign}{st_pnl:,.0f}')}원 " f"\\({esc(str(st_count))}건\\)" ) # AI 리뷰 요약 (있으면) ai_review = review.get("ai_review") if ai_review: review_text = ai_review.get("review_text", "") grade = ai_review.get("grade", "") if grade: lines.append(f"") lines.append(f"\U0001F3C6 {esc('AI 평가')}: {esc(grade)}") if review_text: # 최대 300자 (Telegram 메시지 길이 제한 고려) truncated = str(review_text)[:300] lines.append(f"\U0001F4DD {esc(truncated)}") lessons = ai_review.get("lessons", []) if lessons: lines.append(f"") lines.append(f"\U0001F4A1 {esc('교훈')}:") for lesson in lessons[:3]: lines.append(f" \\- {esc(str(lesson))}") market_info = review.get("market_summary", {}) regime = market_info.get("regime", "") if regime: lines.append(f"") lines.append(f"\U0001F30D {esc('시장 국면')}: {esc(regime)}") text = "\n".join(lines) await telegram_notifier.send_message(text) except Exception as e: logger.warning("[AI-Review] Telegram 리뷰 발송 실패: %s", e)