diff --git a/backend/app/api/endpoints/recommendations.py b/backend/app/api/endpoints/recommendations.py index a5b6b18..7c10f69 100644 --- a/backend/app/api/endpoints/recommendations.py +++ b/backend/app/api/endpoints/recommendations.py @@ -1,43 +1,32 @@ from fastapi import APIRouter, HTTPException, Query, Depends from typing import List, Optional -from pydantic import BaseModel from datetime import datetime, date, timedelta -from decimal import Decimal import asyncio import logging import httpx -import pytz -import time from collections import namedtuple -from app.services.korea_investment_api import korea_investment_api from app.core.database import get_db, AsyncSessionLocal -from app.models.models import TradeHistory, Portfolio, StockAnalysis +from app.models.models import StockAnalysis from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import select, and_, desc +from sqlalchemy import select from app.core.config import settings -from app.core.market_schedule import is_market_open -from app.services.mean_reversion_strategy import ( - MarketRegime, detect_market_regime, check_buy_signal, - calculate_atr, calculate_adx, calculate_rsi, calculate_bollinger, is_reversal_candle, - calculate_position_size_v2, +from app.services.market_analyzer import ( + check_market_regime, get_all_tradable_stocks, analyze_stock, + RecommendationCriteria, StockRecommendation, KST ) -from app.services.mdd_manager import mdd_manager -from app.services.loss_tracker import loss_tracker +from app.services.trade_executor import execute_auto_trade_background, execute_auto_trade +from app.services.mean_reversion_strategy import MarketRegime router = APIRouter() logger = logging.getLogger(__name__) -# 매매 규칙 namedtuple (get_rule_based_recommendations_from_db/realtime 공용) +# 매매 규칙 namedtuple Rule = namedtuple('Rule', ['id', 'name', 'description', 'rule_type', 'priority', 'conditions', 'actions', 'is_active']) - def _stock_analysis_to_dict(stock, rule=None, actions=None) -> dict: - """StockAnalysis ORM 행 → stock_dict 변환 헬퍼 (중복 제거). - - rule=None 이면 '고점수 분석 종목' 폴백 값을 채운다. - """ + """StockAnalysis ORM 행 → stock_dict 변환 헬퍼 (중복 제거).""" base = { "stock_code": stock.stock_code, "stock_name": stock.stock_name, @@ -81,977 +70,6 @@ def _stock_analysis_to_dict(stock, rule=None, actions=None) -> dict: }) return base -# 한국 시간대 -KST = pytz.timezone('Asia/Seoul') - -def _get_kst_now(): - """현재 한국 시간 반환 (timezone-naive datetime)""" - return datetime.now(KST).replace(tzinfo=None) - -# 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: - """급등 과열 필터 - 단기 급등 종목 매수 차단 - - 3일/5일/10일 상승률과 20일선 이격도를 종합 판단하여 - 고점 추격 매수를 원천 차단합니다. - - Returns: - {"blocked": bool, "reason": str, "details": 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} - - -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 # 평균 대비 거래량 배수 (1.0 = 필터 없음) - max_disparity: float = 200.0 # 최대 이격도 (20일선 대비 %) - 완화 - for_batch: bool = False # 배치 분석 모드: 필터 완화, score=0도 반환 - - -class StockRecommendation(BaseModel): - """추천 종목 정보""" - stock_code: str - stock_name: str - current_price: float - change_rate: float - signals: List[str] - score: int - price_date: str = "" # 가격 기준일 (YYYYMMDD 형식) - # 추가 지표 (배치 저장 및 API 응답에 사용) - 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 앙상블 스코어 - ml_score: Optional[int] = None # 앙상블 스코어 (0~100) - ml_confidence: Optional[float] = None # 신뢰도 (0~1) - ml_signal: Optional[str] = None # BUY/HOLD/SELL - # ATR (배치 저장용 — 이중 API 호출 방지) - atr_14: Optional[float] = None - # ADX (추세 강도 — 평균회귀 필터용) - adx_14: Optional[float] = None - - -# 공식 기반 신호 분석 함수 (이동평균 정배열/역배열, 눌림목, 손절/익절) -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 - - # 🚫 급등 종목 필터: 최근 3일 상승률 체크 - 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: # 3일간 10% 이상 급등 시 매수 안함 - return {"signal": None, "description": f"급등 종목 추격 금지 (3일 +{recent_gain:.1f}%)"} - - # 정배열 매수 - 이격도 조건 강화 (105% -> 102%) - 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": "역배열 하락 매도타점"} - - # 눌림목 진입 - 이격도 조건 강화 (103% -> 101%) - 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: - """MACD 분석 (골든크로스/데드크로스)""" - if len(chart_data) < 26: - return {"signal": None, "description": "데이터 부족"} - - # 간단한 MACD 계산 (12일, 26일, 9일) - prices = [float(item.get("stck_clpr", 0)) for item in chart_data[-26:]] - - if len(prices) < 26: - return {"signal": None, "description": "데이터 부족"} - - # EMA 계산 - 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": "데이터 부족"} - - # 골든크로스: MACD가 시그널선을 상향 돌파 - if macd[-2] <= signal[-2] and macd[-1] > signal[-1]: - return {"signal": "buy", "description": "MACD 골든크로스"} - # 데드크로스: 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: - """RSI 분석 (과매수/과매도) - 바닥 탈출 전략""" - if len(chart_data) < 15: - return {"signal": None, "description": "데이터 부족"} - - # 현재 RSI 계산 - 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)) - - # 이전 RSI 계산 (추세 확인용) - 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 - - # 🔥 Falling Knife 방지: 바닥 탈출 확인 - if rsi < 30: - # 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:]] - - # 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: - """이동평균선 분석 (5, 20, 60일)""" - 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] - - # 조건 완화: 5일선이 20일선 위에만 있어도 매수 - if ma5 > ma20 and current_price > ma5: - return {"signal": "buy", "description": "5일선 > 20일선"} - # 20일선이 60일선 위에 있으면 추가 점수 - 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 - - # 조건 완화: 30 이하 매수, 70 이상 매도 - 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: - """🔥 변동성 돌파 전략 (래리 윌리엄스) - 승률이 매우 높은 단타 전략 - - 공식: 현재가 > 시가 + (전일 고가 - 전일 저가) * K - """ - 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": "데이터 부족"} - - # 목표 돌파가 = 시가 + (전일 변동폭 * K) - target_price = open_price + (prev_range * k) - - if current_price >= target_price: - return {"signal": "buy", "description": f"변동성 돌파 발생 (목표가: {target_price:.0f})"} - - return {"signal": None, "description": "변동성 돌파 미달"} - - -# check_market_regime() 인메모리 캐시 (KIS API 반복 호출 방지) -_market_regime_cache: dict = {"result": None, "ts": 0.0} -_MARKET_REGIME_CACHE_TTL = 120 # 2분 - - -async def check_market_regime() -> dict: - """시장 레짐 판별 (CLAUDE.md 섹션 7.1: 200일 MA + 기울기) - - 상승장(Bull): 코스피 > MA200 AND MA200 기울기 양(+) - 하락장(Bear): 그 외 - - 하위 분류 (기존 호환): - - strong_bearish: 코스피 < MA60 또는 5일 연속 하락 - - bearish: 코스피 < MA20 - - bullish: 코스피 > MA200 + 기울기 상승 - """ - try: - # 장외 시간에는 모의투자 API가 차트 데이터를 반환하지 않음 → neutral 반환 - if not is_market_open(): - return {"regime": "neutral", "regime_200d": "bear", "message": "장외 시간 - 시장 레짐 판별 스킵", "position_multiplier": 0.5} - - # 캐시 히트: TTL 내 이전 결과 반환 (KIS API 반복 호출 방지) - now_ts = time.monotonic() - if _market_regime_cache["result"] is not None and (now_ts - _market_regime_cache["ts"]) < _MARKET_REGIME_CACHE_TTL: - logger.debug("check_market_regime: cache hit") - return _market_regime_cache["result"] - - # KODEX 200 (069500) = KOSPI 200 추종 ETF 사용 (지수코드 0001은 daily-price API 미지원) - logger.info("check_market_regime: KIS API 호출 (캐시 미스)") - kospi_chart = await korea_investment_api.get_daily_chart("069500", "D", limit=240) - - if "output2" not in kospi_chart or not kospi_chart["output2"]: - logger.warning("KOSPI 지수 데이터 없음 - 시장 필터 비활성화") - 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 - - # 200일 MA 기반 레짐 판별 (섹션 7.1) - 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 # 200일 데이터 부족 시 MA20 폴백 - ma200 = ma60 - - # 5일 연속 하락 체크 - 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 - - # 강한 하락장: 코스피 < MA60 또는 5일 연속 하락 - 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), - }) - - # 약세장: 코스피 < MA20 - 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), - }) - - # 상승장: 200일 MA 기반 - 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_adx(chart_data: list) -> dict: - """ADX 분석 — Wilder's ADX(14) 기반 추세 강도 측정 - - 평균회귀 전략에서의 해석: - - ADX < 20: 횡보장 → 평균회귀 유리 (BUY 신호) - - ADX 20~30: 중간 추세 → 중립 - - ADX > 30: 강한 추세 → 평균회귀 위험 (SELL/SKIP 신호) - - chart_data는 시간 순(오래된 → 최신) 정렬을 가정. - """ - if len(chart_data) < 29: # period * 2 + 1 - 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: - """거래량(OBV) 분석""" - 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": "거래량 중립"} - - -async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: - """개별 종목 분석""" - try: - logger.info(f"Analyzing stock: {stock_code}") - - # 차트 데이터 조회 (현재가 포함) - chart_response = await korea_investment_api.get_daily_chart(stock_code, "D") - if "output2" not in chart_response or not chart_response["output2"]: - logger.warning(f"No chart data for {stock_code}") - return None - - # KIS API는 최신순(newest-first)으로 반환 → 시간순(chronological)으로 뒤집어 분석 - chart_data = list(reversed(chart_response["output2"])) - logger.info(f"Got {len(chart_data)} chart data points for {stock_code}") - - # 최신 데이터에서 현재가 및 종목명 추출 (시간순 기준 마지막 = 최신) - 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.warning(f"Zero price for {stock_code}") - 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 - - # 🔥 1. 거래량 분석 (Volume Analysis) - 필수 조건 - 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: - logger.debug(f"{stock_code}: 거래량 부족으로 제외 (평균 거래량: {avg_vol_20:.0f})") - return None - - # 🔥 2. 이격도 계산 (Disparity Filter) - 추격 매수 방지 - disparity = (current_price / ma20) * 100 if ma20 else 100 - - # Trend filter check (non-blocking here; used for scoring/tagging) - trend_ok = None - if TREND_FILTER_ENABLED: - trend_ok = _passes_uptrend(closes, TREND_MA_SHORT, TREND_MA_LONG) - - # RSI 계산 (단순 방식) - 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, %D 간단 계산 - 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: - k = ((closes_for_stoch[-1] - lowest_low) / (highest_high - lowest_low)) * 100 - stochastic_k = k - # %D는 간단히 최근 3일 %K 평균 - 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: - slice_k = ((float(recent[j].get("stck_clpr", 0)) - slice_low) / (slice_high - slice_low)) * 100 - stochastic_d_vals.append(slice_k) - if stochastic_d_vals: - stochastic_d = sum(stochastic_d_vals) / len(stochastic_d_vals) - - # Bollinger bands - 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 - - # 🔥 3. 이격도 과열 체크 (max_disparity 조건) - if disparity > criteria.max_disparity: - score -= 5 - signals.append(f"이격도 과열({disparity:.1f}% > {criteria.max_disparity}%) - 매수금지") - logger.debug(f"{stock_code}: 이격도 과열로 점수 감점") - - # 🔥 3.5. 급등 과열 필터 (고점 추격 매수 차단) - if criteria.use_overheat_filter and OVERHEAT_FILTER_ENABLED: - overheat = analyze_overheat(closes, ma20) - if overheat["blocked"]: - score -= 10 - signals.append(overheat["reason"]) - logger.info(f"{stock_code}: 급등 과열 차단 - {overheat['reason']} {overheat['details']}") - - # 🔥 4. 변동성 돌파 전략 (고승률) - 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 - - # 🔥 5. 거래량 급증 보너스 (매수 신호 강도 증가) - if volume_ratio >= 1.5 and change_rate > 0: - signals.append(f"거래량 실린 상승 ({volume_ratio:.1f}배)") - score += 3 - elif volume_ratio < 0.8: - # 거래량이 평소보다 적으면 신뢰도 감소 - score -= 1 - logger.debug(f"{stock_code}: 거래량 부족 (평소의 {volume_ratio:.1f}배)") - - # Tag trend state to signals and adjust score slightly - 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)") - - # 🔥 6. 최종 거래량 필터 (min_volume_ratio) - if criteria.min_volume_ratio > 0 and volume_ratio < criteria.min_volume_ratio: - logger.debug(f"{stock_code}: 거래량 필터 미달 ({volume_ratio:.2f} < {criteria.min_volume_ratio})") - return None - - logger.info(f"{stock_code} ({stock_name}): score={score}, signals={signals}, volume_ratio={volume_ratio:.2f}, disparity={disparity:.1f}%") - - # 🤖 ML 앙상블 스코어 계산 - 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"] - - # 룰베이스 + ML 가중 결합 - rule_score_normalized = (score / 15.0) * 100.0 # 0~15 → 0~100 - combined_score = int( - rule_score_normalized * settings.ML_ENSEMBLE_RULE_WEIGHT + - ml_score_val * settings.ML_ENSEMBLE_ML_WEIGHT - ) - # 0~100 → 0~15 스케일로 재변환 (기존 호환) - score = max(0, int(combined_score / 100.0 * 15.0)) - - signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})") - logger.info(f" ML 앙상블: score={ml_score_val}, confidence={ml_confidence_val:.2f}, signal={ml_signal_val}") - except Exception as e: - logger.debug(f"ML 앙상블 스코어 계산 실패 (정상 - 모델 미학습): {e}") - - # ATR / ADX 계산 (이미 가져온 차트 데이터 재사용 — 이중 API 호출 방지) - 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: # Wilder ADX 최소 데이터 (period * 2 + 1) - adx_14_val = calculate_adx(_h, _l, _c) - except Exception: - pass - - # 신호가 있거나 점수가 0 이상이면 추천 (배치 모드에서는 score=0도 반환) - if criteria.for_batch or len(signals) > 0 or score > 0: - # 신호가 있으면 최소 1점 보장 - 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, - ) - - logger.debug(f"{stock_code}: No recommendation (score={score})") - 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]: - """배치 DB 저장용: 거래량/점수 필터 없이 지표만 계산하여 전 종목 저장. - - 모든 지표를 계산하되, 매매 필터(min_volume_ratio, 거래량 부족, score=0)를 적용하지 않습니다. - score=0이어도 반환하여 배치에서 DB에 저장됩니다. - """ - 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, # score=0 · 저유동성 종목도 반환 - ) - return await analyze_stock(stock_code, stock_name, criteria) - - -async def analyze_stock_with_indicators(stock_code: str, stock_name: str) -> Optional[StockRecommendation]: - """호출 편의성: 모든 지표를 켜서 분석한 결과(Pydantic 모델)를 반환합니다. - - 이 함수는 배치 스크립트에서 사용됩니다. - - 🔥 승률 향상 기능 포함: - - 변동성 돌파 전략 - - 거래량 필터 (평균 대비 1.2배) - - 이격도 필터 (110% 제한) - """ - 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, # 🔥 거래량 필터 (평균의 1.2배 이상) - max_disparity=110.0 # 🔥 이격도 필터 (20일선 대비 110% 이하) - ) - return await analyze_stock(stock_code, stock_name, criteria) - - -async def get_all_tradable_stocks() -> List[tuple]: - """KIS API로 전체 거래 가능 종목 조회""" - try: - logger.info("Fetching all tradable stocks from KIS API...") - - # 이전 API 호출과의 충돌을 피하기 위해 대기 - await asyncio.sleep(3.0) - - # KOSPI 거래대금 상위 종목 조회 - kospi_result = await korea_investment_api.get_stock_list("J") - logger.info(f"KOSPI result: {kospi_result}") - - # Rate limit을 위한 대기 (초당 1건 제한) - 안전하게 3초 대기 - await asyncio.sleep(3.0) - - # KOSDAQ 거래대금 상위 종목 조회 ("Q" = KOSDAQ, KIS API 규격) - kosdaq_result = await korea_investment_api.get_stock_list("Q") - logger.info(f"KOSDAQ result: {kosdaq_result}") - - stocks = [] - - # KOSPI 종목 추출 - 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)) - - # KOSDAQ 종목 추가 - 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)) - - logger.info(f"Found {len(stocks)} tradable stocks from KIS API") - - # API에서 종목을 못 가져온 경우 빈 리스트 반환 (테스트 데이터 제거) - if not stocks: - logger.warning("No stocks from KIS API, returning empty list") - - return stocks - - except Exception as e: - logger.error(f"Error fetching stock list from KIS API: {e}", exc_info=True) - logger.warning("Returning empty list due to API error") - return [] - - @router.get("/recommendations") async def get_recommendations( use_macd: bool = Query(False, description="MACD 지표 사용"), @@ -1066,18 +84,6 @@ async def get_recommendations( max_disparity: float = Query(200.0, description="최대 이격도 (20일선 대비 %)"), max_stocks: int = Query(100, description="분석할 최대 종목 수 (기본 100개)"), ): - """기술적 지표 기반 추천 종목 조회 - - 선택한 기술적 지표 조건에 부합하는 종목을 KOSPI 시가총액 상위 종목 중에서 찾아 추천합니다. - - 🔥 승률 향상 기능: - - 변동성 돌파: 래리 윌리엄스 전략 (고승률) - - 거래량 필터: 평균 대비 거래량 확인 - - 이격도 필터: 20일선과의 거리 제한 (추격 매수 방지) - - 시장 지수 필터: 코스피 지수 확인 (하락장 보수적 대응) - """ - - # 🔥 시장 지수 확인 (가장 중요한 승률 결정 요소) market_regime = await check_market_regime() logger.info(f"📊 시장 상태: {market_regime['message']}") @@ -1094,123 +100,77 @@ async def get_recommendations( max_disparity=max_disparity ) - # 모든 지표가 꺼져있으면 기본값 사용 if not any([use_macd, use_rsi, use_bollinger, use_moving_average, use_stochastic, use_adx, use_obv, use_volatility_breakout]): criteria.use_rsi = True criteria.use_moving_average = True - criteria.use_volatility_breakout = True # 기본값에 변동성 돌파 추가 + criteria.use_volatility_breakout = True - # 분석할 종목 리스트 가져오기 (KIS API에서 실시간 조회) all_stocks = await get_all_tradable_stocks() if not all_stocks: - logger.warning("No stocks found from KIS API") return { "total": 0, "analyzed_stocks": 0, - "criteria": criteria.dict(), + "criteria": criteria.model_dump(), "recommendations": [], "message": "거래 가능한 종목을 조회할 수 없습니다" } stocks_to_analyze = all_stocks[:max_stocks] + logger.info(f"Analyzing {len(stocks_to_analyze)} stocks with criteria: {criteria.model_dump()}") - logger.info(f"Analyzing {len(stocks_to_analyze)} stocks with criteria: {criteria.dict()}") - - # Rate Limit 방지: 배치 처리 (초당 최대 5개) batch_size = 5 - delay_between_batches = 1.2 # 1.2초 대기 - + delay_between_batches = 1.2 all_results = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] - batch_num = i // batch_size + 1 - total_batches = (len(stocks_to_analyze) + batch_size - 1) // batch_size - logger.info(f"Processing batch {batch_num}/{total_batches} (stocks {i+1}-{min(i+batch_size, len(stocks_to_analyze))})") - - # 배치 단위로 병렬 처리 tasks = [analyze_stock(code, name, criteria) for code, name in batch] batch_results = await asyncio.gather(*tasks, return_exceptions=True) all_results.extend(batch_results) - # 다음 배치 전 대기 (마지막 배치는 제외) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) - # 예외 로깅 - errors = [r for r in all_results if isinstance(r, Exception)] - if errors: - logger.error(f"Encountered {len(errors)} errors during analysis") - - # None 및 예외 제거 후 3000원 이하 종목 필터링, 점수순 정렬 regime_min_score = market_regime.get("min_score", 1) recommendations = [ r for r in all_results if r is not None and not isinstance(r, Exception) - and r.current_price >= 3000 # 🚫 동전주 제외 - and r.score >= regime_min_score # 🐻 하락장 최소 점수 필터 + and r.current_price >= 3000 + and r.score >= regime_min_score ] recommendations.sort(key=lambda x: x.score, reverse=True) - logger.info(f"Found {len(recommendations)} recommendations out of {len(stocks_to_analyze)} analyzed stocks (min_score={regime_min_score})") - - if not recommendations: - logger.warning("No recommendations found! Criteria might be too strict or API failed") - return { "total": len(recommendations), "analyzed_stocks": len(stocks_to_analyze), - "criteria": criteria.dict(), - "market_regime": market_regime, # 🔥 시장 상태 정보 추가 - "recommendations": recommendations[:20] # 상위 20개만 반환 + "criteria": criteria.model_dump(), + "market_regime": market_regime, + "recommendations": recommendations[:20] } - @router.get("/rule-based") async def get_rule_based_recommendations( realtime: bool = Query(False, description="실시간 분석 여부 (느림, 30초)"), auto_trade: bool = Query(True, description="자동매매 실행 여부"), db: AsyncSession = Depends(get_db) ): - """ - 매매 규칙 기반 자동 추천 - - - realtime=false (기본): DB에서 조회 (빠름, 0.1초) - 매일 아침 9시 데이터 - - realtime=true: KIS API 실시간 분석 (느림, 30초) - - auto_trade=true (기본): 조건에 맞는 종목을 자동으로 매매 - - 활성화된 모든 매매 규칙을 확인하고, 조건에 맞는 종목을 자동으로 추천합니다. - """ - - # 실시간 모드: 기존 방식 (KIS API 직접 호출) if realtime: - logger.info("🔴 Real-time mode: Fetching live data from KIS API (slow)") result = await get_rule_based_recommendations_realtime() else: - # DB 모드: 빠른 조회 - logger.info("🟢 DB mode: Fetching from cached analysis (fast)") result = await get_rule_based_recommendations_from_db(db) - # 자동매매 실행 (백그라운드) - # 주의: 엔드포인트 반환 후 FastAPI가 db 세션을 닫으므로, - # 백그라운드 태스크는 독립적인 새 세션을 열어서 사용해야 한다. if auto_trade and result.get("recommendations_by_rule"): rec_snapshot = result async def _bg_trade(): async with AsyncSessionLocal() as bg_db: await execute_auto_trade_background(rec_snapshot, bg_db) asyncio.create_task(_bg_trade()) - logger.info("🤖 Auto-trade task created in background") return result - async def _load_active_rules_from_db(db: AsyncSession) -> list: - """trading_rules 테이블에서 활성 규칙 로드. - DB가 비어있으면 JSON 파일 → 하드코딩 폴백 순으로 시도한다. - """ from app.models.models import TradingRule as TradingRuleModel try: result = await db.execute( @@ -1220,7 +180,6 @@ async def _load_active_rules_from_db(db: AsyncSession) -> list: ) db_rules = result.scalars().all() if db_rules: - logger.info(f"trading_rules DB에서 {len(db_rules)}개 활성 규칙 로드") return [ Rule(id=r.id, name=r.name, description=r.description or "", rule_type=r.rule_type, priority=r.priority, @@ -1230,7 +189,6 @@ async def _load_active_rules_from_db(db: AsyncSession) -> list: except Exception as e: logger.warning(f"trading_rules DB 조회 실패: {e}") - # 폴백 1: JSON 파일 try: import json as _json from pathlib import Path as _Path @@ -1239,7 +197,6 @@ async def _load_active_rules_from_db(db: AsyncSession) -> list: config = _json.load(f) rules_data = [r for r in config.get("rules", []) if r.get("is_active")] if rules_data: - logger.info(f"trading_rules DB 비어있음 — JSON 파일에서 {len(rules_data)}개 규칙 로드") return [ Rule(id=i + 1, name=r["name"], description=r.get("description", ""), rule_type=r["rule_type"].upper(), priority=r.get("priority", i + 1), @@ -1250,11 +207,9 @@ async def _load_active_rules_from_db(db: AsyncSession) -> list: except Exception as e: logger.warning(f"JSON 규칙 파일 로드 실패: {e}") - # 폴백 2: 하드코딩 기본 규칙 - logger.warning("규칙 소스 없음 — 하드코딩 기본 규칙 사용") return [ Rule( - id=1, name="RSI+볼린저 평균회귀", description="기본 평균회귀 규칙 (CLAUDE.md 섹션 6)", + id=1, name="RSI+볼린저 평균회귀", description="기본 평균회귀 규칙", rule_type="MEAN_REVERSION", priority=1, conditions={"rsi_bull_max": 25, "rsi_bear_max": 35, "bb_lower_pct": 3.0, "bb_upper_pct": 1.5}, @@ -1265,14 +220,7 @@ async def _load_active_rules_from_db(db: AsyncSession) -> list: ) ] - def is_mean_reversion_candidate(stock, conditions: dict, market_regime_str: str) -> bool: - """RSI+볼린저 평균회귀 매수 조건 확인 (CLAUDE.md 섹션 7.2) - - Bull: RSI ≤ 25 + BB 하단 -3% ~ +1.5% - (거래량 200% OR 조건은 real-time check_buy_signal에서 처리) - Bear: RSI ≤ 35 + BB 하단 이하 - """ rsi_val = float(stock.rsi) if stock.rsi else None close = int(stock.current_price) if stock.current_price else 0 bb_lower = float(stock.bollinger_lower) if stock.bollinger_lower else None @@ -1282,49 +230,38 @@ def is_mean_reversion_candidate(stock, conditions: dict, market_regime_str: str) if not all([rsi_val, close, bb_lower, bb_upper, bb_middle]): return False - # 가격 필터 (섹션 6.4) if close < settings.MIN_PRICE_KRW: return False regime = MarketRegime.BULL if market_regime_str == "bull" else MarketRegime.BEAR - - # ADX 필터: 강한 추세 중 평균회귀 진입 차단 (ADX > 30이면 횡보 아님) adx_val = float(stock.adx_14) if getattr(stock, "adx_14", None) else None adx_strong_trend = adx_val is not None and adx_val > 30 if regime == MarketRegime.BULL: rsi_threshold = conditions.get("rsi_bull_max", 25) - bb_lower_pct = conditions.get("bb_lower_pct", 3.0) / 100 # -3% - bb_upper_pct = conditions.get("bb_upper_pct", 1.5) / 100 # +1.5% + bb_lower_pct = conditions.get("bb_lower_pct", 3.0) / 100 + bb_upper_pct = conditions.get("bb_upper_pct", 1.5) / 100 in_bb_zone = bb_lower * (1 - bb_lower_pct) <= close <= bb_lower * (1 + bb_upper_pct) if adx_strong_trend: - return False # 강추세 중 평균회귀 차단 + return False return rsi_val <= rsi_threshold and in_bb_zone else: rsi_threshold = conditions.get("rsi_bear_max", 35) if adx_strong_trend: - return False # 강추세 중 평균회귀 차단 + return False return rsi_val <= rsi_threshold and close <= bb_lower - async def get_rule_based_recommendations_from_db(db: AsyncSession): - """DB에서 사전 분석된 데이터 조회 (빠름 - 0.1초)""" - - logger.info("Rule-based recommendations from DB") - active_rules = await _load_active_rules_from_db(db) - logger.info(f"Found {len(active_rules)} active trading rules") - # 오늘·어제 분석 데이터 조회 (KST 날짜 기준, 배치 미실행 시 전날 데이터 사용) today_kst = datetime.now(KST).date() - since_date = today_kst - timedelta(days=1) # 어제~오늘 포함 + since_date = today_kst - timedelta(days=1) result = await db.execute( select(StockAnalysis) .where(StockAnalysis.analysis_date >= since_date) .order_by(StockAnalysis.analysis_date.desc(), StockAnalysis.score.desc()) ) raw_stocks = result.scalars().all() - # 종목코드 기준 중복 제거 (score 내림차순 정렬 후 첫 번째만) seen = set() all_analyzed_stocks = [] for s in raw_stocks: @@ -1332,9 +269,7 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): seen.add(s.stock_code) all_analyzed_stocks.append(s) - # DB에 데이터가 없으면 빈 결과 반환 if not all_analyzed_stocks or len(all_analyzed_stocks) < 1: - logger.warning(f"No analysis data found ({len(all_analyzed_stocks) if all_analyzed_stocks else 0} stocks). Run batch analysis first.") return { "total": 0, "active_rules": len(active_rules), @@ -1346,59 +281,32 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): "message": "⚠️ 배치 분석 데이터 없음. 'make batch-test' 명령어로 분석 실행 필요." } - logger.info(f"Found {len(all_analyzed_stocks)} stocks in DB for {today_kst}") - - # 규칙별로 매칭되는 종목 찾기 rule_matches = [] recommendations_by_rule = [] - stock_to_rules = {} # 종목코드 -> 매칭된 규칙 리스트 매핑 + stock_to_rules = {} - # 시장 레짐 판별 (섹션 7.1: 200일 MA 기반) market_regime_info = await check_market_regime() current_market_regime = market_regime_info.get("regime_200d", "bear") for rule in active_rules: - logger.info(f"Applying rule: {rule.name}") - conditions = rule.conditions actions = rule.actions - rule_matched_stocks = [] for stock in all_analyzed_stocks: matches = True - - # === 전략 타입별 분기 === - - # 1) 평균회귀 전략 (RSI + 볼린저 반등) if rule.rule_type == "MEAN_REVERSION": matches = is_mean_reversion_candidate(stock, conditions, current_market_regime) - - # 2) 추세/골든크로스 전략 (나중에 별도 함수로 분리 예정) - elif rule.rule_type == "TREND": - # 여기서는 당장 비활성화하거나, 매우 보수적으로만 사용 - # matches = is_trend_candidate(stock, conditions) - matches = False # 일단 평균회귀 안정화 전까지 비활성화 - - # 3) 박스권 돌파 / 모멘텀 전략 (위험 축) - elif rule.rule_type in ("BREAKOUT", "MOMENTUM"): - # matches = is_breakout_or_momentum_candidate(stock, conditions) - matches = False # 초기에는 끄고, 나중에 별도 엔진으로 - - # 기타 타입은 현재 스킵 else: matches = False if not matches: continue - # 여기까지 왔으면 rule_matched_stocks에 추가 rule_matched_stocks.append(stock) - # 점수순 정렬 (score=None 안전 처리) rule_matched_stocks.sort(key=lambda x: x.score or 0, reverse=True) - # 이하 로직은 그대로 (Top N, rule_matches, stock_to_rules, 실시간 가격 업데이트 등) rule_stocks = [] for stock in rule_matched_stocks[:10]: stock_dict = _stock_analysis_to_dict(stock, rule=rule, actions=actions) @@ -1427,11 +335,7 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): "stocks": rule_stocks, }) - logger.info(f"Rule '{rule.name}' matched {len(rule_matched_stocks)} stocks, showing top {len(rule_stocks)}") - - # 규칙 매칭 결과가 없으면 분석 종목 상위 20개를 폴백으로 추가 if not rule_matches and all_analyzed_stocks: - logger.info("No rule matches found. Using top scored analyzed stocks as fallback recommendations.") already_added = set() fallback_stocks = [] for stock in all_analyzed_stocks[:20]: @@ -1452,7 +356,6 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): "matched_count": len(fallback_stocks), "stocks": fallback_stocks, }) - logger.info(f"Added {len(fallback_stocks)} fallback stocks from analyzed_stock_list") return { "total": len(rule_matches), @@ -1474,43 +377,25 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): "recommendations": rule_matches, "recommendations_by_rule": recommendations_by_rule, "rules_applied": [{"id": r.id, "name": r.name, "priority": r.priority} for r in active_rules], - # DB에 저장된 실제 배치 분석 날짜 (TIMESTAMP/DATE 모두 YYYY-MM-DD 반환) "batch_time": (lambda d: d.date() if isinstance(d, datetime) else d)( max((s.analysis_date for s in all_analyzed_stocks), default=date.today()) ).isoformat(), } - async def get_rule_based_recommendations_realtime(): - """ - 실시간 KIS API 분석 (느림 - 배치 주기에 호출됨) - 규칙은 trading_rules DB → JSON 파일 → 하드코딩 순으로 로드한다. - """ - logger.info("Rule-based recommendations requested (realtime mode)") - async with AsyncSessionLocal() as _db: active_rules = await _load_active_rules_from_db(_db) - logger.info(f"Found {len(active_rules)} active trading rules (realtime)") - # active_rules가 비어있으면 기본 지표로 분석 if not active_rules: - logger.info("No rules in database, using default indicators") - - # 규칙이 없으면 기본 RSI + 이동평균으로 분석 - if not active_rules: - logger.info("Using default criteria: RSI + Moving Average") criteria = RecommendationCriteria( use_rsi=True, use_moving_average=True ) - - # 분석할 종목 리스트 all_stocks = await get_all_tradable_stocks() - stocks_to_analyze = all_stocks[:20] # 20개 종목으로 단축 (속도 개선) + stocks_to_analyze = all_stocks[:20] - # 배치 처리 (속도 개선) - batch_size = 5 # 배치 크기 - delay_between_batches = 0.3 # 딜레이 추가 감소 + batch_size = 5 + delay_between_batches = 0.3 batch_results = [] for i in range(0, len(stocks_to_analyze), batch_size): @@ -1522,14 +407,12 @@ async def get_rule_based_recommendations_realtime(): if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) - # 유효한 추천만 필터링 valid_recommendations = [ r for r in batch_results if r is not None and not isinstance(r, Exception) ] valid_recommendations.sort(key=lambda x: x.score, reverse=True) - # 기본 규칙으로 그룹핑 default_stocks = [] for rec in valid_recommendations[:10]: stock_dict = { @@ -1567,14 +450,10 @@ async def get_rule_based_recommendations_realtime(): "rules_applied": [], "message": "기본 지표로 분석되었습니다 (규칙 미설정)" } - - logger.info(f"Found {len(active_rules)} active trading rules") - # 분석할 종목 리스트 all_stocks = await get_all_tradable_stocks() stocks_to_analyze = all_stocks[:settings.ANALYSIS_STOCK_COUNT] - # 보유 포트폴리오 종목 추가 (보유 종목은 거래대금 순위 무관하게 항상 분석) try: from app.models.models import Portfolio as _PortfolioModel async with AsyncSessionLocal() as _pf_db: @@ -1588,24 +467,20 @@ async def get_rule_based_recommendations_realtime(): stocks_to_analyze = list(stocks_to_analyze) stocks_to_analyze.append((_pf.stock_code, _pf.stock_name or _pf.stock_code)) existing_codes.add(_pf.stock_code) - logger.info(f"Added portfolio stocks → total {len(stocks_to_analyze)} stocks to analyze") except Exception as _e: logger.warning(f"포트폴리오 종목 추가 실패 (non-critical): {_e}") - # ✅ 최적화: 모든 종목을 한 번만 분석 (모든 지표 포함) - logger.info(f"Analyzing {len(stocks_to_analyze)} stocks once with all indicators...") criteria = RecommendationCriteria( use_rsi=True, use_moving_average=True, use_macd=True, use_bollinger=True, use_stochastic=True, - for_batch=True, # score=0 종목도 반환하여 모든 지표 DB 저장 + for_batch=True, ) - # 배치 처리로 종목 분석 (Rate Limiting) - batch_size = 1 # KIS API rate limit: 초당 1건 - delay_between_batches = 1.5 # 배치 간 1.5초 대기 + batch_size = 1 + delay_between_batches = 1.5 all_analyzed_stocks = [] for i in range(0, len(stocks_to_analyze), batch_size): @@ -1614,59 +489,45 @@ async def get_rule_based_recommendations_realtime(): results = await asyncio.gather(*tasks, return_exceptions=True) all_analyzed_stocks.extend(results) - # 다음 배치 전 대기 (Rate Limiting) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) - # 유효한 분석 결과만 필터링 valid_analyzed_stocks = [ r for r in all_analyzed_stocks if r is not None and not isinstance(r, Exception) ] - logger.info(f"Successfully analyzed {len(valid_analyzed_stocks)} stocks") - - # 시장 레짐 판별 (DB 모드와 동일한 로직) market_regime_info = await check_market_regime() current_market_regime_str = market_regime_info.get("regime_200d", "bear") - # 규칙별로 매칭되는 종목 찾기 (이미 분석된 데이터 재사용) rule_matches = [] recommendations_by_rule = [] for rule in active_rules: - logger.info(f"Applying rule: {rule.name}") - conditions = rule.conditions actions = rule.actions - rule_matched_stocks = [] for stock in valid_analyzed_stocks: - # ETF/레버리지/인버스/KODEX/선물 종목 제외 etf_keywords = ["ETF", "etf", "레버리지", "인버스", "KODEX", "선물"] if any(k in stock.stock_name for k in etf_keywords): continue - # === 통합 규칙 조건 매칭 (DB 모드와 동일한 로직) === if rule.rule_type == "MEAN_REVERSION": matches = is_mean_reversion_candidate(stock, conditions, current_market_regime_str) elif rule.rule_type in ("TREND", "BREAKOUT", "MOMENTUM"): - matches = False # 미구현 (DB 모드와 동일) + matches = False else: - # BUY 타입: RSI 임계값 체크 rsi_threshold = conditions.get("rsi_threshold", 40) matches = (stock.rsi is not None and stock.rsi <= rsi_threshold) if matches: rule_matched_stocks.append(stock) - # 점수순 정렬 (score=None 안전 처리) rule_matched_stocks.sort(key=lambda x: x.score or 0, reverse=True) - # 규칙별로 종목 저장 rule_stocks = [] - for rec in rule_matched_stocks[:10]: # 규칙당 상위 10개 + for rec in rule_matched_stocks[:10]: stock_with_rule = { **rec.model_dump(), "rule_id": rule.id, @@ -1683,7 +544,6 @@ async def get_rule_based_recommendations_realtime(): rule_stocks.append(stock_with_rule) rule_matches.append(stock_with_rule) - # 규칙별 그룹 추가 if rule_stocks: recommendations_by_rule.append({ "rule_id": rule.id, @@ -1694,40 +554,29 @@ async def get_rule_based_recommendations_realtime(): "matched_count": len(rule_matched_stocks), "stocks": rule_stocks }) - - logger.info(f"Rule '{rule.name}' matched {len(rule_matched_stocks)} stocks, showing top {len(rule_stocks)}") - # 우선순위와 점수로 정렬 rule_matches.sort(key=lambda x: (x["rule_priority"], -x["score"])) - # 실시간 현재가 업데이트 (상위 20개만) - # 토큰 재요청을 줄이기 위해 공유 인스턴스를 사용합니다. from app.services.korea_investment_api import korea_investment_api as kis_api - top_recommendations = rule_matches[:20] for rec in top_recommendations: try: - await asyncio.sleep(0.2) # Rate limit 고려 + await asyncio.sleep(0.2) price_data = await kis_api.get_current_price(rec["stock_code"]) if price_data and price_data.get("output"): output = price_data["output"] rec["current_price"] = int(output.get("stck_prpr", rec["current_price"])) rec["change_rate"] = float(output.get("prdy_ctrt", rec.get("change_rate", 0))) - logger.info(f"Updated real-time price for {rec['stock_code']}: {rec['current_price']}원") except Exception as e: logger.warning(f"Failed to get real-time price for {rec['stock_code']}: {e}") - # recommendations_by_rule의 종목들도 실시간 가격 업데이트 for rule_group in recommendations_by_rule: for stock in rule_group["stocks"]: - # top_recommendations에서 업데이트된 가격 찾기 updated = next((r for r in top_recommendations if r["stock_code"] == stock["stock_code"]), None) if updated: stock["current_price"] = updated["current_price"] stock["change_rate"] = updated["change_rate"] - # 실시간 분석 결과를 DB에 저장 (규칙 매칭 여부와 무관하게 분석된 전 종목 저장) - # N+1 SELECT 루프 → 단일 bulk UPSERT로 대체 (성능 개선) from app.services.analysis_db_service import bulk_upsert_analysis as _bulk_upsert from datetime import date as _date _today = _date.today() @@ -1735,793 +584,26 @@ async def get_rule_based_recommendations_realtime(): async with AsyncSessionLocal() as _db: saved = await _bulk_upsert(_db, _items, _today) await _db.commit() - logger.info(f"✅ [Batch] DB upsert 완료: {saved}종목 저장") + return { "total": len(rule_matches), "active_rules": len(active_rules), "analyzed_stocks": len(stocks_to_analyze), - "recommendations": top_recommendations, # 전체 통합 리스트 (실시간 가격 포함) - "recommendations_by_rule": recommendations_by_rule, # 규칙별 그룹핑 (실시간 가격 포함) + "recommendations": top_recommendations, + "recommendations_by_rule": recommendations_by_rule, "rules_applied": [{"id": r.id, "name": r.name, "priority": r.priority} for r in active_rules] } - -async def _mcp_validate_buy(stock: dict) -> tuple: - """MCP 서버에 AI 매수 검증 요청 - - Returns: - (approved: bool, reason: str) - - 오류/타임아웃 시 (True, "mcp_unavailable") — fail-open (매매 차단 안 함) - """ - if not settings.MCP_VALIDATION_ENABLED: - return True, "mcp_disabled" - - try: - payload = { - "analysis_type": "RULE_VALIDATION", - "data": { - "rule": { - "id": stock.get("rule_id"), - "name": stock.get("rule_name", "auto_trade"), - }, - "market_data": { - "stock_code": stock.get("stock_code"), - "stock_name": stock.get("stock_name"), - "current_price": stock.get("current_price"), - "signals": stock.get("signals", []), - "score": stock.get("score"), - "ml_signal": stock.get("ml_signal"), - "ml_confidence": stock.get("ml_confidence"), - "ma20": stock.get("ma20"), - "ma60": stock.get("ma60"), - } - } - } - - async with httpx.AsyncClient(timeout=10.0) as client: - resp = await client.post( - f"{settings.MCP_SERVER_URL}/analyze", - json=payload - ) - resp.raise_for_status() - data = resp.json() - - result = data.get("result", {}) - decision = result.get("decision", "HOLD") - confidence = float(result.get("confidence", 0.0)) - - logger.info( - f"🤖 MCP 검증: {stock.get('stock_name')} → " - f"decision={decision}, confidence={confidence:.2f}" - ) - - if decision == "BUY" and confidence >= settings.MCP_CONFIDENCE_THRESHOLD: - return True, f"mcp_approved(conf={confidence:.2f})" - else: - return False, f"mcp_rejected(decision={decision}, conf={confidence:.2f})" - - except Exception as e: - logger.warning(f"⚠️ MCP 검증 실패 — 매매 차단: {e}") - return False, "mcp_unavailable" # 장애 시 매매 차단 (fail-closed) - - -async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession): - """백그라운드에서 자동매매 실행""" - try: - trade_results = [] - total_processed = 0 - skipped_held = 0 # 이미 보유 중인 종목 수 - - # 🐻 하락장 방어: 시장 레짐에 따라 매매 건수 제한 - market_regime = None - if settings.BEAR_MARKET_FILTER_ENABLED: - market_regime = await check_market_regime() - logger.info(f"🐻 자동매매 시장 상태: {market_regime['message']}") - - regime_max_trades = market_regime.get("max_trades", 20) if market_regime else 20 - position_multiplier = market_regime.get("position_multiplier", 1.0) if market_regime else 1.0 - max_trades = min(20, regime_max_trades) - - # 현재 보유 중인 종목 조회 - from app.models.models import Portfolio - result = await db.execute( - select(Portfolio).where(Portfolio.quantity > 0) - ) - held_stocks = {p.stock_code for p in result.scalars().all()} - if held_stocks: - logger.info(f"현재 보유 중: {len(held_stocks)}개 종목") - - # ✅ 중복 제거: 모든 규칙 그룹의 종목을 합쳐서 중복 제거 - unique_stocks = {} - for rule_group in recommendations_response["recommendations_by_rule"]: - for stock in rule_group["stocks"]: - stock_code = stock.get("stock_code") - # 우선순위가 높은 규칙의 종목을 우선 (priority가 낮을수록 우선순위 높음) - if stock_code not in unique_stocks: - unique_stocks[stock_code] = stock - elif stock.get("rule_priority", 999) < unique_stocks[stock_code].get("rule_priority", 999): - unique_stocks[stock_code] = stock - - # 우선순위와 점수로 정렬 - sorted_stocks = sorted( - unique_stocks.values(), - key=lambda s: (s.get("rule_priority", 999), -s.get("score", 0)) - ) - - logger.info(f"📊 자동매매 대상: 총 {len(sorted_stocks)}개 종목 (중복 제거 완료)") - - for stock in sorted_stocks[:max_trades]: - # 🔄 이미 보유 중인 종목도 추가 매수 허용 (일일 2회 제한 내에서) - # if stock.get("stock_code") in held_stocks: - # skipped_held += 1 - # logger.debug(f"⏭️ {stock.get('stock_name')} - 이미 보유중, 건너뜀") - # continue - - try: - # 시장 레짐 정보 전달 (포지션 사이징에서 regime 복원에 사용) - stock["_position_multiplier"] = position_multiplier - stock["_regime_str"] = market_regime.get("regime_200d", "bear") if market_regime else "bear" - result = await execute_auto_trade(stock, db) - total_processed += 1 - - if result["action"] in ["BUY", "SELL"]: - trade_results.append(result) - logger.info(f"✅ {result['action']} {result['stock']} - {result.get('quantity')}주") - - # Rate Limiting: KIS API 호출 간 0.5초 대기 - await asyncio.sleep(0.5) - - except Exception as e: - logger.error(f"Error executing auto-trade for {stock.get('stock_code')}: {e}") - continue - - logger.info(f"🎯 자동매매 완료: {len(trade_results)}건 실행 / {total_processed}개 분석 / {skipped_held}개 보유중") - - except Exception as e: - logger.error(f"❌ Background auto-trade error: {e}") - - -async def _background_sync_portfolio(): - """백그라운드 포트폴리오 동기화 (새 세션 사용)""" - try: - from app.api.endpoints.portfolio import sync_portfolio - async with AsyncSessionLocal() as session: - await sync_portfolio(session, force=True) - except Exception as e: - logger.error(f"Background portfolio sync failed: {e}") - - -async def _delayed_sync_portfolio(delay_seconds: int = 3): - """지연 포트폴리오 동기화 (KIS API 업데이트 대기)""" - try: - await asyncio.sleep(delay_seconds) - from app.api.endpoints.portfolio import sync_portfolio - from app.core.database import AsyncSessionLocal - - async with AsyncSessionLocal() as session: - try: - await sync_portfolio(session, force=True) - logger.info(f"🔄 Delayed portfolio sync completed after {delay_seconds}s") - except Exception as inner_e: - logger.error(f"Sync execution failed: {inner_e}") - except Exception as e: - logger.error(f"Delayed portfolio sync failed: {e}") - - -# 자동 매매 실행 함수 -async def execute_auto_trade(stock: dict, db: AsyncSession): - """ - 추천 종목에 대해 자동 매매 실행 (국내 + 해외 주식) - - 매수 신호: 정배열, RSI < 30 등 - - 매도 신호: 역배열, RSI > 70, 손익률 기준 - """ - try: - stock_code = stock.get("stock_code") - stock_name = stock.get("stock_name") - current_price = float(stock.get("current_price", 0)) - signals = stock.get("signals", []) - - if not stock_code or current_price <= 0: - logger.debug(f"Invalid stock data for auto trade: {stock}") - return {"action": "SKIP", "reason": "invalid_data"} - - # 국내/해외 구분 (6자리 숫자면 국내, 알파벳 포함이면 해외) - is_domestic = stock_code.isdigit() and len(stock_code) == 6 - exchange_code = stock.get("exchange_code", "NASD") # 기본값: 나스닥 - - # 가격 필터 (CLAUDE.md 섹션 6.4) - if is_domestic: - if current_price < settings.MIN_PRICE_KRW: - logger.info(f"가격 필터: {stock_name} ({current_price:,.0f}원 < {settings.MIN_PRICE_KRW:,}원)") - return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} - else: - if current_price < settings.MIN_PRICE_USD: - logger.info(f"가격 필터: {stock_name} (${current_price:.2f} < ${settings.MIN_PRICE_USD:.2f})") - return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} - - # MDD 체크 (섹션 6.3) — async lock으로 thread-safe - if not await mdd_manager.can_open_new_position(): - logger.info(f"MDD 제한: 신규 진입 중단 상태 - {stock_name} 스킵") - return {"action": "SKIP", "reason": "mdd_pause", "stock": stock_name} - - # 연속 손실 금지 체크 (섹션 7.4) - if loss_tracker.is_banned(stock_code): - ban_info = loss_tracker.get_ban_info(stock_code) - logger.info(f"연속 손실 금지: {stock_name} ({ban_info})") - return {"action": "SKIP", "reason": "consecutive_loss_ban", "stock": stock_name} - - # 보유 종목 수 체크 (섹션 6.1: 최대 20종목) - held_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) - held_count = len(held_result.scalars().all()) - if held_count >= settings.MAX_HOLDINGS_COUNT: - logger.info(f"보유 종목 한도: {held_count}/{settings.MAX_HOLDINGS_COUNT} - {stock_name} 스킵") - return {"action": "SKIP", "reason": "max_holdings_reached", "stock": stock_name} - - # 0. 매도 후 재진입 금지 (Cooldown) - 최근 24시간 내 매도 이력 확인 - # 같은 날 손절친 종목을 다시 사는 것을 방지 - from app.models.models import TradeHistory - cutoff_time = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) - - recent_sell = await db.execute( - select(TradeHistory).where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "SELL", - TradeHistory.status == "EXECUTED", - TradeHistory.executed_at >= cutoff_time - ) - ) - if recent_sell.first(): - logger.info(f"🚫 Cooldown active: {stock_name}({stock_code}) was sold today. Skipping BUY.") - return {"action": "SKIP", "reason": "cooldown_sell_today", "stock": stock_name} - - # 이미 보유 중인지 확인 - result = await db.execute( - select(Portfolio).where( - Portfolio.stock_code == stock_code, - Portfolio.quantity > 0 - ) - ) - holding = result.scalar_one_or_none() - - # 신호 문자열 기반 매수/매도 판단 (실제 분석 함수의 문구와 일치시킴) - text_signals = " ".join(signals) - # 점수 숫자형 보정 - try: - score_val = float(stock.get("score", 0) or 0) - except Exception: - score_val = 0.0 - - buy_keywords = [ - "정배열 눌림목 매수", # "정배열 눌림목 매수 (이격도 안정)" - "눌림목 진입 매수타점", # "눌림목 진입 매수타점" - "RSI 바닥 탈출 반등", # "RSI 바닥 탈출 반등 (XX)" - "RSI 안정 상승", # "RSI 안정 상승 (XX)" - "스토캐스틱 과매도", # "스토캐스틱 과매도 (%K: XX)" - "5일선 > 20일선", - "20일선 > 60일선", - "골든크로스", # "MACD 골든크로스" - "볼린저 하한선 터치", - "횡보장 — 평균회귀 유리", # "ADX(XX<20) 횡보장 — 평균회귀 유리" - "거래량 증가와 상승", - ] - should_buy = any(k in text_signals for k in buy_keywords) - - sell_keywords = [ - "역배열 하락 매도타점", - "RSI 과매수", # RSI > 70 (rsi_signal 함수 출력과 일치) - "스토캐스틱 과매수", # stochastic K > 70 (stochastic_signal 함수 출력과 일치) - "데드크로스", # "MACD 데드크로스" 부분 매칭 - "이동평균 역배열", - "볼린저 상한선 돌파", # price > upper_band (bollinger_signal 함수 출력과 일치) - "손절라인 이탈", - ] - should_sell = any(k in text_signals for k in sell_keywords) - - # 🤖 ML 앙상블 기반 매수/매도 판단 (설정 활성화 시) - ml_score_trade = stock.get("ml_score") - ml_confidence_trade = stock.get("ml_confidence") - ml_signal_trade = stock.get("ml_signal") - - if settings.ML_ENSEMBLE_ENABLED and ml_score_trade is not None: - # ML 신호가 BUY이고 신뢰도/스코어 충분하면 매수 - if (ml_signal_trade == "BUY" - and ml_score_trade >= settings.ML_ENSEMBLE_BUY_THRESHOLD - and (ml_confidence_trade or 0) >= settings.ML_ENSEMBLE_CONFIDENCE_MIN): - should_buy = True - logger.info(f"🤖 ML 매수 신호: {stock_name} (score={ml_score_trade}, conf={ml_confidence_trade:.2f})") - # ML 신호가 SELL이면 매도 강화 - elif ml_signal_trade == "SELL" and (ml_confidence_trade or 0) >= 0.5: - should_sell = True - logger.info(f"🤖 ML 매도 신호: {stock_name} (score={ml_score_trade}, conf={ml_confidence_trade:.2f})") - # ML 신뢰도가 낮으면 매수 억제 - elif ml_signal_trade == "HOLD" and (ml_confidence_trade or 0) < settings.ML_ENSEMBLE_CONFIDENCE_MIN: - should_buy = False - logger.info(f"🤖 ML 관망 신호로 매수 억제: {stock_name}") - - # ✅ 규칙 기반 추천 종목도 안전장치 적용 (이격도 과열, 급등 체크) - # rule_id가 있을 때만 실제 규칙 매칭으로 간주 (폴백 "고점수 분석 종목"은 제외) - has_rule = stock.get("rule_id") is not None - if has_rule: - # 이격도 과열 시 매수 금지 - if "이격도 과열" in text_signals or "매수금지" in text_signals: - logger.info(f"🚫 규칙 추천이지만 이격도 과열로 매수 금지: {stock_name}") - return {"action": "HOLD", "stock": stock_name, "reason": "disparity_overheated"} - - # 급등 종목 추격 금지 체크 - if "급등 종목 추격 금지" in text_signals or "급등과열" in text_signals: - logger.info(f"🚫 규칙 추천이지만 급등/과열 금지: {stock_name}") - return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} - - # ML이 명시적으로 HOLD를 내렸으면 규칙 기반 강제매수 금지 - # (ML 활성화 AND ML=HOLD AND 이전 단계에서 should_buy가 False로 설정됨) - ml_hold_active = ( - settings.ML_ENSEMBLE_ENABLED - and ml_signal_trade == "HOLD" - and not should_buy - ) - if ml_hold_active: - logger.info(f"🤖 ML 관망으로 규칙 기반 강제매수 차단: {stock_name}") - else: - should_buy = True - logger.info(f"📋 규칙 기반 추천: {stock_name} - {stock.get('rule_name', '매수규칙')}") - - # 폴백 종목(규칙 미매칭): 강한 신호 1개 이상 OR 약한 신호 2개 이상 + score ≥ 2 필요 - # 규칙 매칭 종목(has_rule=True)은 is_mean_reversion_candidate()를 통과했으므로 그대로 진행 - if should_buy and not has_rule: - strong_buy_kw = [ - "RSI 바닥 탈출 반등", - "볼린저 하한선 터치", - "눌림목 진입 매수타점", - "정배열 눌림목 매수", - ] - has_strong_signal = any(k in text_signals for k in strong_buy_kw) - matched_count = sum(1 for k in buy_keywords if k in text_signals) - if not has_strong_signal and (matched_count < 2 or score_val < 2): - should_buy = False - logger.info( - f"📉 복합 신호 부족으로 매수 억제: {stock_name}" - f" (신호={matched_count}개, score={score_val:.1f})" - ) - - # 매수 로직 (명시적 매수 신호 필요 — score 단독으로는 진입 불가) - # 프론트 "관망/0%" 표시와 실행 로직 일치: should_buy=True여야만 진입 - if should_buy: - # 🚫 추격매수 방지: 급등 종목 필터 - try: - if is_domestic: - chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10) - chart_data = chart_resp.get("output2", []) - else: - chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") - chart_data = chart_resp.get("output2", []) - - if len(chart_data) >= 4: - # 국내/해외 데이터 키 차이 처리 - if is_domestic: - closes = [float(item.get("stck_clpr", 0)) for item in chart_data[-4:]] - else: - closes = [float(item.get("clos", 0)) for item in chart_data[-4:]] - - if len(closes) >= 4: - price_3days_ago = closes[0] - recent_gain = ((current_price - price_3days_ago) / price_3days_ago) * 100 if price_3days_ago > 0 else 0 - if recent_gain > 10: # 3일간 10% 이상 급등 - logger.info(f"🚫 급등 종목 추격 금지: {stock_name} (3일 +{recent_gain:.1f}%)") - return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} - except Exception as e: - logger.warning(f"⚠️ 급등 체크 실패, 계속 진행: {e}") - - # Enforce trend filter for buys - if TREND_FILTER_ENABLED: - try: - ma20 = stock.get("ma20") - ma60 = stock.get("ma60") - # If moving averages not provided, fetch quick chart to compute - if ma20 is None or ma60 is None: - if is_domestic: - chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D") - closes_local = [float(it.get("stck_clpr", 0)) for it in (chart_resp.get("output") or [])] - else: - chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") - closes_local = [float(it.get("clos", 0)) for it in (chart_resp.get("output2") or [])] - - if len(closes_local) >= max(TREND_MA_SHORT, TREND_MA_LONG): - ma20 = _ma(closes_local, TREND_MA_SHORT) - ma60 = _ma(closes_local, TREND_MA_LONG) - # 허용 오차 적용 - if not (ma20 and ma60 and (float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT)) and (current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT))): - logger.info(f"🚫 Trend filter rejected BUY: {stock_name} (price={current_price}, ma20={ma20}, ma60={ma60}, tol={TREND_TOLERANCE_PCT})") - return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"} - except Exception as e: - logger.info(f"⚠️ Trend filter check failed, skipping buy as safe default: {e}") - return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"} - # 포트폴리오 요약 정보 - try: - from app.api.endpoints.portfolio import portfolio_cache - available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0 - total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0 - except Exception: - available_cash, total_assets = 0.0, 0.0 - - # 🔥 하루 매매 횟수 제한 (당일 0시부터 카운트) - daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) - daily_buy_check = await db.execute( - select(TradeHistory) - .where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), # COMPLETED도 포함 - TradeHistory.created_at >= daily_cutoff, - ) - ) - daily_buys = daily_buy_check.scalars().all() - daily_buy_count = len(daily_buys) - - # 🚫 하루 최대 2회 매매 제한 (신규 + 추가) - max_daily_trades = 2 - if daily_buy_count >= max_daily_trades: - logger.info(f"🚫 일일 매매 한도 초과: {stock_name} 당일 {daily_buy_count}/{max_daily_trades}회") - return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"} - - cooldown_min = int(stock.get("add_buy_cooldown_min", 30) or 30) # 추가매수 쿨다운(분) - - # 🔥 쿨다운 체크 (보유 여부와 관계없이 항상 체크) - # 포트폴리오 동기화가 느릴 수 있으므로, DB 거래 내역 기준으로 체크 - cooldown_check = await db.execute( - select(TradeHistory) - .where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), - TradeHistory.created_at >= daily_cutoff, - ) - .order_by(desc(TradeHistory.created_at)) - ) - recent_buys = cooldown_check.scalars().all() - last_buy_at = recent_buys[0].created_at if recent_buys else None - - # 쿨다운 체크 - if last_buy_at and (_get_kst_now() - last_buy_at).total_seconds() < cooldown_min * 60: - logger.info(f"⏳ 추가매수 대기: {stock_name} {cooldown_min}분 쿨다운 미경과") - return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} - - # 목표/최대 비중 기반 투자금 계산 (이전에 최근 매수 기록 체크) - # 보유중이 아니더라도(아직 싱크 안됨) 최근 매수 시도가 있으면 스킵 - recent_buy_cutoff = _get_kst_now() - timedelta(minutes=5) - check_recent = await db.execute( - select(TradeHistory).where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.created_at >= recent_buy_cutoff, - TradeHistory.status.in_(['EXECUTED', 'COMPLETED', 'PENDING']) - ) - ) - if check_recent.first(): - logger.info(f"⏳ 최근 5분 내 매수 이력 존재: {stock_name} - 스킵") - return {"action": "SKIP", "stock": stock_name, "reason": "recent_buy_5min"} - - # ✅ MCP AI 검증 (BUY 최종 확인 - 모든 사전 필터 통과 후) - mcp_ok, mcp_reason = await _mcp_validate_buy(stock) - if not mcp_ok: - logger.info(f"🚫 MCP AI 매수 거부: {stock_name} ({mcp_reason})") - return {"action": "HOLD", "stock": stock_name, "reason": mcp_reason} - - # 동적 포지션 사이징 (CLAUDE.md 섹션 6.1: 종목당 최대 4%, regime/RSI/ADX 반영) - current_value = float(holding.quantity * current_price) if holding else 0.0 - regime_str = stock.get("_regime_str", "bear") - regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR - invest_amount = calculate_position_size_v2( - total_assets=total_assets, - available_cash=available_cash, - current_strategy_allocation=current_value, - regime=regime, - rsi=stock.get("rsi"), - adx=stock.get("adx_14"), - ) - if invest_amount <= 0 and total_assets <= 0: - # 총자산 미확정 시 보수적으로 현금의 4% - invest_amount = available_cash * 0.04 - - quantity = int(invest_amount / current_price) - - logger.info(f"💰 {stock_name}({stock_code}): 예수금={available_cash:,.0f}원, 투자금액={invest_amount:,.0f}원, 수량={quantity}주, 가격={current_price:,.0f}원") - - if quantity > 0: - logger.info(f"🟢 매수 신호 감지: {stock_name}({stock_code}) - {signals}") - - # KIS API로 매수 주문 (지정가: 현재가로 주문) - try: - if is_domestic: - order_result = await korea_investment_api.place_order( - stock_code=stock_code, - order_type="BUY", - quantity=quantity, - price=int(current_price) # 지정가 주문 (현재가) - ) - else: - order_result = await korea_investment_api.place_overseas_order( - symbol=stock_code, - order_type="BUY", - quantity=quantity, - price=float(current_price), # 해외 주식은 달러 단위 - exchange_code=exchange_code - ) - - # 주문 결과 상세 로깅 - logger.info(f"📝 매수 주문 결과: {stock_name}({stock_code}) - rt_cd={order_result.get('rt_cd')}, msg={order_result.get('msg1')}, odno={order_result.get('odno')}") - logger.debug(f"Full order result: {order_result}") - - # 중복 거래 방지: 최근 10초 이내 동일 종목 동일 타입 주문 확인 - recent_cutoff = _get_kst_now() - timedelta(seconds=10) - duplicate_check = await db.execute( - select(TradeHistory).where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.created_at >= recent_cutoff - ) - ) - recent_duplicate = duplicate_check.scalar_one_or_none() - - if recent_duplicate: - logger.warning(f"⚠️ 중복 매수 방지: {stock_name} - 10초 이내 동일 주문 존재") - return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} - - # 거래 이력 저장 (상세 정보 포함) - total_amount = Decimal(str(current_price)) * quantity - entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점" - - success = order_result.get("rt_cd") == "0" - error_msg = order_result.get("msg1", "") if not success else "" - - # 주문 상태 결정 - if success: - status = "COMPLETED" # 체결 완료 - elif order_result.get("rt_cd") == "1": - status = "PENDING" # 주문 접수됨, 체결 대기 - else: - status = "FAILED" # 주문 실패 - - trade = TradeHistory( - rule_id=stock.get("rule_id"), - stock_code=stock_code, - stock_name=stock_name, - order_type="BUY", - order_price=Decimal(str(current_price)), - order_quantity=quantity, - executed_price=Decimal(str(current_price)) if success else None, - executed_quantity=quantity if success else 0, - total_amount=total_amount, - status=status, - order_id=order_result.get("odno"), - message=f"자동매수: {'추가매수' if holding else '신규'} | {' | '.join(signals[:2])} | {error_msg if not success else '체결완료'}", - entry_reason=entry_reason, - signals=signals, - created_at=_get_kst_now(), - executed_at=_get_kst_now() if success else None - ) - db.add(trade) - await db.commit() - - if success: - logger.info(f"✅ 매수 체결 완료: {stock_name} {quantity}주 @ {current_price}원") - elif status == "PENDING": - logger.info(f"⏳ 매수 주문 접수: {stock_name} {quantity}주 @ {current_price}원 - 체결 대기중") - else: - logger.warning(f"❌ 매수 주문 실패: {stock_name} {quantity}주 - {error_msg}") - # Optimistic local 포트폴리오 업데이트 - if success: - if holding: - # 가중 평균 재계산 - new_total_qty = holding.quantity + quantity - prev_cost = float(holding.average_price or 0) * holding.quantity - new_cost = prev_cost + current_price * quantity - holding.quantity = new_total_qty - holding.average_price = Decimal(str(new_cost / new_total_qty)) - holding.current_price = Decimal(str(current_price)) - else: - new_holding = Portfolio( - stock_code=stock_code, - stock_name=stock_name, - quantity=quantity, - average_price=Decimal(str(current_price)), - current_price=Decimal(str(current_price)), - profit_loss=Decimal("0"), - profit_loss_rate=Decimal("0") - ) - db.add(new_holding) - await db.commit() - # 포트폴리오 캐시 즉시 현금 차감 (미수 방지) - # 캐시 완전 초기화 대신 available_cash만 즉시 차감 → 동일 사이클 내 연속 매수 차단 - try: - from app.api.endpoints.portfolio import portfolio_cache - order_cost = quantity * current_price - if portfolio_cache.summary: - old_cash = float(portfolio_cache.summary.get("available_cash", 0)) - portfolio_cache.summary["available_cash"] = max(0.0, old_cash - order_cost) - logger.info( - f"💸 예수금 즉시 차감: {stock_name}({stock_code}) " - f"{order_cost:,.0f}원 차감 → 잔여 {portfolio_cache.summary['available_cash']:,.0f}원" - ) - portfolio_cache.change_flag = True - portfolio_cache.last_trade_id = None # 다음 sync 강제 - except Exception as e: - logger.error(f"Portfolio cache cash deduction failed after BUY: {e}") - return {"action": "BUY", "stock": stock_name, "quantity": quantity, "price": current_price, "success": success} - - except Exception as e: - logger.error(f"❌ 매수 주문 실패: {stock_name} - {e}") - return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)} - else: - logger.debug(f"⚠️ {stock_name}({stock_code}): 매수 수량 0 (가격={current_price:,.0f}원, 예수금={available_cash:,.0f}원)") - return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name} - - # 매도 로직 - elif should_sell and holding: - quantity = holding.quantity - # DB 저장값은 동기화 지연으로 stale할 수 있으므로 실시간 가격으로 재계산 - _avg = float(holding.average_price or 0) - if _avg > 0 and current_price > 0: - profit_loss_rate = ((current_price - _avg) / _avg) * 100 - else: - profit_loss_rate = float(holding.profit_loss_rate or 0) - - # 익절/손절 조건 확인 - take_profit = stock.get("take_profit", 3.0) # 3% 익절(권장) - stop_loss = stock.get("stop_loss", -1.5) # -1.5% 손절(권장) - - should_execute_sell = ( - profit_loss_rate >= take_profit or # 익절 - profit_loss_rate <= stop_loss or # 손절 - should_sell # 신호 기반 매도 - ) - - if should_execute_sell: - logger.info(f"🔴 매도 신호 감지: {stock_name}({stock_code}) - 손익률: {profit_loss_rate:.2f}% - {signals}") - - try: - if is_domestic: - order_result = await korea_investment_api.place_order( - stock_code=stock_code, - order_type="SELL", - quantity=quantity, - price=int(current_price) # 지정가 주문 (현재가) - ) - else: - order_result = await korea_investment_api.place_overseas_order( - symbol=stock_code, - order_type="SELL", - quantity=quantity, - price=float(current_price), # 해외 주식은 달러 단위 - exchange_code=exchange_code - ) - - # 주문 결과 상세 로깅 - logger.info(f"📝 매도 주문 결과: {stock_name}({stock_code}) - rt_cd={order_result.get('rt_cd')}, msg={order_result.get('msg1')}, odno={order_result.get('odno')}") - logger.debug(f"Full order result: {order_result}") - - # 중복 거래 방지: 최근 10초 이내 동일 종목 동일 타입 주문 확인 - recent_cutoff = _get_kst_now() - timedelta(seconds=10) - duplicate_check = await db.execute( - select(TradeHistory).where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "SELL", - TradeHistory.created_at >= recent_cutoff - ) - ) - recent_duplicate = duplicate_check.scalar_one_or_none() - - if recent_duplicate: - logger.warning(f"⚠️ 중복 매도 방지: {stock_name} - 10초 이내 동일 주문 존재") - return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} - - # 거래 이력 저장 (상세 정보 포함) - total_amount = Decimal(str(current_price)) * quantity - avg_price = float(holding.average_price or current_price) - profit_loss = (current_price - avg_price) * quantity - - # 매도 이유 상세 작성 - exit_reasons = [] - if profit_loss_rate >= take_profit: - exit_reasons.append(f"익절 달성 ({profit_loss_rate:.2f}% >= {take_profit}%)") - if profit_loss_rate <= stop_loss: - exit_reasons.append(f"손절 실행 ({profit_loss_rate:.2f}% <= {stop_loss}%)") - if should_sell: - exit_reasons.append(f"매도 신호: {', '.join(signals[:2])}") - - exit_reason = ". ".join(exit_reasons) + f". 평균단가: {avg_price:,.0f}원, 매도가: {current_price:,.0f}원, 손익: {profit_loss:,.0f}원" - - success = order_result.get("rt_cd") == "0" - error_msg = order_result.get("msg1", "") if not success else "" - - # 주문 상태 결정 - if success: - status = "COMPLETED" # 체결 완료 - elif order_result.get("rt_cd") == "1": - status = "PENDING" # 주문 접수됨, 체결 대기 - else: - status = "FAILED" # 주문 실패 - - trade = TradeHistory( - rule_id=stock.get("rule_id"), - stock_code=stock_code, - stock_name=stock_name, - order_type="SELL", - order_price=Decimal(str(current_price)), - order_quantity=quantity, - executed_price=Decimal(str(current_price)) if success else None, - executed_quantity=quantity if success else 0, - total_amount=total_amount, - status=status, - order_id=order_result.get("odno"), - message=f"자동매도: 손익률 {profit_loss_rate:.2f}% | {' | '.join(signals[:2])} | {error_msg if not success else '체결완료'}", - exit_reason=exit_reason, - signals=signals, - net_profit_loss=Decimal(str(profit_loss)) if success else None, - profit_loss_rate=Decimal(str(profit_loss_rate)) if success else None, - created_at=_get_kst_now(), - executed_at=_get_kst_now() if success else None - ) - db.add(trade) - await db.commit() - - if success: - logger.info(f"✅ 매도 주문 성공: {stock_name} {quantity}주 @ {current_price}원") - else: - logger.warning(f"❌ 매도 주문 실패: {stock_name} {quantity}주 - {error_msg}") - - if success: - # 포트폴리오에서 완전 삭제 - await db.delete(holding) - await db.commit() - logger.info(f"🗑️ Deleted {stock_name}({stock_code}) from portfolio DB") - - # 포트폴리오 캐시 완전 초기화 - try: - from app.api.endpoints.portfolio import portfolio_cache - # 캐시 완전 초기화 - 다음 조회시 강제로 KIS API 호출 - portfolio_cache.holdings = [] - portfolio_cache.summary = None - portfolio_cache.last_synced = None - portfolio_cache.last_trade_id = None - portfolio_cache.change_flag = True - logger.info(f"🔄 Portfolio cache CLEARED after SELL {stock_name}({stock_code})") - except Exception as e: - logger.error(f"Portfolio cache clear failed after SELL: {e}") - return {"action": "SELL", "stock": stock_name, "quantity": quantity, "price": current_price, "profit_loss": profit_loss_rate, "success": success} - - except Exception as e: - logger.error(f"❌ 매도 주문 실패: {stock_name} - {e}") - return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)} - - # 아무 것도 실행 안됨 - logger.info( - f"⏸️ HOLD: {stock_name}({stock_code}) - 매수신호={should_buy}, 매도신호={should_sell}, 보유중={bool(holding)}, 점수={score_val}" - ) - return {"action": "HOLD", "stock": stock_name} - - except Exception as e: - logger.error(f"❌ 자동 매매 오류: {e}") - return {"action": "ERROR", "error": str(e)} - - @router.post("/auto-trade") async def execute_auto_trading(db: AsyncSession = Depends(get_db)): - """ - 규칙 기반 추천 종목에 대해 자동 매매 실행 - """ try: - # 규칙 기반 추천 조회 (실시간 기준으로 매매 판단 일치화) - # DB 캐시의 신호 문자열 불일치로 매수가 안 걸리는 문제를 방지하기 위해 - # 실시간 분석 결과를 우선 사용합니다. recommendations_response = await get_rule_based_recommendations(realtime=True) - if not recommendations_response.get("recommendations_by_rule"): return {"message": "추천 종목이 없습니다", "trades": []} - # 각 규칙별 추천 종목에 대해 자동 매매 실행 trade_results = [] total_processed = 0 - max_trades = 5 # 한 번에 최대 5개만 처리 + max_trades = 5 for rule_group in recommendations_response["recommendations_by_rule"]: if total_processed >= max_trades: @@ -2538,7 +620,6 @@ async def execute_auto_trading(db: AsyncSession = Depends(get_db)): if result["action"] in ["BUY", "SELL"]: trade_results.append(result) - # Rate Limiting: KIS API 호출 간 0.5초 대기 await asyncio.sleep(0.5) except Exception as e: @@ -2555,19 +636,16 @@ async def execute_auto_trading(db: AsyncSession = Depends(get_db)): logger.error(f"❌ 자동 매매 실행 오류: {e}") raise HTTPException(status_code=500, detail=str(e)) - @router.get("/ai/{stock_code}") async def get_ai_recommendation(stock_code: str): - """MCP 서버를 통한 AI 종목 분석 및 뉴스 조회""" + from app.services.korea_investment_api import korea_investment_api try: - # 1. 시장 데이터 조회 (Backend에서 직접 KIS API 호출) price_data = {} try: price_data = await korea_investment_api.get_current_price(stock_code) except Exception as e: logger.warning(f"Failed to fetch price data locally: {e}") - # 2. MCP 서버에 뉴스 조회 요청 news_data = [] try: async with httpx.AsyncClient() as client: @@ -2580,22 +658,18 @@ async def get_ai_recommendation(stock_code: str): if response.status_code == 200: result = response.json() if result.get("status") == "success": - # MCP might wrap it in 'result' -> 'news' or just 'news' data = result.get("result", {}) if "news" in data: news_data = data["news"] else: - news_data = data # Fallback if direct list + news_data = data except Exception as e: logger.warning(f"Failed to fetch news via MCP: {e}") - # 3. AI 분석 요청 (MCP Analyze) analysis_result = "AI 분석을 수행할 수 없습니다." try: async with httpx.AsyncClient() as client: analyze_url = f"{settings.MCP_SERVER_URL}/analyze" - - # 시장 데이터 단순화 market_summary = {} if "output" in price_data: output = price_data["output"] @@ -2615,10 +689,9 @@ async def get_ai_recommendation(stock_code: str): "news": news_data } } - response = await client.post(analyze_url, json=payload, timeout=45.0) # Analysis takes time + response = await client.post(analyze_url, json=payload, timeout=45.0) if response.status_code == 200: analysis_res = response.json() - # If successful, returns {"analysis_type":..., "result": "Markdown text/JSON"} if "result" in analysis_res: analysis_result = analysis_res["result"] except Exception as e: