diff --git a/backend/app/main.py b/backend/app/main.py index 112a2c9..abac302 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -48,14 +48,46 @@ logging.basicConfig( logger = logging.getLogger(__name__) +def _normalize_recommendations(rec: dict) -> dict: + """사로 다른 포맷의 추청 결과를 recommendations_by_rule 포맷으로 정규화. + + - get_rule_based_recommendations_realtime() → {"recommendations": [...], ...} + - get_rule_based_recommendations_from_db() → {"recommendations_by_rule": {...}, ...} + + 둘 다 recommendations_by_rule 키를 보장한 상태로 돌려준다. + Redis 캐시 데이터와 DB 데이터 모두 이 함수를 통해 정규화. + """ + if not rec: + return rec + # 이미 recommendations_by_rule 키가 있으면 그대로 사용 + if rec.get("recommendations_by_rule"): + return rec + # recommendations 리스트를 rule 기준으로 그룹화 + raw_list = rec.get("recommendations", []) + if not raw_list: + return rec + by_rule: dict = {} + for item in raw_list: + rule = ( + item.get("rule_name") + or item.get("signal") + or item.get("strategy") + or "general" + ) + by_rule.setdefault(rule, []).append(item) + return {**rec, "recommendations_by_rule": by_rule} + + async def batch_analysis_scheduler(): """10분 간격 배치 분석 스케줄러. 장중 10분마다 KIS API로 종목 분석 후 stock_analysis DB에 upsert. - auto_trade_scheduler가 DB를 빠르게 읽어 매매 판단에 활용한다. + 분석 완료 후 Redis에 recommendations:latest 캐시를 저장해 + auto_trade_scheduler가 DB 조회 없이 바로 매매 판단에 활용할 수 있도록 한다. """ await asyncio.sleep(15) # 부팅 안정화 (auto_trade_scheduler보다 늦게 시작) from app.api.endpoints.recommendations import get_rule_based_recommendations_realtime + from app.core.redis import setex_json, TTL_RECOMMENDATIONS interval = int(getattr(settings, "BATCH_ANALYSIS_INTERVAL", 600)) # 기본 10분 logger.info(f"📊 Batch analysis scheduler started (every {interval}s)") @@ -69,6 +101,22 @@ async def batch_analysis_scheduler(): result = await get_rule_based_recommendations_realtime() count = len(result.get("recommendations", [])) if result else 0 logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB") + + # ── Redis 캐시 업데이트 ───────────────────────────────────────── + if result: + # 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환) + normalized = _normalize_recommendations(result) + ok = await setex_json( + "recommendations:latest", TTL_RECOMMENDATIONS, normalized + ) + if ok: + logger.info( + f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest " + f"(rule 수: {len(normalized.get('recommendations_by_rule', {}))}, " + f"TTL: {TTL_RECOMMENDATIONS}s)" + ) + # ─────────────────────────────────────────────────────── + except Exception as e: logger.error(f"❌ Batch analysis error: {e}", exc_info=True) await asyncio.sleep(interval) @@ -79,9 +127,10 @@ async def batch_analysis_scheduler(): async def auto_trade_scheduler(): """1분 간격 자동매매 스케줄러. - DB 캐시에서 추천 종목을 읽어 손절/익절 및 신규 매수를 실행한다. - 주문 직전에 MCP RISK_ASSESSMENT를 호출해 승인된 경우에만 주문한다. - 실제 KIS API 분석은 batch_analysis_scheduler(10분)가 담당한다. + + Redis 캐시(recommendations:latest)를 먼저 확인 → 캐시 히트 시 DB 조회 없이 매매 실행. + 케시 미스(30분 TTL 만료) 시만 get_rule_based_recommendations_from_db() 호출. + 주문 직전 MCP RISK_ASSESSMENT 호출 → 승인 시만 주문. """ await asyncio.sleep(2) from app.api.endpoints.recommendations import ( @@ -89,10 +138,10 @@ async def auto_trade_scheduler(): execute_auto_trade_background, ) from app.api.endpoints.portfolio import check_and_execute_exit_trades - # 공통 MCP 리스크 평가 함수 (이슈 4) from app.api.endpoints.mcp import request_mcp_risk_assessment + from app.core.redis import get_json - logger.info("⏱️ Auto-trade scheduler started (every 60s, DB mode)") + logger.info("⏱️ Auto-trade scheduler started (every 60s, Redis→DB fallback mode)") try: while True: try: @@ -109,10 +158,22 @@ async def auto_trade_scheduler(): except Exception as e: logger.error(f"Exit trade check error: {e}", exc_info=True) - # 2. 자동매매: DB 캐시에서 추천 조회 + # 2. 자동매매: Redis 캐시 우선 → cache miss 시만 DB 조회 if settings.AUTO_TRADE_ENABLED: - logger.info("🔄 Checking DB recommendations for auto-trade...") - rec = await get_rule_based_recommendations_from_db(session) + # ── Redis 캐시 확인 (10분 배치 결과, TTL 30분) ──────────────── + rec = await get_json("recommendations:latest") + if rec: + logger.info( + "⚡ Redis cache hit: recommendations:latest " + f"(rule 수: {len(rec.get('recommendations_by_rule', {}))}) " + "→ DB 조회 생략" + ) + # 포맷 정규화 (다른 키 조합 대응) + rec = _normalize_recommendations(rec) + else: + logger.info("🔄 Redis cache miss → DB에서 추천 조회") + rec = await get_rule_based_recommendations_from_db(session) + # ────────────────────────────────────────────────────── if rec and rec.get("recommendations_by_rule"): # ── MCP RISK_ASSESSMENT (공통 함수, 이슈 4) ────────── @@ -134,7 +195,7 @@ async def auto_trade_scheduler(): ) ][:10], } - portfolio: dict = {} # 확장 시 실제 포트폴리오 요약 전달 + portfolio: dict = {} signals = { "source": "db_recommendations", "rules_triggered": list(recommendations.keys()), @@ -163,7 +224,7 @@ async def auto_trade_scheduler(): logger.info("📊 Found recommendations, executing auto-trade") await execute_auto_trade_background(rec, session) else: - logger.info("⏭️ Auto-trade skipped: no recommendations in DB") + logger.info("⏭️ Auto-trade skipped: no recommendations") else: logger.info("⏸️ Auto-trade disabled - only monitoring stop-loss/take-profit") except Exception as e: @@ -530,13 +591,11 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str): client_id, ) # 구독 종목 차트 캐시 사전 워밍업 (백그라운드, 논블로킹) - # 사용자가 종목 클릭 전에 캐시가 완성되도록 미리 KIS API 호출 if codes: async def _warmup_chart_cache(stock_codes: list): from app.api.endpoints.market_data import get_chart_data_cached, chart_cache from datetime import datetime for code in stock_codes: - # 이미 유효한 캐시가 있으면 건너뜀 cached = chart_cache.get(f"{code}__D") if cached and (datetime.now() - cached["timestamp"]).total_seconds() < 600: continue