""" 시장 분석 서비스 - KIS API를 통해 관심 종목(watchlist)의 기술적 지표 일괄 분석 - 시장 국면(bull/bear/sideways) 판별 및 추천 종목 선별 - OVERHEAT(급등 과열) 필터, 추세 필터 적용 - 분석 결과를 StockAnalysis DB에 upsert하고 Redis 캐시에 저장 """ import asyncio import logging import time import pytz from datetime import datetime from typing import List, Optional from pydantic import BaseModel from app.core.config import settings from app.core.market_schedule import is_market_open from app.services.korea_investment_api import korea_investment_api, real_market_api as _real_market_api from app.services.mean_reversion_strategy import ( MarketRegime, detect_market_regime, calculate_atr, calculate_adx ) from app.services.technical_indicators import ( TREND_FILTER_ENABLED, TREND_MA_SHORT, TREND_MA_LONG, TREND_TOLERANCE_PCT, OVERHEAT_FILTER_ENABLED, _passes_uptrend, analyze_overheat, analyze_custom_signals, analyze_macd, analyze_rsi, analyze_bollinger, analyze_moving_average, analyze_stochastic, analyze_volatility_breakout, analyze_adx, analyze_obv ) logger = logging.getLogger(__name__) KST = pytz.timezone('Asia/Seoul') def _get_kst_now(): return datetime.now(KST).replace(tzinfo=None) 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 max_disparity: float = 200.0 for_batch: bool = False class StockRecommendation(BaseModel): stock_code: str stock_name: str current_price: float change_rate: float signals: List[str] score: int price_date: str = "" 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_score: Optional[int] = None ml_confidence: Optional[float] = None ml_signal: Optional[str] = None atr_14: Optional[float] = None adx_14: Optional[float] = None _market_regime_cache: dict = {"result": None, "ts": 0.0} _MARKET_REGIME_CACHE_TTL = 120 async def check_market_regime() -> dict: try: if not is_market_open(): return {"regime": "neutral", "regime_200d": "bear", "message": "장외 시간 - 시장 레짐 판별 스킵", "position_multiplier": 0.5} now_ts = time.monotonic() if _market_regime_cache["result"] is not None and (now_ts - _market_regime_cache["ts"]) < _MARKET_REGIME_CACHE_TTL: return _market_regime_cache["result"] _api = _real_market_api or korea_investment_api kospi_chart = await _api.get_daily_chart("069500", "D", limit=240) if "output2" not in kospi_chart or not kospi_chart["output2"]: 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} def _cache_and_return(result: dict) -> dict: _market_regime_cache["result"] = result _market_regime_cache["ts"] = time.monotonic() return result # ── 수동 비상 정지 ──────────────────────────────────────────── if settings.TRADING_HALTED: return _cache_and_return({ "regime": "halted", "regime_200d": "bear", "message": "⛔ TRADING_HALTED=true — 수동 비상 정지 중 (매수 전면 차단)", "position_multiplier": 0.0, "min_score": 9999, "max_trades": 0, }) current_price = closes[0] ma20 = sum(closes[:20]) / 20 ma60 = sum(closes[:60]) / 60 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 ma200 = ma60 # ── 당일 급락 감지 (사이드카/전쟁 등) ─────────────────────── # closes[0]이 오늘 데이터일 때만 적용 (일봉 미확정 시 전일 종가가 closes[0]이 되어 오탐 발생) latest_date = chart_data[0].get("stck_bsop_date", "") if chart_data else "" today_str = _get_kst_now().strftime("%Y%m%d") if latest_date == today_str and len(closes) >= 2 and closes[1] > 0: today_change_pct = (closes[0] - closes[1]) / closes[1] * 100 if today_change_pct <= settings.MARKET_CRASH_HALT_PCT: return _cache_and_return({ "regime": "crash", "regime_200d": "bear", "message": f"🚨 당일 급락 감지 ({today_change_pct:.1f}% ≤ {settings.MARKET_CRASH_HALT_PCT}%) — 매수 전면 차단", "position_multiplier": 0.0, "min_score": 9999, "max_trades": 0, "today_change_pct": round(today_change_pct, 2), }) 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 if current_price < ma60 or consecutive_down >= 5: return _cache_and_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), }) if current_price < ma20: return _cache_and_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), }) return _cache_and_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_chart_items( chart_data: List[dict], stock_code: str, stock_name: str, criteria: "RecommendationCriteria", ) -> Optional[StockRecommendation]: """KIS 형식 chart_data (oldest first)로 기술적 분석 수행 — API 호출 없음. analyze_stock / analyze_stock_for_batch_db 양쪽에서 공유하는 순수 분석 로직. """ 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.debug("필터[가격0] %s", stock_code) return None # 가격 필터: 5,000원 미만 종목 제외 (CLAUDE.md 섹션 5.4) if current_price < settings.MIN_PRICE_KRW: logger.info("필터[저가] %s: %.0f원 < %d원", stock_code, current_price, settings.MIN_PRICE_KRW) return None # 전일 거래량 필터: 100만주 미만 종목 제외 (유동성 부족) if volume < 1_000_000: logger.info("필터[거래량] %s: %d주 < 100만주", stock_code, volume) 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 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: return None disparity = (current_price / ma20) * 100 if ma20 else 100 trend_ok = None if TREND_FILTER_ENABLED: trend_ok = _passes_uptrend(closes, TREND_MA_SHORT, TREND_MA_LONG) 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 = 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: stochastic_k = ((closes_for_stoch[-1] - lowest_low) / (highest_high - lowest_low)) * 100 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: stochastic_d_vals.append(((float(recent[j].get("stck_clpr", 0)) - slice_low) / (slice_high - slice_low)) * 100) if stochastic_d_vals: stochastic_d = sum(stochastic_d_vals) / len(stochastic_d_vals) 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 if disparity > criteria.max_disparity: score -= 5 signals.append(f"이격도 과열({disparity:.1f}% > {criteria.max_disparity}%) - 매수금지") if criteria.use_overheat_filter and OVERHEAT_FILTER_ENABLED: overheat = analyze_overheat(closes, ma20) if overheat["blocked"]: score -= 10 signals.append(overheat["reason"]) 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 if volume_ratio >= 1.5 and change_rate > 0: signals.append(f"거래량 실린 상승 ({volume_ratio:.1f}배)") score += 3 elif volume_ratio < 0.8: score -= 1 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)") if criteria.min_volume_ratio > 0 and volume_ratio < criteria.min_volume_ratio: return None 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"] rule_score_normalized = (score / 15.0) * 100.0 combined_score = int( rule_score_normalized * settings.ML_ENSEMBLE_RULE_WEIGHT + ml_score_val * settings.ML_ENSEMBLE_ML_WEIGHT ) score = max(0, int(combined_score / 100.0 * 15.0)) signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})") except Exception: pass 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: adx_14_val = calculate_adx(_h, _l, _c) except Exception as _e: logger.debug(f"ATR/ADX 계산 실패 {stock_code}: {_e}") if criteria.for_batch or len(signals) > 0 or score > 0: 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, ) return None async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: """KIS API로 차트를 조회한 뒤 _analyze_chart_items에 위임.""" try: _api = _real_market_api or korea_investment_api chart_response = await _api.get_daily_chart(stock_code, "D") if "output2" not in chart_response or not chart_response["output2"]: return None chart_data = list(reversed(chart_response["output2"])) return _analyze_chart_items(chart_data, stock_code, stock_name, criteria) except Exception as e: logger.error(f"Error analyzing {stock_code}: {str(e)}") return None async def analyze_stock_for_batch_db( chart_data_newest_first: List[dict], stock_code: str, stock_name: str, ) -> Optional[StockRecommendation]: """DB에서 가져온 chart_data (newest first)로 배치 분석 — KIS API 호출 없음.""" 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, ) try: chart_data = list(reversed(chart_data_newest_first)) # oldest first return _analyze_chart_items(chart_data, stock_code, stock_name, criteria) 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]: 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, ) return await analyze_stock(stock_code, stock_name, criteria) async def analyze_stock_with_indicators(stock_code: str, stock_name: str) -> Optional[StockRecommendation]: 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, max_disparity=110.0 ) return await analyze_stock(stock_code, stock_name, criteria) async def get_all_tradable_stocks() -> List[tuple]: SKIP_KEYWORDS = [ "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG", "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "RISE", "선물", "스팩", "리츠", "SPAC", ] try: _list_api = _real_market_api or korea_investment_api await asyncio.sleep(3.0) kospi_result = await _list_api.get_stock_list("J") await asyncio.sleep(3.0) kosdaq_result = await _list_api.get_stock_list("Q") stocks = [] for result in (kospi_result, kosdaq_result): if "output" in result: for item in result["output"]: stock_code = item.get("mksc_shrn_iscd", "") stock_name = item.get("hts_kor_isnm", "") if not stock_code or not stock_name: continue if any(kw in stock_name for kw in SKIP_KEYWORDS): continue stocks.append((stock_code, stock_name)) return stocks except Exception as e: logger.error(f"Error fetching stock list from KIS API: {e}", exc_info=True) return []