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 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"] kospi_chart = await korea_investment_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} 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 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 def _cache_and_return(result: dict) -> dict: _market_regime_cache["result"] = result _market_regime_cache["ts"] = time.monotonic() return result 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} async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: try: chart_response = await korea_investment_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"])) 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: return None # 전일 거래량 필터: 100만주 미만 종목 제외 (유동성 부족) if volume < 1_000_000: 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 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]: try: await asyncio.sleep(3.0) kospi_result = await korea_investment_api.get_stock_list("J") await asyncio.sleep(3.0) kosdaq_result = await korea_investment_api.get_stock_list("Q") stocks = [] 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)) 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)) return stocks except Exception as e: logger.error(f"Error fetching stock list from KIS API: {e}", exc_info=True) return []