From ab75dc66f829de6a81027c25d5f59538a8305e6b Mon Sep 17 00:00:00 2001 From: shuim Date: Fri, 6 Mar 2026 15:22:40 +0900 Subject: [PATCH] =?UTF-8?q?=EC=BD=94=EB=93=9C=EB=A6=AC=EB=B7=B0=20?= =?UTF-8?q?=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/core/config.py | 37 +++ backend/app/core/logging_config.py | 31 +- backend/app/main.py | 267 ++++++++++++------ backend/app/services/korea_investment_api.py | 78 +++++- backend/app/services/trade_executor.py | 280 +++++++++++++++---- 5 files changed, 553 insertions(+), 140 deletions(-) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 8cbb853..ec7bd38 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -306,6 +306,43 @@ class Settings(BaseSettings): extra = "ignore" # .env 파일의 추가 필드 무시 env_file_encoding = "utf-8" + def validate_required(self) -> list[str]: + """필수 환경변수 누락 여부를 검사하고 경고 목록을 반환한다. + + 시작 시 lifespan에서 호출하여 운영자에게 조기 경보를 제공한다. + 누락 시 예외를 발생시키지 않으므로 서비스 기동은 계속된다. + """ + warnings: list[str] = [] + + # KIS API 키: 현재 모드에 맞는 키 필수 + if not self.KOREA_INVESTMENT_APP_KEY: + warnings.append( + f"KIS_{self.KIS_MODE.upper()}_APP_KEY 미설정 — " + f"{'KIS_VIRTUAL_APP_KEY' if self.is_virtual_trading else 'KIS_REAL_APP_KEY'} 확인 필요" + ) + if not self.KOREA_INVESTMENT_APP_SECRET: + warnings.append( + f"KIS_{self.KIS_MODE.upper()}_APP_SECRET 미설정" + ) + if not self.KOREA_INVESTMENT_ACCOUNT_NUMBER: + warnings.append( + f"KIS_{self.KIS_MODE.upper()}_ACCOUNT_NO 미설정" + ) + + # DB / Redis + if "localhost" in self.DATABASE_URL and os.path.exists("/.dockerenv"): + warnings.append("DATABASE_URL에 localhost 포함 — Docker 환경에서 postgres 호스트명 확인 필요") + + # MCP 검증 활성화 시 AI 키 필요 + if self.MCP_VALIDATION_ENABLED and not self.ANTHROPIC_API_KEY: + from pydantic_settings import BaseSettings as _BS # lazy import + # GEMINI_API_KEY는 config에 없으므로 환경변수 직접 확인 + gemini_key = os.environ.get("GEMINI_API_KEY", "") + if not gemini_key: + warnings.append("MCP_VALIDATION_ENABLED=True 이지만 ANTHROPIC_API_KEY / GEMINI_API_KEY 미설정") + + return warnings + @lru_cache() def get_settings() -> Settings: diff --git a/backend/app/core/logging_config.py b/backend/app/core/logging_config.py index fae277e..c20e4b6 100644 --- a/backend/app/core/logging_config.py +++ b/backend/app/core/logging_config.py @@ -1,14 +1,36 @@ -"""공통 로깅 설정 — KST(Asia/Seoul) 타임존 포매터 제공.""" +"""공통 로깅 설정 — KST(Asia/Seoul) 타임존 포매터 + 민감 정보 마스킹 제공.""" import logging import datetime +import re import pytz KST = pytz.timezone("Asia/Seoul") +# ── 민감 정보 마스킹 패턴 (API 키, 시크릿, 토큰, 계좌번호) ────────────────────── +_MASK_PATTERNS = [ + # Bearer 토큰 + (re.compile(r'(Bearer\s+)[A-Za-z0-9\-_.]{20,}', re.I), r'\1***'), + # access_token 값 + (re.compile(r'(access_?token[":\s=]+)[A-Za-z0-9\-_.]{20,}', re.I), r'\1***'), + # appsecret / app_secret 값 + (re.compile(r'(app_?secret[":\s=]+)[A-Za-z0-9+/=]{10,}', re.I), r'\1***'), + # appkey / app_key 값 (8자 이상) + (re.compile(r'(app_?key[":\s=]+)[A-Za-z0-9]{8,}', re.I), r'\1***'), + # 계좌번호 (account_no / account_number) + (re.compile(r'(account[_\s]?(?:no|number)[":\s=]+)\d{6,}', re.I), r'\1***'), +] + + +def _mask_sensitive(text: str) -> str: + """로그 텍스트에서 API 키·토큰·계좌번호 등 민감 정보를 마스킹한다.""" + for pattern, replacement in _MASK_PATTERNS: + text = pattern.sub(replacement, text) + return text + class KSTFormatter(logging.Formatter): - """%(asctime)s 를 KST(UTC+9) 기준으로 출력하는 포매터.""" + """%(asctime)s 를 KST(UTC+9) 기준으로 출력하고, 민감 정보를 마스킹하는 포매터.""" def formatTime(self, record: logging.LogRecord, datefmt: str | None = None) -> str: ct = datetime.datetime.fromtimestamp(record.created, tz=KST) @@ -16,3 +38,8 @@ class KSTFormatter(logging.Formatter): return ct.strftime(datefmt) t = ct.strftime(self.default_time_format) return self.default_msec_format % (t, record.msecs) + + def format(self, record: logging.LogRecord) -> str: + """포맷팅 후 민감 정보 마스킹 적용.""" + formatted = super().format(record) + return _mask_sensitive(formatted) diff --git a/backend/app/main.py b/backend/app/main.py index 3df208a..c222bd4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -50,48 +50,54 @@ logger = logging.getLogger(__name__) def _normalize_recommendations(rec: dict) -> dict: - """사로 다른 포맷의 추청 결과를 recommendations_by_rule 포맷으로 정규화. + """서로 다른 포맷의 추천 결과를 recommendations_by_rule 포맷으로 정규화. - get_rule_based_recommendations_realtime() → {"recommendations": [...], ...} - get_rule_based_recommendations_from_db() → {"recommendations_by_rule": {...}, ...} 둘 다 recommendations_by_rule 키를 보장한 상태로 돌려준다. Redis 캐시 데이터와 DB 데이터 모두 이 함수를 통해 정규화. + 변환 실패 시 원본 rec를 그대로 반환 (런타임 에러 방지). """ if not rec: + return rec or {} + + try: + # Case 1: Already normalized (dict) + if isinstance(rec.get("recommendations_by_rule"), dict): + return rec + + # Determine the source of the recommendation list + raw_list = [] + if "recommendations" in rec and isinstance(rec["recommendations"], list): + raw_list = rec["recommendations"] + elif "recommendations_by_rule" in rec and isinstance(rec["recommendations_by_rule"], list): + # This is the case that causes the bug + raw_list = rec["recommendations_by_rule"] + + if not raw_list: + # if there is no list to process, make sure the output has a dict + rec["recommendations_by_rule"] = {} + return rec + + # Group the list by rule + by_rule: dict = {} + for item in raw_list: + rule = ( + item.get("rule_name") + or item.get("signal") + or item.get("strategy") + or "general" + ) + by_rule.setdefault(rule, []).append(item) + + # Return the original dict with the new/replaced 'recommendations_by_rule' + return {**rec, "recommendations_by_rule": by_rule} + + except Exception as e: + logger.warning(f"⚠️ _normalize_recommendations 변환 실패, 원본 반환: {e}") return rec - # Case 1: Already normalized (dict) - if isinstance(rec.get("recommendations_by_rule"), dict): - return rec - - # Determine the source of the recommendation list - raw_list = [] - if "recommendations" in rec and isinstance(rec["recommendations"], list): - raw_list = rec["recommendations"] - elif "recommendations_by_rule" in rec and isinstance(rec["recommendations_by_rule"], list): - # This is the case that causes the bug - raw_list = rec["recommendations_by_rule"] - - if not raw_list: - # if there is no list to process, make sure the output has a dict - rec["recommendations_by_rule"] = {} - return rec - - # Group the list by rule - by_rule: dict = {} - for item in raw_list: - rule = ( - item.get("rule_name") - or item.get("signal") - or item.get("strategy") - or "general" - ) - by_rule.setdefault(rule, []).append(item) - - # Return the original dict with the new/replaced 'recommendations_by_rule' - return {**rec, "recommendations_by_rule": by_rule} - async def batch_analysis_scheduler(): """10분 간격 배치 분석 스케줄러. @@ -99,57 +105,92 @@ async def batch_analysis_scheduler(): 장중 10분마다 KIS API로 종목 분석 후 stock_analysis DB에 upsert. 분석 완료 후 Redis에 recommendations:latest 캐시를 저장해 auto_trade_scheduler가 DB 조회 없이 바로 매매 판단에 활용할 수 있도록 한다. + + 개선: + - 15:30 이후 장 종료 감지 → 익일 08:50까지 자동 중단 + - 분석 실패 시 exponential backoff 재시도 (최대 3회: 10s/20s/40s) """ await asyncio.sleep(15) # 부팅 안정화 (auto_trade_scheduler보다 늦게 시작) from app.api.endpoints.recommendations import get_rule_based_recommendations_realtime from app.core.redis import setex_json, TTL_RECOMMENDATIONS, TTL_ANALYSIS + from datetime import timedelta as _td interval = int(getattr(settings, "BATCH_ANALYSIS_INTERVAL", 600)) # 기본 10분 + _KST = pytz.timezone("Asia/Seoul") logger.info(f"📊 Batch analysis scheduler started (every {interval}s)") try: while True: + now_kst = pytz.datetime.datetime.now(_KST) + + # 15:30 이후 → 장 종료, 익일 08:50까지 대기 + if now_kst.hour > 15 or (now_kst.hour == 15 and now_kst.minute >= 30): + tomorrow_start = ( + _KST.localize( + pytz.datetime.datetime(now_kst.year, now_kst.month, now_kst.day, 8, 50, 0) + ) + _td(days=1) + ) + sleep_sec = max((tomorrow_start - now_kst).total_seconds(), 3600) + logger.info(f"⏸️ [Batch] 장 종료 (15:30 이후). {sleep_sec / 3600:.1f}시간 후 재개") + await asyncio.sleep(sleep_sec) + continue + if not is_market_open(): await asyncio.sleep(1800) continue - try: - logger.info("🔄 [Batch] 10-min analysis cycle starting...") - result = await get_rule_based_recommendations_realtime() - count = len(result.get("recommendations", [])) if result else 0 - logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB") - # ── Redis 캐시 업데이트 ───────────────────────────────────────── - if result: - # 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환) - normalized = _normalize_recommendations(result) - ok = await setex_json( - "recommendations:latest", TTL_RECOMMENDATIONS, normalized - ) - if ok: - rec_by_rule = normalized.get('recommendations_by_rule', []) - rec_count = len(rec_by_rule) if isinstance(rec_by_rule, (list, dict)) else 0 - logger.info( - f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest " - f"(rule 수: {rec_count}, " - f"TTL: {TTL_RECOMMENDATIONS}s)" + # ── exponential backoff 재시도 (최대 3회) ──────────────────────── + max_retries = 3 + for attempt in range(max_retries): + try: + suffix = f" (재시도 {attempt + 1}/{max_retries})" if attempt > 0 else "" + logger.info(f"🔄 [Batch] 10-min analysis cycle starting...{suffix}") + result = await get_rule_based_recommendations_realtime() + count = len(result.get("recommendations", [])) if result else 0 + logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB") + + # ── Redis 캐시 업데이트 ───────────────────────────────────────── + if result: + # 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환) + normalized = _normalize_recommendations(result) + ok = await setex_json( + "recommendations:latest", TTL_RECOMMENDATIONS, normalized ) - - # ===== 개별 종목별 캐시 (병렬 처리 적용) ===== - raw_list = result.get("recommendations", []) - cache_tasks = [] - - for item in raw_list: - code = item.get("stock_code") or item.get("code") - if code: - cache_tasks.append(setex_json(f"analysis:{code}", TTL_ANALYSIS, item)) - - if cache_tasks: - await asyncio.gather(*cache_tasks, return_exceptions=True) - logger.info(f"💾 [Batch] 개별 종목 분석 데이터 캐시 갱신 완료: {len(cache_tasks)}건 (TTL: {TTL_ANALYSIS}s)") - # =============================================== - # ─────────────────────────────────────────────────────── + if ok: + rec_by_rule = normalized.get('recommendations_by_rule', []) + rec_count = len(rec_by_rule) if isinstance(rec_by_rule, (list, dict)) else 0 + logger.info( + f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest " + f"(rule 수: {rec_count}, " + f"TTL: {TTL_RECOMMENDATIONS}s)" + ) + + # ===== 개별 종목별 캐시 (병렬 처리 적용) ===== + raw_list = result.get("recommendations", []) + cache_tasks = [] + + for item in raw_list: + code = item.get("stock_code") or item.get("code") + if code: + cache_tasks.append(setex_json(f"analysis:{code}", TTL_ANALYSIS, item)) + + if cache_tasks: + await asyncio.gather(*cache_tasks, return_exceptions=True) + logger.info(f"💾 [Batch] 개별 종목 분석 데이터 캐시 갱신 완료: {len(cache_tasks)}건 (TTL: {TTL_ANALYSIS}s)") + # =============================================== + # ─────────────────────────────────────────────────────── + break # 성공 시 재시도 루프 탈출 + + except Exception as e: + wait = 10 * (2 ** attempt) # 10s → 20s → 40s + if attempt < max_retries - 1: + logger.warning( + f"⚠️ [Batch] 분석 실패 ({attempt + 1}/{max_retries}), " + f"{wait}초 후 재시도: {e}" + ) + await asyncio.sleep(wait) + else: + logger.error(f"❌ Batch analysis error (최대 재시도 초과): {e}", exc_info=True) - except Exception as e: - logger.error(f"❌ Batch analysis error: {e}", exc_info=True) await asyncio.sleep(interval) except asyncio.CancelledError: logger.info("⏹️ Batch analysis scheduler stopped") @@ -381,16 +422,26 @@ async def krx_data_collector_scheduler(): await asyncio.sleep(3600) continue - # 수집 실행 - try: - logger.info("📥 [KRX] 일봉 데이터 증분 수집 시작 (update mode)...") - from collect_krx_data import collect_all - await collect_all(update_mode=True) - _last_collected = today - logger.info("✅ [KRX] 일봉 데이터 수집 완료") - except Exception as e: - logger.error(f"❌ [KRX] 데이터 수집 실패: {e}", exc_info=True) - await asyncio.sleep(1800) # 실패 시 30분 후 재시도 + # 수집 실행 (최대 3회 재시도, 30분 간격) + max_krx_retries = 3 + for krx_attempt in range(max_krx_retries): + try: + suffix = f" (재시도 {krx_attempt + 1}/{max_krx_retries})" if krx_attempt > 0 else "" + logger.info(f"📥 [KRX] 일봉 데이터 증분 수집 시작 (update mode){suffix}...") + from collect_krx_data import collect_all + await collect_all(update_mode=True) + _last_collected = today + logger.info("✅ [KRX] 일봉 데이터 수집 완료") + break + except Exception as e: + if krx_attempt < max_krx_retries - 1: + logger.warning( + f"⚠️ [KRX] 수집 실패 ({krx_attempt + 1}/{max_krx_retries}), " + f"30분 후 재시도: {e}" + ) + await asyncio.sleep(1800) + else: + logger.error(f"❌ [KRX] 데이터 수집 최대 재시도 초과: {e}", exc_info=True) except asyncio.CancelledError: logger.info("⏹️ KRX data collector scheduler stopped") @@ -405,6 +456,10 @@ async def lifespan(app: FastAPI): logger.info(f"KIS Base URL: {settings.KOREA_INVESTMENT_API_BASE_URL}") logger.info(f"KIS Account: {settings.KOREA_INVESTMENT_ACCOUNT_NUMBER}") + # ── 필수 환경변수 검증 (조기 경보) ────────────────────────────────────────── + for warn_msg in settings.validate_required(): + logger.warning(f"⚠️ [Config] {warn_msg}") + # DB 테이블 자동 생성 # 컬럼 추가/테이블 생성 DDL 은 Alembic migration 으로 관리됨. # 배포 전 `cd backend && alembic upgrade head` 를 실행할 것. @@ -595,6 +650,21 @@ async def log_requests(request: Request, call_next): app.include_router(api_router, prefix="/api/v1") +# ── 전역 예외 핸들러 ──────────────────────────────────────────────────────────── +from fastapi import Request as _Request +from fastapi.responses import JSONResponse as _JSONResponse + + +@app.exception_handler(Exception) +async def global_exception_handler(_req: _Request, exc: Exception) -> _JSONResponse: + """처리되지 않은 예외를 500 JSON으로 변환해 일관된 에러 응답을 보장한다.""" + logger.error(f"Unhandled exception: {_req.method} {_req.url} — {exc}", exc_info=True) + return _JSONResponse( + status_code=500, + content={"detail": "내부 서버 오류가 발생했습니다. 잠시 후 다시 시도해 주세요."}, + ) + + @app.get("/") async def root(): return {"message": "Auto Stock Trading System API", "version": "1.0.0", "status": "running"} @@ -602,7 +672,34 @@ async def root(): @app.get("/health") async def health_check(): - return {"status": "healthy"} + """DB / Redis 연결 상태를 포함한 헬스체크.""" + from sqlalchemy import text as _sa_text + from app.core.redis import get_redis + + db_ok = False + redis_ok = False + + try: + async with engine.connect() as conn: + await conn.execute(_sa_text("SELECT 1")) + db_ok = True + except Exception: + pass + + try: + r = get_redis() + if r: + await r.ping() + redis_ok = True + except Exception: + pass + + status = "healthy" if (db_ok and redis_ok) else "degraded" + return { + "status": status, + "db": "ok" if db_ok else "error", + "redis": "ok" if redis_ok else "error", + } @app.websocket("/ws/{client_id}") @@ -621,7 +718,17 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str): await websocket_manager.connect(websocket, client_id) try: while True: - data = await websocket.receive_text() + try: + data = await asyncio.wait_for(websocket.receive_text(), timeout=60.0) + except asyncio.TimeoutError: + # 60초 동안 메시지 없음 → ping 전송으로 연결 유지 확인 + try: + await websocket_manager.send_personal_message( + json.dumps({"type": "ping"}), client_id + ) + except Exception: + break + continue try: msg = json.loads(data) except (json.JSONDecodeError, AttributeError): diff --git a/backend/app/services/korea_investment_api.py b/backend/app/services/korea_investment_api.py index b6aea07..1adec60 100644 --- a/backend/app/services/korea_investment_api.py +++ b/backend/app/services/korea_investment_api.py @@ -19,6 +19,53 @@ TTL_MINUTE_FULLDAY = 60 # 당일 전체 분봉 캐시 1분 TTL_OVERSEAS_CHART = 300 # 해외 일봉 캐시 5분 +class CircuitBreaker: + """KIS API Circuit Breaker — 연속 N회 실패 시 일정 시간 요청 차단. + + States: + CLOSED : 정상 운영 (요청 허용) + OPEN : 차단 (recovery_timeout 경과 전까지 요청 거부) + HALF_OPEN: 복구 시도 (1개 요청 허용 → 성공 시 CLOSED, 실패 시 OPEN) + """ + + def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 600): + self.failure_threshold = failure_threshold # 연속 실패 허용 횟수 + self.recovery_timeout = recovery_timeout # 차단 후 복구 대기 시간 (초) + self.failure_count = 0 + self.last_failure_time: Optional[float] = None + self.state = "CLOSED" + + def can_request(self) -> bool: + """요청 가능 여부 반환.""" + if self.state == "CLOSED": + return True + if self.state == "OPEN": + if self.last_failure_time and time.time() - self.last_failure_time >= self.recovery_timeout: + self.state = "HALF_OPEN" + logger.info("⚡ Circuit breaker HALF_OPEN: 복구 시도") + return True + return False + return True # HALF_OPEN + + def record_success(self) -> None: + """성공 기록 — 차단 해제.""" + if self.state != "CLOSED": + logger.info("✅ Circuit breaker CLOSED: 복구 완료") + self.failure_count = 0 + self.state = "CLOSED" + + def record_failure(self) -> None: + """실패 기록 — 임계치 초과 시 차단.""" + self.failure_count += 1 + self.last_failure_time = time.time() + if self.state == "HALF_OPEN" or self.failure_count >= self.failure_threshold: + self.state = "OPEN" + logger.warning( + f"⚡ Circuit breaker OPEN: {self.failure_count}회 연속 실패. " + f"{self.recovery_timeout}초 후 재시도" + ) + + class APICallMonitor: """API 호출 모니터링 및 통계""" @@ -125,6 +172,7 @@ class KoreaInvestmentAPI: instance.semaphore = asyncio.Semaphore(settings.API_MAX_CONCURRENT_REQUESTS) instance._token_lock = asyncio.Lock() instance.monitor = APICallMonitor(window_size=settings.API_MONITORING_WINDOW) + instance.circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=600) logger.info( "🔧 KIS 실계좌 시세조회 API 초기화: base_url=%s, app_key=%s...", instance.base_url, instance.app_key[:6] @@ -157,6 +205,9 @@ class KoreaInvestmentAPI: # 모니터링 self.monitor = APICallMonitor(window_size=settings.API_MONITORING_WINDOW) + # Circuit Breaker (5회 연속 실패 → 10분 차단) + self.circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=600) + # 시작 시 설정 로깅 (키 마스킹) logger.info(f"🔧 KIS API 초기화: mode={settings.KIS_MODE}, " f"base_url={self.base_url}, " @@ -287,8 +338,14 @@ class KoreaInvestmentAPI: raise async def _make_request(self, method: str, endpoint: str, tr_id: str, extra_headers: Optional[Dict[str, str]] = None, **kwargs) -> (Dict[str, Any], Any): - """API 요청 실행 (안전한 재시도 + 쓰로틀링 + 모니터링)""" - + """API 요청 실행 (Circuit Breaker + 세마포어 + 재시도 + 모니터링)""" + + # Circuit Breaker 상태 확인 + if not self.circuit_breaker.can_request(): + wait_sec = self.circuit_breaker.recovery_timeout + logger.warning(f"⚡ Circuit breaker OPEN — {wait_sec}초 후 재시도 가능. 요청 거부: {endpoint}") + raise Exception(f"Circuit breaker open: KIS API 일시 차단 ({wait_sec}s)") + # 동시 요청 수 제한 (세마포어) async with self.semaphore: return await self._make_request_with_retry(method, endpoint, tr_id, extra_headers=extra_headers, **kwargs) @@ -337,23 +394,24 @@ class KoreaInvestmentAPI: if response.status_code == 200: duration = time.time() - start_time self.monitor.record_call(success=True, duration=duration, retries=retry_count) - + self.circuit_breaker.record_success() + # 오류율 경고 if self.monitor.should_alert(threshold=settings.API_ERROR_THRESHOLD): logger.warning( f"⚠️ API 오류율 높음: {self.monitor.get_error_rate():.1%} " f"(임계값: {settings.API_ERROR_THRESHOLD:.1%})" ) - + return response.json(), response.headers - + # API 호출 제한 초과 (429) 또는 서버 오류 (5xx) elif response.status_code in [429, 500, 502, 503, 504]: retry_count += 1 last_exception = Exception( f"API rate limit or server error: {response.status_code} - {response.text}" ) - + if attempt < self.max_retries: # 지수 백오프 적용 delay = self.retry_delay * (self.retry_backoff ** attempt) @@ -364,6 +422,7 @@ class KoreaInvestmentAPI: await asyncio.sleep(delay) continue else: + self.circuit_breaker.record_failure() logger.error(f"❌ API 최대 재시도 횟수 초과: {response.status_code}") raise last_exception @@ -401,19 +460,22 @@ class KoreaInvestmentAPI: else: duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) + self.circuit_breaker.record_failure() logger.error(f"❌ API 타임아웃 최대 재시도 초과") raise - + except Exception as e: # 예상치 못한 오류 duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) + self.circuit_breaker.record_failure() logger.error(f"❌ API 요청 중 예외 발생: {str(e)}") raise - + # 여기까지 왔다면 모든 재시도 실패 duration = time.time() - start_time self.monitor.record_call(success=False, duration=duration, retries=retry_count) + self.circuit_breaker.record_failure() if last_exception: raise last_exception raise Exception("API 요청 실패 (알 수 없는 오류)") diff --git a/backend/app/services/trade_executor.py b/backend/app/services/trade_executor.py index 6dd451c..e2ba524 100644 --- a/backend/app/services/trade_executor.py +++ b/backend/app/services/trade_executor.py @@ -1,5 +1,6 @@ import asyncio import logging +import time from decimal import Decimal from datetime import timedelta from sqlalchemy.ext.asyncio import AsyncSession @@ -19,14 +20,124 @@ from app.services.technical_indicators import ( logger = logging.getLogger(__name__) -# 동시 주문 1개로 제한 — 중복 주문 및 레이스 컨디션 방지 -_order_semaphore = asyncio.Semaphore(1) +# ── 종목별 세마포어 — 동일 종목 동시 주문 방지 ──────────────────────────────── +_order_semaphores: dict = {} -async def _mcp_validate_buy(stock: dict) -> tuple: - # 전략 엔진(RSI/BB/MA/ATR 등)이 이미 신호를 검증하므로 AI 호출 불필요. - # RISK_ASSESSMENT는 매매 직전 mcp.py에서 별도 수행. - return True, "strategy_validated" +def _get_order_semaphore(stock_code: str) -> asyncio.Semaphore: + """종목별 세마포어 반환. 없으면 생성.""" + if stock_code not in _order_semaphores: + _order_semaphores[stock_code] = asyncio.Semaphore(1) + return _order_semaphores[stock_code] + + +# ── Redis 분산락 — 중복 주문 완전 방지 (TTL 60초) ──────────────────────────── +_ORDER_LOCK_TTL = 60 # seconds + + +async def _acquire_order_lock(stock_code: str, order_type: str) -> bool: + """Redis SETNX 기반 분산락 획득. 성공 시 True.""" + try: + from app.core.redis import get_redis + client = get_redis() + if not client: + return True # Redis 미연결 → 락 없이 진행 + key = f"order_lock:{stock_code}:{order_type}" + result = await client.set(key, "1", nx=True, ex=_ORDER_LOCK_TTL) + return result is True + except Exception as e: + logger.debug(f"[order_lock] 획득 오류 (패스): {e}") + return True + + +async def _release_order_lock(stock_code: str, order_type: str) -> None: + """Redis 분산락 해제.""" + try: + from app.core.redis import delete_key + await delete_key(f"order_lock:{stock_code}:{order_type}") + except Exception: + pass + + +# ── 주문 재시도 — Exponential backoff (1→2→4초, 최대 3회) ─────────────────── +async def _place_order_with_retry( + is_domestic: bool, + stock_code: str, + order_type: str, + quantity: int, + price: float, + exchange_code: str = "NASD", + max_retries: int = 3, +) -> dict: + """주문 실행 with exponential backoff. 시스템 오류에만 재시도.""" + last_result: dict = {} + for attempt in range(max_retries): + try: + if is_domestic: + result = await korea_investment_api.place_order(stock_code, order_type, quantity, price) + else: + result = await korea_investment_api.place_overseas_order( + stock_code, order_type, quantity, float(price), exchange_code + ) + # rt_cd "0"=성공, "1"=PENDING → 재시도 불필요 + if result.get("rt_cd") in ("0", "1"): + return result + last_result = result + if attempt < max_retries - 1: + wait = 1 * (2 ** attempt) # 1s → 2s → 4s + logger.warning( + f"⚠️ 주문 실패 (rt_cd={result.get('rt_cd')}) " + f"→ {wait}초 후 재시도 ({attempt + 1}/{max_retries}): {stock_code}" + ) + await asyncio.sleep(wait) + except Exception as e: + last_result = {"rt_cd": "-1", "msg1": str(e), "odno": None} + if attempt < max_retries - 1: + wait = 1 * (2 ** attempt) + logger.warning(f"⚠️ 주문 예외 ({attempt + 1}/{max_retries}), {wait}초 후 재시도: {e}") + await asyncio.sleep(wait) + else: + logger.error(f"❌ 주문 최대 재시도 초과: {stock_code} {order_type} — {e}") + return last_result or {"rt_cd": "-1", "msg1": "max_retries_exceeded", "odno": None} + + +# ── 미체결 주문 1분 후 자동 취소 ───────────────────────────────────────────── +async def _auto_cancel_if_pending(order_id: str, stock_code: str, quantity: int, delay: int = 60) -> None: + """delay초 후에도 PENDING 상태면 자동 취소.""" + await asyncio.sleep(delay) + try: + result = await korea_investment_api.cancel_order(order_id, quantity) + if result.get("rt_cd") == "0": + logger.info(f"미체결 자동 취소 완료: {stock_code} order_id={order_id}") + else: + logger.debug(f"[auto_cancel] {stock_code} — {result.get('msg1')} (이미 체결됐을 수 있음)") + except Exception as e: + logger.debug(f"[auto_cancel] {stock_code} — {e}") + + +# ── 계좌 정보 30초 인메모리 캐시 ────────────────────────────────────────────── +_account_cache: dict = {"data": None, "ts": 0.0} +_ACCOUNT_CACHE_TTL = 30 # seconds + + +def _get_cached_account() -> tuple | None: + """(available_cash, total_assets) 30초 캐시. 만료 시 None 반환.""" + if _account_cache["data"] and time.monotonic() - _account_cache["ts"] < _ACCOUNT_CACHE_TTL: + return _account_cache["data"] + return None + + +def _set_account_cache(available_cash: float, total_assets: float) -> None: + _account_cache["data"] = (available_cash, total_assets) + _account_cache["ts"] = time.monotonic() + + +# ── 지정가 ±0.5% 조정 — 체결률 향상 ────────────────────────────────────────── +def _adjust_order_price(price: float, order_type: str) -> int: + """BUY +0.5% / SELL -0.5% 지정가 조정.""" + if order_type == "BUY": + return int(price * 1.005) + return int(price * 0.995) async def _execute_buy( @@ -43,7 +154,7 @@ async def _execute_buy( ) -> dict: """BUY 주문 실행 로직 — 선체크 → semaphore → 주문 → TradeHistory 기록 → sync.""" - # ── 1. 3일 급등 체크 ───────────────────────────────────────────────────── + # ── 1. 3일 급등 체크 (ATR 기반 동적 임계값) ─────────────────────────────── try: if is_domestic: chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10) @@ -64,7 +175,30 @@ async def _execute_buy( ((current_price - price_3days_ago) / price_3days_ago) * 100 if price_3days_ago > 0 else 0 ) - if recent_gain > 10: + # ATR 기반 동적 임계값: 최근 3일 평균 일간 변동폭 * 2, 최소 10% + try: + highs = ( + [float(item.get("stck_hgpr", 0)) for item in chart_data[-4:]] + if is_domestic + else [float(item.get("high", 0)) for item in chart_data[-4:]] + ) + lows = ( + [float(item.get("stck_lwpr", 0)) for item in chart_data[-4:]] + if is_domestic + else [float(item.get("low", 0)) for item in chart_data[-4:]] + ) + daily_ranges = [ + abs(highs[i] - lows[i]) for i in range(len(closes)) + if closes[i] > 0 and highs[i] > 0 and lows[i] > 0 + ] + atr_pct = ( + (sum(daily_ranges) / len(daily_ranges) / price_3days_ago * 100) + if daily_ranges and price_3days_ago > 0 else 0 + ) + surge_threshold = max(10.0, atr_pct * 2) + except Exception: + surge_threshold = 10.0 + if recent_gain > surge_threshold: return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} except Exception as e: logger.debug(f"[chase_buy] 차트 조회 실패, 우회: {stock_code} — {e}") @@ -95,13 +229,18 @@ async def _execute_buy( except Exception: return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"} - # ── 3. 현금·자산 조회 ──────────────────────────────────────────────────── - 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 + # ── 3. 현금·자산 조회 (30초 캐시 활용) ────────────────────────────────── + cached = _get_cached_account() + if cached: + available_cash, total_assets = cached + else: + 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 + _set_account_cache(available_cash, total_assets) + except Exception: + available_cash, total_assets = 0.0, 0.0 daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) @@ -145,12 +284,7 @@ async def _execute_buy( if recent_buys and (_get_kst_now() - recent_buys[0].created_at).total_seconds() < cooldown_min * 60: return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} - # ── 7. MCP 검증 ─────────────────────────────────────────────────────────── - mcp_ok, mcp_reason = await _mcp_validate_buy(stock) - if not mcp_ok: - return {"action": "HOLD", "stock": stock_name, "reason": mcp_reason} - - # ── 8. 포지션 사이징 ────────────────────────────────────────────────────── + # ── 7. 포지션 사이징 ────────────────────────────────────────────────────── 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 @@ -165,26 +299,32 @@ async def _execute_buy( if invest_amount <= 0 and total_assets <= 0: invest_amount = available_cash * 0.04 - quantity = int(invest_amount / current_price) + quantity = int(Decimal(str(invest_amount)) / Decimal(str(current_price))) if quantity <= 0: return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name} - # ── 9. 주문 실행 (semaphore로 동시성 제어) ─────────────────────────────── + # ── 8. Redis 분산락 획득 ────────────────────────────────────────────────── + if not await _acquire_order_lock(stock_code, "BUY"): + return {"action": "SKIP", "stock": stock_name, "reason": "order_lock_busy"} + + # ── 9. 주문 실행 (종목별 세마포어 + 재시도) ─────────────────────────────── try: - async with _order_semaphore: - if is_domestic: - order_result = await korea_investment_api.place_order( - stock_code, "BUY", quantity, int(current_price) - ) - else: - order_result = await korea_investment_api.place_overseas_order( - stock_code, "BUY", quantity, float(current_price), exchange_code - ) + order_price = _adjust_order_price(current_price, "BUY") + async with _get_order_semaphore(stock_code): + order_result = await _place_order_with_retry( + is_domestic=is_domestic, + stock_code=stock_code, + order_type="BUY", + quantity=quantity, + price=order_price, + exchange_code=exchange_code, + ) total_amount = Decimal(str(current_price)) * quantity entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점" success = order_result.get("rt_cd") == "0" - status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") + is_pending = order_result.get("rt_cd") == "1" + status = "COMPLETED" if success else ("PENDING" if is_pending else "FAILED") trade = TradeHistory( rule_id=stock.get("rule_id"), @@ -214,8 +354,12 @@ async def _execute_buy( portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price) except Exception: pass - # KIS 체결 확인 후 포트폴리오 DB 동기화 (3초 딜레이) asyncio.create_task(_delayed_sync_portfolio(delay_seconds=3)) + elif is_pending: + # PENDING 주문 → 1분 후 자동 취소 + order_id = order_result.get("odno") + if order_id: + asyncio.create_task(_auto_cancel_if_pending(order_id, stock_code, quantity)) return { "action": "BUY", @@ -227,6 +371,8 @@ async def _execute_buy( except Exception as e: return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)} + finally: + await _release_order_lock(stock_code, "BUY") async def _execute_sell( @@ -251,11 +397,28 @@ async def _execute_sell( take_profit = stock.get("take_profit", 3.0) stop_loss = stock.get("stop_loss", -1.5) - if not (profit_loss_rate >= take_profit or profit_loss_rate <= stop_loss or True): - return {"action": "HOLD", "stock": stock_name} + # ATR 기반 동적 손절/트레일링 스탑 조정 + atr = float(stock.get("atr", 0) or 0) + _avg_price = float(holding.average_price or 0) + exit_reason_detail = f"손익률 {profit_loss_rate:.2f}%" + if atr > 0 and _avg_price > 0: + # ATR 손절: 진입가 - 2.5*ATR (CLAUDE.md 규칙) + atr_stop_price = _avg_price - 2.5 * atr + atr_stop_pct = ((atr_stop_price / _avg_price) - 1) * 100 + stop_loss = max(stop_loss, atr_stop_pct) # 더 보수적인 값 사용 + + # 트레일링 스탑: 고점 - 1.5*ATR (CLAUDE.md 규칙) + high_price = float(stock.get("high_price") or current_price) + trailing_stop_price = high_price - 1.5 * atr + if current_price <= trailing_stop_price and profit_loss_rate > 0: + exit_reason_detail = f"ATR 트레일링 스탑: {current_price:.0f} ≤ {trailing_stop_price:.0f} | {exit_reason_detail}" + + # Redis 분산락 획득 + if not await _acquire_order_lock(stock_code, "SELL"): + return {"action": "SKIP", "stock": stock_name, "reason": "order_lock_busy"} try: - # ── 1. 중복 주문 체크 — 주문 전에 먼저 확인 ──────────────────────── + # ── 1. 중복 주문 체크 (DB + Redis 락 이중 방어) ────────────────────── recent_cutoff = _get_kst_now() - timedelta(seconds=10) duplicate_check = await db.execute( select(TradeHistory).where( @@ -267,21 +430,23 @@ async def _execute_sell( if duplicate_check.scalar_one_or_none(): return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} - # ── 2. 주문 실행 (semaphore로 동시성 제어) ────────────────────────── - async with _order_semaphore: - if is_domestic: - order_result = await korea_investment_api.place_order( - stock_code, "SELL", quantity, int(current_price) - ) - else: - order_result = await korea_investment_api.place_overseas_order( - stock_code, "SELL", quantity, float(current_price), exchange_code - ) + # ── 2. 주문 실행 (종목별 세마포어 + 재시도) ────────────────────────── + order_price = _adjust_order_price(current_price, "SELL") + async with _get_order_semaphore(stock_code): + order_result = await _place_order_with_retry( + is_domestic=is_domestic, + stock_code=stock_code, + order_type="SELL", + quantity=quantity, + price=order_price, + exchange_code=exchange_code, + ) total_amount = Decimal(str(current_price)) * quantity profit_loss = (current_price - float(holding.average_price or current_price)) * quantity success = order_result.get("rt_cd") == "0" - status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") + is_pending = order_result.get("rt_cd") == "1" + status = "COMPLETED" if success else ("PENDING" if is_pending else "FAILED") trade = TradeHistory( rule_id=stock.get("rule_id"), @@ -295,7 +460,7 @@ async def _execute_sell( total_amount=total_amount, status=status, order_id=order_result.get("odno"), - message=f"자동매도: 손익률 {profit_loss_rate:.2f}%", + message=f"자동매도: {exit_reason_detail}", exit_reason=f"손익: {profit_loss:,.0f}원", signals=signals, net_profit_loss=Decimal(str(profit_loss)) if success else None, @@ -317,8 +482,11 @@ async def _execute_sell( ] except Exception: pass - # KIS 체결 확인 후 포트폴리오 DB 동기화 asyncio.create_task(_delayed_sync_portfolio(delay_seconds=3)) + elif is_pending: + order_id = order_result.get("odno") + if order_id: + asyncio.create_task(_auto_cancel_if_pending(order_id, stock_code, quantity)) return { "action": "SELL", @@ -331,6 +499,8 @@ async def _execute_sell( except Exception as e: return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)} + finally: + await _release_order_lock(stock_code, "SELL") async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession): @@ -562,6 +732,16 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): if not has_strong_signal and (matched_count < 2 or score_val < 2): should_buy = False + # ── ATR 기반 강제 손절 (신호 없어도 발동) ──────────────────────────── + if holding and not should_sell: + atr = float(stock.get("atr", 0) or 0) + avg_price = float(holding.average_price or 0) + if atr > 0 and avg_price > 0: + atr_stop_price = avg_price - 2.5 * atr + if current_price <= atr_stop_price: + should_sell = True + signals = list(signals) + [f"ATR 손절: {current_price:.0f} ≤ {atr_stop_price:.0f}"] + # ── 매매 실행 위임 ──────────────────────────────────────────────────── if should_buy: return await _execute_buy(