From d822dbb9ca91d45803d0c24c0e8fa6f86a69e10e Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Wed, 25 Feb 2026 11:21:11 +0900 Subject: [PATCH] refactor: create technical_indicators.py --- backend/app/services/technical_indicators.py | 346 +++++++++++++++++++ 1 file changed, 346 insertions(+) create mode 100644 backend/app/services/technical_indicators.py diff --git a/backend/app/services/technical_indicators.py b/backend/app/services/technical_indicators.py new file mode 100644 index 0000000..adfed7b --- /dev/null +++ b/backend/app/services/technical_indicators.py @@ -0,0 +1,346 @@ +import logging +from typing import Optional + +from app.core.config import settings +from app.services.mean_reversion_strategy import calculate_adx + +logger = logging.getLogger(__name__) + +# 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: + """급등 과열 필터 - 단기 급등 종목 매수 차단""" + 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} + + +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 if ma20 else 100 + + 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: + return {"signal": None, "description": f"급등 종목 추격 금지 (3일 +{recent_gain:.1f}%)"} + + 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": "역배열 하락 매도타점"} + + 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: + if len(chart_data) < 26: + return {"signal": None, "description": "데이터 부족"} + + prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-26:]] + + if len(prices) < 26: + return {"signal": None, "description": "데이터 부족"} + + 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": "데이터 부족"} + + if macd[-2] <= signal[-2] and macd[-1] > signal[-1]: + return {"signal": "buy", "description": "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: + if len(chart_data) < 15: + return {"signal": None, "description": "데이터 부족"} + + 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)) + + 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 + + if rsi < 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:]] + 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: + 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] + + if ma5 > ma20 and current_price > ma5: + return {"signal": "buy", "description": "5일선 > 20일선"} + 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 + + 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: + 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": "데이터 부족"} + + target_price = open_price + (prev_range * k) + + if current_price >= target_price: + return {"signal": "buy", "description": f"변동성 돌파 발생 (목표가: {target_price:.0f})"} + + return {"signal": None, "description": "변동성 돌파 미달"} + + +def analyze_adx(chart_data: list) -> dict: + if len(chart_data) < 29: + 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: + 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": "거래량 중립"}