diff --git a/backend/app/services/market_analyzer.py b/backend/app/services/market_analyzer.py new file mode 100644 index 0000000..c1396b0 --- /dev/null +++ b/backend/app/services/market_analyzer.py @@ -0,0 +1,472 @@ +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 +) +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 + + 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: + pass + + 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 [] \ No newline at end of file