from fastapi import APIRouter, HTTPException, Query, Depends from typing import List, Optional from pydantic import BaseModel from datetime import datetime, date, timedelta from decimal import Decimal import asyncio import logging import httpx import pytz from app.services.korea_investment_api import korea_investment_api from app.core.database import get_db, AsyncSessionLocal from app.models.models import TradeHistory, Portfolio, StockAnalysis from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, and_, desc from app.core.config import settings from app.core.market_schedule import is_market_open from app.services.mean_reversion_strategy import ( MarketRegime, detect_market_regime, check_buy_signal, calculate_atr, calculate_adx, calculate_rsi, calculate_bollinger, is_reversal_candle, calculate_position_size_v2, ) from app.services.mdd_manager import mdd_manager from app.services.loss_tracker import loss_tracker router = APIRouter() logger = logging.getLogger(__name__) # 한국 시간대 KST = pytz.timezone('Asia/Seoul') def _get_kst_now(): """현재 한국 시간 반환 (timezone-naive datetime)""" return datetime.now(KST).replace(tzinfo=None) # Trend filter defaults (configurable via settings) TREND_FILTER_ENABLED = getattr(settings, "TREND_FILTER_ENABLED", True) TREND_MA_SHORT = int(getattr(settings, "TREND_MA_SHORT", 20)) TREND_MA_LONG = int(getattr(settings, "TREND_MA_LONG", 60)) TREND_TOLERANCE_PCT = float(getattr(settings, "TREND_TOLERANCE_PCT", 0.005)) # 급등 과열 필터 defaults OVERHEAT_FILTER_ENABLED = getattr(settings, "OVERHEAT_FILTER_ENABLED", True) OVERHEAT_3DAY_MAX = float(getattr(settings, "OVERHEAT_3DAY_MAX", 15.0)) OVERHEAT_5DAY_MAX = float(getattr(settings, "OVERHEAT_5DAY_MAX", 20.0)) OVERHEAT_10DAY_MAX = float(getattr(settings, "OVERHEAT_10DAY_MAX", 30.0)) OVERHEAT_DISPARITY_MAX = float(getattr(settings, "OVERHEAT_DISPARITY_MAX", 108.0)) def _ma(values: list, window: int) -> Optional[float]: try: if len(values) < window: return None return sum(values[-window:]) / window except Exception: return None def _passes_uptrend(closes: list, ma_short: int = TREND_MA_SHORT, ma_long: int = TREND_MA_LONG, tol: float = TREND_TOLERANCE_PCT) -> Optional[bool]: if not closes or len(closes) < max(ma_short, ma_long): return None ma_s = _ma(closes, ma_short) ma_l = _ma(closes, ma_long) close = closes[-1] if ma_s is None or ma_l is None: return None # 허용 오차 적용: 거의 골든크로스/이탈 직후도 허용 return (ma_s >= ma_l * (1 - tol)) and (close >= ma_s * (1 - tol)) def analyze_overheat(closes: list, ma20: Optional[float]) -> dict: """급등 과열 필터 - 단기 급등 종목 매수 차단 3일/5일/10일 상승률과 20일선 이격도를 종합 판단하여 고점 추격 매수를 원천 차단합니다. Returns: {"blocked": bool, "reason": str, "details": dict} """ details = {} reasons = [] if len(closes) >= 4: gain_3d = ((closes[-1] - closes[-4]) / closes[-4]) * 100 if closes[-4] > 0 else 0 details["gain_3d"] = round(gain_3d, 1) if gain_3d > OVERHEAT_3DAY_MAX: reasons.append(f"3일 +{gain_3d:.1f}% (>{OVERHEAT_3DAY_MAX}%)") if len(closes) >= 6: gain_5d = ((closes[-1] - closes[-6]) / closes[-6]) * 100 if closes[-6] > 0 else 0 details["gain_5d"] = round(gain_5d, 1) if gain_5d > OVERHEAT_5DAY_MAX: reasons.append(f"5일 +{gain_5d:.1f}% (>{OVERHEAT_5DAY_MAX}%)") if len(closes) >= 11: gain_10d = ((closes[-1] - closes[-11]) / closes[-11]) * 100 if closes[-11] > 0 else 0 details["gain_10d"] = round(gain_10d, 1) if gain_10d > OVERHEAT_10DAY_MAX: reasons.append(f"10일 +{gain_10d:.1f}% (>{OVERHEAT_10DAY_MAX}%)") if ma20 and ma20 > 0: disparity = (closes[-1] / ma20) * 100 details["disparity_20"] = round(disparity, 1) if disparity > OVERHEAT_DISPARITY_MAX: reasons.append(f"이격도 {disparity:.1f}% (>{OVERHEAT_DISPARITY_MAX}%)") blocked = len(reasons) > 0 reason = "급등과열: " + ", ".join(reasons) if blocked else "" return {"blocked": blocked, "reason": reason, "details": details} class RecommendationCriteria(BaseModel): """추천 종목 조건""" use_macd: bool = False use_rsi: bool = False use_bollinger: bool = False use_moving_average: bool = False use_stochastic: bool = False use_adx: bool = False use_obv: bool = False use_volatility_breakout: bool = False # 변동성 돌파 전략 use_overheat_filter: bool = True # 급등 과열 필터 (고점 추격 방지) min_volume_ratio: float = 1.0 # 평균 대비 거래량 배수 (1.0 = 필터 없음) max_disparity: float = 200.0 # 최대 이격도 (20일선 대비 %) - 완화 for_batch: bool = False # 배치 분석 모드: 필터 완화, score=0도 반환 class StockRecommendation(BaseModel): """추천 종목 정보""" stock_code: str stock_name: str current_price: float change_rate: float signals: List[str] score: int price_date: str = "" # 가격 기준일 (YYYYMMDD 형식) # 추가 지표 (배치 저장 및 API 응답에 사용) volume: Optional[int] = None rsi: Optional[float] = None ma5: Optional[float] = None ma20: Optional[float] = None ma60: Optional[float] = None stochastic_k: Optional[float] = None stochastic_d: Optional[float] = None bollinger_upper: Optional[float] = None bollinger_lower: Optional[float] = None bollinger_middle: Optional[float] = None # ML 앙상블 스코어 ml_score: Optional[int] = None # 앙상블 스코어 (0~100) ml_confidence: Optional[float] = None # 신뢰도 (0~1) ml_signal: Optional[str] = None # BUY/HOLD/SELL # ATR (배치 저장용 — 이중 API 호출 방지) atr_14: Optional[float] = None # ADX (추세 강도 — 평균회귀 필터용) adx_14: Optional[float] = None # 공식 기반 신호 분석 함수 (이동평균 정배열/역배열, 눌림목, 손절/익절) def analyze_custom_signals(chart_data): """커스텀 매매 공식 기반 신호 분석 - 이격도 필터 적용""" closes = [float(item.get("stck_clpr", 0)) for item in chart_data] if len(closes) < 120: return None ma20 = sum(closes[-20:]) / 20 ma60 = sum(closes[-60:]) / 60 ma120 = sum(closes[-120:]) / 120 close = closes[-1] # 🔥 이격도 계산 (추격 매수 방지) - 더 엄격하게 disparity = (close / ma20) * 100 # 🚫 급등 종목 필터: 최근 3일 상승률 체크 if len(closes) >= 4: price_3days_ago = closes[-4] recent_gain = ((close - price_3days_ago) / price_3days_ago) * 100 if price_3days_ago > 0 else 0 if recent_gain > 10: # 3일간 10% 이상 급등 시 매수 안함 return {"signal": None, "description": f"급등 종목 추격 금지 (3일 +{recent_gain:.1f}%)"} # 정배열 매수 - 이격도 조건 강화 (105% -> 102%) if ma20 > ma60 > ma120 and close > ma20: if disparity <= 102: # 더 엄격하게 return {"signal": "buy", "description": "정배열 눌림목 매수 (이격도 안정)"} else: return {"signal": None, "description": f"정배열이나 이격 과열 ({disparity:.1f}%)"} # 역배열 매도 elif ma20 < ma60 < ma120 and close < ma20: return {"signal": "sell", "description": "역배열 하락 매도타점"} # 눌림목 진입 - 이격도 조건 강화 (103% -> 101%) recent_lows = [float(item.get("stck_lwpr", 0)) for item in chart_data[-5:]] if any(low < ma20 for low in recent_lows) and close > ma20: if disparity <= 101: # 더 엄격하게 return {"signal": "buy", "description": "눌림목 진입 매수타점"} else: return {"signal": None, "description": f"눌림목이나 이격 과열 ({disparity:.1f}%)"} # 손절/익절 stop_loss_line = ma60 take_profit_line = ma20 * 1.05 if close < stop_loss_line: return {"signal": "exit", "description": "손절라인 이탈"} if close > take_profit_line: return {"signal": "take_profit", "description": "익절라인 돌파"} return None def analyze_macd(chart_data: list) -> dict: """MACD 분석 (골든크로스/데드크로스)""" if len(chart_data) < 26: return {"signal": None, "description": "데이터 부족"} # 간단한 MACD 계산 (12일, 26일, 9일) prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-26:]] if len(prices) < 26: return {"signal": None, "description": "데이터 부족"} # EMA 계산 def calculate_ema(prices, period): ema = [sum(prices[:period]) / period] multiplier = 2 / (period + 1) for price in prices[period:]: ema.append((price - ema[-1]) * multiplier + ema[-1]) return ema ema12 = calculate_ema(prices, 12) ema26 = calculate_ema(prices, 26) if len(ema12) < 2 or len(ema26) < 2: return {"signal": None, "description": "데이터 부족"} macd = [ema12[i] - ema26[i] for i in range(len(ema26))] signal = calculate_ema(macd, 9) if len(signal) < 2: return {"signal": None, "description": "데이터 부족"} # 골든크로스: MACD가 시그널선을 상향 돌파 if macd[-2] <= signal[-2] and macd[-1] > signal[-1]: return {"signal": "buy", "description": "MACD 골든크로스"} # 데드크로스: MACD가 시그널선을 하향 돌파 elif macd[-2] >= signal[-2] and macd[-1] < signal[-1]: return {"signal": "sell", "description": "MACD 데드크로스"} return {"signal": None, "description": "신호 없음"} def analyze_rsi(chart_data: list) -> dict: """RSI 분석 (과매수/과매도) - 바닥 탈출 전략""" if len(chart_data) < 15: return {"signal": None, "description": "데이터 부족"} # 현재 RSI 계산 prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-15:]] gains = [] losses = [] for i in range(1, len(prices)): change = prices[i] - prices[i-1] if change > 0: gains.append(change) losses.append(0) else: gains.append(0) losses.append(abs(change)) avg_gain = sum(gains[-14:]) / 14 avg_loss = sum(losses[-14:]) / 14 if avg_loss == 0: rsi = 100 else: rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) # 이전 RSI 계산 (추세 확인용) if len(chart_data) >= 16: prev_prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-16:-1]] prev_gains = [] prev_losses = [] for i in range(1, len(prev_prices)): change = prev_prices[i] - prev_prices[i-1] if change > 0: prev_gains.append(change) prev_losses.append(0) else: prev_gains.append(0) prev_losses.append(abs(change)) prev_avg_gain = sum(prev_gains[-14:]) / 14 prev_avg_loss = sum(prev_losses[-14:]) / 14 if prev_avg_loss == 0: rsi_prev = 100 else: prev_rs = prev_avg_gain / prev_avg_loss rsi_prev = 100 - (100 / (1 + prev_rs)) else: rsi_prev = rsi # 🔥 Falling Knife 방지: 바닥 탈출 확인 if rsi < 30: # 30 미만은 관망 (하락세가 강함 - 떨어지는 칼날) return {"signal": None, "description": f"RSI 과매도 심화 ({rsi:.1f}) - 반등 대기"} elif 30 <= rsi <= 45 and rsi > rsi_prev: # 바닥을 찍고 올라오는 구간 + 상승 중 return {"signal": "buy", "description": f"RSI 바닥 탈출 반등 ({rsi:.1f})"} elif rsi > 70: return {"signal": "sell", "description": f"RSI 과매수 ({rsi:.1f})"} elif 45 <= rsi <= 60: # 중립 상승 구간 return {"signal": "buy", "description": f"RSI 안정 상승 ({rsi:.1f})"} return {"signal": None, "description": f"RSI 중립 ({rsi:.1f})"} def analyze_bollinger(chart_data: list) -> dict: """볼린저밴드 분석""" if len(chart_data) < 20: return {"signal": None, "description": "데이터 부족"} prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-20:]] # 20일 이동평균 ma20 = sum(prices) / len(prices) # 표준편차 variance = sum((p - ma20) ** 2 for p in prices) / len(prices) std_dev = variance ** 0.5 upper_band = ma20 + (2 * std_dev) lower_band = ma20 - (2 * std_dev) current_price = prices[-1] if current_price < lower_band: return {"signal": "buy", "description": "볼린저 하한선 터치"} elif current_price > upper_band: return {"signal": "sell", "description": "볼린저 상한선 돌파"} return {"signal": None, "description": "볼린저 중립"} def analyze_moving_average(chart_data: list) -> dict: """이동평균선 분석 (5, 20, 60일)""" if len(chart_data) < 60: return {"signal": None, "description": "데이터 부족"} prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-60:]] ma5 = sum(prices[-5:]) / 5 ma20 = sum(prices[-20:]) / 20 ma60 = sum(prices[-60:]) / 60 current_price = prices[-1] # 조건 완화: 5일선이 20일선 위에만 있어도 매수 if ma5 > ma20 and current_price > ma5: return {"signal": "buy", "description": "5일선 > 20일선"} # 20일선이 60일선 위에 있으면 추가 점수 elif ma20 > ma60 and current_price > ma20: return {"signal": "buy", "description": "20일선 > 60일선"} # 역배열 elif ma5 < ma20 < ma60 and current_price < ma5: return {"signal": "sell", "description": "이동평균 역배열"} return {"signal": None, "description": "이동평균 중립"} def analyze_stochastic(chart_data: list) -> dict: """스토캐스틱 분석""" if len(chart_data) < 14: return {"signal": None, "description": "데이터 부족"} recent_data = chart_data[-14:] highs = [float(item.get("stck_hgpr", 0)) for item in recent_data] lows = [float(item.get("stck_lwpr", 0)) for item in recent_data] closes = [float(item.get("stck_clpr", 0)) for item in recent_data] highest_high = max(highs) lowest_low = min(lows) current_close = closes[-1] if highest_high == lowest_low: return {"signal": None, "description": "데이터 부족"} k = ((current_close - lowest_low) / (highest_high - lowest_low)) * 100 # 조건 완화: 30 이하 매수, 70 이상 매도 if k < 30: return {"signal": "buy", "description": f"스토캐스틱 과매도 (%K: {k:.1f})"} elif k > 70: return {"signal": "sell", "description": f"스토캐스틱 과매수 (%K: {k:.1f})"} return {"signal": None, "description": f"스토캐스틱 중립 (%K: {k:.1f})"} def analyze_volatility_breakout(chart_data: list, k: float = 0.5) -> dict: """🔥 변동성 돌파 전략 (래리 윌리엄스) - 승률이 매우 높은 단타 전략 공식: 현재가 > 시가 + (전일 고가 - 전일 저가) * K """ if len(chart_data) < 2: return {"signal": None, "description": "데이터 부족"} today = chart_data[-1] # 최신 데이터 (시간순 정렬 기준 마지막) yesterday = chart_data[-2] # 전일 데이터 # 오늘 데이터 current_price = float(today.get("stck_clpr", 0)) # 현재가 open_price = float(today.get("stck_oprc", 0)) # 시가 # 전일 데이터 prev_high = float(yesterday.get("stck_hgpr", 0)) prev_low = float(yesterday.get("stck_lwpr", 0)) prev_range = prev_high - prev_low if prev_range == 0 or open_price == 0: return {"signal": None, "description": "데이터 부족"} # 목표 돌파가 = 시가 + (전일 변동폭 * K) target_price = open_price + (prev_range * k) if current_price >= target_price: return {"signal": "buy", "description": f"변동성 돌파 발생 (목표가: {target_price:.0f})"} return {"signal": None, "description": "변동성 돌파 미달"} async def check_market_regime() -> dict: """시장 레짐 판별 (CLAUDE.md 섹션 7.1: 200일 MA + 기울기) 상승장(Bull): 코스피 > MA200 AND MA200 기울기 양(+) 하락장(Bear): 그 외 하위 분류 (기존 호환): - strong_bearish: 코스피 < MA60 또는 5일 연속 하락 - bearish: 코스피 < MA20 - bullish: 코스피 > MA200 + 기울기 상승 """ try: # 장외 시간에는 모의투자 API가 차트 데이터를 반환하지 않음 → neutral 반환 if not is_market_open(): return {"regime": "neutral", "regime_200d": "bear", "message": "장외 시간 - 시장 레짐 판별 스킵", "position_multiplier": 0.5} # KODEX 200 (069500) = KOSPI 200 추종 ETF 사용 (지수코드 0001은 daily-price API 미지원) kospi_chart = await korea_investment_api.get_daily_chart("069500", "D", limit=240) if "output2" not in kospi_chart or not kospi_chart["output2"]: logger.warning("KOSPI 지수 데이터 없음 - 시장 필터 비활성화") return {"regime": "neutral", "message": "시장 지수 확인 불가", "position_multiplier": 0.5} chart_data = kospi_chart["output2"] closes = [float(item.get("stck_clpr", 0)) for item in chart_data] if len(closes) < 60: return {"regime": "neutral", "message": "데이터 부족", "position_multiplier": 0.5} current_price = closes[0] ma20 = sum(closes[:20]) / 20 ma60 = sum(closes[:60]) / 60 # 200일 MA 기반 레짐 판별 (섹션 7.1) if len(closes) >= 205: regime_enum = detect_market_regime(list(reversed(closes))) is_bull_200 = regime_enum == MarketRegime.BULL ma200 = sum(closes[:200]) / 200 else: is_bull_200 = current_price > ma20 # 200일 데이터 부족 시 MA20 폴백 ma200 = ma60 # 5일 연속 하락 체크 consecutive_down = 0 for i in range(len(closes) - 1): if closes[i] < closes[i + 1]: consecutive_down += 1 else: break drop_from_ma20 = ((current_price - ma20) / ma20) * 100 # 강한 하락장: 코스피 < MA60 또는 5일 연속 하락 if current_price < ma60 or consecutive_down >= 5: return { "regime": "strong_bearish", "regime_200d": "bear", "message": f"강한 하락장 (코스피: {current_price:.0f}, MA200: {ma200:.0f}, {consecutive_down}일 연속하락)", "position_multiplier": 0.1, "min_score": settings.BEAR_MARKET_MIN_SCORE + 3, "max_trades": 1, "consecutive_down": consecutive_down, "drop_from_ma20": round(drop_from_ma20, 1), } # 약세장: 코스피 < MA20 if current_price < ma20: return { "regime": "bearish", "regime_200d": "bear", "message": f"하락장 (코스피: {current_price:.0f} < MA20: {ma20:.0f}, MA200: {ma200:.0f})", "position_multiplier": settings.BEAR_MARKET_POSITION_RATIO, "min_score": settings.BEAR_MARKET_MIN_SCORE, "max_trades": settings.BEAR_MARKET_MAX_TRADES, "consecutive_down": consecutive_down, "drop_from_ma20": round(drop_from_ma20, 1), } # 상승장: 200일 MA 기반 return { "regime": "bullish", "regime_200d": "bull" if is_bull_200 else "bear", "message": f"상승장 (코스피: {current_price:.0f}, MA200: {ma200:.0f}, Bull={is_bull_200})", "position_multiplier": 1.0, "min_score": 1, "max_trades": settings.MAX_DAILY_TRADES, "consecutive_down": consecutive_down, "drop_from_ma20": round(drop_from_ma20, 1), } except Exception as e: logger.error(f"시장 지수 확인 실패: {e}") return {"regime": "neutral", "message": "시장 지수 확인 실패", "position_multiplier": 0.5} def analyze_adx(chart_data: list) -> dict: """ADX 분석 — Wilder's ADX(14) 기반 추세 강도 측정 평균회귀 전략에서의 해석: - ADX < 20: 횡보장 → 평균회귀 유리 (BUY 신호) - ADX 20~30: 중간 추세 → 중립 - ADX > 30: 강한 추세 → 평균회귀 위험 (SELL/SKIP 신호) chart_data는 시간 순(오래된 → 최신) 정렬을 가정. """ if len(chart_data) < 29: # period * 2 + 1 return {"signal": None, "description": "데이터 부족", "adx": None} highs = [float(item.get("stck_hgpr", 0)) for item in chart_data] lows = [float(item.get("stck_lwpr", 0)) for item in chart_data] closes = [float(item.get("stck_clpr", 0)) for item in chart_data] adx = calculate_adx(highs, lows, closes) if adx is None: return {"signal": None, "description": "ADX 계산 불가", "adx": None} if adx < 20: return {"signal": "buy", "description": f"ADX({adx:.0f}<20) 횡보장 — 평균회귀 유리", "adx": adx} if adx > 30: return {"signal": "sell", "description": f"ADX({adx:.0f}>30) 강추세 — 평균회귀 위험", "adx": adx} return {"signal": None, "description": f"ADX({adx:.0f}) 중립", "adx": adx} def analyze_obv(chart_data: list) -> dict: """거래량(OBV) 분석""" if len(chart_data) < 2: return {"signal": None, "description": "데이터 부족"} volumes = [int(item.get("acml_vol", 0)) for item in chart_data[-5:]] closes = [float(item.get("stck_clpr", 0)) for item in chart_data[-5:]] if len(volumes) < 2 or len(closes) < 2: return {"signal": None, "description": "데이터 부족"} # 거래량 증가와 가격 상승 if volumes[-1] > volumes[-2] * 1.5 and closes[-1] > closes[-2]: return {"signal": "buy", "description": "거래량 증가와 상승"} # 거래량 감소와 가격 하락 elif volumes[-1] < volumes[-2] * 0.5 and closes[-1] < closes[-2]: return {"signal": "sell", "description": "거래량 감소와 하락"} return {"signal": None, "description": "거래량 중립"} async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: """개별 종목 분석""" try: logger.info(f"Analyzing stock: {stock_code}") # 차트 데이터 조회 (현재가 포함) chart_response = await korea_investment_api.get_daily_chart(stock_code, "D") if "output2" not in chart_response or not chart_response["output2"]: logger.warning(f"No chart data for {stock_code}") return None # KIS API는 최신순(newest-first)으로 반환 → 시간순(chronological)으로 뒤집어 분석 chart_data = list(reversed(chart_response["output2"])) logger.info(f"Got {len(chart_data)} chart data points for {stock_code}") # 최신 데이터에서 현재가 및 종목명 추출 (시간순 기준 마지막 = 최신) latest = chart_data[-1] if chart_data else None if not latest: return None # 종목명이 없으면 차트 데이터에서 추출 시도 if not stock_name: stock_name = latest.get("hts_kor_isnm", stock_code) current_price = float(latest.get("stck_clpr", 0)) price_date = latest.get("stck_bsop_date", "") # 가격 기준일 volume = int(latest.get("acml_vol", 0) or 0) if current_price == 0: logger.warning(f"Zero price for {stock_code}") return None # 등락률 계산 if len(chart_data) >= 2: prev_price = float(chart_data[-2].get("stck_clpr", current_price)) change_rate = ((current_price - prev_price) / prev_price) * 100 if prev_price > 0 else 0 else: change_rate = 0 # 기본 수치형 지표 계산 (배치 저장용) closes = [float(item.get("stck_clpr", 0)) for item in chart_data] volumes = [int(item.get("acml_vol", 0)) for item in chart_data] ma5 = sum(closes[-5:]) / 5 if len(closes) >= 5 else None ma20 = sum(closes[-20:]) / 20 if len(closes) >= 20 else None ma60 = sum(closes[-60:]) / 60 if len(closes) >= 60 else None # 🔥 1. 거래량 분석 (Volume Analysis) - 필수 조건 avg_vol_20 = sum(volumes[-20:]) / 20 if len(volumes) >= 20 else 0 volume_ratio = (volume / avg_vol_20) if avg_vol_20 > 0 else 0 # 거래량이 너무 적으면 추천 제외 (유동성 부족) - 배치 모드에서는 제외 안 함 if not criteria.for_batch and avg_vol_20 < 50000 and current_price < 50000: logger.debug(f"{stock_code}: 거래량 부족으로 제외 (평균 거래량: {avg_vol_20:.0f})") return None # 🔥 2. 이격도 계산 (Disparity Filter) - 추격 매수 방지 disparity = (current_price / ma20) * 100 if ma20 else 100 # Trend filter check (non-blocking here; used for scoring/tagging) trend_ok = None if TREND_FILTER_ENABLED: trend_ok = _passes_uptrend(closes, TREND_MA_SHORT, TREND_MA_LONG) # RSI 계산 (단순 방식) rsi = None if len(closes) >= 15: prices_for_rsi = closes[-15:] gains = [] losses = [] for i in range(1, len(prices_for_rsi)): diff = prices_for_rsi[i] - prices_for_rsi[i-1] gains.append(max(diff, 0)) losses.append(max(-diff, 0)) avg_gain = sum(gains[-14:]) / 14 avg_loss = sum(losses[-14:]) / 14 if avg_loss == 0: rsi = 100.0 else: rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) # Stochastic %K, %D 간단 계산 stochastic_k = None stochastic_d = None if len(chart_data) >= 14: recent = chart_data[-14:] highs = [float(item.get("stck_hgpr", 0)) for item in recent] lows = [float(item.get("stck_lwpr", 0)) for item in recent] closes_for_stoch = [float(item.get("stck_clpr", 0)) for item in recent] highest_high = max(highs) lowest_low = min(lows) if highest_high != lowest_low: k = ((closes_for_stoch[-1] - lowest_low) / (highest_high - lowest_low)) * 100 stochastic_k = k # %D는 간단히 최근 3일 %K 평균 stochastic_d_vals = [] for j in range(len(recent)-3, len(recent)): slice_high = max([float(item.get("stck_hgpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]]) slice_low = min([float(item.get("stck_lwpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]]) if slice_high != slice_low: slice_k = ((float(recent[j].get("stck_clpr", 0)) - slice_low) / (slice_high - slice_low)) * 100 stochastic_d_vals.append(slice_k) if stochastic_d_vals: stochastic_d = sum(stochastic_d_vals) / len(stochastic_d_vals) # Bollinger bands bollinger_upper = bollinger_lower = bollinger_middle = None if len(closes) >= 20: window = closes[-20:] middle = sum(window) / len(window) variance = sum((p - middle) ** 2 for p in window) / len(window) stddev = variance ** 0.5 bollinger_middle = middle bollinger_upper = middle + 2 * stddev bollinger_lower = middle - 2 * stddev # 지표 분석 signals = [] score = 0 # 🔥 3. 이격도 과열 체크 (max_disparity 조건) if disparity > criteria.max_disparity: score -= 5 signals.append(f"이격도 과열({disparity:.1f}% > {criteria.max_disparity}%) - 매수금지") logger.debug(f"{stock_code}: 이격도 과열로 점수 감점") # 🔥 3.5. 급등 과열 필터 (고점 추격 매수 차단) if criteria.use_overheat_filter and OVERHEAT_FILTER_ENABLED: overheat = analyze_overheat(closes, ma20) if overheat["blocked"]: score -= 10 signals.append(overheat["reason"]) logger.info(f"{stock_code}: 급등 과열 차단 - {overheat['reason']} {overheat['details']}") # 🔥 4. 변동성 돌파 전략 (고승률) if criteria.use_volatility_breakout: vb_result = analyze_volatility_breakout(chart_data) if vb_result and vb_result["signal"] == "buy": signals.append(vb_result["description"]) score += 4 # 매우 강력한 신호 # 커스텀 공식 기반 신호 분석 적용 (정배열, 역배열, 눌림목, 손절/익절) custom_result = analyze_custom_signals(chart_data) if custom_result: if custom_result["signal"] == "buy": signals.append(custom_result["description"]) score += 3 # 높은 가중치 elif custom_result["signal"] == "sell": # 매도 신호는 추천에서 제외하거나 음수 점수 signals.append(custom_result["description"]) score -= 2 elif custom_result["signal"] == "exit": # 손절 신호는 추천에서 제외 signals.append(custom_result["description"]) score -= 5 elif custom_result["signal"] == "take_profit": # 익절 신호는 양수 점수 signals.append(custom_result["description"]) score += 2 if criteria.use_macd: result = analyze_macd(chart_data) if result["signal"] == "buy": signals.append(result["description"]) score += 2 if criteria.use_rsi: result = analyze_rsi(chart_data) if result["signal"] == "buy": # 🔥 거래량이 동반되어야 진짜 반등 if volume_ratio >= 1.2: signals.append(result["description"] + " + 거래량증가") score += 3 else: signals.append(result["description"] + " (거래량부족)") score += 1 # 점수 낮게 책정 if criteria.use_bollinger: result = analyze_bollinger(chart_data) if result["signal"] == "buy": signals.append(result["description"]) score += 1 if criteria.use_moving_average: result = analyze_moving_average(chart_data) if result["signal"] == "buy": signals.append(result["description"]) score += 2 if criteria.use_stochastic: result = analyze_stochastic(chart_data) if result["signal"] == "buy": signals.append(result["description"]) score += 1 if criteria.use_adx: result = analyze_adx(chart_data) if result["signal"] == "buy": signals.append(result["description"]) score += 1 if criteria.use_obv: result = analyze_obv(chart_data) if result["signal"] == "buy": signals.append(result["description"]) score += 1 # 🔥 5. 거래량 급증 보너스 (매수 신호 강도 증가) if volume_ratio >= 1.5 and change_rate > 0: signals.append(f"거래량 실린 상승 ({volume_ratio:.1f}배)") score += 3 elif volume_ratio < 0.8: # 거래량이 평소보다 적으면 신뢰도 감소 score -= 1 logger.debug(f"{stock_code}: 거래량 부족 (평소의 {volume_ratio:.1f}배)") # Tag trend state to signals and adjust score slightly if TREND_FILTER_ENABLED and trend_ok is not None: if trend_ok: signals.append(f"추세상승(>{TREND_MA_SHORT}MA & {TREND_MA_SHORT}MA>{TREND_MA_LONG}MA, tol={int(TREND_TOLERANCE_PCT*10000)/100:.2f}%)") score += 1 else: signals.append(f"추세약함({TREND_MA_SHORT}MA≤{TREND_MA_LONG}MA 또는 종가≤{TREND_MA_SHORT}MA)") # 🔥 6. 최종 거래량 필터 (min_volume_ratio) if criteria.min_volume_ratio > 0 and volume_ratio < criteria.min_volume_ratio: logger.debug(f"{stock_code}: 거래량 필터 미달 ({volume_ratio:.2f} < {criteria.min_volume_ratio})") return None logger.info(f"{stock_code} ({stock_name}): score={score}, signals={signals}, volume_ratio={volume_ratio:.2f}, disparity={disparity:.1f}%") # 🤖 ML 앙상블 스코어 계산 ml_score_val = None ml_confidence_val = None ml_signal_val = None if settings.ML_ENSEMBLE_ENABLED: try: from app.services.ml_ensemble import get_ensemble ensemble = get_ensemble() if ensemble.is_trained: ml_result = ensemble.predict(chart_data) if "error" not in ml_result: ml_score_val = ml_result["ensemble_score"] ml_confidence_val = ml_result["confidence"] ml_signal_val = ml_result["signal"] # 룰베이스 + ML 가중 결합 rule_score_normalized = (score / 15.0) * 100.0 # 0~15 → 0~100 combined_score = int( rule_score_normalized * settings.ML_ENSEMBLE_RULE_WEIGHT + ml_score_val * settings.ML_ENSEMBLE_ML_WEIGHT ) # 0~100 → 0~15 스케일로 재변환 (기존 호환) score = max(0, int(combined_score / 100.0 * 15.0)) signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})") logger.info(f" ML 앙상블: score={ml_score_val}, confidence={ml_confidence_val:.2f}, signal={ml_signal_val}") except Exception as e: logger.debug(f"ML 앙상블 스코어 계산 실패 (정상 - 모델 미학습): {e}") # ATR / ADX 계산 (이미 가져온 차트 데이터 재사용 — 이중 API 호출 방지) atr_14_val = None adx_14_val = None try: if len(chart_data) >= 15: _h = [float(d.get("stck_hgpr", 0)) for d in chart_data] _l = [float(d.get("stck_lwpr", 0)) for d in chart_data] _c = [float(d.get("stck_clpr", 0)) for d in chart_data] atr_14_val = calculate_atr(_h, _l, _c) if len(chart_data) >= 29: # Wilder ADX 최소 데이터 (period * 2 + 1) adx_14_val = calculate_adx(_h, _l, _c) except Exception: pass # 신호가 있거나 점수가 0 이상이면 추천 (배치 모드에서는 score=0도 반환) if criteria.for_batch or len(signals) > 0 or score > 0: # 신호가 있으면 최소 1점 보장 final_score = max(1, score) if len(signals) > 0 else score return StockRecommendation( stock_code=stock_code, stock_name=stock_name, current_price=current_price, change_rate=change_rate, signals=signals if signals else ["조건 분석 중"], score=final_score, price_date=price_date, volume=volume, rsi=rsi, ma5=ma5, ma20=ma20, ma60=ma60, stochastic_k=stochastic_k, stochastic_d=stochastic_d, bollinger_upper=bollinger_upper, bollinger_lower=bollinger_lower, bollinger_middle=bollinger_middle, ml_score=ml_score_val, ml_confidence=ml_confidence_val, ml_signal=ml_signal_val, atr_14=atr_14_val, adx_14=adx_14_val, ) logger.debug(f"{stock_code}: No recommendation (score={score})") return None except Exception as e: logger.error(f"Error analyzing {stock_code}: {str(e)}") return None async def analyze_stock_for_batch(stock_code: str, stock_name: str) -> Optional[StockRecommendation]: """배치 DB 저장용: 거래량/점수 필터 없이 지표만 계산하여 전 종목 저장. 모든 지표를 계산하되, 매매 필터(min_volume_ratio, 거래량 부족, score=0)를 적용하지 않습니다. score=0이어도 반환하여 배치에서 DB에 저장됩니다. """ criteria = RecommendationCriteria( use_macd=True, use_rsi=True, use_bollinger=True, use_moving_average=True, use_stochastic=True, use_adx=True, use_obv=True, use_volatility_breakout=True, min_volume_ratio=0, # 거래량 필터 없음 max_disparity=200.0, # 이격도 필터 없음 for_batch=True, # score=0 · 저유동성 종목도 반환 ) return await analyze_stock(stock_code, stock_name, criteria) async def analyze_stock_with_indicators(stock_code: str, stock_name: str) -> Optional[StockRecommendation]: """호출 편의성: 모든 지표를 켜서 분석한 결과(Pydantic 모델)를 반환합니다. 이 함수는 배치 스크립트에서 사용됩니다. 🔥 승률 향상 기능 포함: - 변동성 돌파 전략 - 거래량 필터 (평균 대비 1.2배) - 이격도 필터 (110% 제한) """ criteria = RecommendationCriteria( use_macd=True, use_rsi=True, use_bollinger=True, use_moving_average=True, use_stochastic=True, use_adx=True, use_obv=True, use_volatility_breakout=True, # 🔥 변동성 돌파 추가 min_volume_ratio=1.2, # 🔥 거래량 필터 (평균의 1.2배 이상) max_disparity=110.0 # 🔥 이격도 필터 (20일선 대비 110% 이하) ) return await analyze_stock(stock_code, stock_name, criteria) async def get_all_tradable_stocks() -> List[tuple]: """KIS API로 전체 거래 가능 종목 조회""" try: logger.info("Fetching all tradable stocks from KIS API...") # 이전 API 호출과의 충돌을 피하기 위해 대기 await asyncio.sleep(3.0) # KOSPI 거래대금 상위 종목 조회 kospi_result = await korea_investment_api.get_stock_list("J") logger.info(f"KOSPI result: {kospi_result}") # Rate limit을 위한 대기 (초당 1건 제한) - 안전하게 3초 대기 await asyncio.sleep(3.0) # KOSDAQ 거래대금 상위 종목 조회 ("Q" = KOSDAQ, KIS API 규격) kosdaq_result = await korea_investment_api.get_stock_list("Q") logger.info(f"KOSDAQ result: {kosdaq_result}") stocks = [] # KOSPI 종목 추출 if "output" in kospi_result: for item in kospi_result["output"]: stock_code = item.get("mksc_shrn_iscd", "") stock_name = item.get("hts_kor_isnm", "") if stock_code and stock_name: stocks.append((stock_code, stock_name)) # KOSDAQ 종목 추가 if "output" in kosdaq_result: for item in kosdaq_result["output"]: stock_code = item.get("mksc_shrn_iscd", "") stock_name = item.get("hts_kor_isnm", "") if stock_code and stock_name: stocks.append((stock_code, stock_name)) logger.info(f"Found {len(stocks)} tradable stocks from KIS API") # API에서 종목을 못 가져온 경우 빈 리스트 반환 (테스트 데이터 제거) if not stocks: logger.warning("No stocks from KIS API, returning empty list") return stocks except Exception as e: logger.error(f"Error fetching stock list from KIS API: {e}", exc_info=True) logger.warning("Returning empty list due to API error") return [] @router.get("/recommendations") async def get_recommendations( use_macd: bool = Query(False, description="MACD 지표 사용"), use_rsi: bool = Query(False, description="RSI 지표 사용"), use_bollinger: bool = Query(False, description="볼린저밴드 사용"), use_moving_average: bool = Query(False, description="이동평균선 사용"), use_stochastic: bool = Query(False, description="스토캐스틱 사용"), use_adx: bool = Query(False, description="ADX 사용"), use_obv: bool = Query(False, description="거래량(OBV) 사용"), use_volatility_breakout: bool = Query(False, description="변동성 돌파 전략 사용"), min_volume_ratio: float = Query(1.0, description="최소 거래량 비율 (평균 대비)"), max_disparity: float = Query(200.0, description="최대 이격도 (20일선 대비 %)"), max_stocks: int = Query(100, description="분석할 최대 종목 수 (기본 100개)"), ): """기술적 지표 기반 추천 종목 조회 선택한 기술적 지표 조건에 부합하는 종목을 KOSPI 시가총액 상위 종목 중에서 찾아 추천합니다. 🔥 승률 향상 기능: - 변동성 돌파: 래리 윌리엄스 전략 (고승률) - 거래량 필터: 평균 대비 거래량 확인 - 이격도 필터: 20일선과의 거리 제한 (추격 매수 방지) - 시장 지수 필터: 코스피 지수 확인 (하락장 보수적 대응) """ # 🔥 시장 지수 확인 (가장 중요한 승률 결정 요소) market_regime = await check_market_regime() logger.info(f"📊 시장 상태: {market_regime['message']}") criteria = RecommendationCriteria( use_macd=use_macd, use_rsi=use_rsi, use_bollinger=use_bollinger, use_moving_average=use_moving_average, use_stochastic=use_stochastic, use_adx=use_adx, use_obv=use_obv, use_volatility_breakout=use_volatility_breakout, min_volume_ratio=min_volume_ratio, max_disparity=max_disparity ) # 모든 지표가 꺼져있으면 기본값 사용 if not any([use_macd, use_rsi, use_bollinger, use_moving_average, use_stochastic, use_adx, use_obv, use_volatility_breakout]): criteria.use_rsi = True criteria.use_moving_average = True criteria.use_volatility_breakout = True # 기본값에 변동성 돌파 추가 # 분석할 종목 리스트 가져오기 (KIS API에서 실시간 조회) all_stocks = await get_all_tradable_stocks() if not all_stocks: logger.warning("No stocks found from KIS API") return { "total": 0, "analyzed_stocks": 0, "criteria": criteria.dict(), "recommendations": [], "message": "거래 가능한 종목을 조회할 수 없습니다" } stocks_to_analyze = all_stocks[:max_stocks] logger.info(f"Analyzing {len(stocks_to_analyze)} stocks with criteria: {criteria.dict()}") # Rate Limit 방지: 배치 처리 (초당 최대 5개) batch_size = 5 delay_between_batches = 1.2 # 1.2초 대기 all_results = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] batch_num = i // batch_size + 1 total_batches = (len(stocks_to_analyze) + batch_size - 1) // batch_size logger.info(f"Processing batch {batch_num}/{total_batches} (stocks {i+1}-{min(i+batch_size, len(stocks_to_analyze))})") # 배치 단위로 병렬 처리 tasks = [analyze_stock(code, name, criteria) for code, name in batch] batch_results = await asyncio.gather(*tasks, return_exceptions=True) all_results.extend(batch_results) # 다음 배치 전 대기 (마지막 배치는 제외) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) # 예외 로깅 errors = [r for r in all_results if isinstance(r, Exception)] if errors: logger.error(f"Encountered {len(errors)} errors during analysis") # None 및 예외 제거 후 3000원 이하 종목 필터링, 점수순 정렬 regime_min_score = market_regime.get("min_score", 1) recommendations = [ r for r in all_results if r is not None and not isinstance(r, Exception) and r.current_price >= 3000 # 🚫 동전주 제외 and r.score >= regime_min_score # 🐻 하락장 최소 점수 필터 ] recommendations.sort(key=lambda x: x.score, reverse=True) logger.info(f"Found {len(recommendations)} recommendations out of {len(stocks_to_analyze)} analyzed stocks (min_score={regime_min_score})") if not recommendations: logger.warning("No recommendations found! Criteria might be too strict or API failed") return { "total": len(recommendations), "analyzed_stocks": len(stocks_to_analyze), "criteria": criteria.dict(), "market_regime": market_regime, # 🔥 시장 상태 정보 추가 "recommendations": recommendations[:20] # 상위 20개만 반환 } @router.get("/rule-based") async def get_rule_based_recommendations( realtime: bool = Query(False, description="실시간 분석 여부 (느림, 30초)"), auto_trade: bool = Query(True, description="자동매매 실행 여부"), db: AsyncSession = Depends(get_db) ): """ 매매 규칙 기반 자동 추천 - realtime=false (기본): DB에서 조회 (빠름, 0.1초) - 매일 아침 9시 데이터 - realtime=true: KIS API 실시간 분석 (느림, 30초) - auto_trade=true (기본): 조건에 맞는 종목을 자동으로 매매 활성화된 모든 매매 규칙을 확인하고, 조건에 맞는 종목을 자동으로 추천합니다. """ # 실시간 모드: 기존 방식 (KIS API 직접 호출) if realtime: logger.info("🔴 Real-time mode: Fetching live data from KIS API (slow)") result = await get_rule_based_recommendations_realtime() else: # DB 모드: 빠른 조회 logger.info("🟢 DB mode: Fetching from cached analysis (fast)") result = await get_rule_based_recommendations_from_db(db) # 자동매매 실행 (백그라운드) # 주의: 엔드포인트 반환 후 FastAPI가 db 세션을 닫으므로, # 백그라운드 태스크는 독립적인 새 세션을 열어서 사용해야 한다. if auto_trade and result.get("recommendations_by_rule"): rec_snapshot = result async def _bg_trade(): async with AsyncSessionLocal() as bg_db: await execute_auto_trade_background(rec_snapshot, bg_db) asyncio.create_task(_bg_trade()) logger.info("🤖 Auto-trade task created in background") return result def is_mean_reversion_candidate(stock, conditions: dict, market_regime_str: str) -> bool: """RSI+볼린저 평균회귀 매수 조건 확인 (CLAUDE.md 섹션 7.2) Bull: RSI ≤ 25 + BB 하단 -3% ~ +1.5% (거래량 200% OR 조건은 real-time check_buy_signal에서 처리) Bear: RSI ≤ 35 + BB 하단 이하 """ rsi_val = float(stock.rsi) if stock.rsi else None close = int(stock.current_price) if stock.current_price else 0 bb_lower = float(stock.bollinger_lower) if stock.bollinger_lower else None bb_upper = float(stock.bollinger_upper) if stock.bollinger_upper else None bb_middle = float(stock.bollinger_middle) if stock.bollinger_middle else None if not all([rsi_val, close, bb_lower, bb_upper, bb_middle]): return False # 가격 필터 (섹션 6.4) if close < settings.MIN_PRICE_KRW: return False regime = MarketRegime.BULL if market_regime_str == "bull" else MarketRegime.BEAR # ADX 필터: 강한 추세 중 평균회귀 진입 차단 (ADX > 30이면 횡보 아님) adx_val = float(stock.adx_14) if getattr(stock, "adx_14", None) else None adx_strong_trend = adx_val is not None and adx_val > 30 if regime == MarketRegime.BULL: rsi_threshold = conditions.get("rsi_bull_max", 25) bb_lower_pct = conditions.get("bb_lower_pct", 3.0) / 100 # -3% bb_upper_pct = conditions.get("bb_upper_pct", 1.5) / 100 # +1.5% in_bb_zone = bb_lower * (1 - bb_lower_pct) <= close <= bb_lower * (1 + bb_upper_pct) if adx_strong_trend: return False # 강추세 중 평균회귀 차단 return rsi_val <= rsi_threshold and in_bb_zone else: rsi_threshold = conditions.get("rsi_bear_max", 35) if adx_strong_trend: return False # 강추세 중 평균회귀 차단 return rsi_val <= rsi_threshold and close <= bb_lower async def get_rule_based_recommendations_from_db(db: AsyncSession): """DB에서 사전 분석된 데이터 조회 (빠름 - 0.1초)""" logger.info("Rule-based recommendations from DB") # JSON 파일에서 규칙 로드 import json from pathlib import Path from collections import namedtuple Rule = namedtuple('Rule', ['id', 'name', 'description', 'rule_type', 'priority', 'conditions', 'actions', 'is_active']) try: config_path = Path(__file__).parent.parent.parent.parent / "trading_rules_config.json" with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) rules_data = config.get("rules", []) active_rules = [] for idx, rule_data in enumerate(rules_data, start=1): if rule_data.get("is_active", False): active_rules.append(Rule( id=idx, name=rule_data["name"], description=rule_data["description"], rule_type=rule_data["rule_type"].upper(), priority=rule_data.get("priority", idx), conditions=rule_data.get("conditions", {}), actions=rule_data.get("actions", {}), is_active=True )) logger.info(f"Loaded {len(active_rules)} active trading rules from config file") except Exception as e: logger.error(f"Failed to load trading rules config: {e}") # 폴백: 기본 규칙 사용 active_rules = [ Rule( id=1, name="RSI 과매도 반등", description="RSI 30 이하 과매도 종목", rule_type="BUY", priority=1, conditions={"rsi_threshold": 30}, actions={ "take_profit": 5.0, "stop_loss": 3.0, "position_sizing": {"target_ratio": 0.15} }, is_active=True ) ] logger.warning(f"Using fallback rule: {len(active_rules)} rule(s)") logger.info(f"Found {len(active_rules)} active trading rules") # 오늘 날짜의 분석 데이터 조회 (중복 종목은 score 높은 것 우선) today = date.today() result = await db.execute( select(StockAnalysis) .where(StockAnalysis.analysis_date >= today) .order_by(StockAnalysis.score.desc()) ) raw_stocks = result.scalars().all() # 종목코드 기준 중복 제거 (score 내림차순 정렬 후 첫 번째만) seen = set() all_analyzed_stocks = [] for s in raw_stocks: if s.stock_code not in seen: seen.add(s.stock_code) all_analyzed_stocks.append(s) # DB에 데이터가 없으면 빈 결과 반환 if not all_analyzed_stocks or len(all_analyzed_stocks) < 1: logger.warning(f"No analysis data found ({len(all_analyzed_stocks) if all_analyzed_stocks else 0} stocks). Run batch analysis first.") return { "total": 0, "active_rules": len(active_rules), "analyzed_stocks": 0, "analyzed_stock_list": [], "recommendations": [], "recommendations_by_rule": [], "rules_applied": [], "message": "⚠️ 배치 분석 데이터 없음. 'make batch-test' 명령어로 분석 실행 필요." } logger.info(f"Found {len(all_analyzed_stocks)} stocks in DB for {today}") # 규칙별로 매칭되는 종목 찾기 rule_matches = [] recommendations_by_rule = [] stock_to_rules = {} # 종목코드 -> 매칭된 규칙 리스트 매핑 # 시장 레짐 판별 (섹션 7.1: 200일 MA 기반) market_regime_info = await check_market_regime() current_market_regime = market_regime_info.get("regime_200d", "bear") for rule in active_rules: logger.info(f"Applying rule: {rule.name}") conditions = rule.conditions actions = rule.actions rule_matched_stocks = [] for stock in all_analyzed_stocks: matches = True # === 전략 타입별 분기 === # 1) 평균회귀 전략 (RSI + 볼린저 반등) if rule.rule_type == "MEAN_REVERSION": matches = is_mean_reversion_candidate(stock, conditions, current_market_regime) # 2) 추세/골든크로스 전략 (나중에 별도 함수로 분리 예정) elif rule.rule_type == "TREND": # 여기서는 당장 비활성화하거나, 매우 보수적으로만 사용 # matches = is_trend_candidate(stock, conditions) matches = False # 일단 평균회귀 안정화 전까지 비활성화 # 3) 박스권 돌파 / 모멘텀 전략 (위험 축) elif rule.rule_type in ("BREAKOUT", "MOMENTUM"): # matches = is_breakout_or_momentum_candidate(stock, conditions) matches = False # 초기에는 끄고, 나중에 별도 엔진으로 # 기타 타입은 현재 스킵 else: matches = False if not matches: continue # 여기까지 왔으면 rule_matched_stocks에 추가 rule_matched_stocks.append(stock) # 점수순 정렬 rule_matched_stocks.sort(key=lambda x: x.score, reverse=True) # 이하 로직은 그대로 (Top N, rule_matches, stock_to_rules, 실시간 가격 업데이트 등) rule_stocks = [] for stock in rule_matched_stocks[:10]: stock_dict = { "stock_code": stock.stock_code, "stock_name": stock.stock_name, "current_price": int(stock.current_price) if stock.current_price else 0, "change_rate": round(float(stock.change_rate), 2) if stock.change_rate else 0, "volume": stock.volume, "rsi": round(float(stock.rsi), 2) if stock.rsi else None, "adx_14": round(float(stock.adx_14), 2) if stock.adx_14 else None, "ma5": int(stock.ma5) if stock.ma5 else None, "ma20": int(stock.ma20) if stock.ma20 else None, "stochastic_k": round(float(stock.stochastic_k), 2) if stock.stochastic_k else None, "signals": stock.signals or [], "score": stock.score, "ml_score": stock.ml_score, "ml_confidence": round(float(stock.ml_confidence), 4) if stock.ml_confidence else None, "ml_signal": stock.ml_signal, "price_date": stock.analysis_date.strftime("%Y%m%d"), "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_priority": rule.priority, "take_profit": actions.get("take_profit", 3.0), "stop_loss": actions.get("stop_loss", -1.5), "max_add_buys": actions.get("max_add_buys", 3), "add_buy_cooldown_min": actions.get("add_buy_cooldown_min", 15), } rule_stocks.append(stock_dict) rule_matches.append(stock_dict) if stock.stock_code not in stock_to_rules: stock_to_rules[stock.stock_code] = [] rule_exists = any(r["rule_id"] == rule.id for r in stock_to_rules[stock.stock_code]) if not rule_exists: stock_to_rules[stock.stock_code].append({ "rule_id": rule.id, "rule_name": rule.name, "priority": rule.priority, }) if rule_stocks: recommendations_by_rule.append({ "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_type": rule.rule_type, "priority": rule.priority, "matched_count": len(rule_matched_stocks), "stocks": rule_stocks, }) logger.info(f"Rule '{rule.name}' matched {len(rule_matched_stocks)} stocks, showing top {len(rule_stocks)}") # 규칙 매칭 결과가 없으면 분석 종목 상위 20개를 폴백으로 추가 if not rule_matches and all_analyzed_stocks: logger.info("No rule matches found. Using top scored analyzed stocks as fallback recommendations.") already_added = set() fallback_stocks = [] for stock in all_analyzed_stocks[:20]: if stock.stock_code in already_added: continue already_added.add(stock.stock_code) stock_dict = { "stock_code": stock.stock_code, "stock_name": stock.stock_name, "current_price": int(stock.current_price) if stock.current_price else 0, "change_rate": round(float(stock.change_rate), 2) if stock.change_rate else 0, "volume": stock.volume, "rsi": round(float(stock.rsi), 2) if stock.rsi else None, "adx_14": round(float(stock.adx_14), 2) if stock.adx_14 else None, "ma5": int(stock.ma5) if stock.ma5 else None, "ma20": int(stock.ma20) if stock.ma20 else None, "stochastic_k": round(float(stock.stochastic_k), 2) if stock.stochastic_k else None, "signals": stock.signals or [], "score": stock.score, "ml_score": stock.ml_score, "ml_confidence": round(float(stock.ml_confidence), 4) if stock.ml_confidence else None, "ml_signal": stock.ml_signal, "price_date": stock.analysis_date.strftime("%Y%m%d"), "rule_id": 0, "rule_name": "고점수 분석 종목", "rule_description": "매매 규칙 조건 미충족 - 분석 점수 기준 상위 종목", "rule_priority": 99, "take_profit": 3.0, "stop_loss": -1.5, "max_add_buys": 2, "add_buy_cooldown_min": 30, } fallback_stocks.append(stock_dict) rule_matches.append(stock_dict) if fallback_stocks: recommendations_by_rule.append({ "rule_id": 0, "rule_name": "고점수 분석 종목", "rule_description": "현재 매매 규칙 조건 미충족 - 분석 점수 기준 상위 종목", "rule_type": "ANALYSIS", "priority": 99, "matched_count": len(fallback_stocks), "stocks": fallback_stocks, }) logger.info(f"Added {len(fallback_stocks)} fallback stocks from analyzed_stock_list") return { "total": len(rule_matches), "active_rules": len(active_rules), "analyzed_stocks": len(all_analyzed_stocks), "analyzed_stock_list": [ { "stock_code": s.stock_code, "stock_name": s.stock_name, "current_price": int(s.current_price) if s.current_price else 0, "change_rate": round(float(s.change_rate), 2) if s.change_rate else 0, "rsi": round(float(s.rsi), 2) if s.rsi else None, "stochastic_k": round(float(s.stochastic_k), 2) if s.stochastic_k else None, "signals": s.signals or [], "score": s.score or 0, } for s in all_analyzed_stocks ], "recommendations": rule_matches, "recommendations_by_rule": recommendations_by_rule, "rules_applied": [{"id": r.id, "name": r.name, "priority": r.priority} for r in active_rules], } async def get_rule_based_recommendations_realtime(): """ 실시간 KIS API 분석 (느림 - 30초) 기존 방식 유지 """ # TODO: 데이터베이스 연결 후 실제 규칙 조회 # 현재는 샘플 규칙으로 동작 logger.info("Rule-based recommendations requested (realtime mode)") # 샘플 규칙 생성 (실제로는 DB에서 조회) from collections import namedtuple Rule = namedtuple('Rule', ['id', 'name', 'description', 'rule_type', 'priority', 'conditions', 'actions', 'is_active']) active_rules = [ Rule( id=1, name="RSI 과매도 반등", description="RSI 40 이하 과매도 종목", rule_type="BUY", priority=1, conditions={"rsi_threshold": 40}, actions={ "take_profit": 3.0, "stop_loss": -1.5, "position_sizing": {"target_ratio": 0.10}, "max_position_pct": 0.12, "max_add_buys": 3, "add_buy_cooldown_min": 15 }, is_active=True ), Rule( id=2, name="골든크로스 추세", description="5일선이 20일선 돌파", rule_type="BUY", priority=2, conditions={"sma_short": 5, "sma_long": 20}, actions={ "take_profit": 3.0, "stop_loss": -1.5, "position_sizing": {"target_ratio": 0.10}, "max_position_pct": 0.12, "max_add_buys": 3, "add_buy_cooldown_min": 15 }, is_active=True ), Rule( id=3, name="볼린저밴드 하단", description="볼린저밴드 하단 근처 반등", rule_type="BUY", priority=3, conditions={"bb_threshold": 0.2}, actions={ "take_profit": 3.0, "stop_loss": -1.5, "position_sizing": {"target_ratio": 0.10}, "max_position_pct": 0.12, "max_add_buys": 3, "add_buy_cooldown_min": 15 }, is_active=True ), ] # active_rules가 비어있으면 기본 지표로 분석 if not active_rules: logger.info("No rules in database, using default indicators") # 규칙이 없으면 기본 RSI + 이동평균으로 분석 if not active_rules: logger.info("Using default criteria: RSI + Moving Average") criteria = RecommendationCriteria( use_rsi=True, use_moving_average=True ) # 분석할 종목 리스트 all_stocks = await get_all_tradable_stocks() stocks_to_analyze = all_stocks[:20] # 20개 종목으로 단축 (속도 개선) # 배치 처리 (속도 개선) batch_size = 5 # 배치 크기 delay_between_batches = 0.3 # 딜레이 추가 감소 batch_results = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] tasks = [analyze_stock(code, name, criteria) for code, name in batch] results = await asyncio.gather(*tasks, return_exceptions=True) batch_results.extend(results) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) # 유효한 추천만 필터링 valid_recommendations = [ r for r in batch_results if r is not None and not isinstance(r, Exception) ] valid_recommendations.sort(key=lambda x: x.score, reverse=True) # 기본 규칙으로 그룹핑 default_stocks = [] for rec in valid_recommendations[:10]: stock_dict = { "stock_code": rec.stock_code, "stock_name": rec.stock_name, "current_price": rec.current_price, "change_rate": rec.change_rate, "signals": rec.signals, "score": rec.score, "price_date": rec.price_date, "rule_id": 0, "rule_name": "기본 추천", "rule_description": "RSI + 이동평균 기반 추천", "rule_priority": 1, "position_ratio": 0.15, "take_profit": 5.0, "stop_loss": 3.0 } default_stocks.append(stock_dict) return { "total": len(default_stocks), "active_rules": 0, "analyzed_stocks": len(stocks_to_analyze), "recommendations": default_stocks, "recommendations_by_rule": [{ "rule_id": 0, "rule_name": "기본 추천 (RSI + 이동평균)", "rule_description": "RSI 40 이하 과매도 + 이동평균 정배열 종목", "rule_type": "BUY", "priority": 1, "matched_count": len(valid_recommendations), "stocks": default_stocks }] if default_stocks else [], "rules_applied": [], "message": "기본 지표로 분석되었습니다 (규칙 미설정)" } logger.info(f"Found {len(active_rules)} active trading rules") # 분석할 종목 리스트 (타임아웃 방지를 위해 10개로 제한) all_stocks = await get_all_tradable_stocks() stocks_to_analyze = all_stocks[:10] # 상위 10개 종목으로 제한 # ✅ 최적화: 모든 종목을 한 번만 분석 (모든 지표 포함) logger.info(f"Analyzing {len(stocks_to_analyze)} stocks once with all indicators...") criteria = RecommendationCriteria( use_rsi=True, use_moving_average=True, use_macd=True, use_bollinger=True, use_stochastic=True ) # 배치 처리로 종목 분석 (Rate Limiting) batch_size = 1 # KIS API rate limit: 초당 1건 delay_between_batches = 1.5 # 배치 간 1.5초 대기 all_analyzed_stocks = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] tasks = [analyze_stock(code, name, criteria) for code, name in batch] results = await asyncio.gather(*tasks, return_exceptions=True) all_analyzed_stocks.extend(results) # 다음 배치 전 대기 (Rate Limiting) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) # 유효한 분석 결과만 필터링 valid_analyzed_stocks = [ r for r in all_analyzed_stocks if r is not None and not isinstance(r, Exception) ] logger.info(f"Successfully analyzed {len(valid_analyzed_stocks)} stocks") # 규칙별로 매칭되는 종목 찾기 (이미 분석된 데이터 재사용) rule_matches = [] recommendations_by_rule = [] for rule in active_rules: logger.info(f"Applying rule: {rule.name}") # 규칙 조건 파싱 conditions = rule.conditions actions = rule.actions # 이미 분석된 종목들 중에서 규칙에 맞는 종목 필터링 rule_matched_stocks = [] for stock in valid_analyzed_stocks: # ETF/레버리지/인버스/KODEX/선물 종목 제외 etf_keywords = ["ETF", "etf", "레버리지", "인버스", "KODEX", "선물"] if any(k in stock.stock_name for k in etf_keywords): continue # 규칙 조건 체크 matches = True # RSI 조건 체크 if "rsi_threshold" in conditions: rsi_threshold = conditions["rsi_threshold"] if not any(f"RSI" in signal for signal in stock.signals): matches = False # Trend filter: only allow long entries in uptrend if TREND_FILTER_ENABLED: try: t_ok = False if stock.ma20 is not None and stock.ma60 is not None: t_ok = (stock.ma20 > stock.ma60) and (stock.current_price > stock.ma20) else: # Fallback compute from price history is skipped here (already done in analyze_stock) t_ok = False if not t_ok: matches = False except Exception: matches = False if matches: rule_matched_stocks.append(stock) # 점수순 정렬 rule_matched_stocks.sort(key=lambda x: x.score, reverse=True) # 규칙별로 종목 저장 rule_stocks = [] for rec in rule_matched_stocks[:10]: # 규칙당 상위 10개 stock_with_rule = { **rec.model_dump(), "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_priority": rule.priority, "position_ratio": actions.get("position_sizing", {}).get("target_ratio", 0.10), "take_profit": actions.get("take_profit", 3.0), "stop_loss": actions.get("stop_loss", -1.5), "max_position_pct": actions.get("max_position_pct", 0.12), "max_add_buys": actions.get("max_add_buys", 3), "add_buy_cooldown_min": actions.get("add_buy_cooldown_min", 15) } rule_stocks.append(stock_with_rule) rule_matches.append(stock_with_rule) # 규칙별 그룹 추가 if rule_stocks: recommendations_by_rule.append({ "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_type": rule.rule_type, "priority": rule.priority, "matched_count": len(rule_matched_stocks), "stocks": rule_stocks }) logger.info(f"Rule '{rule.name}' matched {len(rule_matched_stocks)} stocks, showing top {len(rule_stocks)}") # 우선순위와 점수로 정렬 rule_matches.sort(key=lambda x: (x["rule_priority"], -x["score"])) # 실시간 현재가 업데이트 (상위 20개만) # 토큰 재요청을 줄이기 위해 공유 인스턴스를 사용합니다. from app.services.korea_investment_api import korea_investment_api as kis_api top_recommendations = rule_matches[:20] for rec in top_recommendations: try: await asyncio.sleep(0.2) # Rate limit 고려 price_data = await kis_api.get_current_price(rec["stock_code"]) if price_data and price_data.get("output"): output = price_data["output"] rec["current_price"] = int(output.get("stck_prpr", rec["current_price"])) rec["change_rate"] = float(output.get("prdy_ctrt", rec.get("change_rate", 0))) logger.info(f"Updated real-time price for {rec['stock_code']}: {rec['current_price']}원") except Exception as e: logger.warning(f"Failed to get real-time price for {rec['stock_code']}: {e}") # recommendations_by_rule의 종목들도 실시간 가격 업데이트 for rule_group in recommendations_by_rule: for stock in rule_group["stocks"]: # top_recommendations에서 업데이트된 가격 찾기 updated = next((r for r in top_recommendations if r["stock_code"] == stock["stock_code"]), None) if updated: stock["current_price"] = updated["current_price"] stock["change_rate"] = updated["change_rate"] # 실시간 분석 결과를 DB에 저장 from app.models.models import StockAnalysis from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import AsyncSessionLocal import datetime db = AsyncSessionLocal() today = datetime.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) for rec in top_recommendations: # 중복 저장 방지: 오늘자+종목코드 기준 exists = await db.execute( StockAnalysis.__table__.select().where( StockAnalysis.stock_code == rec["stock_code"], StockAnalysis.analysis_date == today ) ) if not exists.first(): stock_obj = StockAnalysis( stock_code=rec["stock_code"], stock_name=rec["stock_name"], analysis_date=today, current_price=rec["current_price"], change_rate=rec.get("change_rate", 0), volume=rec.get("volume", 0), rsi=rec.get("rsi", None), ma5=rec.get("ma5", None), ma20=rec.get("ma20", None), ma60=rec.get("ma60", None), stochastic_k=rec.get("stochastic_k", None), stochastic_d=rec.get("stochastic_d", None), bollinger_upper=rec.get("bollinger_upper", None), bollinger_lower=rec.get("bollinger_lower", None), bollinger_middle=rec.get("bollinger_middle", None), signals=rec.get("signals", []), score=rec.get("score", 0), ml_score=rec.get("ml_score"), ml_confidence=rec.get("ml_confidence"), ml_signal=rec.get("ml_signal"), ) db.add(stock_obj) try: await db.commit() finally: await db.close() return { "total": len(rule_matches), "active_rules": len(active_rules), "analyzed_stocks": len(stocks_to_analyze), "recommendations": top_recommendations, # 전체 통합 리스트 (실시간 가격 포함) "recommendations_by_rule": recommendations_by_rule, # 규칙별 그룹핑 (실시간 가격 포함) "rules_applied": [{"id": r.id, "name": r.name, "priority": r.priority} for r in active_rules] } async def _mcp_validate_buy(stock: dict) -> tuple: """MCP 서버에 AI 매수 검증 요청 Returns: (approved: bool, reason: str) - 오류/타임아웃 시 (True, "mcp_unavailable") — fail-open (매매 차단 안 함) """ if not settings.MCP_VALIDATION_ENABLED: return True, "mcp_disabled" try: payload = { "analysis_type": "RULE_VALIDATION", "data": { "rule": { "id": stock.get("rule_id"), "name": stock.get("rule_name", "auto_trade"), }, "market_data": { "stock_code": stock.get("stock_code"), "stock_name": stock.get("stock_name"), "current_price": stock.get("current_price"), "signals": stock.get("signals", []), "score": stock.get("score"), "ml_signal": stock.get("ml_signal"), "ml_confidence": stock.get("ml_confidence"), "ma20": stock.get("ma20"), "ma60": stock.get("ma60"), } } } async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json=payload ) resp.raise_for_status() data = resp.json() result = data.get("result", {}) decision = result.get("decision", "HOLD") confidence = float(result.get("confidence", 0.0)) logger.info( f"🤖 MCP 검증: {stock.get('stock_name')} → " f"decision={decision}, confidence={confidence:.2f}" ) if decision == "BUY" and confidence >= settings.MCP_CONFIDENCE_THRESHOLD: return True, f"mcp_approved(conf={confidence:.2f})" else: return False, f"mcp_rejected(decision={decision}, conf={confidence:.2f})" except Exception as e: logger.warning(f"⚠️ MCP 검증 실패 (스킵하고 계속): {e}") return True, "mcp_unavailable" # 장애 시 매매 차단 안 함 async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession): """백그라운드에서 자동매매 실행""" try: trade_results = [] total_processed = 0 skipped_held = 0 # 이미 보유 중인 종목 수 # 🐻 하락장 방어: 시장 레짐에 따라 매매 건수 제한 market_regime = None if settings.BEAR_MARKET_FILTER_ENABLED: market_regime = await check_market_regime() logger.info(f"🐻 자동매매 시장 상태: {market_regime['message']}") regime_max_trades = market_regime.get("max_trades", 20) if market_regime else 20 position_multiplier = market_regime.get("position_multiplier", 1.0) if market_regime else 1.0 max_trades = min(20, regime_max_trades) # 현재 보유 중인 종목 조회 from app.models.models import Portfolio result = await db.execute( select(Portfolio).where(Portfolio.quantity > 0) ) held_stocks = {p.stock_code for p in result.scalars().all()} if held_stocks: logger.info(f"현재 보유 중: {len(held_stocks)}개 종목") # ✅ 중복 제거: 모든 규칙 그룹의 종목을 합쳐서 중복 제거 unique_stocks = {} for rule_group in recommendations_response["recommendations_by_rule"]: for stock in rule_group["stocks"]: stock_code = stock.get("stock_code") # 우선순위가 높은 규칙의 종목을 우선 (priority가 낮을수록 우선순위 높음) if stock_code not in unique_stocks: unique_stocks[stock_code] = stock elif stock.get("rule_priority", 999) < unique_stocks[stock_code].get("rule_priority", 999): unique_stocks[stock_code] = stock # 우선순위와 점수로 정렬 sorted_stocks = sorted( unique_stocks.values(), key=lambda s: (s.get("rule_priority", 999), -s.get("score", 0)) ) logger.info(f"📊 자동매매 대상: 총 {len(sorted_stocks)}개 종목 (중복 제거 완료)") for stock in sorted_stocks[:max_trades]: # 🔄 이미 보유 중인 종목도 추가 매수 허용 (일일 2회 제한 내에서) # if stock.get("stock_code") in held_stocks: # skipped_held += 1 # logger.debug(f"⏭️ {stock.get('stock_name')} - 이미 보유중, 건너뜀") # continue try: # 시장 레짐 정보 전달 (포지션 사이징에서 regime 복원에 사용) stock["_position_multiplier"] = position_multiplier stock["_regime_str"] = market_regime.get("regime_200d", "bear") if market_regime else "bear" result = await execute_auto_trade(stock, db) total_processed += 1 if result["action"] in ["BUY", "SELL"]: trade_results.append(result) logger.info(f"✅ {result['action']} {result['stock']} - {result.get('quantity')}주") # Rate Limiting: KIS API 호출 간 0.5초 대기 await asyncio.sleep(0.5) except Exception as e: logger.error(f"Error executing auto-trade for {stock.get('stock_code')}: {e}") continue logger.info(f"🎯 자동매매 완료: {len(trade_results)}건 실행 / {total_processed}개 분석 / {skipped_held}개 보유중") except Exception as e: logger.error(f"❌ Background auto-trade error: {e}") async def _background_sync_portfolio(): """백그라운드 포트폴리오 동기화 (새 세션 사용)""" try: from app.api.endpoints.portfolio import sync_portfolio async with AsyncSessionLocal() as session: await sync_portfolio(session, force=True) except Exception as e: logger.error(f"Background portfolio sync failed: {e}") async def _delayed_sync_portfolio(delay_seconds: int = 3): """지연 포트폴리오 동기화 (KIS API 업데이트 대기)""" try: await asyncio.sleep(delay_seconds) from app.api.endpoints.portfolio import sync_portfolio from app.core.database import AsyncSessionLocal async with AsyncSessionLocal() as session: try: await sync_portfolio(session, force=True) logger.info(f"🔄 Delayed portfolio sync completed after {delay_seconds}s") except Exception as inner_e: logger.error(f"Sync execution failed: {inner_e}") except Exception as e: logger.error(f"Delayed portfolio sync failed: {e}") # 자동 매매 실행 함수 async def execute_auto_trade(stock: dict, db: AsyncSession): """ 추천 종목에 대해 자동 매매 실행 (국내 + 해외 주식) - 매수 신호: 정배열, RSI < 30 등 - 매도 신호: 역배열, RSI > 70, 손익률 기준 """ try: stock_code = stock.get("stock_code") stock_name = stock.get("stock_name") current_price = float(stock.get("current_price", 0)) signals = stock.get("signals", []) if not stock_code or current_price <= 0: logger.debug(f"Invalid stock data for auto trade: {stock}") return {"action": "SKIP", "reason": "invalid_data"} # 국내/해외 구분 (6자리 숫자면 국내, 알파벳 포함이면 해외) is_domestic = stock_code.isdigit() and len(stock_code) == 6 exchange_code = stock.get("exchange_code", "NASD") # 기본값: 나스닥 # 가격 필터 (CLAUDE.md 섹션 6.4) if is_domestic: if current_price < settings.MIN_PRICE_KRW: logger.info(f"가격 필터: {stock_name} ({current_price:,.0f}원 < {settings.MIN_PRICE_KRW:,}원)") return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} else: if current_price < settings.MIN_PRICE_USD: logger.info(f"가격 필터: {stock_name} (${current_price:.2f} < ${settings.MIN_PRICE_USD:.2f})") return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} # MDD 체크 (섹션 6.3) — async lock으로 thread-safe if not await mdd_manager.can_open_new_position(): logger.info(f"MDD 제한: 신규 진입 중단 상태 - {stock_name} 스킵") return {"action": "SKIP", "reason": "mdd_pause", "stock": stock_name} # 연속 손실 금지 체크 (섹션 7.4) if loss_tracker.is_banned(stock_code): ban_info = loss_tracker.get_ban_info(stock_code) logger.info(f"연속 손실 금지: {stock_name} ({ban_info})") return {"action": "SKIP", "reason": "consecutive_loss_ban", "stock": stock_name} # 보유 종목 수 체크 (섹션 6.1: 최대 20종목) held_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) held_count = len(held_result.scalars().all()) if held_count >= settings.MAX_HOLDINGS_COUNT: logger.info(f"보유 종목 한도: {held_count}/{settings.MAX_HOLDINGS_COUNT} - {stock_name} 스킵") return {"action": "SKIP", "reason": "max_holdings_reached", "stock": stock_name} # 0. 매도 후 재진입 금지 (Cooldown) - 최근 24시간 내 매도 이력 확인 # 같은 날 손절친 종목을 다시 사는 것을 방지 from app.models.models import TradeHistory cutoff_time = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) recent_sell = await db.execute( select(TradeHistory).where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "SELL", TradeHistory.status == "EXECUTED", TradeHistory.executed_at >= cutoff_time ) ) if recent_sell.first(): logger.info(f"🚫 Cooldown active: {stock_name}({stock_code}) was sold today. Skipping BUY.") return {"action": "SKIP", "reason": "cooldown_sell_today", "stock": stock_name} # 이미 보유 중인지 확인 result = await db.execute( select(Portfolio).where( Portfolio.stock_code == stock_code, Portfolio.quantity > 0 ) ) holding = result.scalar_one_or_none() # 신호 문자열 기반 매수/매도 판단 (실제 분석 함수의 문구와 일치시킴) text_signals = " ".join(signals) # 점수 숫자형 보정 try: score_val = float(stock.get("score", 0) or 0) except Exception: score_val = 0.0 buy_keywords = [ "정배열 눌림목 매수", # "정배열 눌림목 매수 (이격도 안정)" "눌림목 진입 매수타점", # "눌림목 진입 매수타점" "RSI 바닥 탈출 반등", # "RSI 바닥 탈출 반등 (XX)" "RSI 안정 상승", # "RSI 안정 상승 (XX)" "스토캐스틱 과매도", # "스토캐스틱 과매도 (%K: XX)" "5일선 > 20일선", "20일선 > 60일선", "골든크로스", # "MACD 골든크로스" "볼린저 하한선 터치", "횡보장 — 평균회귀 유리", # "ADX(XX<20) 횡보장 — 평균회귀 유리" "거래량 증가와 상승", ] should_buy = any(k in text_signals for k in buy_keywords) sell_keywords = [ "역배열 하락 매도타점", "RSI 과매수", "데드크로스", "이동평균 역배열", "볼린저 상한선 돌파", "손절라인 이탈", ] should_sell = any(k in text_signals for k in sell_keywords) # 🤖 ML 앙상블 기반 매수/매도 판단 (설정 활성화 시) ml_score_trade = stock.get("ml_score") ml_confidence_trade = stock.get("ml_confidence") ml_signal_trade = stock.get("ml_signal") if settings.ML_ENSEMBLE_ENABLED and ml_score_trade is not None: # ML 신호가 BUY이고 신뢰도/스코어 충분하면 매수 if (ml_signal_trade == "BUY" and ml_score_trade >= settings.ML_ENSEMBLE_BUY_THRESHOLD and (ml_confidence_trade or 0) >= settings.ML_ENSEMBLE_CONFIDENCE_MIN): should_buy = True logger.info(f"🤖 ML 매수 신호: {stock_name} (score={ml_score_trade}, conf={ml_confidence_trade:.2f})") # ML 신호가 SELL이면 매도 강화 elif ml_signal_trade == "SELL" and (ml_confidence_trade or 0) >= 0.5: should_sell = True logger.info(f"🤖 ML 매도 신호: {stock_name} (score={ml_score_trade}, conf={ml_confidence_trade:.2f})") # ML 신뢰도가 낮으면 매수 억제 elif ml_signal_trade == "HOLD" and (ml_confidence_trade or 0) < settings.ML_ENSEMBLE_CONFIDENCE_MIN: should_buy = False logger.info(f"🤖 ML 관망 신호로 매수 억제: {stock_name}") # ✅ 규칙 기반 추천 종목도 안전장치 적용 (이격도 과열, 급등 체크) has_rule = stock.get("rule_id") is not None or stock.get("rule_name") is not None if has_rule: # 이격도 과열 시 매수 금지 if "이격도 과열" in text_signals or "매수금지" in text_signals: logger.info(f"🚫 규칙 추천이지만 이격도 과열로 매수 금지: {stock_name}") return {"action": "HOLD", "stock": stock_name, "reason": "disparity_overheated"} # 급등 종목 추격 금지 체크 if "급등 종목 추격 금지" in text_signals or "급등과열" in text_signals: logger.info(f"🚫 규칙 추천이지만 급등/과열 금지: {stock_name}") return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} should_buy = True logger.info(f"📋 규칙 기반 추천: {stock_name} - {stock.get('rule_name', '매수규칙')}") # 매수 로직 (신호 또는 점수 기반) # 추가매수 허용: 보유 중이어도 비중 한도 내에서 추매 가능 if (should_buy or (score_val >= 1)): # 🚫 추격매수 방지: 급등 종목 필터 try: if is_domestic: chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10) chart_data = chart_resp.get("output2", []) else: chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") chart_data = chart_resp.get("output2", []) if len(chart_data) >= 4: # 국내/해외 데이터 키 차이 처리 if is_domestic: closes = [float(item.get("stck_clpr", 0)) for item in chart_data[-4:]] else: closes = [float(item.get("clos", 0)) for item in chart_data[-4:]] if len(closes) >= 4: price_3days_ago = closes[0] recent_gain = ((current_price - price_3days_ago) / price_3days_ago) * 100 if price_3days_ago > 0 else 0 if recent_gain > 10: # 3일간 10% 이상 급등 logger.info(f"🚫 급등 종목 추격 금지: {stock_name} (3일 +{recent_gain:.1f}%)") return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} except Exception as e: logger.warning(f"⚠️ 급등 체크 실패, 계속 진행: {e}") # Enforce trend filter for buys if TREND_FILTER_ENABLED: try: ma20 = stock.get("ma20") ma60 = stock.get("ma60") # If moving averages not provided, fetch quick chart to compute if ma20 is None or ma60 is None: if is_domestic: chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D") closes_local = [float(it.get("stck_clpr", 0)) for it in (chart_resp.get("output") or [])] else: chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") closes_local = [float(it.get("clos", 0)) for it in (chart_resp.get("output2") or [])] if len(closes_local) >= max(TREND_MA_SHORT, TREND_MA_LONG): ma20 = _ma(closes_local, TREND_MA_SHORT) ma60 = _ma(closes_local, TREND_MA_LONG) # 허용 오차 적용 if not (ma20 and ma60 and (float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT)) and (current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT))): logger.info(f"🚫 Trend filter rejected BUY: {stock_name} (price={current_price}, ma20={ma20}, ma60={ma60}, tol={TREND_TOLERANCE_PCT})") return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"} except Exception as e: logger.info(f"⚠️ Trend filter check failed, skipping buy as safe default: {e}") return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"} # 포트폴리오 요약 정보 try: from app.api.endpoints.portfolio import portfolio_cache available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0 total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0 except Exception: available_cash, total_assets = 0.0, 0.0 # 🔥 하루 매매 횟수 제한 (당일 0시부터 카운트) daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) daily_buy_check = await db.execute( select(TradeHistory) .where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "BUY", TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), # COMPLETED도 포함 TradeHistory.created_at >= daily_cutoff, ) ) daily_buys = daily_buy_check.scalars().all() daily_buy_count = len(daily_buys) # 🚫 하루 최대 2회 매매 제한 (신규 + 추가) max_daily_trades = 2 if daily_buy_count >= max_daily_trades: logger.info(f"🚫 일일 매매 한도 초과: {stock_name} 당일 {daily_buy_count}/{max_daily_trades}회") return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"} cooldown_min = int(stock.get("add_buy_cooldown_min", 30) or 30) # 추가매수 쿨다운(분) # 🔥 쿨다운 체크 (보유 여부와 관계없이 항상 체크) # 포트폴리오 동기화가 느릴 수 있으므로, DB 거래 내역 기준으로 체크 cooldown_check = await db.execute( select(TradeHistory) .where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "BUY", TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), TradeHistory.created_at >= daily_cutoff, ) .order_by(desc(TradeHistory.created_at)) ) recent_buys = cooldown_check.scalars().all() last_buy_at = recent_buys[0].created_at if recent_buys else None # 쿨다운 체크 if last_buy_at and (_get_kst_now() - last_buy_at).total_seconds() < cooldown_min * 60: logger.info(f"⏳ 추가매수 대기: {stock_name} {cooldown_min}분 쿨다운 미경과") return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} # 목표/최대 비중 기반 투자금 계산 (이전에 최근 매수 기록 체크) # 보유중이 아니더라도(아직 싱크 안됨) 최근 매수 시도가 있으면 스킵 recent_buy_cutoff = _get_kst_now() - timedelta(minutes=5) check_recent = await db.execute( select(TradeHistory).where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "BUY", TradeHistory.created_at >= recent_buy_cutoff, TradeHistory.status.in_(['EXECUTED', 'COMPLETED', 'PENDING']) ) ) if check_recent.first(): logger.info(f"⏳ 최근 5분 내 매수 이력 존재: {stock_name} - 스킵") return {"action": "SKIP", "stock": stock_name, "reason": "recent_buy_5min"} # ✅ MCP AI 검증 (BUY 최종 확인 - 모든 사전 필터 통과 후) mcp_ok, mcp_reason = await _mcp_validate_buy(stock) if not mcp_ok: logger.info(f"🚫 MCP AI 매수 거부: {stock_name} ({mcp_reason})") return {"action": "HOLD", "stock": stock_name, "reason": mcp_reason} # 동적 포지션 사이징 (CLAUDE.md 섹션 6.1: 종목당 최대 4%, regime/RSI/ADX 반영) current_value = float(holding.quantity * current_price) if holding else 0.0 regime_str = stock.get("_regime_str", "bear") regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR invest_amount = calculate_position_size_v2( total_assets=total_assets, available_cash=available_cash, current_strategy_allocation=current_value, regime=regime, rsi=stock.get("rsi"), adx=stock.get("adx_14"), ) if invest_amount <= 0 and total_assets <= 0: # 총자산 미확정 시 보수적으로 현금의 4% invest_amount = available_cash * 0.04 quantity = int(invest_amount / current_price) logger.info(f"💰 {stock_name}({stock_code}): 예수금={available_cash:,.0f}원, 투자금액={invest_amount:,.0f}원, 수량={quantity}주, 가격={current_price:,.0f}원") if quantity > 0: logger.info(f"🟢 매수 신호 감지: {stock_name}({stock_code}) - {signals}") # KIS API로 매수 주문 (지정가: 현재가로 주문) try: if is_domestic: order_result = await korea_investment_api.place_order( stock_code=stock_code, order_type="BUY", quantity=quantity, price=int(current_price) # 지정가 주문 (현재가) ) else: order_result = await korea_investment_api.place_overseas_order( symbol=stock_code, order_type="BUY", quantity=quantity, price=float(current_price), # 해외 주식은 달러 단위 exchange_code=exchange_code ) # 주문 결과 상세 로깅 logger.info(f"📝 매수 주문 결과: {stock_name}({stock_code}) - rt_cd={order_result.get('rt_cd')}, msg={order_result.get('msg1')}, odno={order_result.get('odno')}") logger.debug(f"Full order result: {order_result}") # 중복 거래 방지: 최근 10초 이내 동일 종목 동일 타입 주문 확인 recent_cutoff = _get_kst_now() - timedelta(seconds=10) duplicate_check = await db.execute( select(TradeHistory).where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "BUY", TradeHistory.created_at >= recent_cutoff ) ) recent_duplicate = duplicate_check.scalar_one_or_none() if recent_duplicate: logger.warning(f"⚠️ 중복 매수 방지: {stock_name} - 10초 이내 동일 주문 존재") return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} # 거래 이력 저장 (상세 정보 포함) total_amount = Decimal(str(current_price)) * quantity entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점" success = order_result.get("rt_cd") == "0" error_msg = order_result.get("msg1", "") if not success else "" # 주문 상태 결정 if success: status = "COMPLETED" # 체결 완료 elif order_result.get("rt_cd") == "1": status = "PENDING" # 주문 접수됨, 체결 대기 else: status = "FAILED" # 주문 실패 trade = TradeHistory( rule_id=stock.get("rule_id"), stock_code=stock_code, stock_name=stock_name, order_type="BUY", order_price=Decimal(str(current_price)), order_quantity=quantity, executed_price=Decimal(str(current_price)) if success else None, executed_quantity=quantity if success else 0, total_amount=total_amount, status=status, order_id=order_result.get("odno"), message=f"자동매수: {'추가매수' if holding else '신규'} | {' | '.join(signals[:2])} | {error_msg if not success else '체결완료'}", entry_reason=entry_reason, signals=signals, created_at=_get_kst_now(), executed_at=_get_kst_now() if success else None ) db.add(trade) await db.commit() if success: logger.info(f"✅ 매수 체결 완료: {stock_name} {quantity}주 @ {current_price}원") elif status == "PENDING": logger.info(f"⏳ 매수 주문 접수: {stock_name} {quantity}주 @ {current_price}원 - 체결 대기중") else: logger.warning(f"❌ 매수 주문 실패: {stock_name} {quantity}주 - {error_msg}") # Optimistic local 포트폴리오 업데이트 if success: if holding: # 가중 평균 재계산 new_total_qty = holding.quantity + quantity prev_cost = float(holding.average_price or 0) * holding.quantity new_cost = prev_cost + current_price * quantity holding.quantity = new_total_qty holding.average_price = Decimal(str(new_cost / new_total_qty)) holding.current_price = Decimal(str(current_price)) else: new_holding = Portfolio( stock_code=stock_code, stock_name=stock_name, quantity=quantity, average_price=Decimal(str(current_price)), current_price=Decimal(str(current_price)), profit_loss=Decimal("0"), profit_loss_rate=Decimal("0") ) db.add(new_holding) await db.commit() # 포트폴리오 캐시 즉시 업데이트 try: from app.api.endpoints.portfolio import portfolio_cache # 캐시 완전 초기화 - 다음 조회시 강제로 KIS API 호출 portfolio_cache.holdings = [] portfolio_cache.summary = None portfolio_cache.last_synced = None portfolio_cache.last_trade_id = None portfolio_cache.change_flag = True logger.info(f"🔄 Portfolio cache CLEARED after BUY {stock_name}({stock_code})") except Exception as e: logger.error(f"Portfolio cache clear failed after BUY: {e}") return {"action": "BUY", "stock": stock_name, "quantity": quantity, "price": current_price, "success": success} except Exception as e: logger.error(f"❌ 매수 주문 실패: {stock_name} - {e}") return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)} else: logger.debug(f"⚠️ {stock_name}({stock_code}): 매수 수량 0 (가격={current_price:,.0f}원, 예수금={available_cash:,.0f}원)") return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name} # 매도 로직 elif should_sell and holding: quantity = holding.quantity profit_loss_rate = float(holding.profit_loss_rate or 0) # 익절/손절 조건 확인 take_profit = stock.get("take_profit", 3.0) # 3% 익절(권장) stop_loss = stock.get("stop_loss", -1.5) # -1.5% 손절(권장) should_execute_sell = ( profit_loss_rate >= take_profit or # 익절 profit_loss_rate <= stop_loss or # 손절 should_sell # 신호 기반 매도 ) if should_execute_sell: logger.info(f"🔴 매도 신호 감지: {stock_name}({stock_code}) - 손익률: {profit_loss_rate:.2f}% - {signals}") try: if is_domestic: order_result = await korea_investment_api.place_order( stock_code=stock_code, order_type="SELL", quantity=quantity, price=int(current_price) # 지정가 주문 (현재가) ) else: order_result = await korea_investment_api.place_overseas_order( symbol=stock_code, order_type="SELL", quantity=quantity, price=float(current_price), # 해외 주식은 달러 단위 exchange_code=exchange_code ) # 주문 결과 상세 로깅 logger.info(f"📝 매도 주문 결과: {stock_name}({stock_code}) - rt_cd={order_result.get('rt_cd')}, msg={order_result.get('msg1')}, odno={order_result.get('odno')}") logger.debug(f"Full order result: {order_result}") # 중복 거래 방지: 최근 10초 이내 동일 종목 동일 타입 주문 확인 recent_cutoff = _get_kst_now() - timedelta(seconds=10) duplicate_check = await db.execute( select(TradeHistory).where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "SELL", TradeHistory.created_at >= recent_cutoff ) ) recent_duplicate = duplicate_check.scalar_one_or_none() if recent_duplicate: logger.warning(f"⚠️ 중복 매도 방지: {stock_name} - 10초 이내 동일 주문 존재") return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} # 거래 이력 저장 (상세 정보 포함) total_amount = Decimal(str(current_price)) * quantity avg_price = float(holding.average_price or current_price) profit_loss = (current_price - avg_price) * quantity # 매도 이유 상세 작성 exit_reasons = [] if profit_loss_rate >= take_profit: exit_reasons.append(f"익절 달성 ({profit_loss_rate:.2f}% >= {take_profit}%)") if profit_loss_rate <= stop_loss: exit_reasons.append(f"손절 실행 ({profit_loss_rate:.2f}% <= {stop_loss}%)") if should_sell: exit_reasons.append(f"매도 신호: {', '.join(signals[:2])}") exit_reason = ". ".join(exit_reasons) + f". 평균단가: {avg_price:,.0f}원, 매도가: {current_price:,.0f}원, 손익: {profit_loss:,.0f}원" success = order_result.get("rt_cd") == "0" error_msg = order_result.get("msg1", "") if not success else "" # 주문 상태 결정 if success: status = "COMPLETED" # 체결 완료 elif order_result.get("rt_cd") == "1": status = "PENDING" # 주문 접수됨, 체결 대기 else: status = "FAILED" # 주문 실패 trade = TradeHistory( rule_id=stock.get("rule_id"), stock_code=stock_code, stock_name=stock_name, order_type="SELL", order_price=Decimal(str(current_price)), order_quantity=quantity, executed_price=Decimal(str(current_price)) if success else None, executed_quantity=quantity if success else 0, total_amount=total_amount, status=status, order_id=order_result.get("odno"), message=f"자동매도: 손익률 {profit_loss_rate:.2f}% | {' | '.join(signals[:2])} | {error_msg if not success else '체결완료'}", exit_reason=exit_reason, signals=signals, net_profit_loss=Decimal(str(profit_loss)) if success else None, profit_loss_rate=Decimal(str(profit_loss_rate)) if success else None, created_at=_get_kst_now(), executed_at=_get_kst_now() if success else None ) db.add(trade) await db.commit() if success: logger.info(f"✅ 매도 주문 성공: {stock_name} {quantity}주 @ {current_price}원") else: logger.warning(f"❌ 매도 주문 실패: {stock_name} {quantity}주 - {error_msg}") if success: # 포트폴리오에서 완전 삭제 await db.delete(holding) await db.commit() logger.info(f"🗑️ Deleted {stock_name}({stock_code}) from portfolio DB") # 포트폴리오 캐시 완전 초기화 try: from app.api.endpoints.portfolio import portfolio_cache # 캐시 완전 초기화 - 다음 조회시 강제로 KIS API 호출 portfolio_cache.holdings = [] portfolio_cache.summary = None portfolio_cache.last_synced = None portfolio_cache.last_trade_id = None portfolio_cache.change_flag = True logger.info(f"🔄 Portfolio cache CLEARED after SELL {stock_name}({stock_code})") except Exception as e: logger.error(f"Portfolio cache clear failed after SELL: {e}") return {"action": "SELL", "stock": stock_name, "quantity": quantity, "price": current_price, "profit_loss": profit_loss_rate, "success": success} except Exception as e: logger.error(f"❌ 매도 주문 실패: {stock_name} - {e}") return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)} # 아무 것도 실행 안됨 logger.info( f"⏸️ HOLD: {stock_name}({stock_code}) - 매수신호={should_buy}, 매도신호={should_sell}, 보유중={bool(holding)}, 점수={score_val}" ) return {"action": "HOLD", "stock": stock_name} except Exception as e: logger.error(f"❌ 자동 매매 오류: {e}") return {"action": "ERROR", "error": str(e)} @router.post("/auto-trade") async def execute_auto_trading(db: AsyncSession = Depends(get_db)): """ 규칙 기반 추천 종목에 대해 자동 매매 실행 """ try: # 규칙 기반 추천 조회 (실시간 기준으로 매매 판단 일치화) # DB 캐시의 신호 문자열 불일치로 매수가 안 걸리는 문제를 방지하기 위해 # 실시간 분석 결과를 우선 사용합니다. recommendations_response = await get_rule_based_recommendations(realtime=True) if not recommendations_response.get("recommendations_by_rule"): return {"message": "추천 종목이 없습니다", "trades": []} # 각 규칙별 추천 종목에 대해 자동 매매 실행 trade_results = [] total_processed = 0 max_trades = 5 # 한 번에 최대 5개만 처리 for rule_group in recommendations_response["recommendations_by_rule"]: if total_processed >= max_trades: break for stock in rule_group["stocks"][:2]: # 규칙당 상위 2개만 if total_processed >= max_trades: break try: result = await execute_auto_trade(stock, db) total_processed += 1 if result["action"] in ["BUY", "SELL"]: trade_results.append(result) # Rate Limiting: KIS API 호출 간 0.5초 대기 await asyncio.sleep(0.5) except Exception as e: logger.error(f"Error executing auto-trade for {stock.get('stock_code')}: {e}") continue return { "message": f"자동 매매 완료: {len(trade_results)}건 실행 / {total_processed}개 분석", "trades": trade_results, "total_analyzed": total_processed } except Exception as e: logger.error(f"❌ 자동 매매 실행 오류: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/ai/{stock_code}") async def get_ai_recommendation(stock_code: str): """MCP 서버를 통한 AI 종목 분석 및 뉴스 조회""" try: # 1. 시장 데이터 조회 (Backend에서 직접 KIS API 호출) price_data = {} try: price_data = await korea_investment_api.get_current_price(stock_code) except Exception as e: logger.warning(f"Failed to fetch price data locally: {e}") # 2. MCP 서버에 뉴스 조회 요청 news_data = [] try: async with httpx.AsyncClient() as client: mcp_url = f"{settings.MCP_SERVER_URL}/tools/call" payload = { "tool_name": "get_news", "parameters": {"stock_code": stock_code, "limit": 3} } response = await client.post(mcp_url, json=payload, timeout=10.0) if response.status_code == 200: result = response.json() if result.get("status") == "success": # MCP might wrap it in 'result' -> 'news' or just 'news' data = result.get("result", {}) if "news" in data: news_data = data["news"] else: news_data = data # Fallback if direct list except Exception as e: logger.warning(f"Failed to fetch news via MCP: {e}") # 3. AI 분석 요청 (MCP Analyze) analysis_result = "AI 분석을 수행할 수 없습니다." try: async with httpx.AsyncClient() as client: analyze_url = f"{settings.MCP_SERVER_URL}/analyze" # 시장 데이터 단순화 market_summary = {} if "output" in price_data: output = price_data["output"] market_summary = { "current_price": output.get("stck_prpr"), "change_rate": output.get("prdy_ctrt"), "volume": output.get("acml_vol"), "high": output.get("stck_hgpr"), "low": output.get("stck_lwpr") } payload = { "analysis_type": "MARKET_ANALYSIS", "data": { "stock_code": stock_code, "market_data": market_summary, "news": news_data } } response = await client.post(analyze_url, json=payload, timeout=45.0) # Analysis takes time if response.status_code == 200: analysis_res = response.json() # If successful, returns {"analysis_type":..., "result": "Markdown text/JSON"} if "result" in analysis_res: analysis_result = analysis_res["result"] except Exception as e: logger.error(f"Failed to fetch AI analysis: {e}") return { "stock_code": stock_code, "analysis": analysis_result, "news": news_data, "market_data": price_data.get("output", {}) } except Exception as e: logger.error(f"AI Recommendation error: {e}") raise HTTPException(status_code=500, detail=str(e))