""" 시세 데이터 API 엔드포인트 - GET /price/{stock_code}: 실시간 현재가 조회 - GET /chart/{stock_code}: 차트 데이터 (일봉/주봉/월봉) - GET /orderbook/{stock_code}: 호가 정보 조회 - GET /ai-analysis/{stock_code}: AI 기반 종목 분석 """ # 필요한 라이브러리 및 모듈 임포트 from datetime import datetime, timedelta import random import logging import pytz _KST = pytz.timezone("Asia/Seoul") from fastapi import APIRouter, HTTPException, Query, Depends from typing import List, Dict from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select # 내부 모듈 임포트 from app.core.config import settings from app.core.database import get_db from app.services.korea_investment_api import korea_investment_api # APIRouter 인스턴스 생성 router = APIRouter() # 로거 설정 logger = logging.getLogger(__name__) # 주식 가격 응답을 위한 Pydantic 모델 정의 class StockPriceResponse(BaseModel): stock_code: str current_price: float change_rate: float volume: int high_price: float low_price: float open_price: float @router.get("/price/{stock_code}") async def get_stock_price(stock_code: str): """주식 현재가 조회 엔드포인트""" try: # 한국투자증권 API를 통해 현재가 데이터 조회 data = await korea_investment_api.get_current_price(stock_code) # API 응답에 'output' 필드가 없는 경우 예외 처리 if "output" not in data: raise HTTPException(status_code=404, detail="Stock data not found") output = data["output"] # 필요한 데이터를 추출하여 딕셔너리 형태로 반환 return { "stock_code": stock_code, "current_price": float(output.get("stck_prpr", 0)), "change_rate": float(output.get("prdy_ctrt", 0)), "volume": int(output.get("acml_vol", 0)), "high_price": float(output.get("stck_hgpr", 0)), "low_price": float(output.get("stck_lwpr", 0)), "open_price": float(output.get("stck_oprc", 0)) } except Exception as e: # 예외 발생 시 500 에러와 함께 상세 내용 반환 raise HTTPException(status_code=500, detail=f"Failed to fetch stock price: {str(e)}") @router.get("/prices") async def get_stock_prices(codes: str): """여러 종목의 현재가를 일괄 조회하는 엔드포인트 Query: - codes: 콤마로 구분된 종목코드들 (예: 005930,000660,035420) """ try: # codes 파라미터가 없는 경우 빈 데이터 반환 if not codes: return {"prices": {}, "count": 0, "timestamp": datetime.now().isoformat()} # 콤마로 구분된 종목 코드 문자열을 리스트로 변환 code_list: List[str] = [c.strip() for c in codes.split(",") if c.strip()] prices: Dict[str, float] = {} # 각 종목 코드에 대해 순차적으로 현재가 조회 # 클래스 내부에 API 호출 간격을 조절하는 쓰로틀링 기능이 내장되어 있음 for code in code_list: try: data = await korea_investment_api.get_current_price(code) if data and data.get("output"): o = data["output"] # 콤마가 포함된 가격 문자열을 float으로 변환 current_price = float(str(o.get("stck_prpr", "0")).replace(",", "")) prices[code] = current_price except Exception as e: # 특정 종목 조회 실패 시 경고 로그 기록 logger.warning(f"Failed to fetch price for {code}: {e}") # 조회 결과를 딕셔너리 형태로 반환 return { "prices": prices, "count": len(prices), "requested": len(code_list), "timestamp": datetime.now().isoformat(), } except Exception as e: # 예외 발생 시 500 에러와 함께 상세 내용 반환 raise HTTPException(status_code=500, detail=f"Failed to fetch prices: {str(e)}") def _generate_sample_chart(stock_code: str, period: str, limit: int = 100) -> dict: """로컬 개발 환경 및 API 키가 없을 때 사용할 샘플 차트 데이터를 생성하는 함수""" # 유효한 주기(period)인지 확인, 아니면 'D'(일봉)으로 설정 if period not in {"D", "W", "M"}: period = "D" # 주기에 따라 데이터 포인트 수와 날짜 간격 설정 if period == "D": points = min(limit, 240) # 일봉은 최대 240개 step = 1 elif period == "W": points = min(limit, 240) # 주봉도 최대 240개 step = 7 else: # "M" (월봉) points = min(limit, 240) # 월봉도 최대 240개 step = 30 # 차트 데이터의 시작 날짜 계산 today = datetime.now() start_date = today - timedelta(days=(points - 1) * step) # 종목코드와 주기에 기반하여 일관된 랜덤 데이터 생성을 위한 시드 설정 rng = random.Random(f"{stock_code}:{period}") # 랜덤 가격 및 변동성 파라미터 초기화 base_price = rng.randint(30000, 90000) drift = rng.uniform(-0.04, 0.06) volatility = rng.uniform(0.004, 0.018) volume_base = rng.randint(80000, 180000) # 가격을 10단위로 반올림하는 헬퍼 함수 def round_price(value: float) -> int: return max(1000, int(round(value / 10)) * 10) price = float(base_price) chart_data = [] # 설정된 포인트 수만큼 반복하여 캔들 데이터 생성 for idx in range(points): current_date = start_date + timedelta(days=idx * step) # 일일 가격 변동률 계산 daily_change = drift / points + rng.uniform(-volatility, volatility) # 시가, 종가, 고가, 저가 생성 open_price = price close_price = max(1000.0, open_price * (1 + daily_change)) high_price = max(open_price, close_price) * (1 + rng.uniform(0, volatility * 1.8)) low_price = min(open_price, close_price) * (1 - rng.uniform(0, volatility * 1.8)) # 저가가 0 이하로 내려가지 않도록 보정 if low_price <= 0: low_price = min(open_price, close_price) * 0.98 # 거래량 생성 volume = int(volume_base * (1 + rng.uniform(-0.35, 0.35))) # 가격 반올림 처리 open_value = round_price(open_price) close_value = round_price(close_price) high_value = max(round_price(high_price), open_value, close_value) low_value = min(round_price(low_price), open_value, close_value) # 생성된 캔들 데이터를 리스트에 추가 chart_data.append({ "date": current_date.strftime("%Y%m%d"), "open": float(open_value), "high": float(high_value), "low": float(low_value), "close": float(close_value), "volume": max(1000, volume), }) # 다음 캔들의 시가를 위해 현재 종가 저장 price = close_price # 최종 샘플 차트 데이터를 딕셔너리 형태로 반환 return { "stock_code": stock_code, "period": period, "data": chart_data, "is_sample": True, } # 차트 데이터를 위한 간단한 인메모리 캐시 딕셔너리 chart_cache = {} @router.get("/chart/{stock_code}") async def get_stock_chart( stock_code: str, period: str = "D", realtime: bool = Query(False, description="최신 현재가를 포함하여 실시간으로 보정된 마지막 캔들 반환"), limit: int = Query(100, description="조회할 데이터 개수 (기본: 100, 최대: 240)") ): """주식 차트 데이터 조회 엔드포인트 (캐싱 적용) Args: stock_code: 종목 코드 period: D(일봉), W(주봉), M(월봉) limit: 조회할 데이터 개수 (일봉 100개는 약 3개월치 데이터) """ try: # 캐시 키 생성 (예: "005930__D") cache_key = f"{stock_code}__{period}" current_time = datetime.now() # 캐시에서 데이터 조회 cached_data = chart_cache.get(cache_key) use_cache = False chart_data = [] # 캐시 데이터가 있고, 10분(600초) 이내에 생성된 경우 캐시 사용 if cached_data: last_updated = cached_data.get("timestamp") if last_updated and (current_time - last_updated).total_seconds() < 600: chart_data = list(cached_data.get("data", [])) # 원본 수정을 방지하기 위해 데이터 복사 use_cache = True # logger.debug(f"Chart cache hit for {stock_code}") # 캐시를 사용하지 않는 경우 (캐시가 없거나 만료됨) if not use_cache: # API 키가 설정되지 않은 경우 샘플 데이터 생성 if not settings.KOREA_INVESTMENT_APP_KEY or not settings.KOREA_INVESTMENT_APP_SECRET: return _generate_sample_chart(stock_code, period, limit) # 한국투자증권 API를 통해 차트 데이터 조회 data = await korea_investment_api.get_daily_chart(stock_code, period, limit=limit) if "output2" not in data: raise HTTPException(status_code=404, detail="Chart data not found") chart_data = [] # 데이터 리스트 초기화 for item in data["output2"]: # KIS API 응답에서 가격 데이터를 안전하게 파싱 try: # 콤마가 포함될 수 있는 문자열 형태의 가격을 가져옴 close_str = str(item.get("stck_clpr", "0")) open_str = str(item.get("stck_oprc", "0")) high_str = str(item.get("stck_hgpr", "0")) low_str = str(item.get("stck_lwpr", "0")) # 콤마를 제거하고 float으로 변환 close_val = float(close_str.replace(",", "")) open_val = float(open_str.replace(",", "")) high_val = float(high_str.replace(",", "")) low_val = float(low_str.replace(",", "")) # 파싱된 데이터를 딕셔너리 형태로 리스트에 추가 chart_data.append({ "date": item.get("stck_bsop_date"), "open": open_val, "high": high_val, "low": low_val, "close": close_val, "volume": int(item.get("acml_vol", 0)) }) except (ValueError, TypeError) as e: # 가격 파싱 중 오류 발생 시 로그 기록하고 해당 데이터는 건너뜀 logger.error(f"가격 파싱 오류: {item} - {e}") continue # 조회한 데이터를 캐시에 저장 chart_cache[cache_key] = { "timestamp": current_time, "data": chart_data } # KIS API는 최신 데이터가 가장 먼저 오는 경향이 있으므로, 프론트엔드에서 날짜순으로 정렬(reverse)하여 사용 latest_candle = None # 'realtime' 쿼리 파라미터가 True인 경우, 현재가를 반영한 실시간 캔들 생성 if realtime: try: price_data = await korea_investment_api.get_current_price(stock_code) if price_data and price_data.get("output"): o = price_data["output"] # 현재가, 고가, 저가, 시가 등의 정보를 파싱하여 실시간 캔들 데이터 구성 current_price = float(str(o.get("stck_prpr", "0")).replace(",", "")) high_price = float(str(o.get("stck_hgpr", "0")).replace(",", "")) low_price = float(str(o.get("stck_lwpr", "0")).replace(",", "")) open_price = float(str(o.get("stck_oprc", current_price)).replace(",", "")) volume = int(o.get("acml_vol", 0)) today_str = o.get("stck_bsop_date") or datetime.now().strftime("%Y%m%d") latest_candle = { "date": today_str, "open": open_price, "high": high_price, "low": low_price, "close": current_price, "volume": volume, "realtime": True } # 기존 차트 데이터의 첫 번째(가장 최신) 캔들이 오늘 날짜와 같으면 교체, 아니면 맨 앞에 추가 if chart_data and chart_data[0]["date"] == latest_candle["date"]: chart_data[0] = latest_candle else: chart_data.insert(0, latest_candle) except Exception as e: # 실시간 캔들 생성 실패 시 경고 로그 기록 logger.warning(f"실시간 현재가 캔들 생성 실패: {e}") # 최종 차트 데이터를 딕셔너리 형태로 반환 return { "stock_code": stock_code, "period": period, "data": chart_data, "is_sample": False, "realtime": realtime, "latest": latest_candle } except Exception as e: # API 호출 등에서 예외 발생 시, 개발 편의를 위해 샘플 데이터를 반환 sample = _generate_sample_chart(stock_code, period) sample["error"] = str(e) # 에러 메시지 포함 sample["is_sample"] = True sample["realtime"] = False sample["latest"] = None return sample @router.get("/news") async def get_market_news(limit: int = 10): """시장 뉴스 조회 엔드포인트 (Google News RSS 활용)""" import feedparser # RSS 파싱을 위한 라이브러리 from datetime import datetime try: # Google News에서 '한국 증시 주식' 키워드로 검색한 결과의 RSS 주소 rss_url = "https://news.google.com/rss/search?q=한국+증시+주식&hl=ko&gl=KR&ceid=KR:ko" # RSS 피드 파싱 feed = feedparser.parse(rss_url) news_list = [] # 조회할 개수(limit)만큼 반복하여 뉴스 항목 추출 for idx, entry in enumerate(feed.entries[:limit]): news_list.append({ "id": idx + 1, "title": entry.get("title", "제목 없음"), "summary": entry.get("summary", entry.get("description", ""))[:200] + "..." if entry.get("summary", entry.get("description", "")) else "", "source": entry.get("source", {}).get("title", "Google News") if hasattr(entry.get("source", {}), "get") else "Google News", "published_at": entry.get("published", datetime.now().isoformat()), "url": entry.get("link", "#") }) return news_list except Exception as e: # 에러 발생 시 미리 정의된 샘플 뉴스 데이터 반환 return [ { "id": 1, "title": "코스피, 외국인 매수세에 상승 마감", "summary": "외국인 투자자들의 순매수세가 이어지면서 코스피가 상승 마감했습니다.", "source": "한국경제", "published_at": datetime.now().isoformat(), "url": "#" }, { "id": 2, "title": "반도체 업종 강세, 삼성전자 상승", "summary": "반도체 업종이 강세를 보이며 삼성전자 주가가 상승했습니다.", "source": "매일경제", "published_at": datetime.now().isoformat(), "url": "#" }, { "id": 3, "title": "배터리 3사 동반 상승", "summary": "LG에너지솔루션, 삼성SDI, SK온이 동반 상승했습니다.", "source": "이데일리", "published_at": datetime.now().isoformat(), "url": "#" } ][:limit] @router.get("/orderbook/{stock_code}") async def get_orderbook(stock_code: str): """호가 정보 조회 엔드포인트 (매수/매도 10단계 호가)""" try: logger.info(f"🔍 Fetching orderbook for stock: {stock_code}") # 한국투자증권 API를 통해 호가 데이터 조회 data = await korea_investment_api.get_orderbook(stock_code) # API 응답 코드가 '0'(성공)이 아닌 경우 에러 처리 if data.get("rt_cd") != "0": logger.error(f"❌ Orderbook API error for {stock_code}: rt_cd={data.get('rt_cd')}, msg={data.get('msg1')}") raise HTTPException(status_code=404, detail=data.get("msg1", "Orderbook data not found")) # API 응답에서 output1(호가), output2(체결가) 데이터 추출 output1 = data.get("output1", {}) output2 = data.get("output2", {}) # 현재가와 누적 거래량 정보 추출 current_price = 0 current_volume = 0 if output2 and output2.get("stck_prpr"): current_price = int(output2.get("stck_prpr", 0)) current_volume = int(output2.get("acml_vol", 0) or 0) elif output1.get("stck_prpr"): current_price = int(output1.get("stck_prpr", 0)) current_volume = int(output1.get("acml_vol", 0) or 0) # 호가 데이터를 담을 딕셔너리 초기화 orderbook = { "stock_code": stock_code, "stock_name": output1.get("hts_kor_isnm", ""), "current_price": current_price, "current_volume": current_volume, "asks": [], # 매도 호가 (낮은 가격 -> 높은 가격 순으로 임시 저장 후 reverse) "bids": [], # 매수 호가 (높은 가격 -> 낮은 가격 순) "total_ask_volume": 0, "total_bid_volume": 0, } # 디버깅을 위한 원본 매도호가 데이터 로그 logger.info(f"📊 [{stock_code}] 매도호가 원본 데이터:") for i in range(1, 11): price = output1.get(f"askp{i}", 0) volume = output1.get(f"askp_rsqn{i}", 0) logger.info(f" askp{i}: {price} ({volume}주)") # 디버깅을 위한 원본 매수호가 데이터 로그 logger.info(f"📊 [{stock_code}] 매수호가 원본 데이터:") for i in range(1, 11): price = output1.get(f"bidp{i}", 0) volume = output1.get(f"bidp_rsqn{i}", 0) logger.info(f" bidp{i}: {price} ({volume}주)") # 매도호가 10단계 파싱 (askp1 ~ askp10) temp_asks = [] for i in range(1, 11): price = int(output1.get(f"askp{i}", 0) or 0) volume = int(output1.get(f"askp_rsqn{i}", 0) or 0) if price > 0: temp_asks.append({"price": price, "volume": volume}) orderbook["total_ask_volume"] += volume # UI 표기를 위해 매도호가는 높은 가격이 위로 가도록 리스트를 뒤집음 orderbook["asks"] = list(reversed(temp_asks)) # 매수호가 10단계 파싱 (bidp1 ~ bidp10) for i in range(1, 11): price = int(output1.get(f"bidp{i}", 0) or 0) volume = int(output1.get(f"bidp_rsqn{i}", 0) or 0) if price > 0: orderbook["bids"].append({"price": price, "volume": volume}) orderbook["total_bid_volume"] += volume # 최종 파싱 결과 로그 logger.info(f"✅ Orderbook {stock_code}: current={current_price}, asks={len(orderbook['asks'])}, bids={len(orderbook['bids'])}") logger.info(f" 파싱된 매도호가: {[ask['price'] for ask in orderbook['asks']]}") logger.info(f" 파싱된 매수호가: {[bid['price'] for bid in orderbook['bids']]}") if orderbook['asks']: logger.info(f" Best ask: {orderbook['asks'][-1]['price']}") # 가장 낮은 매도호가 if orderbook['bids']: logger.info(f" Best bid: {orderbook['bids'][0]['price']}") # 가장 높은 매수호가 return orderbook except HTTPException: # HTTPException은 그대로 다시 발생시킴 raise except Exception as e: # 그 외 예외는 500 에러로 처리 logger.error(f"Error fetching orderbook for {stock_code}: {e}") raise HTTPException(status_code=500, detail=f"Failed to fetch orderbook: {str(e)}") @router.get("/ai-analysis/{stock_code}") async def get_ai_analysis(stock_code: str, db: AsyncSession = Depends(get_db)): """AI 기반 종목 분석 엔드포인트 (기술적 지표 + 규칙 기반 분석)""" try: import httpx from app.core.config import settings logger.info(f"🤖 AI 분석 요청: {stock_code}") # 1. 최신 시장 데이터(현재가 등) 가져오기 price_data = await korea_investment_api.get_current_price(stock_code) if price_data.get("rt_cd") != "0": raise HTTPException(status_code=404, detail="시장 데이터를 가져올 수 없습니다") output = price_data.get("output", {}) current_price = int(output.get("stck_prpr", 0)) high_price = int(output.get("stck_hgpr", 0)) low_price = int(output.get("stck_lwpr", 0)) volume = int(output.get("acml_vol", 0)) # 2. DB에서 가장 최근의 기술적 지표 분석 데이터 가져오기 from app.models.models import StockAnalysis result = await db.execute( select(StockAnalysis) .filter(StockAnalysis.stock_code == stock_code) .order_by(StockAnalysis.analysis_date.desc()) .limit(1) ) analysis_record = result.scalar_one_or_none() indicators = {} if analysis_record: indicators = { "rsi": float(analysis_record.rsi) if analysis_record.rsi else None, "ma5": float(analysis_record.ma5) if analysis_record.ma5 else None, "ma20": float(analysis_record.ma20) if analysis_record.ma20 else None, "ma60": float(analysis_record.ma60) if analysis_record.ma60 else None, "stochastic_k": float(analysis_record.stochastic_k) if analysis_record.stochastic_k else None, "stochastic_d": float(analysis_record.stochastic_d) if analysis_record.stochastic_d else None, "bollinger_upper": float(analysis_record.bollinger_upper) if analysis_record.bollinger_upper else None, "bollinger_middle": float(analysis_record.bollinger_middle) if analysis_record.bollinger_middle else None, "bollinger_lower": float(analysis_record.bollinger_lower) if analysis_record.bollinger_lower else None, } else: # DB에 분석 데이터 없으면 KIS 일봉 차트로 직접 계산 (fallback) try: from app.services.mean_reversion_strategy import calculate_rsi, calculate_bollinger chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=60) chart_data = chart_resp.get("output2", []) if len(chart_data) >= 20: closes = [float(c.get("stck_clpr", 0)) for c in reversed(chart_data)] highs = [float(c.get("stck_hgpr", 0)) for c in reversed(chart_data)] lows = [float(c.get("stck_lwpr", 0)) for c in reversed(chart_data)] rsi_val = calculate_rsi(closes) bb = calculate_bollinger(closes) 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 indicators = { "rsi": round(rsi_val, 4) if rsi_val else None, "ma5": round(ma5, 2) if ma5 else None, "ma20": round(ma20, 2) if ma20 else None, "ma60": round(ma60, 2) if ma60 else None, "stochastic_k": None, "stochastic_d": None, "bollinger_upper": round(bb["upper"], 2) if bb else None, "bollinger_middle": round(bb["middle"], 2) if bb else None, "bollinger_lower": round(bb["lower"], 2) if bb else None, } logger.info(f"📊 {stock_code} 지표 실시간 계산 (DB 없음 fallback)") except Exception as fe: logger.debug(f"지표 fallback 계산 실패 ({stock_code}): {fe}") # 3. 규칙 기반 분석 로직 recommendation = "HOLD" # 기본 추천: 보유 confidence = 0.0 # 신뢰도 trend = "SIDEWAYS" # 추세 analysis_text = "" # 분석 요약 텍스트 # 매수/매도 신호 점수 계산 buy_signals = 0 sell_signals = 0 total_signals = 0 # 기술적 지표 데이터가 있는 경우 분석 시작 if indicators and any(indicators.values()): signals = [] # 분석된 신호들을 담을 리스트 # RSI 분석 if indicators.get("rsi") is not None: rsi = indicators["rsi"] total_signals += 1 if rsi < 30: # 과매도 buy_signals += 1 signals.append(f"RSI 과매도({rsi:.1f})") elif rsi > 70: # 과매수 sell_signals += 1 signals.append(f"RSI 과매수({rsi:.1f})") elif rsi < 40: # 매수 고려 buy_signals += 0.5 signals.append(f"RSI 매수권장({rsi:.1f})") elif rsi > 60: # 매도 고려 sell_signals += 0.5 signals.append(f"RSI 매도권장({rsi:.1f})") # 이동평균선(MA) 분석 ma5 = indicators.get("ma5") ma20 = indicators.get("ma20") ma60 = indicators.get("ma60") if ma5 and ma20: # 현재가와 20일 이동평균선의 괴리율을 계산하여 데이터의 신선도 체크 ma_deviation = abs(current_price - ma20) / ma20 if ma20 > 0 else 999 if ma_deviation < 0.5: # 괴리율이 50% 미만일 때만 유효한 데이터로 간주 total_signals += 1 if ma60 and ma5 > ma20 > ma60: # 정배열 (강한 상승 추세) buy_signals += 1 trend = "UP" signals.append("이동평균 정배열(상승)") elif ma60 and ma5 < ma20 < ma60: # 역배열 (강한 하락 추세) sell_signals += 1 trend = "DOWN" signals.append("이동평균 역배열(하락)") elif ma5 < ma20 and current_price < ma20: # 데드크로스 (하락 신호) sell_signals += 0.5 trend = "DOWN" signals.append("데드크로스 신호") else: logger.debug(f"이동평균선 데이터 오래됨 (현재가: {current_price}, MA20: {ma20}, 괴리율: {ma_deviation*100:.1f}%)") # 볼린저 밴드 분석 bb_upper = indicators.get("bollinger_upper") bb_lower = indicators.get("bollinger_lower") bb_middle = indicators.get("bollinger_middle") if bb_upper and bb_lower and bb_middle and bb_upper > bb_lower: # 현재가가 볼린저 밴드 내 어느 위치에 있는지 계산 (0~1) bb_position = (current_price - bb_lower) / (bb_upper - bb_lower) # 밴드 위치가 -0.2 ~ 1.2 범위 내에 있을 때만 유효한 데이터로 간주 (괴리율 체크) if -0.2 <= bb_position <= 1.2: total_signals += 1 if bb_position > 0.8: # 상단 근접 (과매수) sell_signals += 1 signals.append(f"볼린저 밴드 상단근접({bb_position*100:.0f}%)") elif bb_position < 0.2: # 하단 근접 (과매도) buy_signals += 1 signals.append(f"볼린저 밴드 하단근접({bb_position*100:.0f}%)") else: logger.debug(f"볼린저 밴드 데이터 오래됨 (현재가: {current_price}, 밴드: {bb_lower:.0f}~{bb_upper:.0f}, 위치: {bb_position*100:.0f}%)") # 스토캐스틱 분석 stoch_k = indicators.get("stochastic_k") stoch_d = indicators.get("stochastic_d") if stoch_k is not None and stoch_d is not None: total_signals += 1 if stoch_k < 20 and stoch_d < 20: # 과매도 buy_signals += 1 signals.append(f"스토캐스틱 과매도({stoch_k:.1f})") elif stoch_k > 80 and stoch_d > 80: # 과매수 sell_signals += 1 signals.append(f"스토캐스틱 과매수({stoch_k:.1f})") elif stoch_k > stoch_d and stoch_k < 50: # 상승 전환 신호 buy_signals += 0.3 signals.append("스토캐스틱 상승전환") elif stoch_k < stoch_d and stoch_k > 50: # 하락 전환 신호 sell_signals += 0.3 signals.append("스토캐스틱 하락전환") # 최종 추천 결정 if total_signals > 0: # (매수 신호 - 매도 신호) / 전체 신호 수로 점수 계산 signal_score = (buy_signals - sell_signals) / total_signals confidence = min(abs(signal_score), 1.0) # 신뢰도는 점수의 절대값 if signal_score > 0.3: # 점수가 0.3보다 크면 '매수' recommendation = "BUY" elif signal_score < -0.3: # 점수가 -0.3보다 작으면 '매도' recommendation = "SELL" else: # 그 외에는 '보유' recommendation = "HOLD" analysis_text = " | ".join(signals) if signals else "충분한 신호 없음" else: analysis_text = "기술적 지표 데이터 부족" else: analysis_text = "기술적 지표 데이터 없음" # 추세 최종 결정 (이동평균선 없으면 가격 변동으로) if trend == "SIDEWAYS" and high_price and low_price and current_price: daily_change_pct = ((current_price - low_price) / low_price) * 100 if low_price > 0 else 0 if daily_change_pct > 10: trend = "UP" elif daily_change_pct < -10: trend = "DOWN" # 4. 종합 결과 반환 return { "stock_code": stock_code, "stock_name": output.get("hts_kor_isnm", "") or output.get("stck_prdt_name", ""), "current_price": current_price, "high_price": high_price, "low_price": low_price, "volume": volume, "indicators": indicators, "ai_analysis": analysis_text, "recommendation": recommendation, "confidence": confidence, "trend": trend, "analyzed_at": datetime.now(_KST).isoformat(), # KST +09:00 포함 } except HTTPException: # HTTPException은 그대로 다시 발생시킴 raise except Exception as e: # 그 외 예외는 500 에러로 처리 logger.error(f"Error in AI analysis for {stock_code}: {e}") raise HTTPException(status_code=500, detail=f"AI 분석 실패: {str(e)}")