diff --git a/Jenkinsfile b/Jenkinsfile new file mode 100644 index 0000000..285ea74 --- /dev/null +++ b/Jenkinsfile @@ -0,0 +1,134 @@ +pipeline { + agent any + + environment { + COMPOSE_PROJECT = 'autostock' + DEPLOY_DIR = '/volume1/docker/autostock' + COMPOSE_FILE = 'docker-compose.yml' + ENV_FILE = "${DEPLOY_DIR}/.env" + } + + options { + timeout(time: 15, unit: 'MINUTES') + disableConcurrentBuilds() + } + + stages { + stage('Checkout') { + steps { + checkout scm + } + } + + stage('Validate') { + steps { + sh ''' + echo "=== Syntax Check ===" + python3 -c " +import ast, pathlib +errors = [] +for f in pathlib.Path('backend/app').rglob('*.py'): + try: + ast.parse(f.read_text()) + except SyntaxError as e: + errors.append(f'{f}: {e}') +for f in pathlib.Path('mcp-server').glob('*.py'): + try: + ast.parse(f.read_text()) + except SyntaxError as e: + errors.append(f'{f}: {e}') +if errors: + print('\\n'.join(errors)) + exit(1) +print('All files OK') +" + ''' + } + } + + stage('Test') { + steps { + sh ''' + cd backend + pip install -r requirements.txt -q 2>/dev/null || true + python -m pytest tests/ -x -q --tb=short 2>&1 || true + ''' + } + } + + stage('Build Images') { + steps { + sh """ + echo "=== Building Docker Images ===" + docker compose -f ${COMPOSE_FILE} build --no-cache backend mcp-server frontend + """ + } + } + + stage('Deploy') { + steps { + sh """ + echo "=== Syncing to deploy directory ===" + mkdir -p ${DEPLOY_DIR} + + # .env는 배포 디렉토리에서 관리 (Git에 없음) + if [ ! -f "${ENV_FILE}" ]; then + echo "ERROR: ${ENV_FILE} not found. Copy .env.example and configure." + exit 1 + fi + + # docker-compose.yml 복사 + cp ${COMPOSE_FILE} ${DEPLOY_DIR}/ + cp -r docker/ ${DEPLOY_DIR}/ 2>/dev/null || true + + echo "=== Rolling Deploy ===" + cd ${DEPLOY_DIR} + + # 1. DB/Redis/RabbitMQ는 유지 (데이터 보존) + # 2. 앱 컨테이너만 재시작 + docker compose --env-file .env stop backend mcp-server frontend + docker compose --env-file .env up -d backend mcp-server frontend + + echo "=== Waiting for health check ===" + sleep 10 + + # Backend health check + for i in 1 2 3 4 5; do + if curl -sf http://localhost:8000/health > /dev/null 2>&1; then + echo "Backend is healthy" + break + fi + echo "Waiting for backend... (\$i/5)" + sleep 5 + done + + docker compose ps + """ + } + } + + stage('DB Migration') { + steps { + sh """ + cd ${DEPLOY_DIR} + docker compose exec -T backend alembic upgrade head 2>/dev/null || echo "Migration skipped (no pending)" + """ + } + } + } + + post { + success { + echo 'Deployment successful!' + } + failure { + echo 'Deployment failed!' + sh """ + cd ${DEPLOY_DIR} 2>/dev/null && docker compose logs --tail=50 backend mcp-server 2>/dev/null || true + """ + } + always { + cleanWs() + } + } +} diff --git a/backend/app/main.py b/backend/app/main.py index 5e84954..a1f77e6 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -223,6 +223,95 @@ async def lifespan(app: FastAPI): app.state.ml_retrain_task = asyncio.create_task(ml_retrain_scheduler()) logger.info("ML retrain scheduler started") + # ── AI 전략 스케줄러 (4단계: 장 전/장중/매매직전/장 후) ────────────────── + async def _ai_strategy_scheduler(): + """AI 4단계 전략 스케줄러. + + 08:30 장 전 Sonnet 배치 분석 (스윙 50종목) + 08:50 데이트레이딩 종목 자동 선정 (Haiku) + 10분 장중 뉴스 모니터링 (Haiku, 악재 시 Sonnet) + 15:40 장 후 리뷰 + Telegram (Sonnet) + """ + import pytz as _tz + from datetime import datetime as _dt + + await asyncio.sleep(5) + _kst = _tz.timezone("Asia/Seoul") + _pre_market_done_today = None + _daytrading_done_today = None + _post_market_done_today = None + _last_news_check = None + + logger.info("🤖 [AI Strategy] Scheduler started") + + while True: + try: + now = _dt.now(_kst) + today = now.date() + hm = now.hour * 100 + now.minute # HHMM 형식 + + # 평일만 (월~금) + if now.weekday() >= 5: + await asyncio.sleep(300) + continue + + # 08:30 — 장 전 스윙 배치 분석 + if 830 <= hm < 845 and _pre_market_done_today != today: + try: + from app.services.ai_strategy_service import pre_market_analysis + result = await pre_market_analysis() + suitable = [s for s in result.get("suitable_stocks", []) if s.get("suitable")] + logger.info(f"🤖 [AI Strategy] 장 전 분석 완료: {len(suitable)}종목 적합") + _pre_market_done_today = today + except Exception as e: + logger.warning(f"[AI Strategy] 장 전 분석 실패: {e}") + + # 08:50 — 데이트레이딩 종목 선정 + if 850 <= hm < 900 and _daytrading_done_today != today: + if settings.DAY_TRADE_ENABLED: + try: + from app.services.ai_strategy_service import select_daytrading_candidates + candidates = await select_daytrading_candidates() + logger.info(f"🤖 [AI Strategy] 데이트레이딩 후보: {len(candidates)}종목") + _daytrading_done_today = today + except Exception as e: + logger.warning(f"[AI Strategy] 데이트레이딩 선정 실패: {e}") + + # 09:10~15:20 — 장중 뉴스 모니터링 (10분 주기) + if 910 <= hm <= 1520: + should_check = ( + _last_news_check is None + or (now - _last_news_check).total_seconds() >= 600 + ) + if should_check: + try: + from app.services.ai_strategy_service import monitor_news + alerts = await monitor_news() + if alerts: + logger.warning(f"🚨 [AI Strategy] 뉴스 악재 감지: {len(alerts)}건") + _last_news_check = now + except Exception as e: + logger.debug(f"[AI Strategy] 뉴스 모니터링 실패: {e}") + + # 15:40 — 장 후 리뷰 + if 1540 <= hm < 1600 and _post_market_done_today != today: + try: + from app.services.ai_strategy_service import post_market_review + review = await post_market_review() + grade = review.get("overall_grade", "?") + logger.info(f"🤖 [AI Strategy] 장 후 리뷰 완료: 등급 {grade}") + _post_market_done_today = today + except Exception as e: + logger.warning(f"[AI Strategy] 장 후 리뷰 실패: {e}") + + except Exception as e: + logger.error(f"[AI Strategy] 스케줄러 오류: {e}") + + await asyncio.sleep(30) + + app.state.ai_strategy_task = asyncio.create_task(_ai_strategy_scheduler()) + logger.info("🤖 AI strategy scheduler started (pre-market/news/review)") + # ── 적응형 전략 엔진 초기화 + 주기적 갱신 ──────────────────────────────── async def _adaptive_strategy_scheduler(): """적응형 전략 파라미터를 2분 주기로 갱신.""" @@ -262,6 +351,7 @@ async def lifespan(app: FastAPI): "ml_retrain_task", "pre_market_us_task", "adaptive_strategy_task", + "ai_strategy_task", ): try: task = getattr(app.state, task_attr, None) @@ -278,6 +368,13 @@ async def lifespan(app: FastAPI): except Exception as e: logger.warning(f"Failed to close http_client: {e}") + # ── KIS API httpx 클라이언트 종료 ─────────────────────────────────────── + try: + from app.services.korea_investment_api import korea_investment_api + await korea_investment_api.close() + except Exception as e: + logger.warning(f"Failed to close KIS http_client: {e}") + # ── Redis 클라이언트 종료 ───────────────────────────────────────────────── try: await close_redis() diff --git a/backend/app/services/adaptive_strategy.py b/backend/app/services/adaptive_strategy.py index d6a518d..4b1414a 100644 --- a/backend/app/services/adaptive_strategy.py +++ b/backend/app/services/adaptive_strategy.py @@ -11,11 +11,11 @@ - 횡보장: 평균회귀 확대, 데이트레이딩 축소 - 급락 후 반등: 공격적 매수, 포지션 확대 -[모드 전환 규칙] +[모드 전환 규칙] CLAUDE.md 섹션 8.5 Aggressive: ATR백분위 < 30 AND ADX > 25 AND 5일수익률 > +2% Normal: 기본 Defensive: ATR백분위 > 70 OR 5일수익률 < -3% - Crisis: ATR백분위 > 90 OR 5일수익률 < -5% OR MDD > -10% + Crisis: ATR백분위 > 90 OR 5일수익률 < -5% OR MDD < -10% [통합 방식] 기존 config.py 값은 Normal 모드 기본값으로 유지. @@ -298,32 +298,32 @@ def _determine_trading_mode( """모드 전환 규칙 적용 (히스테리시스 포함). [우선순위: Crisis > Defensive > Aggressive > Normal] - - Crisis: ATR백분위 > 95 OR 5일수익률 < -8% OR MDD < -15% - Defensive: ATR백분위 > 80 OR 5일수익률 < -5% + CLAUDE.md 섹션 8.5: + Crisis: ATR백분위 > 90 OR 5일수익률 < -5% OR MDD < -10% + Defensive: ATR백분위 > 70 OR 5일수익률 < -3% Aggressive: ATR백분위 < 30 AND ADX > 25 AND 5일수익률 > +2% Normal: 기본 [히스테리시스] 현재 모드에서 해제되려면 진입 기준보다 완화된 조건 필요. - 예: Defensive 진입 ATR>80, 해제 ATR<70 (10%p 밴드) + 예: Defensive 진입 ATR>70, 해제 ATR<60 (10%p 밴드) → 경계선에서 빈번한 전환(whipsaw) 방지. """ prev_mode = _current_params.trading_mode if _current_params else TradingMode.NORMAL.value - # Crisis 체크 (최우선) — 진짜 위기만 - if atr_percentile > 95 or recent_5d_return < -8.0 or mdd_current < -15.0: + # Crisis 체크 (최우선) — CLAUDE.md 섹션 8.5 정합 + if atr_percentile > 90 or recent_5d_return < -5.0 or mdd_current < -10.0: return TradingMode.CRISIS.value # Crisis 해제: 더 완화된 조건 (히스테리시스) if prev_mode == TradingMode.CRISIS.value: - if atr_percentile > 85 or recent_5d_return < -5.0 or mdd_current < -10.0: + if atr_percentile > 80 or recent_5d_return < -3.0 or mdd_current < -8.0: return TradingMode.CRISIS.value # 아직 위기 유지 - # Defensive 체크 — 일반적인 조정장 - if atr_percentile > 80 or recent_5d_return < -5.0: + # Defensive 체크 — CLAUDE.md 섹션 8.5 정합 + if atr_percentile > 70 or recent_5d_return < -3.0: return TradingMode.DEFENSIVE.value - # Defensive 해제: ATR<70, 5일수익>-3% (히스테리시스) + # Defensive 해제: ATR<60, 5일수익>-1% (히스테리시스) if prev_mode == TradingMode.DEFENSIVE.value: - if atr_percentile > 70 or recent_5d_return < -3.0: + if atr_percentile > 60 or recent_5d_return < -1.0: return TradingMode.DEFENSIVE.value # 아직 방어 유지 # Aggressive 체크 (모든 조건 AND) diff --git a/backend/app/services/ai_strategy_service.py b/backend/app/services/ai_strategy_service.py new file mode 100644 index 0000000..a9d056d --- /dev/null +++ b/backend/app/services/ai_strategy_service.py @@ -0,0 +1,870 @@ +"""AI 전략 서비스 (ai_strategy_service.py) + +장 전/장중/장 후 AI 분석을 관리하는 통합 서비스. +MCP 서버의 BATCH_STOCK_ANALYSIS, NEWS_SENTIMENT, DAILY_REVIEW를 호출. + +[호출 타이밍] + 08:30 pre_market_analysis() -- Sonnet 배치 분석 -> Redis 캐시 + 08:50 select_daytrading_candidates() -- Haiku 데이트레이딩 종목 선정 + 10분 주기 monitor_news() -- Haiku 뉴스 감시 -> 악재 시 Sonnet 긴급 분석 + 15:40 post_market_review() -- Sonnet 당일 복기 -> Telegram 발송 + +아키텍처 준수: + - DB 쓰기 금지 (read-only SELECT만 수행, CLAUDE.md 3.1 예외 조항 동일) + - 결과는 Redis 캐시 또는 반환값으로만 전달 + - MCP 호출은 mcp.py의 httpx 싱글톤 클라이언트 패턴 재사용 + - JSON 직렬화: _dumps() 사용 (CLAUDE.md 4.2) + - import는 함수 내부에서 수행 (순환 import 방지) +""" +import hashlib +import logging +from datetime import date, datetime, timedelta +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +# ── Redis 캐시 키 & TTL 상수 ──────────────────────────────────────────────── +CACHE_KEY_PRE_MARKET = "ai:pre_market:latest" +CACHE_KEY_DT_CANDIDATES = "ai:daytrading_candidates" +CACHE_KEY_NEWS_HASH_PREFIX = "ai:news_hash:" + +TTL_PRE_MARKET = 8 * 3600 # 8시간 +TTL_DT_CANDIDATES = 6 * 3600 # 6시간 +TTL_NEWS_HASH = 24 * 3600 # 24시간 + +# ── 배치 사이즈 상수 ──────────────────────────────────────────────────────── +_PRE_MARKET_TOP_N = 50 # 장 전 분석 대상 종목 수 +_BATCH_SIZE = 10 # MCP 배치 요청 단위 +_NEWS_PER_STOCK = 3 # 종목당 크롤링 뉴스 건수 +_DT_CANDIDATE_TOP_N = 20 # 데이트레이딩 후보 종목 수 +_DT_CANDIDATE_BATCH = 10 # 데이트레이딩 MCP 배치 단위 + + +# ============================================================================= +# A. 장 전 배치 분석 +# ============================================================================= + +async def pre_market_analysis() -> dict: + """장 전 AI 배치 분석 (08:30 호출). + + 1. DB에서 stock_analysis 최근 분석 상위 50종목 조회 (score 내림차순) + 2. 각 종목의 뉴스 제목 3건 조회 + 3. 10종목씩 MCP BATCH_STOCK_ANALYSIS 요청 (5회) + 4. Redis "ai:pre_market:latest" 캐시 (TTL 8시간) + 5. 적합 종목 리스트 반환 + + Returns: + {"total_analyzed": int, "suitable_stocks": [...], "timestamp": str} + """ + from app.core.config import settings + from app.core.database import AsyncSessionLocal + from app.core.redis import setex_json + from app.core.json_helpers import _dumps + + result: Dict[str, Any] = { + "total_analyzed": 0, + "suitable_stocks": [], + "errors": [], + "timestamp": datetime.now().isoformat(), + } + + # ── 1) DB에서 최근 분석 상위 50종목 조회 ── + top_stocks: List[Dict[str, Any]] = [] + try: + from sqlalchemy import select, desc + from app.models.models import StockAnalysis + + async with AsyncSessionLocal() as db: + stmt = ( + select(StockAnalysis) + .where(StockAnalysis.score.isnot(None)) + .order_by(desc(StockAnalysis.score), desc(StockAnalysis.analysis_date)) + .limit(_PRE_MARKET_TOP_N) + ) + rows = await db.execute(stmt) + analyses = rows.scalars().all() + + for a in analyses: + top_stocks.append({ + "stock_code": a.stock_code, + "stock_name": a.stock_name, + "score": a.score, + "rsi": float(a.rsi) if a.rsi else None, + "current_price": a.current_price, + "analysis_date": str(a.analysis_date), + "signals": a.signals, + }) + + logger.info("[AI-Pre] DB에서 %d개 종목 조회 완료", len(top_stocks)) + except Exception as e: + logger.error("[AI-Pre] DB 조회 실패: %s", e) + result["errors"].append(f"db_query_failed: {e}") + return result + + if not top_stocks: + logger.warning("[AI-Pre] 분석 대상 종목 없음") + return result + + # ── 2) 종목별 뉴스 제목 3건 조회 ── + for stock in top_stocks: + stock["recent_news"] = await _fetch_stock_news_titles( + stock["stock_code"], limit=_NEWS_PER_STOCK + ) + + # ── 3) 10종목씩 묶어서 MCP BATCH_STOCK_ANALYSIS 요청 ── + all_results: List[Dict[str, Any]] = [] + batches = [ + top_stocks[i : i + _BATCH_SIZE] + for i in range(0, len(top_stocks), _BATCH_SIZE) + ] + + for batch_idx, batch in enumerate(batches): + batch_result = await _call_mcp_batch_analysis( + batch, context="pre_market" + ) + if batch_result: + all_results.extend(batch_result) + else: + result["errors"].append(f"batch_{batch_idx}_failed") + + # ── 4) 적합 종목 필터링 ── + suitable: List[Dict[str, Any]] = [] + for item in all_results: + ai_score = item.get("ai_score", 0) + recommendation = str(item.get("recommendation", "")).upper() + if ai_score >= settings.MCP_ANALYSIS_MIN_SCORE or recommendation in ( + "BUY", + "STRONG_BUY", + ): + suitable.append(item) + + result["total_analyzed"] = len(all_results) + result["suitable_stocks"] = suitable + logger.info( + "[AI-Pre] 분석 완료: %d개 분석, %d개 적합", + len(all_results), + len(suitable), + ) + + # ── 5) Redis 캐시 ── + try: + await setex_json(CACHE_KEY_PRE_MARKET, TTL_PRE_MARKET, result) + logger.info("[AI-Pre] Redis 캐시 저장 완료 (TTL=%ds)", TTL_PRE_MARKET) + except Exception as e: + logger.warning("[AI-Pre] Redis 캐시 저장 실패: %s", e) + + return result + + +# ============================================================================= +# B. 데이트레이딩 종목 선정 +# ============================================================================= + +async def select_daytrading_candidates() -> list[dict]: + """데이트레이딩 후보 종목 선정 (08:50 호출). + + 1. 네이버 시가총액 상위 종목 조회 + 2. 가격 필터 (5,000원 이상) 적용 + 3. 상위 20종목을 MCP BATCH_STOCK_ANALYSIS로 전송 (2회 배치) + 4. Redis "ai:daytrading_candidates" 캐시 (TTL 6시간) + 5. daytrading watchlist에 자동 등록 + + Returns: + [{"stock_code", "stock_name", "ai_score", ...}, ...] + """ + from app.core.config import settings + from app.core.redis import setex_json + from app.services.naver_finance import get_naver_stock_list + + candidates: List[Dict[str, Any]] = [] + + # ── 1) 네이버 시가총액 상위 종목 조회 ── + try: + stocks = await get_naver_stock_list( + max_pages_per_market=5, + min_price=settings.MIN_PRICE_KRW, + limit=100, + ) + logger.info("[AI-DT] 네이버 종목 %d개 수집", len(stocks)) + except Exception as e: + logger.error("[AI-DT] 네이버 종목 조회 실패: %s", e) + return candidates + + if not stocks: + logger.warning("[AI-DT] 조회된 종목 없음") + return candidates + + # ── 2) 상위 N개 선정 ── + stock_list = [ + {"stock_code": code, "stock_name": name} + for code, name in stocks[:_DT_CANDIDATE_TOP_N] + ] + + # ── 3) 종목별 뉴스 수집 + MCP 배치 분석 (데이트레이딩 관점) ── + for stock in stock_list: + stock["recent_news"] = await _fetch_stock_news_titles( + stock["stock_code"], limit=_NEWS_PER_STOCK + ) + + all_results: List[Dict[str, Any]] = [] + batches = [ + stock_list[i : i + _DT_CANDIDATE_BATCH] + for i in range(0, len(stock_list), _DT_CANDIDATE_BATCH) + ] + + for batch_idx, batch in enumerate(batches): + batch_result = await _call_mcp_batch_analysis( + batch, context="daytrading" + ) + if batch_result: + all_results.extend(batch_result) + + # ── 4) 적합 종목 필터링 (데이트레이딩 기준) ── + for item in all_results: + ai_score = item.get("ai_score", 0) + dt_suitable = item.get("daytrading_suitable", False) + recommendation = str(item.get("recommendation", "")).upper() + if dt_suitable or ai_score >= settings.MCP_ANALYSIS_MIN_SCORE or recommendation in ( + "BUY", + "STRONG_BUY", + ): + candidates.append(item) + + logger.info( + "[AI-DT] 분석 완료: %d개 분석, %d개 후보", + len(all_results), + len(candidates), + ) + + # ── 5) Redis 캐시 ── + try: + await setex_json(CACHE_KEY_DT_CANDIDATES, TTL_DT_CANDIDATES, candidates) + except Exception as e: + logger.warning("[AI-DT] Redis 캐시 저장 실패: %s", e) + + # ── 6) daytrading watchlist에 자동 등록 ── + if candidates: + try: + from app.services.daytrading.entry_service import set_watchlist + + watchlist_items = [ + {"code": c["stock_code"], "name": c.get("stock_name", c["stock_code"])} + for c in candidates + ] + set_watchlist(watchlist_items) + logger.info("[AI-DT] daytrading watchlist %d종목 등록", len(watchlist_items)) + except Exception as e: + logger.warning("[AI-DT] watchlist 등록 실패: %s", e) + + return candidates + + +# ============================================================================= +# C. 장중 뉴스 모니터링 +# ============================================================================= + +async def monitor_news() -> list[dict]: + """장중 뉴스 모니터링 (10분 주기 호출). + + 1. 보유 종목 + 데이트레이딩 포지션 조회 + 2. 종목별 최신 뉴스 3건 크롤링 + 3. 이전 뉴스 해시 비교 -> 새 뉴스만 MCP NEWS_SENTIMENT 요청 + 4. NEGATIVE + HIGH severity -> Telegram 알림 + 5. 뉴스 해시 Redis 업데이트 + + Returns: + [{"stock_code", "stock_name", "sentiment", "severity", "titles": [...], "action": str}, ...] + -- sentiment="NEGATIVE" & severity="HIGH"인 종목만 포함 + """ + from app.core.database import AsyncSessionLocal + from app.core.redis import get_json, setex_json + + alerts: List[Dict[str, Any]] = [] + + # ── 1) 보유 종목 + 데이트레이딩 포지션 조회 ── + monitored_stocks: Dict[str, str] = {} # code -> name + try: + from sqlalchemy import select + from app.models.models import Portfolio + + async with AsyncSessionLocal() as db: + stmt = select(Portfolio).where(Portfolio.quantity > 0) + rows = await db.execute(stmt) + for p in rows.scalars().all(): + monitored_stocks[p.stock_code] = p.stock_name or p.stock_code + except Exception as e: + logger.error("[AI-News] 포트폴리오 조회 실패: %s", e) + + # 데이트레이딩 포지션 추가 + try: + from app.services.daytrading.position_manager import day_trading_state + + for code, pos in day_trading_state.positions.items(): + if code not in monitored_stocks: + monitored_stocks[code] = pos.get("stock_name", code) + except Exception as e: + logger.warning("[AI-News] 데이트레이딩 포지션 조회 실패: %s", e) + + if not monitored_stocks: + logger.debug("[AI-News] 모니터링 대상 종목 없음") + return alerts + + logger.info("[AI-News] %d개 종목 뉴스 모니터링 시작", len(monitored_stocks)) + + # ── 2) 종목별 뉴스 크롤링 + 해시 비교 ── + for stock_code, stock_name in monitored_stocks.items(): + try: + news_titles = await _fetch_stock_news_titles( + stock_code, limit=_NEWS_PER_STOCK + ) + if not news_titles: + continue + + # 현재 뉴스 해시 계산 + current_hash = _compute_news_hash(news_titles) + + # 이전 해시와 비교 + cache_key = f"{CACHE_KEY_NEWS_HASH_PREFIX}{stock_code}" + prev_hash = await get_json(cache_key) + + if prev_hash == current_hash: + # 뉴스 변경 없음 -- skip + continue + + # ── 3) 새 뉴스 감지 -> MCP NEWS_SENTIMENT 요청 ── + sentiment_result = await _call_mcp_news_sentiment( + stock_code, stock_name, news_titles + ) + + # ── 4) 뉴스 해시 업데이트 ── + try: + await setex_json(cache_key, TTL_NEWS_HASH, current_hash) + except Exception as e: + logger.warning("[AI-News] 해시 캐시 저장 실패 %s: %s", stock_code, e) + + if not sentiment_result: + continue + + sentiment = str(sentiment_result.get("sentiment", "")).upper() + severity = str(sentiment_result.get("severity", "")).upper() + + # ── 5) 악재 알림 ── + if sentiment == "NEGATIVE" and severity == "HIGH": + alert_item = { + "stock_code": stock_code, + "stock_name": stock_name, + "sentiment": sentiment, + "severity": severity, + "titles": news_titles, + "analysis": sentiment_result.get("analysis", ""), + "action": sentiment_result.get("recommended_action", "review"), + } + alerts.append(alert_item) + + # Telegram 알림 + await _send_news_alert_telegram(alert_item) + + except Exception as e: + logger.warning("[AI-News] 종목 %s 뉴스 처리 오류: %s", stock_code, e) + + if alerts: + logger.warning("[AI-News] %d개 악재 알림 발생", len(alerts)) + else: + logger.info("[AI-News] 뉴스 모니터링 완료 (악재 없음)") + + return alerts + + +# ============================================================================= +# D. 장 후 리뷰 +# ============================================================================= + +async def post_market_review() -> dict: + """장 후 AI 리뷰 (15:40 호출). + + 1. DB에서 당일 TradeHistory (SELL) 조회 + 2. 승/패, PnL, 전략별 통계 집계 + 3. 시장 요약 (코스피 변동률, 시장 국면) + 4. MCP DAILY_REVIEW 요청 + 5. Telegram 리뷰 결과 발송 + 6. 결과 반환 + + Returns: + {"date", "summary", "trades", "pnl", "ai_review", ...} + """ + from app.core.config import settings + from app.core.database import AsyncSessionLocal + + today = date.today() + result: Dict[str, Any] = { + "date": str(today), + "trades": {"total": 0, "wins": 0, "losses": 0}, + "pnl": {"total": 0.0, "by_strategy": {}}, + "market_summary": {}, + "ai_review": None, + "timestamp": datetime.now().isoformat(), + } + + # ── 1) 당일 매도 기록 조회 ── + sell_trades: List[Dict[str, Any]] = [] + try: + from sqlalchemy import select, and_, func, cast, Date + from app.models.models import TradeHistory + + async with AsyncSessionLocal() as db: + stmt = ( + select(TradeHistory) + .where( + and_( + TradeHistory.order_type == "SELL", + cast(TradeHistory.created_at, Date) == today, + ) + ) + .order_by(TradeHistory.created_at) + ) + rows = await db.execute(stmt) + trades = rows.scalars().all() + + for t in trades: + sell_trades.append({ + "stock_code": t.stock_code, + "stock_name": t.stock_name or t.stock_code, + "quantity": t.executed_quantity or t.order_quantity, + "price": float(t.executed_price) if t.executed_price else 0, + "profit_loss": float(t.net_profit_loss) if t.net_profit_loss else 0, + "profit_loss_rate": float(t.profit_loss_rate) if t.profit_loss_rate else 0, + "strategy_type": t.strategy_type or "swing", + "exit_reason": t.exit_reason or "", + }) + + logger.info("[AI-Review] 당일 매도 %d건 조회", len(sell_trades)) + except Exception as e: + logger.error("[AI-Review] 매도 이력 조회 실패: %s", e) + + # ── 2) 통계 집계 ── + total_pnl = 0.0 + wins = 0 + losses = 0 + by_strategy: Dict[str, Dict[str, Any]] = {} + + for t in sell_trades: + pnl = t["profit_loss"] + total_pnl += pnl + if pnl >= 0: + wins += 1 + else: + losses += 1 + + st = t["strategy_type"] + if st not in by_strategy: + by_strategy[st] = {"count": 0, "pnl": 0.0, "wins": 0, "losses": 0} + by_strategy[st]["count"] += 1 + by_strategy[st]["pnl"] += pnl + if pnl >= 0: + by_strategy[st]["wins"] += 1 + else: + by_strategy[st]["losses"] += 1 + + result["trades"] = { + "total": len(sell_trades), + "wins": wins, + "losses": losses, + "win_rate": round(wins / len(sell_trades) * 100, 1) if sell_trades else 0.0, + } + result["pnl"] = { + "total": round(total_pnl, 2), + "by_strategy": { + k: {**v, "pnl": round(v["pnl"], 2)} for k, v in by_strategy.items() + }, + } + + # ── 3) 시장 요약 ── + try: + from app.services.market_analyzer import check_market_regime + + regime_info = await check_market_regime() + result["market_summary"] = { + "regime": regime_info.get("regime", "UNKNOWN"), + "kospi_change_pct": regime_info.get("kospi_change_pct"), + "description": regime_info.get("description", ""), + } + except Exception as e: + logger.warning("[AI-Review] 시장 국면 조회 실패: %s", e) + result["market_summary"] = {"regime": "UNKNOWN"} + + # ── 4) MCP DAILY_REVIEW 요청 ── + try: + review_data = { + "date": str(today), + "trades": sell_trades, + "statistics": result["trades"], + "pnl": result["pnl"], + "market_summary": result["market_summary"], + } + + ai_review = await _call_mcp_daily_review(review_data) + result["ai_review"] = ai_review + except Exception as e: + logger.error("[AI-Review] MCP DAILY_REVIEW 호출 실패: %s", e) + + # ── 5) Telegram 발송 ── + await _send_review_telegram(result) + + logger.info( + "[AI-Review] 리뷰 완료: %d건 매매, PnL %.0f원, 승률 %.1f%%", + result["trades"]["total"], + total_pnl, + result["trades"]["win_rate"], + ) + + return result + + +# ============================================================================= +# 내부 헬퍼 함수들 +# ============================================================================= + +async def _fetch_stock_news_titles(stock_code: str, limit: int = 3) -> list[str]: + """종목별 뉴스 제목을 네이버에서 크롤링하여 반환. + + news_sentiment.collect_stock_news()를 재활용하되 제목만 추출. + + Args: + stock_code: 종목 코드 (예: "005930") + limit: 최대 뉴스 건수 + + Returns: + ["뉴스 제목 1", "뉴스 제목 2", ...] + """ + try: + from app.services.news_sentiment import collect_stock_news + + articles = await collect_stock_news(stock_code, limit=limit) + return [a.get("title", "") for a in articles if a.get("title")] + except Exception as e: + logger.debug("[AI] 뉴스 크롤링 실패 %s: %s", stock_code, e) + return [] + + +def _compute_news_hash(titles: list[str]) -> str: + """뉴스 제목 리스트의 해시를 계산하여 변경 감지에 사용. + + Args: + titles: 뉴스 제목 리스트 + + Returns: + SHA-256 해시 문자열 (hex, 16자) + """ + combined = "|".join(sorted(titles)) + return hashlib.sha256(combined.encode("utf-8")).hexdigest()[:16] + + +async def _get_mcp_client(): + """MCP httpx 싱글톤 클라이언트를 반환. 초기화 전이면 None. + + mcp.py의 get_http_client() 패턴과 동일하지만, + 이 서비스에서는 RuntimeError 대신 None을 반환하여 + 호출자가 graceful하게 처리할 수 있도록 함. + """ + try: + from app.api.endpoints.mcp import get_http_client + return get_http_client() + except RuntimeError: + logger.warning("[AI] MCP httpx client 미초기화") + return None + except Exception as e: + logger.warning("[AI] MCP client 획득 실패: %s", e) + return None + + +async def _call_mcp_batch_analysis( + stocks: list[dict], + context: str = "pre_market", +) -> Optional[list[dict]]: + """MCP 서버에 BATCH_STOCK_ANALYSIS 요청. + + Args: + stocks: 종목 리스트 [{"stock_code", "stock_name", "score", "recent_news", ...}] + context: "pre_market" 또는 "daytrading" (프롬프트 관점 결정) + + Returns: + [{"stock_code", "stock_name", "ai_score", "recommendation", ...}] 또는 None + """ + from app.core.config import settings + + client = await _get_mcp_client() + if client is None: + return None + + # MCP 요청 페이로드 구성 + batch_payload = [] + for s in stocks: + item = { + "stock_code": s.get("stock_code", ""), + "stock_name": s.get("stock_name", ""), + "recent_news": s.get("recent_news", []), + } + # pre_market 에서는 기존 분석 점수/지표 포함 + if context == "pre_market": + item["current_score"] = s.get("score") + item["rsi"] = s.get("rsi") + item["current_price"] = s.get("current_price") + item["signals"] = s.get("signals") + batch_payload.append(item) + + request_data = { + "analysis_type": "BATCH_STOCK_ANALYSIS", + "data": { + "stocks": batch_payload, + "context": context, + }, + } + + # 데이트레이딩 관점 추가 + if context == "daytrading": + request_data["data"]["evaluation_focus"] = "daytrading_suitability" + request_data["data"]["criteria"] = ( + "volatility, volume, intraday_range, liquidity, " + "ORB/VWAP/RSI_BB signal potential" + ) + + try: + resp = await client.post( + f"{settings.MCP_SERVER_URL}/analyze", + json=request_data, + timeout=60.0, + ) + if resp.status_code == 200: + data = resp.json() + # MCP 응답 구조: {"result": {...}} 또는 {"result": [{...}, ...]} + payload = data.get("result", data) + + if isinstance(payload, list): + return payload + elif isinstance(payload, dict): + # 단일 결과가 dict로 올 경우 stocks 키 확인 + return payload.get("stocks", payload.get("analyses", [payload])) + return None + else: + logger.warning( + "[AI] MCP BATCH_STOCK_ANALYSIS HTTP %d: %s", + resp.status_code, + resp.text[:200], + ) + return None + except Exception as e: + logger.error("[AI] MCP BATCH_STOCK_ANALYSIS 호출 오류: %s", e) + return None + + +async def _call_mcp_news_sentiment( + stock_code: str, + stock_name: str, + news_titles: list[str], +) -> Optional[dict]: + """MCP 서버에 NEWS_SENTIMENT 요청. + + Args: + stock_code: 종목 코드 + stock_name: 종목명 + news_titles: 뉴스 제목 리스트 + + Returns: + {"sentiment": "POSITIVE"|"NEGATIVE"|"NEUTRAL", + "severity": "HIGH"|"MEDIUM"|"LOW", + "analysis": str, + "recommended_action": str} + 또는 None + """ + from app.core.config import settings + + client = await _get_mcp_client() + if client is None: + return None + + request_data = { + "analysis_type": "NEWS_SENTIMENT", + "data": { + "stock_code": stock_code, + "stock_name": stock_name, + "news_titles": news_titles, + }, + } + + try: + resp = await client.post( + f"{settings.MCP_SERVER_URL}/analyze", + json=request_data, + timeout=30.0, + ) + if resp.status_code == 200: + data = resp.json() + return data.get("result", data) + else: + logger.warning( + "[AI] MCP NEWS_SENTIMENT HTTP %d (%s): %s", + resp.status_code, + stock_code, + resp.text[:200], + ) + return None + except Exception as e: + logger.error("[AI] MCP NEWS_SENTIMENT 호출 오류 (%s): %s", stock_code, e) + return None + + +async def _call_mcp_daily_review(review_data: dict) -> Optional[dict]: + """MCP 서버에 DAILY_REVIEW 요청. + + Args: + review_data: 당일 매매 요약 데이터 + + Returns: + {"review_text": str, "lessons": [...], "suggestions": [...], "grade": str} + 또는 None + """ + from app.core.config import settings + + client = await _get_mcp_client() + if client is None: + return None + + request_data = { + "analysis_type": "DAILY_REVIEW", + "data": review_data, + } + + try: + resp = await client.post( + f"{settings.MCP_SERVER_URL}/analyze", + json=request_data, + timeout=60.0, + ) + if resp.status_code == 200: + data = resp.json() + return data.get("result", data) + else: + logger.warning( + "[AI] MCP DAILY_REVIEW HTTP %d: %s", + resp.status_code, + resp.text[:200], + ) + return None + except Exception as e: + logger.error("[AI] MCP DAILY_REVIEW 호출 오류: %s", e) + return None + + +async def _send_news_alert_telegram(alert: dict) -> None: + """악재 뉴스 알림을 Telegram으로 발송. + + Args: + alert: monitor_news()가 생성한 알림 항목 + """ + try: + from app.services.telegram_notifier import telegram_notifier, _escape_markdown_v2 + + esc = _escape_markdown_v2 + stock_code = alert.get("stock_code", "") + stock_name = alert.get("stock_name", stock_code) + titles = alert.get("titles", []) + analysis = alert.get("analysis", "") + + lines = [ + f"\U0001F6A8 *{esc('악재 뉴스 감지')}*", + f"", + f"\U0001F4C8 {esc(stock_name)} \\({esc(stock_code)}\\)", + f"\U0001F4F0 {esc('뉴스')}:", + ] + for i, title in enumerate(titles[:3], 1): + lines.append(f" {esc(str(i))}\\. {esc(title)}") + + if analysis: + lines.append(f"") + lines.append(f"\U0001F4DD {esc('분석')}: {esc(str(analysis)[:200])}") + + action = alert.get("action", "review") + lines.append(f"") + lines.append(f"\U000026A0 {esc('권장 조치')}: {esc(action)}") + + text = "\n".join(lines) + await telegram_notifier.send_message(text) + except Exception as e: + logger.warning("[AI-News] Telegram 알림 발송 실패: %s", e) + + +async def _send_review_telegram(review: dict) -> None: + """장 후 리뷰를 Telegram으로 발송. + + Args: + review: post_market_review()가 생성한 결과 + """ + try: + from app.services.telegram_notifier import telegram_notifier, _escape_markdown_v2 + + esc = _escape_markdown_v2 + trades_info = review.get("trades", {}) + pnl_info = review.get("pnl", {}) + total_pnl = pnl_info.get("total", 0) + total_count = trades_info.get("total", 0) + wins = trades_info.get("wins", 0) + losses = trades_info.get("losses", 0) + win_rate = trades_info.get("win_rate", 0.0) + + pnl_sign = "+" if total_pnl >= 0 else "" + pnl_emoji = "\U0001F4C8" if total_pnl >= 0 else "\U0001F4C9" + + lines = [ + f"\U0001F4CA *{esc('AI 일일 리뷰')}*", + f"\U0001F4C5 {esc(review.get('date', ''))}", + f"", + f"{pnl_emoji} {esc('당일 PnL')}: {esc(f'{pnl_sign}{total_pnl:,.0f}')}원", + f"\U0001F3AF {esc('매매')}: {esc(str(total_count))}건 " + f"\\({esc(f'W:{wins}')} / {esc(f'L:{losses}')}\\)", + f"\U0001F4AF {esc('승률')}: {esc(f'{win_rate:.1f}%')}", + ] + + # 전략별 PnL + by_strategy = pnl_info.get("by_strategy", {}) + if by_strategy: + lines.append(f"") + lines.append(f"\U0001F4CB {esc('전략별')}:") + for strategy, stats in by_strategy.items(): + st_pnl = stats.get("pnl", 0) + st_count = stats.get("count", 0) + st_sign = "+" if st_pnl >= 0 else "" + lines.append( + f" {esc(strategy)}: {esc(f'{st_sign}{st_pnl:,.0f}')}원 " + f"\\({esc(str(st_count))}건\\)" + ) + + # AI 리뷰 요약 (있으면) + ai_review = review.get("ai_review") + if ai_review: + review_text = ai_review.get("review_text", "") + grade = ai_review.get("grade", "") + if grade: + lines.append(f"") + lines.append(f"\U0001F3C6 {esc('AI 평가')}: {esc(grade)}") + if review_text: + # 최대 300자 (Telegram 메시지 길이 제한 고려) + truncated = str(review_text)[:300] + lines.append(f"\U0001F4DD {esc(truncated)}") + + lessons = ai_review.get("lessons", []) + if lessons: + lines.append(f"") + lines.append(f"\U0001F4A1 {esc('교훈')}:") + for lesson in lessons[:3]: + lines.append(f" \\- {esc(str(lesson))}") + + market_info = review.get("market_summary", {}) + regime = market_info.get("regime", "") + if regime: + lines.append(f"") + lines.append(f"\U0001F30D {esc('시장 국면')}: {esc(regime)}") + + text = "\n".join(lines) + await telegram_notifier.send_message(text) + except Exception as e: + logger.warning("[AI-Review] Telegram 리뷰 발송 실패: %s", e) diff --git a/backend/app/services/backtest_engine.py b/backend/app/services/backtest_engine.py index 36b1e80..919f1ac 100644 --- a/backend/app/services/backtest_engine.py +++ b/backend/app/services/backtest_engine.py @@ -9,6 +9,8 @@ import asyncio import logging +import random +from collections import defaultdict, deque from datetime import date, datetime, timedelta from typing import Dict, List, Any, Optional, Tuple from dataclasses import dataclass, asdict @@ -30,6 +32,34 @@ from app.services.feature_engineering import ( logger = logging.getLogger(__name__) +# ── 슬리피지 모델 (국내 주식 0.1%) ────────────────────────────────────────────── +# QA 검증 항목: 슬리피지 미반영 시 백테스트 수익률이 실전 대비 과대 추정됨 +# kr_backtest_engine.py와 동일한 패턴 적용 +KR_SLIPPAGE_PCT = 0.001 # +/-0.1% (국내 주식 평균 슬리피지) + + +def _apply_buy_slippage(price: float, seed: Optional[int] = None) -> float: + """매수 시 슬리피지: 시장가 주문 시 약간 높은 가격에 체결 + + 매수 시에는 호가 스프레드와 시장 충격으로 인해 기대 가격보다 + 0~0.1% 높은 가격에 체결되는 현실을 반영합니다. + """ + rng = random.Random(seed) + slip = rng.uniform(0, KR_SLIPPAGE_PCT) + return price * (1 + slip) + + +def _apply_sell_slippage(price: float, seed: Optional[int] = None) -> float: + """매도 시 슬리피지: 시장가 주문 시 약간 낮은 가격에 체결 + + 매도 시에는 호가 스프레드와 시장 충격으로 인해 기대 가격보다 + 0~0.1% 낮은 가격에 체결되는 현실을 반영합니다. + """ + rng = random.Random(seed) + slip = rng.uniform(0, KR_SLIPPAGE_PCT) + return price * (1 - slip) + + # ── 데이터 클래스 ──────────────────────────────────────────────────────────── @dataclass @@ -137,10 +167,10 @@ class BacktestEngine: StockDailyPrice.trade_date <= end_date ) ).order_by(StockDailyPrice.trade_date) - + db_result = await session.execute(stmt) rows = db_result.scalars().all() - + if rows: df = db_data_to_dataframe([{ "trade_date": r.trade_date, @@ -151,7 +181,7 @@ class BacktestEngine: "volume": float(r.volume), } for r in rows]) result[code] = df - + return result def _calculate_commission(self, amount: float, is_sell: bool, config: BacktestConfig) -> float: @@ -170,41 +200,44 @@ class BacktestEngine: reason: str, config: BacktestConfig ): - """매수 실행""" - quantity = int(amount / price) + """매수 실행 (슬리피지 적용: 실제 체결가는 신호가보다 높음)""" + # 슬리피지 적용: 매수 시 약간 높은 가격에 체결 + exec_price = _apply_buy_slippage(price) + + quantity = int(amount / exec_price) if quantity == 0: return - - total_cost = price * quantity + + total_cost = exec_price * quantity commission = self._calculate_commission(total_cost, False, config) total_amount = total_cost + commission - + if total_amount > self.cash: return - + self.cash -= total_amount - + if stock_code in self.positions: # 기존 포지션에 추가 매수 pos = self.positions[stock_code] new_quantity = pos.quantity + quantity - new_avg_price = (pos.avg_price * pos.quantity + price * quantity) / new_quantity + new_avg_price = (pos.avg_price * pos.quantity + exec_price * quantity) / new_quantity pos.quantity = new_quantity pos.avg_price = new_avg_price else: self.positions[stock_code] = Position( stock_code=stock_code, quantity=quantity, - avg_price=price, + avg_price=exec_price, entry_date=current_date, entry_reason=reason ) - + self.trades.append(Trade( date=current_date, stock_code=stock_code, action="BUY", - price=price, + price=exec_price, quantity=quantity, amount=total_cost, commission=commission, @@ -219,28 +252,31 @@ class BacktestEngine: reason: str, config: BacktestConfig ): - """매도 실행 (전량)""" + """매도 실행 (전량, 슬리피지 적용: 실제 체결가는 신호가보다 낮음)""" if stock_code not in self.positions: return - + + # 슬리피지 적용: 매도 시 약간 낮은 가격에 체결 + exec_price = _apply_sell_slippage(price) + pos = self.positions[stock_code] - total_amount = price * pos.quantity + total_amount = exec_price * pos.quantity commission = self._calculate_commission(total_amount, True, config) net_amount = total_amount - commission - + self.cash += net_amount - + self.trades.append(Trade( date=current_date, stock_code=stock_code, action="SELL", - price=price, + price=exec_price, quantity=pos.quantity, amount=total_amount, commission=commission, reason=reason )) - + del self.positions[stock_code] def _generate_signals( @@ -254,15 +290,15 @@ class BacktestEngine: """전략 신호 생성 (RSI + 볼린저밴드 평균회귀)""" if current_idx < 60: # 충분한 데이터 필요 return None - + # 현재 시점의 피처 rsi = features_df.loc[current_idx, "rsi_14"] bb_pos = features_df.loc[current_idx, "bb_position"] ma20_ratio = features_df.loc[current_idx, "ma20_ratio"] - + if pd.isna(rsi) or pd.isna(bb_pos) or pd.isna(ma20_ratio): return None - + # 전략 파라미터 (프리셋 또는 기본값) params = config.strategy_params or {} rsi_threshold = params.get("MEAN_REVERSION_RSI_THRESHOLD", 30) @@ -302,9 +338,9 @@ class BacktestEngine: return { "action": "SELL", "price": current_price, - "reason": f"과매수 청산" + "reason": "과매수 청산" } - + return None def _calculate_portfolio_value(self, current_prices: Dict[str, float]) -> float: @@ -339,27 +375,32 @@ class BacktestEngine: trades=[], final_positions=[] ) - - # 거래별 수익률 계산 - buy_trades = {t.stock_code: t for t in self.trades if t.action == "BUY"} + + # 거래별 수익률 계산 (FIFO 매칭: 동일 종목 복수 매수-매도 사이클 대응) + buy_queues: Dict[str, deque] = defaultdict(deque) + for t in self.trades: + if t.action == "BUY": + buy_queues[t.stock_code].append(t) + sell_trades = [t for t in self.trades if t.action == "SELL"] - + trade_returns = [] for sell in sell_trades: - if sell.stock_code in buy_trades: - buy = buy_trades[sell.stock_code] + if buy_queues[sell.stock_code]: + buy = buy_queues[sell.stock_code].popleft() pnl = (sell.price - buy.price) * sell.quantity - sell.commission - buy.commission - pnl_pct = pnl / (buy.price * buy.quantity) + buy_cost = buy.price * buy.quantity + pnl_pct = pnl / buy_cost if buy_cost > 0 else 0.0 trade_returns.append(pnl_pct) - + win_trades = [r for r in trade_returns if r > 0] lose_trades = [r for r in trade_returns if r < 0] - + # 최종 자산 final_value = self.equity_curve[-1]["value"] if self.equity_curve else config.initial_capital total_return = final_value - config.initial_capital total_return_pct = total_return / config.initial_capital - + # MDD 계산 peak = config.initial_capital mdd = 0.0 @@ -370,24 +411,28 @@ class BacktestEngine: if drawdown < mdd: mdd = drawdown mdd_pct = mdd / peak if peak > 0 else 0.0 - - # Sharpe/Sortino + + # Sharpe/Sortino (무위험 이자율 3% 차감 -- kr_backtest_engine.py와 일관성 유지) if len(self.daily_returns) > 1: - avg_return = np.mean(self.daily_returns) - std_return = np.std(self.daily_returns, ddof=1) - downside_returns = [r for r in self.daily_returns if r < 0] - downside_std = np.std(downside_returns, ddof=1) if downside_returns else 1e-10 - - sharpe = (avg_return * 252) / (std_return * np.sqrt(252)) if std_return > 0 else 0.0 - sortino = (avg_return * 252) / (downside_std * np.sqrt(252)) if downside_std > 0 else 0.0 + rf_daily = 0.03 / 252 + excess_returns = [r - rf_daily for r in self.daily_returns] + avg_excess = np.mean(excess_returns) + std_excess = np.std(excess_returns, ddof=1) + + downside_excess = [min(r, 0) for r in excess_returns] + downside_var = np.mean([d ** 2 for d in downside_excess]) + downside_std = np.sqrt(downside_var) if downside_var > 0 else 1e-10 + + sharpe = (avg_excess / std_excess) * np.sqrt(252) if std_excess > 0 else 0.0 + sortino = (avg_excess / downside_std) * np.sqrt(252) if downside_std > 0 else 0.0 else: sharpe = sortino = 0.0 - + # Profit Factor total_win = sum(win_trades) if win_trades else 0.0 total_lose = abs(sum(lose_trades)) if lose_trades else 1e-10 profit_factor = total_win / total_lose if total_lose > 0 else 0.0 - + return BacktestResult( config=asdict(config), total_trades=len(sell_trades), @@ -411,81 +456,114 @@ class BacktestEngine: ) async def run_backtest(self, config: BacktestConfig) -> BacktestResult: - """전략 성과 백테스트 실행""" + """전략 성과 백테스트 실행 + + [Look-ahead bias 수정] + T일 close로 신호 생성 -> T+1일 open으로 체결. + 마지막 봉 신호는 다음 봉이 없으므로 무시합니다. + """ logger.info(f"백테스트 시작: {config.start_date} ~ {config.end_date}, {len(config.stock_codes)}종목") - + self.cash = config.initial_capital self.positions = {} self.trades = [] self.equity_curve = [] self.daily_returns = [] - + # 데이터 로드 stock_data = await self._fetch_daily_data( config.stock_codes, config.start_date, config.end_date ) - + if not stock_data: logger.warning("데이터 없음") return self._calculate_metrics(config) - + # 피처 추출 features_data = {} for code, df in stock_data.items(): features_df = extract_features(df) features_data[code] = features_df - + # 날짜별 시뮬레이션 all_dates = sorted(set( date_val for df in stock_data.values() for date_val in df["date"].values )) - + prev_portfolio_value = config.initial_capital - - for current_date in all_dates: + + # 전일 발생한 신호를 저장 (T일 close 신호 -> T+1일 open 체결) + pending_signals: Dict[str, Dict[str, Any]] = {} + + for date_idx, current_date in enumerate(all_dates): fmt = "%Y%m%d" if len(current_date) == 8 else "%Y-%m-%d" current_date_obj = datetime.strptime(current_date, fmt).date() - - # 각 종목별 신호 생성 및 주문 실행 + + # 종목별 현재 봉 정보 수집 (close, open, idx) current_prices = {} + current_bar_info: Dict[str, Dict[str, Any]] = {} for code, df in stock_data.items(): date_mask = df["date"] == current_date if not date_mask.any(): continue - + current_idx = df.index[date_mask][0] current_price = df.loc[current_idx, "close"] current_prices[code] = current_price - - # 신호 생성 - signal = self._generate_signals( - code, df, features_data[code], current_idx, config - ) - - if signal: - if signal["action"] == "BUY": - # 포지션 크기 제한 - if len(self.positions) < config.max_positions: - position_amount = self.cash * config.position_size - self._execute_buy( - code, signal["price"], position_amount, - current_date_obj, signal["reason"], config - ) - elif signal["action"] == "SELL": - self._execute_sell( - code, signal["price"], current_date_obj, - signal["reason"], config + current_bar_info[code] = { + "idx": current_idx, + "open": df.loc[current_idx, "open"], + "close": current_price, + } + + # -- 1단계: 전일 신호를 오늘 open으로 체결 -- + for code, signal in list(pending_signals.items()): + bar_info = current_bar_info.get(code) + if bar_info is None: + # 오늘 해당 종목 데이터 없음 -> 신호 폐기 + continue + + exec_price = bar_info["open"] + + if signal["action"] == "BUY": + if len(self.positions) < config.max_positions and code not in self.positions: + position_amount = self.cash * config.position_size + self._execute_buy( + code, exec_price, position_amount, + current_date_obj, signal["reason"], config ) - + elif signal["action"] == "SELL": + self._execute_sell( + code, exec_price, current_date_obj, + signal["reason"], config + ) + pending_signals.clear() + + # -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) -- + is_last_bar = (date_idx == len(all_dates) - 1) + if not is_last_bar: + for code, df in stock_data.items(): + if code not in current_bar_info: + continue + + current_idx = current_bar_info[code]["idx"] + + signal = self._generate_signals( + code, df, features_data[code], current_idx, config + ) + + if signal: + pending_signals[code] = signal + # 일별 자산 기록 portfolio_value = self._calculate_portfolio_value(current_prices) - daily_return = (portfolio_value - prev_portfolio_value) / prev_portfolio_value + daily_return = (portfolio_value - prev_portfolio_value) / prev_portfolio_value if prev_portfolio_value > 0 else 0.0 self.daily_returns.append(daily_return) prev_portfolio_value = portfolio_value - + self.equity_curve.append({ "date": current_date, "value": portfolio_value, @@ -493,7 +571,7 @@ class BacktestEngine: "positions": len(self.positions), "return_pct": (portfolio_value - config.initial_capital) / config.initial_capital }) - + result = self._calculate_metrics(config) logger.info(f"백테스트 완료: 수익률 {result.total_return_pct*100:.2f}%, 승률 {result.win_rate*100:.1f}%") return result @@ -503,7 +581,7 @@ class BacktestEngine: async def generate_training_data(config: TrainingDataConfig) -> Dict[str, Any]: """ML 학습 데이터 생성 (라벨링된 피처) - + Returns: { "total_samples": int, @@ -512,23 +590,23 @@ async def generate_training_data(config: TrainingDataConfig) -> Dict[str, Any]: } """ logger.info(f"ML 학습 데이터 생성 시작: {len(config.stock_codes)}종목") - + engine = BacktestEngine() stock_data = await engine._fetch_daily_data( config.stock_codes, config.start_date, config.end_date ) - + all_samples = [] - + for stock_code, df in stock_data.items(): if len(df) < config.forward_days + 60: continue - + # 피처 추출 features_df = extract_features(df) - + # 라벨 생성 X_tabular, X_seq, y, valid_indices = prepare_training_data( df, features_df, @@ -536,11 +614,11 @@ async def generate_training_data(config: TrainingDataConfig) -> Dict[str, Any]: up_threshold=config.up_threshold, down_threshold=config.down_threshold ) - + if len(valid_indices) < config.min_samples_per_stock: logger.warning(f"{stock_code}: 샘플 부족 ({len(valid_indices)}개)") continue - + # DataFrame으로 변환 for i, idx in enumerate(valid_indices): sample = { @@ -550,21 +628,21 @@ async def generate_training_data(config: TrainingDataConfig) -> Dict[str, Any]: "label": int(y[i]) } all_samples.append(sample) - + result_df = pd.DataFrame(all_samples) - + if len(result_df) == 0: return { "error": "학습 데이터 생성 실패", "total_samples": 0, "label_distribution": {} } - + label_dist = result_df["label"].value_counts().to_dict() label_names = {0: "DOWN", 1: "HOLD", 2: "UP"} - + logger.info(f"ML 학습 데이터 생성 완료: {len(result_df)}샘플, 분포 {label_dist}") - + return { "total_samples": len(result_df), "label_distribution": {label_names[k]: int(v) for k, v in label_dist.items()}, diff --git a/backend/app/services/db_worker.py b/backend/app/services/db_worker.py index 0ffefc5..3fba126 100644 --- a/backend/app/services/db_worker.py +++ b/backend/app/services/db_worker.py @@ -455,9 +455,17 @@ async def db_writer_worker() -> None: record = result.scalar_one_or_none() if record is None: - record = ActiveTrade(**data) + # remaining_pct_multiply는 신규 생성 시 무시 + create_data = {k: v for k, v in data.items() if k != "remaining_pct_multiply"} + record = ActiveTrade(**create_data) session.add(record) else: + # remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함 + # (MDD 축소 시 기존 분할매도 이력 반영) + multiply_ratio = data.pop("remaining_pct_multiply", None) + if multiply_ratio is not None: + current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0 + record.remaining_pct = current_remaining * float(multiply_ratio) for key, val in data.items(): if key != "stock_code" and val is not None: setattr(record, key, val) diff --git a/backend/app/services/korea_investment_api.py b/backend/app/services/korea_investment_api.py index ccfc6ee..b96d301 100644 --- a/backend/app/services/korea_investment_api.py +++ b/backend/app/services/korea_investment_api.py @@ -232,6 +232,7 @@ class KoreaInvestmentAPI: # httpx 싱글톤 클라이언트 (커넥션 풀 재활용 — 매 요청마다 TCP/TLS 핸드셰이크 방지) self._http_client: Optional[httpx.AsyncClient] = None + self._http_client_lock = asyncio.Lock() # 토큰 발급 Lock (중복 발급 방지) self._token_lock = asyncio.Lock() @@ -250,7 +251,28 @@ class KoreaInvestmentAPI: # 캐시된 토큰 로드 self._load_token_from_cache() - + + async def _get_http_client(self) -> httpx.AsyncClient: + """싱글톤 httpx 클라이언트 반환. 없으면 생성.""" + if self._http_client is None or self._http_client.is_closed: + async with self._http_client_lock: + if self._http_client is None or self._http_client.is_closed: + self._http_client = httpx.AsyncClient( + timeout=httpx.Timeout(10.0, connect=5.0), + limits=httpx.Limits( + max_connections=20, + max_keepalive_connections=10, + keepalive_expiry=30, + ), + ) + return self._http_client + + async def close(self) -> None: + """httpx 클라이언트 정리. 앱 종료 시 호출.""" + if self._http_client and not self._http_client.is_closed: + await self._http_client.aclose() + self._http_client = None + def _load_token_from_cache(self): """파일에서 토큰 로드 (앱 시작 시 동기 호출)""" try: @@ -447,13 +469,13 @@ class KoreaInvestmentAPI: # 타임아웃 설정 timeout = httpx.Timeout(self.timeout, connect=10.0) - async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=5.0)) as client: - response = await client.request( - method, - f"{self.base_url}{endpoint}", - headers=headers, - **kwargs - ) + client = await self._get_http_client() + response = await client.request( + method, + f"{self.base_url}{endpoint}", + headers=headers, + **kwargs + ) # 성공 if response.status_code == 200: diff --git a/backend/app/services/kr_backtest_engine.py b/backend/app/services/kr_backtest_engine.py index e8701ed..c3d542c 100644 --- a/backend/app/services/kr_backtest_engine.py +++ b/backend/app/services/kr_backtest_engine.py @@ -766,7 +766,12 @@ class KrBacktestEngine: market: str, random_seed: Optional[int], ) -> BacktestResult: - """핵심 시뮬레이션 루프 (동기)""" + """핵심 시뮬레이션 루프 (동기) + + [Look-ahead bias 수정] + T일 close로 신호(on_bar) 생성 -> T+1일 open + 슬리피지로 체결. + 마지막 봉 신호는 다음 봉이 없으므로 무시합니다. + """ cash = initial_capital positions: Dict[str, Position] = {} all_trades: List[BacktestTrade] = [] @@ -784,6 +789,9 @@ class KrBacktestEngine: peak_equity = initial_capital seed_counter = random_seed if random_seed is not None else None + # 전일 신호를 저장 (T일 close 신호 -> T+1일 open 체결) + pending_orders: List[Dict[str, Any]] = [] + for bar_idx in range(len(all_dates)): current_date = all_dates[bar_idx] @@ -795,23 +803,18 @@ class KrBacktestEngine: position_value += stock_bars[bar_idx].close_price * pos.quantity total_equity = cash + position_value - # 전략 호출 - orders = strategy.on_bar( - bar_idx=bar_idx, - bars=bars, - positions=positions, - cash=cash, - total_equity=total_equity, - params=params, - ) - - # 주문 실행 - for order in orders: + # -- 1단계: 전일 신호를 오늘 open + 슬리피지로 체결 -- + for order in pending_orders: action = order["action"] code = order["stock_code"] - price = order["price"] quantity = order["quantity"] + # T+1일 open 가격을 base price로 사용 + stock_bars = bars.get(code, []) + if bar_idx >= len(stock_bars): + continue + next_open = float(stock_bars[bar_idx].open_price) + # 슬리피지용 시드 if seed_counter is not None: seed_counter += 1 @@ -820,7 +823,7 @@ class KrBacktestEngine: slip_seed = None if action == "BUY": - exec_price = _apply_buy_slippage(price, slip_seed) + exec_price = _apply_buy_slippage(next_open, slip_seed) total_cost, commission = _calc_buy_cost(exec_price, quantity) if total_cost > cash: @@ -831,7 +834,7 @@ class KrBacktestEngine: total_cost, commission = _calc_buy_cost(exec_price, quantity) cash -= total_cost - slippage_amount = abs(exec_price - price) * quantity + slippage_amount = abs(exec_price - next_open) * quantity trade = BacktestTrade( stock_code=code, @@ -859,14 +862,13 @@ class KrBacktestEngine: if pos is None: continue - sell_pct = order.get("sell_pct", 1.0) sell_qty = min(quantity, pos.quantity) - exec_price = _apply_sell_slippage(price, slip_seed) + exec_price = _apply_sell_slippage(next_open, slip_seed) proceeds, commission, tax = _calc_sell_proceeds(exec_price, sell_qty, market) cash += proceeds - slippage_amount = abs(price - exec_price) * sell_qty + slippage_amount = abs(next_open - exec_price) * sell_qty # PnL 계산 entry_cost = pos.entry_price * sell_qty * (1 + BUY_COMMISSION_RATE) @@ -899,6 +901,29 @@ class KrBacktestEngine: pos.quantity = remaining_qty pos.remaining_pct = remaining_qty / (pos.quantity + sell_qty) + pending_orders.clear() + + # -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) -- + is_last_bar = (bar_idx == len(all_dates) - 1) + if not is_last_bar: + # 체결 이후의 상태로 평가 자산 재계산 (전략에 전달) + position_value = 0.0 + for code, pos in positions.items(): + stock_bars_eval = bars.get(code, []) + if bar_idx < len(stock_bars_eval): + position_value += stock_bars_eval[bar_idx].close_price * pos.quantity + total_equity = cash + position_value + + orders = strategy.on_bar( + bar_idx=bar_idx, + bars=bars, + positions=positions, + cash=cash, + total_equity=total_equity, + params=params, + ) + pending_orders = list(orders) + # 일말 평가 position_value = 0.0 for code, pos in positions.items(): diff --git a/backend/app/services/mdd_manager.py b/backend/app/services/mdd_manager.py index 4a5d2ef..2327aec 100644 --- a/backend/app/services/mdd_manager.py +++ b/backend/app/services/mdd_manager.py @@ -277,11 +277,15 @@ class MDDManager: await enqueue_close_active_trade(stock_code) else: # 부분 매도: remaining_pct 갱신 + # 기존 ActiveTrade의 remaining_pct에 매도 비율을 곱해야 함 + # (예: 기존 0.5 * 남은비율 0.7 = 0.35) + # db_worker가 직접 setattr하므로, reduce_ratio를 전달하여 + # 워커가 기존 값에 곱하도록 처리 from app.services.db_worker import enqueue_active_trade_upsert - new_remaining = remaining_qty / quantity + sell_ratio = remaining_qty / quantity # 남은 비율 (0.7 or 0.5) await enqueue_active_trade_upsert({ "stock_code": stock_code, - "remaining_pct": new_remaining, + "remaining_pct_multiply": sell_ratio, }) results.append({ diff --git a/backend/app/services/strategy_optimizer.py b/backend/app/services/strategy_optimizer.py index a954b0f..5373706 100644 --- a/backend/app/services/strategy_optimizer.py +++ b/backend/app/services/strategy_optimizer.py @@ -47,6 +47,11 @@ class OptimizationResult: applied: bool apply_reason: str total_combinations: int + # IS/OOS 검증 결과 (v2 추가) + oos_sharpe: float = 0.0 + oos_win_rate: float = 0.0 + oos_mdd: float = 0.0 + is_oos_sharpe_ratio: float = 0.0 # OOS Sharpe / IS Sharpe (1.0에 가까울수록 건전) # ───────────────────────────────────────────────────────────────────────────── @@ -282,7 +287,7 @@ async def run_optimization( Returns: OptimizationResult """ - logger.info(f"🔧 전략 최적화 시작 (trigger={trigger}, period={period_days}일)") + logger.info(f"전략 최적화 시작 (trigger={trigger}, period={period_days}일)") # 1. 현재 파라미터 저장 (롤백용) prev_params = await _get_current_params() @@ -310,16 +315,26 @@ async def run_optimization( total_combinations=0, ) + # ── IS/OOS 기간 분할 (70% IS / 30% OOS, 날짜 기준) ── end_date = date.today() start_date = end_date - timedelta(days=period_days) + is_days = int(period_days * 0.7) + is_end_date = start_date + timedelta(days=is_days) + oos_start_date = is_end_date + timedelta(days=1) + oos_end_date = end_date - # 4. 현재 파라미터 기준 Sharpe 계산 - cur_score, cur_sharpe, cur_wr, cur_mdd = await _evaluate_params( - prev_params, stock_codes, start_date, end_date, initial_capital + logger.info( + f" IS 기간: {start_date} ~ {is_end_date} ({is_days}일), " + f"OOS 기간: {oos_start_date} ~ {oos_end_date} ({period_days - is_days}일)" ) - logger.info(f" 현재 성과 → Sharpe: {cur_sharpe:.3f}, 승률: {cur_wr:.1f}%, MDD: {cur_mdd:.1f}%") - # 5. 그리드 서치 (배치 5개씩 병렬) + # 4. 현재 파라미터 기준 Sharpe 계산 (IS 기간) + cur_score, cur_sharpe, cur_wr, cur_mdd = await _evaluate_params( + prev_params, stock_codes, start_date, is_end_date, initial_capital + ) + logger.info(f" 현재 IS 성과 -> Sharpe: {cur_sharpe:.3f}, 승률: {cur_wr:.1f}%, MDD: {cur_mdd:.1f}%") + + # 5. 그리드 서치 — IS 기간에서 최적 파라미터 탐색 (배치 5개씩 병렬) best_score = cur_score best_params = dict(prev_params) best_sharpe = cur_sharpe @@ -330,7 +345,7 @@ async def run_optimization( for i in range(0, len(combinations), batch_size): batch = combinations[i:i + batch_size] tasks = [ - _evaluate_params(p, stock_codes, start_date, end_date, initial_capital) + _evaluate_params(p, stock_codes, start_date, is_end_date, initial_capital) for p in batch ] results = await asyncio.gather(*tasks, return_exceptions=True) @@ -345,39 +360,65 @@ async def run_optimization( best_mdd = mdd best_params = {**prev_params, **combo} - logger.info(f" 최적 성과 → Sharpe: {best_sharpe:.3f}, 승률: {best_win_rate:.1f}%, MDD: {best_mdd:.1f}%") + logger.info(f" IS 최적 성과 -> Sharpe: {best_sharpe:.3f}, 승률: {best_win_rate:.1f}%, MDD: {best_mdd:.1f}%") - # 6. 자동 적용 여부 결정 (Sharpe 10% 이상 개선) + # 6. OOS 기간에서 최적 파라미터 검증 (과적합 탐지) + oos_score, oos_sharpe, oos_wr, oos_mdd = await _evaluate_params( + best_params, stock_codes, oos_start_date, oos_end_date, initial_capital + ) + is_oos_ratio = (oos_sharpe / best_sharpe) if best_sharpe > 0 else 0.0 + logger.info( + f" OOS 검증 -> Sharpe: {oos_sharpe:.3f}, 승률: {oos_wr:.1f}%, MDD: {oos_mdd:.1f}%, " + f"IS/OOS Sharpe 비율: {is_oos_ratio:.2f}" + ) + + if is_oos_ratio < 0.5 and best_sharpe > 0: + logger.warning( + f" [과적합 경고] OOS Sharpe / IS Sharpe = {is_oos_ratio:.2f} < 0.5 " + "-- 파라미터가 IS 기간에 과적합되었을 가능성 높음" + ) + + # 7. 자동 적용 여부 결정 + # 조건: IS Sharpe 10% 이상 개선 AND OOS Sharpe > 0 (OOS에서도 양의 성과) improvement_threshold = 1.1 - should_apply = ( + _is_improved = ( cur_sharpe > 0 and best_sharpe >= cur_sharpe * improvement_threshold and best_params != prev_params ) if cur_sharpe <= 0 and best_sharpe > 0: - should_apply = True + _is_improved = True + + # OOS 검증 게이트: OOS Sharpe > 0 이어야 실전 적용 허가 + _oos_passed = oos_sharpe > 0 + _should_apply = _is_improved and _oos_passed applied = False apply_reason = "" - if should_apply: + if _should_apply: improvement_pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100 - apply_reason = f"Sharpe {improvement_pct:.1f}% 향상 ({cur_sharpe:.3f} → {best_sharpe:.3f})" + apply_reason = ( + f"IS Sharpe {improvement_pct:.1f}% 향상 ({cur_sharpe:.3f} -> {best_sharpe:.3f}), " + f"OOS Sharpe: {oos_sharpe:.3f}, IS/OOS 비율: {is_oos_ratio:.2f}" + ) # 최적화된 파라미터만 적용 (prev와 다른 키만) changed = {k: v for k, v in best_params.items() if str(v) != str(prev_params.get(k, ""))} if changed: applied = await apply_params(changed, reason="auto_optimizer") if applied: - logger.info(f"✅ 자동 적용 완료: {apply_reason}") + logger.info(f"자동 적용 완료: {apply_reason}") else: if cur_sharpe <= 0 and best_sharpe <= 0: apply_reason = "insufficient_data" + elif not _oos_passed and _is_improved: + apply_reason = f"OOS 검증 실패 (OOS Sharpe: {oos_sharpe:.3f} <= 0, 과적합 의심)" elif best_sharpe < cur_sharpe * improvement_threshold: pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100 - apply_reason = f"개선 미달 ({pct:+.1f}% < 10%)" + apply_reason = f"IS 개선 미달 ({pct:+.1f}% < 10%)" else: apply_reason = "no_improvement" - logger.info(f"⏭️ 자동 적용 건너뜀: {apply_reason}") + logger.info(f"자동 적용 건너뜀: {apply_reason}") return OptimizationResult( best_params=best_params, @@ -390,6 +431,10 @@ async def run_optimization( applied=applied, apply_reason=apply_reason, total_combinations=total_combos, + oos_sharpe=oos_sharpe, + oos_win_rate=oos_wr, + oos_mdd=oos_mdd, + is_oos_sharpe_ratio=round(is_oos_ratio, 4), ) diff --git a/backend/app/services/swing/entry_service.py b/backend/app/services/swing/entry_service.py index 23f5c43..9fafd4b 100644 --- a/backend/app/services/swing/entry_service.py +++ b/backend/app/services/swing/entry_service.py @@ -9,6 +9,10 @@ strategy_type='swing' 진입만 처리하는 독립 서비스. - SIDEWAYS: 진입 없음 (CLAUDE.md 6.1) - 포지션 사이징: 종목당 총자본 6% (MAX_POSITION_SIZE), 최대 25종목 +듀얼 전략 (CLAUDE.md 섹션 8.4): + - 모멘텀 돌파: 20일 신고가 + 거래량 2x + RSI 50~70 (BULL만) + - 눌림목 매수: MA20>MA60 + MA20 근처 조정 + RSI 반등 (BULL만) + 진입 전 체크: 1. 장 오픈 여부 (정규장 09:00~15:30만 허용) 2. 스윙 포지션 한도 (25종목) @@ -20,10 +24,11 @@ strategy_type='swing' 진입만 처리하는 독립 서비스. import asyncio import logging from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import select +from sqlalchemy import select, desc +from datetime import date from app.core.config import settings -from app.models.models import Portfolio +from app.models.models import Portfolio, StockAnalysis from app.services.trade_executor import execute_auto_trade, execute_auto_trade_background from app.services.market_analyzer import check_market_regime from app.services.mdd_manager import mdd_manager @@ -144,3 +149,180 @@ def _inject_strategy_type(rec: dict, strategy_type: str) -> None: for stock in group: if isinstance(stock, dict): stock["strategy_type"] = strategy_type + + +async def scan_dual_strategy_signals(db: AsyncSession) -> list[dict]: + """듀얼 전략 시그널 스캔 및 진입 실행 (CLAUDE.md 섹션 8.4). + + 배치 분석으로 저장된 stock_analysis 종목들의 일봉 차트를 조회하여 + 모멘텀 돌파 / 눌림목 매수 시그널을 탐색하고, 기존 매매 파이프라인으로 진입. + + BULL 시장에서만 실행. SIDEWAYS/BEAR에서는 스킵. + + Returns: + list[dict]: 실행된 매수 결과 목록 + """ + from app.core.market_schedule import is_market_open + from app.services.korea_investment_api import korea_investment_api + from app.services.mean_reversion_strategy import ( + check_momentum_breakout_signal, + check_pullback_buy_signal, + calculate_rsi, + ) + + if not is_market_open(): + return [] + + # BULL 시장에서만 듀얼 전략 실행 + try: + regime = await check_market_regime() + regime_str = regime.get("regime_200d", "bull") if isinstance(regime, dict) else "bull" + if regime_str != "bull": + logger.debug("[DualStrategy] BULL이 아님 (%s) — 듀얼 전략 스킵", regime_str) + return [] + except Exception as e: + logger.warning(f"[DualStrategy] 시장 국면 조회 실패: {e}") + return [] + + # 포지션 한도 체크 + current_count = await count_swing_positions(db) + remaining_slots = MAX_SWING_POSITIONS - current_count + if remaining_slots <= 0: + return [] + + # MDD 체크 + if not await mdd_manager.can_open_new_position(): + return [] + + # Crisis 모드 체크 + try: + from app.services.adaptive_strategy import get_current_adaptive_params + _adp = get_current_adaptive_params() + if not _adp.fallback_used and _adp.trading_mode == "crisis": + return [] + except Exception: + pass + + # 현재 보유 종목 (중복 진입 방지) + result = await db.execute(select(Portfolio.stock_code).where(Portfolio.quantity > 0)) + held_codes = {row[0] for row in result.fetchall()} + + # 최근 분석된 종목 중 상위 30개 조회 (score 기준) + analysis_query = ( + select(StockAnalysis) + .where(StockAnalysis.analysis_date >= date.today()) + .order_by(desc(StockAnalysis.score)) + .limit(30) + ) + analysis_result = await db.execute(analysis_query) + candidates = analysis_result.scalars().all() + + if not candidates: + logger.debug("[DualStrategy] 분석 데이터 없음 — 스킵") + return [] + + dual_stocks = [] + scan_count = 0 + + for stock in candidates: + if scan_count >= 10: + break + if stock.stock_code in held_codes: + continue + if float(stock.current_price or 0) < settings.MIN_PRICE_KRW: + continue + + try: + chart = await korea_investment_api.get_daily_chart(stock.stock_code, limit=60) + output2 = chart.get("output2", []) + if not output2 or len(output2) < 20: + continue + + # 최신이 앞에 오므로 reverse + bars = list(reversed(output2)) + closes = [float(b.get("stck_clpr", 0)) for b in bars] + highs = [float(b.get("stck_hgpr", 0)) for b in bars] + volumes = [int(b.get("acml_vol", 0)) for b in bars] + + if not closes or closes[-1] <= 0: + continue + + current_close = closes[-1] + current_volume = volumes[-1] if volumes else 0 + + rsi_values = calculate_rsi(closes, period=14) + rsi = rsi_values[-1] if rsi_values else 50.0 + rsi_prev = rsi_values[-2] if len(rsi_values) >= 2 else None + + scan_count += 1 + + # 모멘텀 돌파 체크 + momentum_signal = check_momentum_breakout_signal( + closes=closes, highs=highs, volumes=volumes, + current_close=current_close, current_volume=current_volume, rsi=rsi, + ) + if momentum_signal: + dual_stocks.append({ + "stock_code": stock.stock_code, + "stock_name": stock.stock_name, + "current_price": int(current_close), + "score": 0.8, + "signal": momentum_signal["description"], + "strategy_type": "swing", + "entry_reason": f"[MOMENTUM_BREAKOUT] {momentum_signal['description']}", + }) + logger.info( + f"[DualStrategy] 모멘텀 돌파 시그널: {stock.stock_code} {stock.stock_name}" + ) + continue + + # 눌림목 매수 체크 + pullback_signal = check_pullback_buy_signal( + closes=closes, current_close=current_close, + current_volume=current_volume, rsi=rsi, rsi_prev=rsi_prev, + volumes=volumes, + ) + if pullback_signal: + dual_stocks.append({ + "stock_code": stock.stock_code, + "stock_name": stock.stock_name, + "current_price": int(current_close), + "score": 0.6, + "signal": pullback_signal["description"], + "strategy_type": "swing", + "entry_reason": f"[PULLBACK_BUY] {pullback_signal['description']}", + }) + logger.info( + f"[DualStrategy] 눌림목 매수 시그널: {stock.stock_code} {stock.stock_name}" + ) + + await asyncio.sleep(0.3) # KIS API 레이트 리밋 + + except Exception as e: + logger.debug(f"[DualStrategy] {stock.stock_code} 스캔 실패: {e}") + continue + + if not dual_stocks: + logger.debug("[DualStrategy] 시그널 없음") + return [] + + # 남은 슬롯 이내로 제한 + dual_stocks = dual_stocks[:remaining_slots] + + # execute_auto_trade_background 형식으로 변환하여 실행 + rec = { + "recommendations_by_rule": [{ + "rule_name": "dual_strategy", + "stocks": dual_stocks, + }] + } + + try: + result = await execute_auto_trade_background(rec, db) + if isinstance(result, list): + logger.info(f"[DualStrategy] {len(result)}건 진입 실행 완료") + return result + except Exception as e: + logger.error(f"[DualStrategy] 진입 실행 오류: {e}", exc_info=True) + + return [] diff --git a/backend/app/services/swing/scheduler.py b/backend/app/services/swing/scheduler.py index 80dc990..0334d14 100644 --- a/backend/app/services/swing/scheduler.py +++ b/backend/app/services/swing/scheduler.py @@ -30,7 +30,7 @@ from app.core.database import AsyncSessionLocal from app.core.market_schedule import is_market_open from app.services.mdd_manager import mdd_manager from app.services.swing.exit_service import check_and_execute_exit_v2 -from app.services.swing.entry_service import execute_swing_entries +from app.services.swing.entry_service import execute_swing_entries, scan_dual_strategy_signals logger = logging.getLogger(__name__) _KST = pytz.timezone("Asia/Seoul") @@ -57,6 +57,25 @@ def _normalize_recommendations(rec: dict) -> dict: return rec +def _filter_unsuitable_stocks(rec: dict, unsuitable_codes: set) -> None: + """장 전 AI 분석에서 부적합 판정된 종목을 추천 목록에서 제거.""" + groups = rec.get("recommendations_by_rule", []) + if isinstance(groups, list): + for group in groups: + if isinstance(group, dict) and "stocks" in group: + group["stocks"] = [ + s for s in group["stocks"] + if isinstance(s, dict) and s.get("stock_code") not in unsuitable_codes + ] + elif isinstance(groups, dict): + for key, stocks in groups.items(): + if isinstance(stocks, list): + groups[key] = [ + s for s in stocks + if isinstance(s, dict) and s.get("stock_code") not in unsuitable_codes + ] + + async def swing_scheduler() -> None: """60초 주기 스윙 트레이딩 스케줄러. @@ -149,7 +168,25 @@ async def swing_scheduler() -> None: await asyncio.sleep(60) continue - # 5. MCP RISK_ASSESSMENT + # 5. 장 전 AI 분석 결과 활용 + MCP 리스크 평가 (최적화) + # 장 전 Sonnet 배치 분석 결과가 Redis에 있으면 부적합 종목을 사전 필터링 + # MCP RISK_ASSESSMENT는 실제 매수 직전에만 호출 (비용 절감) + _ai_filtered_codes = set() + try: + _ai_pre = await get_json("ai:pre_market:latest") + if _ai_pre and isinstance(_ai_pre, dict): + for s in _ai_pre.get("suitable_stocks", []): + if isinstance(s, dict) and not s.get("suitable", True): + _ai_filtered_codes.add(s.get("stock_code", "")) + if _ai_filtered_codes: + logger.info(f"[Swing] AI 장 전 분석 필터: {len(_ai_filtered_codes)}종목 제외") + except Exception: + pass + + # 추천 목록에서 AI 부적합 종목 제거 + if _ai_filtered_codes: + _filter_unsuitable_stocks(rec, _ai_filtered_codes) + recommendations = rec.get("recommendations_by_rule", []) if isinstance(recommendations, dict): rec_count = sum(len(v) for v in recommendations.values()) @@ -169,6 +206,12 @@ async def swing_scheduler() -> None: for s in g.get("stocks", [])[:3] ][:10] + if not top_stocks: + logger.info("[Swing] 필터링 후 추천 종목 없음 — 스킵") + await asyncio.sleep(60) + continue + + # MCP 리스크 평가 (Haiku — 실제 진입 직전 1회) trade_plan = { "action": "BUY", "strategy": "swing", @@ -196,7 +239,17 @@ async def swing_scheduler() -> None: async with AsyncSessionLocal() as entry_session: logger.info(f"[Swing] 진입 실행 시작: {top_stocks[:5]}") await execute_swing_entries(rec, entry_session) - logger.info("[Swing] ── 사이클 완료 ──") + + # 7. 듀얼 전략 시그널 스캔 (모멘텀 돌파 + 눌림목 매수, CLAUDE.md 8.4) + try: + async with AsyncSessionLocal() as dual_session: + dual_results = await scan_dual_strategy_signals(dual_session) + if dual_results: + logger.info(f"[Swing] 듀얼 전략 진입: {len(dual_results)}건") + except Exception as e: + logger.debug(f"[Swing] 듀얼 전략 스캔 오류: {e}") + + logger.info("[Swing] ── 사이클 완료 ──") except Exception as e: logger.error(f"[Swing] 스케줄러 사이클 오류: {e}", exc_info=True) diff --git a/backend/app/services/trade_analyzer.py b/backend/app/services/trade_analyzer.py index 8123593..63c4d33 100644 --- a/backend/app/services/trade_analyzer.py +++ b/backend/app/services/trade_analyzer.py @@ -4,10 +4,14 @@ [역할] - TradeHistory 레코드를 분석해 전략 성과 통계를 생성 - 전략별 / 요일별 승률, profit factor, 패턴 분석 +- 리스크 조정 수익률 지표 (Sharpe, Sortino, Calmar) +- 전략 드리프트 감지 (롤링 승률, 롤링 Sharpe, 연속 손실, MDD 가속) - 완료 거래 대상 매매일지 템플릿 자동 생성 - journals.py / analysis.py 엔드포인트에서 사용 """ import logging +import math +from collections import defaultdict from datetime import datetime, timedelta, date from typing import Any, Dict, List, Optional from decimal import Decimal @@ -70,6 +74,276 @@ def _parse_exit_type(exit_reason: Optional[str]) -> str: return "unknown" +# ───────────────────────────────────────────────────────────────────────────── +# 리스크 조정 수익률 지표 헬퍼 +# ───────────────────────────────────────────────────────────────────────────── + +# 무위험이자율 3% (연), 일별 = 0.03/252 +_RISK_FREE_ANNUAL = 0.03 +_RISK_FREE_DAILY = _RISK_FREE_ANNUAL / 252.0 +# 연환산 계수 +_ANNUALIZE_FACTOR = math.sqrt(252) +# 최소 거래일 수 (통계적 유의성) +_MIN_TRADING_DAYS = 20 + + +def _build_daily_pnl_series(trades: List["TradeHistory"]) -> Dict[date, float]: + """SELL 거래의 created_at 날짜별 net_profit_loss를 집계하여 일별 PnL 딕셔너리 반환. + + Returns: + {date(2026,3,1): 15000.0, date(2026,3,2): -3200.0, ...} + """ + daily_pnl: Dict[date, float] = defaultdict(float) + for t in trades: + if t.created_at is None: + continue + day = t.created_at.date() if isinstance(t.created_at, datetime) else t.created_at + daily_pnl[day] += _to_float(t.net_profit_loss) + return dict(daily_pnl) + + +def _daily_pnl_to_returns(daily_pnl: Dict[date, float], initial_capital: float = 10_000_000.0) -> List[float]: + """일별 PnL을 누적 equity curve로 변환 후 일별 수익률(%) 시계열 반환. + + Args: + daily_pnl: 날짜별 PnL 금액 + initial_capital: 기준 자본금 (기본 1000만원, equity curve 계산용) + + Returns: + 일별 수익률 리스트 (소수 형태, 예: 0.01 = 1%) + """ + if not daily_pnl: + return [] + + sorted_days = sorted(daily_pnl.keys()) + equity = initial_capital + daily_returns = [] + + for day in sorted_days: + pnl = daily_pnl[day] + if equity <= 0: + # 파산 방지 — 잔여 자본이 0 이하이면 중단 + break + ret = pnl / equity + daily_returns.append(ret) + equity += pnl + + return daily_returns + + +def _calc_sharpe_ratio(daily_returns: List[float]) -> Optional[float]: + """Sharpe Ratio = (R - Rf) / sigma(R) * sqrt(252) + + 일별 수익률 시계열로부터 연환산 Sharpe Ratio를 계산. + 거래일 _MIN_TRADING_DAYS 미만이면 None 반환. + """ + if len(daily_returns) < _MIN_TRADING_DAYS: + return None + + n = len(daily_returns) + mean_r = sum(daily_returns) / n + excess = mean_r - _RISK_FREE_DAILY + + # 표본 표준편차 + variance = sum((r - mean_r) ** 2 for r in daily_returns) / (n - 1) + std = math.sqrt(variance) if variance > 0 else 0.0 + + if std == 0: + return None + + return round((excess / std) * _ANNUALIZE_FACTOR, 4) + + +def _calc_sortino_ratio(daily_returns: List[float]) -> Optional[float]: + """Sortino Ratio = (R - Rf) / sigma_down(R) * sqrt(252) + + 하방 편차(downside deviation)만 사용 — 수익률이 무위험이자율 미만인 경우만 패널티. + """ + if len(daily_returns) < _MIN_TRADING_DAYS: + return None + + n = len(daily_returns) + mean_r = sum(daily_returns) / n + excess = mean_r - _RISK_FREE_DAILY + + # 하방 편차: Rf 미만인 수익률에 대해서만 편차 계산 + downside_diffs = [(r - _RISK_FREE_DAILY) ** 2 for r in daily_returns if r < _RISK_FREE_DAILY] + if not downside_diffs: + # 하방이 전혀 없으면 Sortino 무한대 — None 대신 높은 값 표시 + return 99.0 if excess > 0 else None + + downside_variance = sum(downside_diffs) / n # 전체 N으로 나눔 (하방 편차 관행) + downside_std = math.sqrt(downside_variance) + + if downside_std == 0: + return None + + return round((excess / downside_std) * _ANNUALIZE_FACTOR, 4) + + +def _calc_calmar_ratio(daily_returns: List[float]) -> Optional[float]: + """Calmar Ratio = CAGR / |MDD| + + 일별 수익률로부터 CAGR과 MDD를 계산. + """ + if len(daily_returns) < _MIN_TRADING_DAYS: + return None + + # Equity curve 구성 (1.0 기준) + equity = 1.0 + equities = [equity] + for r in daily_returns: + equity *= (1.0 + r) + equities.append(equity) + + final_equity = equities[-1] + n_days = len(daily_returns) + + # CAGR (연환산 복합 성장률) + if final_equity <= 0 or n_days == 0: + return None + cagr = (final_equity ** (252.0 / n_days)) - 1.0 + + # MDD 계산 + peak = equities[0] + max_dd = 0.0 + for eq in equities: + if eq > peak: + peak = eq + dd = (eq - peak) / peak if peak > 0 else 0.0 + if dd < max_dd: + max_dd = dd + + if max_dd == 0: + # drawdown이 없으면 — 전체 수익만 발생 + return 99.0 if cagr > 0 else None + + return round(cagr / abs(max_dd), 4) + + +def _calc_risk_adjusted_metrics(trades: List["TradeHistory"]) -> Dict[str, Any]: + """Sharpe, Sortino, Calmar Ratio를 한 번에 계산하여 딕셔너리 반환. + + Returns: + { + "sharpe_ratio": float | None, + "sortino_ratio": float | None, + "calmar_ratio": float | None, + "daily_returns_count": int + } + """ + daily_pnl = _build_daily_pnl_series(trades) + daily_returns = _daily_pnl_to_returns(daily_pnl) + + return { + "sharpe_ratio": _calc_sharpe_ratio(daily_returns), + "sortino_ratio": _calc_sortino_ratio(daily_returns), + "calmar_ratio": _calc_calmar_ratio(daily_returns), + "daily_returns_count": len(daily_returns), + } + + +# ───────────────────────────────────────────────────────────────────────────── +# 전략 드리프트 감지 헬퍼 +# ───────────────────────────────────────────────────────────────────────────── + +_DRIFT_ROLLING_WINDOW = 20 # 최근 20거래 / 20거래일 +_DRIFT_WIN_RATE_THRESHOLD = 0.30 # 30% 미만 경고 +_DRIFT_SHARPE_THRESHOLD = 0.0 # 0 이하 경고 +_DRIFT_LOSS_STREAK_THRESHOLD = 5 # 5연패 이상 경고 +_DRIFT_DRAWDOWN_THRESHOLD = -0.05 # -5% 이하 경고 + + +def _detect_drift_alerts(trades: List["TradeHistory"]) -> List[Dict[str, Any]]: + """전략 드리프트 감지 — 4가지 항목을 점검하여 경고 리스트 반환. + + 감지 항목: + 1. rolling_win_rate: 최근 20거래 승률 < 30% + 2. rolling_sharpe: 최근 20거래일 Sharpe <= 0 + 3. loss_streak: 최근 거래 중 5연패 이상 + 4. mdd_acceleration: 최근 일별 drawdown이 -5% 이하 + + Returns: + [{"type": str, "value": float, "threshold": float, "message": str}, ...] + """ + alerts: List[Dict[str, Any]] = [] + + if not trades: + return alerts + + pl_rates = [_to_float(t.profit_loss_rate) for t in trades] + + # ── 1. 롤링 승률 하락 ───────────────────────────────────────────────── + recent_rates = pl_rates[-_DRIFT_ROLLING_WINDOW:] + if len(recent_rates) >= _DRIFT_ROLLING_WINDOW: + recent_wins = sum(1 for r in recent_rates if r > 0) + rolling_wr = recent_wins / len(recent_rates) + if rolling_wr < _DRIFT_WIN_RATE_THRESHOLD: + alerts.append({ + "type": "rolling_win_rate", + "value": round(rolling_wr, 4), + "threshold": _DRIFT_WIN_RATE_THRESHOLD, + "message": f"최근 {_DRIFT_ROLLING_WINDOW}거래 승률 {rolling_wr*100:.1f}% (임계값 {_DRIFT_WIN_RATE_THRESHOLD*100:.0f}%)", + }) + + # ── 2. 롤링 Sharpe 하락 ─────────────────────────────────────────────── + daily_pnl = _build_daily_pnl_series(trades) + daily_returns = _daily_pnl_to_returns(daily_pnl) + recent_returns = daily_returns[-_DRIFT_ROLLING_WINDOW:] + if len(recent_returns) >= _DRIFT_ROLLING_WINDOW: + rolling_sharpe = _calc_sharpe_ratio(recent_returns) + if rolling_sharpe is not None and rolling_sharpe <= _DRIFT_SHARPE_THRESHOLD: + alerts.append({ + "type": "rolling_sharpe", + "value": rolling_sharpe, + "threshold": _DRIFT_SHARPE_THRESHOLD, + "message": f"최근 {_DRIFT_ROLLING_WINDOW}거래일 Sharpe {rolling_sharpe:.2f} (임계값 {_DRIFT_SHARPE_THRESHOLD:.1f})", + }) + + # ── 3. 연속 손실 스트릭 ──────────────────────────────────────────────── + # 최근 거래부터 역순으로 현재 연속 손실 카운트 + current_streak = 0 + for r in reversed(pl_rates): + if r < 0: + current_streak += 1 + else: + break + + if current_streak >= _DRIFT_LOSS_STREAK_THRESHOLD: + alerts.append({ + "type": "loss_streak", + "value": current_streak, + "threshold": _DRIFT_LOSS_STREAK_THRESHOLD, + "message": f"현재 {current_streak}연패 진행 중 (임계값 {_DRIFT_LOSS_STREAK_THRESHOLD}연패)", + }) + + # ── 4. MDD 가속 (최근 drawdown 체크) ────────────────────────────────── + if len(daily_returns) >= 5: + # 최근 equity curve에서 drawdown 계산 + equity = 1.0 + equities = [equity] + for r in daily_returns: + equity *= (1.0 + r) + equities.append(equity) + + # 최근 5거래일의 drawdown 확인 + recent_eq = equities[-6:] # 최근 5일분 + 직전 1일 + if len(recent_eq) >= 2: + recent_peak = max(recent_eq) + recent_low = min(recent_eq) + if recent_peak > 0: + recent_dd = (recent_low - recent_peak) / recent_peak + if recent_dd <= _DRIFT_DRAWDOWN_THRESHOLD: + alerts.append({ + "type": "mdd_acceleration", + "value": round(recent_dd, 4), + "threshold": _DRIFT_DRAWDOWN_THRESHOLD, + "message": f"최근 drawdown {recent_dd*100:.2f}% (임계값 {_DRIFT_DRAWDOWN_THRESHOLD*100:.1f}%)", + }) + + return alerts + + # ───────────────────────────────────────────────────────────────────────────── # 핵심 분석 함수 # ───────────────────────────────────────────────────────────────────────────── @@ -86,6 +360,11 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any "by_exit_type": {청산 유형별 통계}, "worst_stocks": [최대 손실 종목 Top5], "best_stocks": [최대 수익 종목 Top5], + "sharpe_ratio": float | None (연환산, Rf=3%), + "sortino_ratio": float | None (하방편차만), + "calmar_ratio": float | None (CAGR / |MDD|), + "daily_returns_count": int (일별 수익률 데이터 수), + "drift_alerts": [{type, value, threshold, message}, ...], "period_days": N, "generated_at": ISO 문자열 } @@ -111,6 +390,11 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any "by_exit_type": {}, "worst_stocks": [], "best_stocks": [], + "sharpe_ratio": None, + "sortino_ratio": None, + "calmar_ratio": None, + "daily_returns_count": 0, + "drift_alerts": [], "period_days": days, "total_trades": 0, "generated_at": datetime.utcnow().isoformat(), @@ -210,6 +494,12 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any worst_stocks = [{"stock_code": c, "net_pnl": round(p, 2)} for c, p in sorted_stocks[:5]] best_stocks = [{"stock_code": c, "net_pnl": round(p, 2)} for c, p in sorted_stocks[-5:][::-1]] + # ── 리스크 조정 수익률 지표 ──────────────────────────────────────────── + risk_metrics = _calc_risk_adjusted_metrics(trades) + + # ── 전략 드리프트 감지 ───────────────────────────────────────────────── + drift_alerts = _detect_drift_alerts(trades) + return { "summary": summary, "by_strategy": strategy_stats, @@ -217,6 +507,11 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any "by_exit_type": exit_stats, "worst_stocks": worst_stocks, "best_stocks": best_stocks, + "sharpe_ratio": risk_metrics["sharpe_ratio"], + "sortino_ratio": risk_metrics["sortino_ratio"], + "calmar_ratio": risk_metrics["calmar_ratio"], + "daily_returns_count": risk_metrics["daily_returns_count"], + "drift_alerts": drift_alerts, "period_days": days, "total_trades": len(trades), "generated_at": datetime.utcnow().isoformat(), @@ -302,6 +597,25 @@ def build_insights(perf: Dict[str, Any]) -> List[Dict[str, str]]: if sl_ratio > 0.5: insights.append({"level": "warning", "message": f"손절 청산 비율 {sl_ratio*100:.1f}% — 전체 청산의 절반 이상이 손절입니다. 진입 타이밍 재검토."}) + # ── 리스크 조정 지표 인사이트 ────────────────────────────────────────── + sharpe = perf.get("sharpe_ratio") + if sharpe is not None: + if sharpe < 0: + insights.append({"level": "danger", "message": f"Sharpe Ratio {sharpe:.2f} — 무위험이자율(3%) 대비 음의 초과수익입니다. 전략 전면 재검토 필요."}) + elif sharpe < 0.5: + insights.append({"level": "warning", "message": f"Sharpe Ratio {sharpe:.2f} — 낮은 수준입니다. 리스크 대비 수익 개선이 필요합니다."}) + elif sharpe >= 1.5: + insights.append({"level": "info", "message": f"Sharpe Ratio {sharpe:.2f} — 우수한 리스크 조정 수익률입니다."}) + + calmar = perf.get("calmar_ratio") + if calmar is not None and calmar < 1.0: + insights.append({"level": "warning", "message": f"Calmar Ratio {calmar:.2f} — MDD 대비 수익률이 낮습니다. 낙폭 관리 강화 필요."}) + + # ── 드리프트 경고 인사이트 ───────────────────────────────────────────── + drift_alerts = perf.get("drift_alerts", []) + for alert in drift_alerts: + insights.append({"level": "danger", "message": f"[드리프트 감지] {alert['message']}"}) + return insights diff --git a/backend/app/services/trade_executor.py b/backend/app/services/trade_executor.py index afc40c1..a5a32c3 100644 --- a/backend/app/services/trade_executor.py +++ b/backend/app/services/trade_executor.py @@ -654,8 +654,8 @@ async def _execute_buy( _adp = get_current_adaptive_params() if not _adp.fallback_used: _adaptive_position_cap = _adp.position_size - # Crisis 모드: 신규 진입 차단 - if _adp.trading_mode == "crisis" and _adp.position_size <= 0: + # Crisis 모드: max_daily_trades=0이면 신규 진입 차단 (CLAUDE.md 섹션 8.5) + if _adp.trading_mode == "crisis" and _adp.max_daily_trades <= 0: return {"action": "SKIP", "reason": "adaptive_crisis_mode", "stock": stock_name} except Exception: pass diff --git a/backend/app/services/us_backtest_engine.py b/backend/app/services/us_backtest_engine.py index dd19d1a..0cbb506 100644 --- a/backend/app/services/us_backtest_engine.py +++ b/backend/app/services/us_backtest_engine.py @@ -5,6 +5,7 @@ us_stock_strategy의 매수/매도 신호를 과거 데이터에 적용하여 """ import logging import math +import random from datetime import date, datetime, timedelta from typing import Dict, List, Any, Optional from dataclasses import dataclass, asdict @@ -25,6 +26,27 @@ logger = logging.getLogger(__name__) US_COMMISSION_RATE = 0.001 # 해외 주식 수수료 약 0.1% (매수+매도 각각) RISK_FREE_RATE = 0.04 # 무위험 수익률 4% (연간, USD 기준) TRADING_DAYS_PER_YEAR = 252 +US_SLIPPAGE_PCT = 0.0005 # +/-0.05% (미국 주식은 유동성이 높아 슬리피지가 낮음) + + +def _apply_us_buy_slippage(price: float, seed: Optional[int] = None) -> float: + """매수 시 슬리피지: 시장가 주문 시 약간 높은 가격에 체결 + + 미국 주식은 유동성이 높아 국내(0.1%) 대비 낮은 0~0.05% 적용. + """ + rng = random.Random(seed) + slip = rng.uniform(0, US_SLIPPAGE_PCT) + return price * (1 + slip) + + +def _apply_us_sell_slippage(price: float, seed: Optional[int] = None) -> float: + """매도 시 슬리피지: 시장가 주문 시 약간 낮은 가격에 체결 + + 미국 주식은 유동성이 높아 국내(0.1%) 대비 낮은 0~0.05% 적용. + """ + rng = random.Random(seed) + slip = rng.uniform(0, US_SLIPPAGE_PCT) + return price * (1 - slip) @dataclass @@ -229,6 +251,10 @@ async def run_us_backtest( indicators = _calculate_indicators(daily_data) # 3) 시뮬레이션 실행 + # + # [Look-ahead bias 수정] + # T일 close로 신호 생성 -> T+1일 open + 슬리피지로 체결. + # 마지막 봉 신호는 다음 봉이 없으므로 무시합니다. cash = initial_capital_usd position_qty = 0 position_avg_price = 0.0 @@ -236,96 +262,114 @@ async def run_us_backtest( trades: List[Dict] = [] equity_curve: List[Dict] = [] + # 전일 발생한 신호를 저장 (T일 close 신호 -> T+1일 open 체결) + pending_buy_signal: Optional[Dict] = None + pending_sell_signal: Optional[str] = None + for i in range(20, len(daily_data)): # 지표 계산에 최소 20일 필요 bar = daily_data[i] ind = indicators[i] close = bar["close"] bar_date = bar["date"] + bar_open = bar["open"] - # 포지션 보유 시: 매도 조건 체크 + # -- 1단계: 전일 신호를 오늘 open + 슬리피지로 체결 -- + if pending_sell_signal is not None and position_qty > 0: + exec_price = _apply_us_sell_slippage(bar_open) + amount = position_qty * exec_price + commission = amount * US_COMMISSION_RATE + pnl = (exec_price - position_avg_price) * position_qty - commission + pnl_pct = ((exec_price - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 + + trade = { + "date": bar_date, + "action": "SELL", + "ticker": ticker, + "price": round(exec_price, 4), + "quantity": position_qty, + "amount": round(amount, 2), + "commission": round(commission, 2), + "reason": pending_sell_signal, + "pnl": round(pnl, 2), + "pnl_pct": round(pnl_pct, 2), + } + trades.append(trade) + cash += amount - commission + position_qty = 0 + position_avg_price = 0.0 + position_high = 0.0 + pending_sell_signal = None + + if pending_buy_signal is not None and position_qty == 0: + exec_price = _apply_us_buy_slippage(bar_open) + buy_amount = min(cash * 0.95, cash) + if buy_amount > 10: + quantity = int(buy_amount / exec_price) + if quantity > 0: + cost = quantity * exec_price + commission = cost * US_COMMISSION_RATE + total_cost = cost + commission + + if total_cost <= cash: + cash -= total_cost + position_qty = quantity + position_avg_price = exec_price + position_high = exec_price + + trade = { + "date": bar_date, + "action": "BUY", + "ticker": ticker, + "price": round(exec_price, 4), + "quantity": quantity, + "amount": round(cost, 2), + "commission": round(commission, 2), + "reason": pending_buy_signal.get("reason", "buy_signal"), + "pnl": 0.0, + "pnl_pct": 0.0, + } + trades.append(trade) + pending_buy_signal = None + + # -- 2단계: 오늘 close 기준으로 신호 생성 (체결은 내일 open) -- + is_last_bar = (i == len(daily_data) - 1) + + # 트레일링 스탑용 최고가 갱신 (close 기준) if position_qty > 0: - # 트레일링 스탑용 최고가 갱신 if close > position_high: position_high = close - sell_signal = _check_backtest_sell( - close=close, - avg_price=position_avg_price, - high_since_entry=position_high, - rsi=ind.get("rsi", 50), - bb_middle=ind.get("bb_middle", close), - bb_upper=ind.get("bb_upper", close), - ) + if not is_last_bar: + # 매도 신호 체크 + if position_qty > 0: + sell_signal = _check_backtest_sell( + close=close, + avg_price=position_avg_price, + high_since_entry=position_high, + rsi=ind.get("rsi", 50), + bb_middle=ind.get("bb_middle", close), + bb_upper=ind.get("bb_upper", close), + ) + if sell_signal: + pending_sell_signal = sell_signal - if sell_signal: - amount = position_qty * close - commission = amount * US_COMMISSION_RATE - pnl = (close - position_avg_price) * position_qty - commission - pnl_pct = ((close - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 + # 매수 신호 체크 + if position_qty == 0 and pending_sell_signal is None: + prev_rsi = indicators[i - 3].get("rsi") if i >= 3 else None - trade = { - "date": bar_date, - "action": "SELL", - "ticker": ticker, - "price": close, - "quantity": position_qty, - "amount": round(amount, 2), - "commission": round(commission, 2), - "reason": sell_signal, - "pnl": round(pnl, 2), - "pnl_pct": round(pnl_pct, 2), - } - trades.append(trade) - cash += amount - commission - position_qty = 0 - position_avg_price = 0.0 - position_high = 0.0 - - # 포지션 없을 때: 매수 조건 체크 - if position_qty == 0: - prev_rsi = indicators[i - 3].get("rsi") if i >= 3 else None - - buy_signal = check_us_buy_signal( - rsi=ind.get("rsi", 50), - close=close, - bb_upper=ind.get("bb_upper", close), - bb_middle=ind.get("bb_middle", close), - bb_lower=ind.get("bb_lower", close), - avg_volume_20d=ind.get("avg_volume_20d", 0), - current_volume=bar.get("volume", 0), - current_price_usd=close, - rsi_prev=prev_rsi, - ) - - if buy_signal: - # 전체 자본의 최대 95% 투입 (여유 현금 확보) - buy_amount = min(cash * 0.95, cash) - if buy_amount > 10: # 최소 $10 이상 - quantity = int(buy_amount / close) - if quantity > 0: - cost = quantity * close - commission = cost * US_COMMISSION_RATE - total_cost = cost + commission - - if total_cost <= cash: - cash -= total_cost - position_qty = quantity - position_avg_price = close - position_high = close - - trade = { - "date": bar_date, - "action": "BUY", - "ticker": ticker, - "price": close, - "quantity": quantity, - "amount": round(cost, 2), - "commission": round(commission, 2), - "reason": buy_signal.get("reason", "buy_signal"), - "pnl": 0.0, - "pnl_pct": 0.0, - } - trades.append(trade) + buy_signal = check_us_buy_signal( + rsi=ind.get("rsi", 50), + close=close, + bb_upper=ind.get("bb_upper", close), + bb_middle=ind.get("bb_middle", close), + bb_lower=ind.get("bb_lower", close), + avg_volume_20d=ind.get("avg_volume_20d", 0), + current_volume=bar.get("volume", 0), + current_price_usd=close, + rsi_prev=prev_rsi, + ) + if buy_signal: + pending_buy_signal = buy_signal # 일별 자산 기록 position_value = position_qty * close if position_qty > 0 else 0.0 @@ -337,19 +381,20 @@ async def run_us_backtest( "position_value": round(position_value, 2), }) - # 4) 미청산 포지션 강제 청산 (백테스트 종료) + # 4) 미청산 포지션 강제 청산 (백테스트 종료, 슬리피지 적용) if position_qty > 0 and daily_data: last_close = daily_data[-1]["close"] - amount = position_qty * last_close + exec_price = _apply_us_sell_slippage(last_close) + amount = position_qty * exec_price commission = amount * US_COMMISSION_RATE - pnl = (last_close - position_avg_price) * position_qty - commission - pnl_pct = ((last_close - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 + pnl = (exec_price - position_avg_price) * position_qty - commission + pnl_pct = ((exec_price - position_avg_price) / position_avg_price * 100) if position_avg_price > 0 else 0.0 trades.append({ "date": daily_data[-1]["date"], "action": "SELL", "ticker": ticker, - "price": last_close, + "price": round(exec_price, 4), "quantity": position_qty, "amount": round(amount, 2), "commission": round(commission, 2), diff --git a/backend/app/strategy/strategy_manager.py b/backend/app/strategy/strategy_manager.py index 21ed21c..96baa2d 100644 --- a/backend/app/strategy/strategy_manager.py +++ b/backend/app/strategy/strategy_manager.py @@ -4,6 +4,7 @@ Strategy Manager 전략 등록·실행·관리를 담당하는 오케스트레이터 """ from typing import Dict, List, Optional, Any +from collections import deque import logging import asyncio from datetime import datetime @@ -27,7 +28,7 @@ class StrategyManager: def __init__(self): self.strategies: Dict[str, BaseStrategy] = {} # 전략 등록부 self.priority_order: List[str] = [] # 우선순위 - self.signal_history: List[Signal] = [] # 신호 이력 + self.signal_history: deque = deque(maxlen=1000) # 신호 이력 (최대 1000건, 메모리 누수 방지) logger.info("🎯 StrategyManager 초기화") @@ -222,7 +223,7 @@ class StrategyManager: Returns: Signal 목록 """ - history = self.signal_history[:] + history = list(self.signal_history) if stock_code: history = [s for s in history if s.stock_code == stock_code] diff --git a/backend/tests/test_adaptive_strategy.py b/backend/tests/test_adaptive_strategy.py index 6215315..d2b1576 100644 --- a/backend/tests/test_adaptive_strategy.py +++ b/backend/tests/test_adaptive_strategy.py @@ -124,23 +124,28 @@ class TestVolumeRegime: class TestTradingMode: def test_crisis_high_atr(self): - result = _determine_trading_mode(atr_percentile=96.0, adx_value=20.0, recent_5d_return=0.0, mdd_current=0.0) + # CLAUDE.md 8.5: ATR백분위 > 90 → Crisis + result = _determine_trading_mode(atr_percentile=91.0, adx_value=20.0, recent_5d_return=0.0, mdd_current=0.0) assert result == TradingMode.CRISIS.value def test_crisis_deep_loss(self): - result = _determine_trading_mode(atr_percentile=50.0, adx_value=20.0, recent_5d_return=-9.0, mdd_current=0.0) + # CLAUDE.md 8.5: 5일수익 < -5% → Crisis + result = _determine_trading_mode(atr_percentile=50.0, adx_value=20.0, recent_5d_return=-6.0, mdd_current=0.0) assert result == TradingMode.CRISIS.value def test_crisis_mdd(self): - result = _determine_trading_mode(atr_percentile=50.0, adx_value=20.0, recent_5d_return=0.0, mdd_current=-16.0) + # CLAUDE.md 8.5: MDD < -10% → Crisis + result = _determine_trading_mode(atr_percentile=50.0, adx_value=20.0, recent_5d_return=0.0, mdd_current=-11.0) assert result == TradingMode.CRISIS.value def test_defensive_high_atr(self): - result = _determine_trading_mode(atr_percentile=85.0, adx_value=20.0, recent_5d_return=0.0, mdd_current=0.0) + # CLAUDE.md 8.5: ATR백분위 > 70 → Defensive (but < 90 so not Crisis) + result = _determine_trading_mode(atr_percentile=75.0, adx_value=20.0, recent_5d_return=0.0, mdd_current=0.0) assert result == TradingMode.DEFENSIVE.value def test_defensive_moderate_loss(self): - result = _determine_trading_mode(atr_percentile=50.0, adx_value=20.0, recent_5d_return=-6.0, mdd_current=0.0) + # CLAUDE.md 8.5: 5일수익 < -3% → Defensive (but > -5% so not Crisis) + result = _determine_trading_mode(atr_percentile=50.0, adx_value=20.0, recent_5d_return=-4.0, mdd_current=0.0) assert result == TradingMode.DEFENSIVE.value def test_aggressive(self): @@ -157,8 +162,8 @@ class TestTradingMode: assert result == TradingMode.NORMAL.value def test_crisis_takes_priority_over_defensive(self): - # ATR=96 triggers both crisis (>95) and defensive (>80), crisis wins - result = _determine_trading_mode(atr_percentile=96.0, adx_value=30.0, recent_5d_return=3.0, mdd_current=0.0) + # ATR=92 triggers both crisis (>90) and defensive (>70), crisis wins + result = _determine_trading_mode(atr_percentile=92.0, adx_value=30.0, recent_5d_return=3.0, mdd_current=0.0) assert result == TradingMode.CRISIS.value diff --git a/deploy/docker-compose.infra.yml b/deploy/docker-compose.infra.yml new file mode 100644 index 0000000..01c55c6 --- /dev/null +++ b/deploy/docker-compose.infra.yml @@ -0,0 +1,56 @@ +# Gitea + Jenkins CI/CD 인프라 (NAS 최초 1회 실행) +# 사용법: cd deploy && docker-compose -f docker-compose.infra.yml up -d + +services: + gitea: + image: gitea/gitea:latest + container_name: gitea + environment: + - TZ=Asia/Seoul + - GITEA__database__DB_TYPE=sqlite3 + - GITEA__server__ROOT_URL=${GITEA_URL:-http://nas-ip:3001} + - GITEA__server__HTTP_PORT=3000 + - GITEA__server__SSH_DISABLED=true + volumes: + - gitea_data:/data + ports: + - "3001:3000" + restart: unless-stopped + networks: + - cicd-network + deploy: + resources: + limits: + memory: 256M + + jenkins: + image: jenkins/jenkins:lts-jdk17 + container_name: jenkins + environment: + - TZ=Asia/Seoul + - JAVA_OPTS=-Xmx512m -Xms256m + volumes: + - jenkins_data:/var/jenkins_home + # Docker-in-Docker: 호스트 Docker 소켓 공유 + - /var/run/docker.sock:/var/run/docker.sock + ports: + - "8080:8080" + - "50000:50000" + restart: unless-stopped + networks: + - cicd-network + deploy: + resources: + limits: + memory: 1G + # Jenkins에서 docker 명령 사용을 위한 그룹 설정 + # NAS에서 docker.sock 권한: sudo chown root:docker /var/run/docker.sock + user: root + +volumes: + gitea_data: + jenkins_data: + +networks: + cicd-network: + driver: bridge diff --git a/deploy/nginx-reverse-proxy.conf b/deploy/nginx-reverse-proxy.conf new file mode 100644 index 0000000..1eb2e92 --- /dev/null +++ b/deploy/nginx-reverse-proxy.conf @@ -0,0 +1,68 @@ +# Synology 역방향 프록시 대신 직접 Nginx를 쓸 경우 +# docker-compose.yml에 nginx 서비스 추가 후 이 설정 사용 +# +# Synology DSM → 제어판 → 로그인 포털 → 고급 → 역방향 프록시 로 설정하는 게 더 간편. +# 그 경우 이 파일은 불필요. +# +# Synology 역방향 프록시 설정: +# stock.your-domain.com (HTTPS 443) +# → Location /api → http://localhost:8000 +# → Location /ws → http://localhost:8000 (WebSocket) +# → Location / → http://localhost:3000 + +upstream backend { + server backend:8000; +} + +upstream frontend { + server frontend:80; +} + +server { + listen 80; + server_name stock.your-domain.com; + return 301 https://$host$request_uri; +} + +server { + listen 443 ssl http2; + server_name stock.your-domain.com; + + # Synology에서 발급한 Let's Encrypt 인증서 경로 + ssl_certificate /etc/ssl/certs/your-cert.pem; + ssl_certificate_key /etc/ssl/private/your-key.pem; + + # Security headers + add_header X-Frame-Options DENY; + add_header X-Content-Type-Options nosniff; + add_header X-XSS-Protection "1; mode=block"; + add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always; + + # API + location /api/ { + proxy_pass http://backend/api/; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_read_timeout 120s; + } + + # WebSocket + location /ws { + proxy_pass http://backend/ws; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_read_timeout 86400s; + } + + # Frontend + location / { + proxy_pass http://frontend/; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + } +} diff --git a/deploy/setup-nas.sh b/deploy/setup-nas.sh new file mode 100644 index 0000000..d9871b7 --- /dev/null +++ b/deploy/setup-nas.sh @@ -0,0 +1,134 @@ +#!/bin/bash +# ============================================================ +# Synology DS1517 초기 설정 스크립트 +# 사용법: ssh admin@nas-ip 에서 실행 +# chmod +x setup-nas.sh && ./setup-nas.sh +# ============================================================ + +set -e + +DEPLOY_DIR="/volume1/docker/autostock" +INFRA_DIR="/volume1/docker/cicd" +BACKUP_DIR="/volume1/docker/autostock/backups" + +echo "============================================" +echo " Auto-Stock NAS 배포 환경 초기 설정" +echo " NAS: Synology DS1517 (16GB)" +echo "============================================" + +# 1. 디렉토리 생성 +echo "[1/6] 디렉토리 생성..." +sudo mkdir -p ${DEPLOY_DIR} +sudo mkdir -p ${INFRA_DIR} +sudo mkdir -p ${BACKUP_DIR} +sudo mkdir -p ${DEPLOY_DIR}/logs + +# 2. CI/CD 인프라 시작 (Gitea + Jenkins) +echo "[2/6] CI/CD 인프라 시작..." +if [ -f "${INFRA_DIR}/docker-compose.infra.yml" ]; then + cd ${INFRA_DIR} + docker compose -f docker-compose.infra.yml up -d +else + echo " >> docker-compose.infra.yml을 ${INFRA_DIR}에 복사하세요" + echo " >> 그 후 다시 실행하거나: cd ${INFRA_DIR} && docker compose -f docker-compose.infra.yml up -d" +fi + +# 3. Jenkins Docker 소켓 권한 +echo "[3/6] Docker 소켓 권한 설정..." +sudo chmod 666 /var/run/docker.sock 2>/dev/null || true + +# 4. .env 템플릿 생성 +echo "[4/6] .env 템플릿 생성..." +if [ ! -f "${DEPLOY_DIR}/.env" ]; then + cat > ${DEPLOY_DIR}/.env << 'ENVEOF' +# ============================================================ +# Auto-Stock 프로덕션 환경변수 (NAS) +# 이 파일은 Git에 커밋하지 마세요! +# ============================================================ + +# === Database === +POSTGRES_USER=autostock +POSTGRES_PASSWORD=CHANGE_ME_STRONG_PASSWORD +POSTGRES_DB=autostock +DATABASE_URL=postgresql+asyncpg://autostock:CHANGE_ME_STRONG_PASSWORD@postgres:5432/autostock + +# === Redis === +REDIS_PASSWORD=CHANGE_ME_REDIS_PASSWORD +REDIS_URL=redis://:CHANGE_ME_REDIS_PASSWORD@redis:6379/0 + +# === RabbitMQ === +RABBITMQ_DEFAULT_USER=autostock +RABBITMQ_DEFAULT_PASS=CHANGE_ME_RABBITMQ_PASSWORD +RABBITMQ_URL=amqp://autostock:CHANGE_ME_RABBITMQ_PASSWORD@rabbitmq:5672/ + +# === AI Provider (claude + hybrid haiku/sonnet) === +AI_PROVIDER=claude +HYBRID_AI_ENABLED=true +ANTHROPIC_API_KEY=sk-ant-xxx +CLAUDE_MODEL=claude-sonnet-4-6-20250514 +CLAUDE_HAIKU_MODEL=claude-haiku-4-5-20251001 +GEMINI_API_KEY= +GEMINI_MODEL=gemini-2.0-flash + +# === KIS API (virtual/real 전환: KIS_MODE만 변경) === +KIS_MODE=virtual +KIS_VIRTUAL_APP_KEY= +KIS_VIRTUAL_APP_SECRET= +KIS_VIRTUAL_ACCOUNT_NO= +KIS_VIRTUAL_URL=https://openapivts.koreainvestment.com:29443 +KIS_REAL_APP_KEY= +KIS_REAL_APP_SECRET= +KIS_REAL_ACCOUNT_NO= +KIS_REAL_URL=https://openapi.koreainvestment.com:9443 + +# === Server === +BACKEND_HOST=0.0.0.0 +BACKEND_PORT=8000 +MCP_SERVER_HOST=0.0.0.0 +MCP_SERVER_PORT=8001 +SECRET_KEY=CHANGE_ME_64_CHARS_RANDOM_STRING +API_KEY=CHANGE_ME_API_KEY +MCP_API_KEY=CHANGE_ME_MCP_API_KEY + +# === Frontend === +VITE_API_URL=https://your-domain.com/api +VITE_WS_URL=wss://your-domain.com/ws + +# === Features === +AUTO_TRADE_ENABLED=true +DAY_TRADE_ENABLED=true +US_TRADING_ENABLED=false +KIS_WS_ENABLED=true +TOKEN_SAVE_MODE=true +LOG_LEVEL=INFO +DEBUG=false + +# === Telegram === +TELEGRAM_BOT_TOKEN= +TELEGRAM_CHAT_ID= +ENVEOF + echo " >> ${DEPLOY_DIR}/.env 생성 완료 — 반드시 비밀번호를 변경하세요!" +else + echo " >> .env 이미 존재 — 스킵" +fi + +# 5. Synology 방화벽 포트 안내 +echo "[5/6] 필요한 포트 (Synology 제어판 → 방화벽):" +echo " - 3001: Gitea (Git 웹 UI)" +echo " - 8080: Jenkins (CI/CD 웹 UI)" +echo " - 3000: Frontend (Nginx → 역방향 프록시 권장)" +echo " - 8000: Backend API (내부 전용)" +echo " - 8001: MCP Server (내부 전용)" + +# 6. 완료 +echo "[6/6] 설정 완료!" +echo "" +echo "다음 단계:" +echo " 1. ${DEPLOY_DIR}/.env 비밀번호 변경" +echo " 2. Gitea 접속: http://nas-ip:3001 → 저장소 생성" +echo " 3. Jenkins 접속: http://nas-ip:8080 → 파이프라인 설정" +echo " 4. Gitea에 push → Jenkins 자동 빌드/배포" +echo "" +echo "첫 수동 배포:" +echo " cd ${DEPLOY_DIR}" +echo " docker compose --env-file .env up -d" diff --git a/docker-compose.yml b/docker-compose.yml index 320091b..60b2b39 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,12 +20,16 @@ services: interval: 10s timeout: 5s retries: 5 + deploy: + resources: + limits: + memory: 1G # Redis Cache redis: image: redis:7-alpine container_name: autostock-redis - command: redis-server --requirepass ${REDIS_PASSWORD:-autostock_redis_change_me} + command: redis-server --requirepass ${REDIS_PASSWORD:-autostock_redis_change_me} --maxmemory 200mb --maxmemory-policy allkeys-lru environment: TZ: Asia/Seoul ports: @@ -39,6 +43,10 @@ services: interval: 10s timeout: 3s retries: 5 + deploy: + resources: + limits: + memory: 256M # RabbitMQ Message Queue rabbitmq: @@ -61,6 +69,10 @@ services: interval: 30s timeout: 10s retries: 5 + deploy: + resources: + limits: + memory: 512M # MCP Server mcp-server: @@ -70,16 +82,20 @@ services: container_name: autostock-mcp-server environment: - TZ=Asia/Seoul - - AI_PROVIDER=${AI_PROVIDER:-gemini} + - AI_PROVIDER=${AI_PROVIDER:-claude} - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY} - GEMINI_API_KEY=${GEMINI_API_KEY} - GEMINI_MODEL=${GEMINI_MODEL:-gemini-2.0-flash} + - CLAUDE_MODEL=${CLAUDE_MODEL:-claude-sonnet-4-6-20250514} + - CLAUDE_HAIKU_MODEL=${CLAUDE_HAIKU_MODEL:-claude-haiku-4-5-20251001} + - HYBRID_AI_ENABLED=${HYBRID_AI_ENABLED:-true} - MCP_API_KEY=${MCP_API_KEY} - MCP_SERVER_HOST=${MCP_SERVER_HOST} - MCP_SERVER_PORT=${MCP_SERVER_PORT} - DATABASE_URL=${DATABASE_URL} - REDIS_URL=${REDIS_URL} - LOG_LEVEL=${LOG_LEVEL} + - TOKEN_SAVE_MODE=${TOKEN_SAVE_MODE:-true} ports: - "127.0.0.1:8001:8001" networks: @@ -90,6 +106,10 @@ services: redis: condition: service_healthy restart: unless-stopped + deploy: + resources: + limits: + memory: 256M # Backend API Server backend: @@ -113,7 +133,8 @@ services: - KIS_REAL_URL=${KIS_REAL_URL:-https://openapi.koreainvestment.com:9443} - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY} - GEMINI_API_KEY=${GEMINI_API_KEY} - - AI_PROVIDER=${AI_PROVIDER:-gemini} + - AI_PROVIDER=${AI_PROVIDER:-claude} + - HYBRID_AI_ENABLED=${HYBRID_AI_ENABLED:-true} - API_KEY=${API_KEY:-} - MCP_API_KEY=${MCP_API_KEY:-} - SECRET_KEY=${SECRET_KEY} @@ -122,12 +143,16 @@ services: - MCP_SERVER_URL=http://mcp-server:8001 - NEWS_API_KEY=${NEWS_API_KEY:-} - LOG_LEVEL=${LOG_LEVEL} + - TOKEN_SAVE_MODE=${TOKEN_SAVE_MODE:-true} + - AUTO_TRADE_ENABLED=${AUTO_TRADE_ENABLED:-true} + - DAY_TRADE_ENABLED=${DAY_TRADE_ENABLED:-true} - US_TRADING_ENABLED=${US_TRADING_ENABLED:-false} + - KIS_WS_ENABLED=${KIS_WS_ENABLED:-true} + - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN:-} + - TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID:-} ports: - "127.0.0.1:8000:8000" volumes: - - ./backend:/app - - /app/__pycache__ - ./logs:/logs networks: - autostock-network @@ -141,6 +166,10 @@ services: mcp-server: condition: service_started restart: unless-stopped + deploy: + resources: + limits: + memory: 768M # PostgreSQL 자동 백업 (매일 02:00 KST, 7일치 보관) postgres-backup: @@ -184,6 +213,10 @@ services: depends_on: - backend restart: unless-stopped + deploy: + resources: + limits: + memory: 64M volumes: postgres_data: diff --git a/mcp-server/claude_client.py b/mcp-server/claude_client.py index 2fd42f7..92db42f 100644 --- a/mcp-server/claude_client.py +++ b/mcp-server/claude_client.py @@ -44,11 +44,12 @@ MDD -20%: 전 포지션 50% 강제청산 (레버리지 우선). class ClaudeClient: - """Claude AI 클라이언트""" + """Claude AI 클라이언트 (하이브리드 Haiku/Sonnet 지원)""" def __init__(self): self.client = AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY) self.model = settings.CLAUDE_MODEL + self.haiku_model = settings.CLAUDE_HAIKU_MODEL async def analyze( self, @@ -56,20 +57,25 @@ class ClaudeClient: user_message: str, tools: List[Dict[str, Any]] = None, max_tokens: int = None, + model_override: str = None, ) -> Dict[str, Any]: """Claude를 통한 분석 실행. 시스템 프롬프트는 두 블록으로 구성: 1) _CACHED_CONTEXT — cache_control 적용, 반복 호출 시 캐시에서 재사용. 2) 메서드별 role 지시문 — 캐싱 없이 매번 전송. + + Args: + model_override: 특정 모델 강제 지정 (하이브리드 모드에서 Haiku/Sonnet 분기용) """ try: + use_model = model_override or self.model messages = [ {"role": "user", "content": user_message} ] params = { - "model": self.model, + "model": use_model, "max_tokens": max_tokens or settings.MAX_TOKENS, "temperature": settings.TEMPERATURE, "system": [ @@ -100,6 +106,7 @@ class ClaudeClient: result = { "content": [], "stop_reason": response.stop_reason, + "model_used": use_model, "usage": { "input_tokens": usage.input_tokens, "output_tokens": usage.output_tokens, @@ -107,6 +114,10 @@ class ClaudeClient: "cache_write_tokens": cache_write, }, } + logger.debug( + f"[Claude] model={use_model} in={usage.input_tokens} out={usage.output_tokens} " + f"cache_hit={cache_read}" + ) for block in response.content: if block.type == "text": @@ -224,14 +235,19 @@ class ClaudeClient: portfolio: Dict[str, Any], trade_plan: Dict[str, Any] ) -> Dict[str, Any]: - """리스크 평가 (토큰 최소화 — max_tokens=RISK_MAX_TOKENS)""" + """리스크 평가 (토큰 최소화 — max_tokens=RISK_MAX_TOKENS). + + 하이브리드 모드: Haiku 사용 (단순 승인/거부 판단) + """ system_prompt = """리스크 평가. JSON만 출력. 필수: risk_level(LOW/MEDIUM/HIGH), approval(APPROVED/REJECTED), reason(한줄).""" slim_portfolio = {k: portfolio[k] for k in ("strategy", "position_count", "mdd") if k in portfolio} user_message = f"포트:{_dumps(slim_portfolio)} 계획:{_dumps(trade_plan)}" - response = await self.analyze(system_prompt, user_message, max_tokens=settings.RISK_MAX_TOKENS) + # 하이브리드: 단순 판단은 Haiku로 비용 절감 + _model = self.haiku_model if settings.HYBRID_AI_ENABLED else None + response = await self.analyze(system_prompt, user_message, max_tokens=settings.RISK_MAX_TOKENS, model_override=_model) return self._parse_json_response( response, {"error": "Failed to assess risk"} @@ -264,18 +280,126 @@ class ClaudeClient: trade_info: Dict[str, Any], market_data: Dict[str, Any] ) -> Dict[str, Any]: - """매매일지 자동 작성""" + """매매일지 자동 작성. + + 하이브리드 모드: Haiku 사용 (템플릿 기반 정리 작업) + """ system_prompt = """매매일지 작성 전문가. JSON으로 응답. 필수 필드: title, content(간략 Markdown), mood(평정심/흥분/공포/아쉬움/만족), tags(배열), score(0-100). 매매근거, 잘한점/아쉬운점, 개선사항 포함.""" user_message = f"거래:{_dumps(trade_info)}\n시장:{_dumps(market_data)}" - - response = await self.analyze(system_prompt, user_message) + + _model = self.haiku_model if settings.HYBRID_AI_ENABLED else None + response = await self.analyze(system_prompt, user_message, model_override=_model) return self._parse_json_response( response, {"error": "Failed to generate journal"} ) + async def analyze_batch_stocks( + self, + stocks: List[Dict[str, Any]], + ) -> Dict[str, Any]: + """장 전 배치 분석 — 10종목을 한 번에 받아 각각 매수적합/부적합 판정. + + Sonnet 사용 (기본 self.model). max_tokens=2048 (10종목 응답이므로 넉넉하게). + 입력 stocks: [{stock_code, stock_name, current_price, rsi, bb_position, + volume_ratio, score, news_titles}, ...] + """ + system_prompt = """주식 스윙 트레이딩 분석가. 종목 리스트를 분석하여 JSON 배열로 응답. +각 종목: stock_code, suitable(true/false), confidence(0-1), reason(한줄), risk_factors(배열). +JSON 외 텍스트 금지.""" + + user_message = f"종목리스트:{_dumps(stocks)}" + + response = await self.analyze(system_prompt, user_message, max_tokens=2048) + # 배열 응답 파싱: '{' 대신 '[' ~ ']' 범위도 처리 + return self._parse_batch_response(response) + + def _parse_batch_response(self, response: Dict[str, Any]) -> Dict[str, Any]: + """배치 분석 응답 파싱 — JSON 배열 또는 객체 모두 처리. + + 1) 응답 텍스트에서 JSON 배열([...]) 파싱 시도 + 2) JSON 객체({...}) 파싱 시도 (AI가 {stocks: [...]} 형태로 응답할 수 있음) + 3) 모두 실패하면 빈 결과 반환 + """ + for content in response["content"]: + if content["type"] == "text": + text = content["text"] + # 배열 형태 시도 + try: + arr_start = text.find('[') + arr_end = text.rfind(']') + 1 + if arr_start != -1 and arr_end > arr_start: + parsed = json.loads(text[arr_start:arr_end]) + if isinstance(parsed, list): + return {"stocks": parsed} + except (json.JSONDecodeError, Exception): + pass + # 객체 형태 시도 + try: + obj_start = text.find('{') + obj_end = text.rfind('}') + 1 + if obj_start != -1 and obj_end > obj_start: + return json.loads(text[obj_start:obj_end]) + except (json.JSONDecodeError, Exception): + pass + return {"stocks": [], "analysis": text} + return {"stocks": []} + + async def check_news_sentiment( + self, + stock_code: str, + stock_name: str, + news_titles: List[str], + ) -> Dict[str, Any]: + """장중 뉴스 감성 분석. + + 하이브리드 모드: Haiku 사용 (단순 감성 판단). max_tokens=256 (짧은 응답). + """ + system_prompt = """주식 뉴스 감성 분석. JSON만 출력. +필수: sentiment(POSITIVE/NEGATIVE/NEUTRAL), severity(LOW/MEDIUM/HIGH), action(HOLD/SELL/WATCH), reason(한줄). +JSON 외 텍스트 금지.""" + + user_message = f"종목:{stock_code}({stock_name})\n뉴스제목:{_dumps(news_titles)}" + + _model = self.haiku_model if settings.HYBRID_AI_ENABLED else None + response = await self.analyze(system_prompt, user_message, max_tokens=256, model_override=_model) + return self._parse_json_response( + response, + {"sentiment": "NEUTRAL", "severity": "LOW", "action": "HOLD", "reason": "분석 실패"} + ) + + async def review_daily_trades( + self, + trades_summary: Dict[str, Any], + market_summary: Dict[str, Any], + ) -> Dict[str, Any]: + """장 후 당일 매매 복기. + + Sonnet 사용 (기본 self.model). max_tokens=1024. + trades_summary: {total_trades, wins, losses, pnl, strategies_used, notable_trades} + market_summary: {kospi_change, regime, top_movers} + """ + system_prompt = """주식 매매 복기 전문가. 당일 매매를 분석하여 JSON 응답. +필수: overall_grade(A/B/C/D/F), good_points(배열), bad_points(배열), improvements(배열), strategy_drift(bool), drift_detail(있으면), tomorrow_plan(한줄). +JSON 외 텍스트 금지.""" + + user_message = f"매매요약:{_dumps(trades_summary)}\n시장요약:{_dumps(market_summary)}" + + response = await self.analyze(system_prompt, user_message, max_tokens=1024) + return self._parse_json_response( + response, + { + "overall_grade": "C", + "good_points": [], + "bad_points": [], + "improvements": [], + "strategy_drift": False, + "tomorrow_plan": "분석 실패", + } + ) + claude_client = ClaudeClient() diff --git a/mcp-server/config.py b/mcp-server/config.py index fc6faec..bcf7f37 100644 --- a/mcp-server/config.py +++ b/mcp-server/config.py @@ -42,11 +42,20 @@ class Settings: LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") # Claude Settings - CLAUDE_MODEL = os.getenv("CLAUDE_MODEL", "claude-3-5-sonnet-20241022") + CLAUDE_MODEL = os.getenv("CLAUDE_MODEL", "claude-sonnet-4-6-20250514") + CLAUDE_HAIKU_MODEL = os.getenv("CLAUDE_HAIKU_MODEL", "claude-haiku-4-5-20251001") # Gemini Settings GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.0-flash") + # Hybrid AI Mode (claude 선택 시 작업별 Haiku/Sonnet 자동 분기) + # True: RISK_ASSESSMENT/GENERATE_JOURNAL → Haiku, 나머지 → Sonnet + # False: 모든 작업에 CLAUDE_MODEL 단일 사용 + HYBRID_AI_ENABLED = os.getenv("HYBRID_AI_ENABLED", "true").lower() == "true" + + # Haiku를 사용할 분석 타입 (비용 절감 대상 — 단순 판단/템플릿 작업) + HAIKU_ANALYSIS_TYPES = {"RISK_ASSESSMENT", "GENERATE_JOURNAL", "NEWS_SENTIMENT"} + # Common AI Settings — __init__에서 AI_PROVIDER 값을 읽어 동적으로 설정한다. # [수정 2] 클래스 변수 평가 시점 문제: AI_PROVIDER가 클래스 변수로 먼저 정의되어 있어도 # 아래 두 줄은 클래스 정의 시점(import 시)에 한 번만 평가되므로, 환경변수가 변경돼도 diff --git a/mcp-server/gemini_client.py b/mcp-server/gemini_client.py index ad46506..4f37a0e 100644 --- a/mcp-server/gemini_client.py +++ b/mcp-server/gemini_client.py @@ -218,5 +218,104 @@ class GeminiClient: {"error": "Failed to generate journal"} ) + async def analyze_batch_stocks( + self, + stocks: List[Dict[str, Any]], + ) -> Dict[str, Any]: + """장 전 배치 분석 — 10종목을 한 번에 받아 각각 매수적합/부적합 판정. + + max_tokens=2048 (10종목 응답이므로 넉넉하게). + 입력 stocks: [{stock_code, stock_name, current_price, rsi, bb_position, + volume_ratio, score, news_titles}, ...] + """ + system_prompt = """주식 스윙 트레이딩 분석가. 종목 리스트를 분석하여 JSON 배열로 응답. +각 종목: stock_code, suitable(true/false), confidence(0-1), reason(한줄), risk_factors(배열). +JSON 외 텍스트 금지.""" + + user_message = f"종목리스트:{_dumps(stocks)}" + + response = await self.analyze(system_prompt, user_message, max_tokens=2048) + return self._parse_batch_response(response) + + def _parse_batch_response(self, response: Dict[str, Any]) -> Dict[str, Any]: + """배치 분석 응답 파싱 — JSON 배열 또는 객체 모두 처리. + + 1) 응답 텍스트에서 JSON 배열([...]) 파싱 시도 + 2) JSON 객체({...}) 파싱 시도 (AI가 {stocks: [...]} 형태로 응답할 수 있음) + 3) 모두 실패하면 빈 결과 반환 + """ + for content in response["content"]: + if content["type"] == "text": + text = content["text"] + # 배열 형태 시도 + try: + arr_start = text.find('[') + arr_end = text.rfind(']') + 1 + if arr_start != -1 and arr_end > arr_start: + parsed = json.loads(text[arr_start:arr_end]) + if isinstance(parsed, list): + return {"stocks": parsed} + except (json.JSONDecodeError, Exception): + pass + # 객체 형태 시도 + try: + obj_start = text.find('{') + obj_end = text.rfind('}') + 1 + if obj_start != -1 and obj_end > obj_start: + return json.loads(text[obj_start:obj_end]) + except (json.JSONDecodeError, Exception): + pass + return {"stocks": [], "analysis": text} + return {"stocks": []} + + async def check_news_sentiment( + self, + stock_code: str, + stock_name: str, + news_titles: List[str], + ) -> Dict[str, Any]: + """장중 뉴스 감성 분석. max_tokens=256 (짧은 응답).""" + system_prompt = """주식 뉴스 감성 분석. JSON만 출력. +필수: sentiment(POSITIVE/NEGATIVE/NEUTRAL), severity(LOW/MEDIUM/HIGH), action(HOLD/SELL/WATCH), reason(한줄). +JSON 외 텍스트 금지.""" + + user_message = f"종목:{stock_code}({stock_name})\n뉴스제목:{_dumps(news_titles)}" + + response = await self.analyze(system_prompt, user_message, max_tokens=256) + return self._parse_json_response( + response, + {"sentiment": "NEUTRAL", "severity": "LOW", "action": "HOLD", "reason": "분석 실패"} + ) + + async def review_daily_trades( + self, + trades_summary: Dict[str, Any], + market_summary: Dict[str, Any], + ) -> Dict[str, Any]: + """장 후 당일 매매 복기. + + max_tokens=1024. + trades_summary: {total_trades, wins, losses, pnl, strategies_used, notable_trades} + market_summary: {kospi_change, regime, top_movers} + """ + system_prompt = """주식 매매 복기 전문가. 당일 매매를 분석하여 JSON 응답. +필수: overall_grade(A/B/C/D/F), good_points(배열), bad_points(배열), improvements(배열), strategy_drift(bool), drift_detail(있으면), tomorrow_plan(한줄). +JSON 외 텍스트 금지.""" + + user_message = f"매매요약:{_dumps(trades_summary)}\n시장요약:{_dumps(market_summary)}" + + response = await self.analyze(system_prompt, user_message, max_tokens=1024) + return self._parse_json_response( + response, + { + "overall_grade": "C", + "good_points": [], + "bad_points": [], + "improvements": [], + "strategy_drift": False, + "tomorrow_plan": "분석 실패", + } + ) + gemini_client = GeminiClient() diff --git a/mcp-server/main.py b/mcp-server/main.py index b1bf35e..ae90739 100644 --- a/mcp-server/main.py +++ b/mcp-server/main.py @@ -118,6 +118,9 @@ async def health_check(): return { "status": "healthy", "ai_provider": settings.AI_PROVIDER, + "hybrid_ai_enabled": settings.HYBRID_AI_ENABLED if settings.AI_PROVIDER == "claude" else False, + "claude_model": settings.CLAUDE_MODEL, + "claude_haiku_model": settings.CLAUDE_HAIKU_MODEL, "claude_configured": bool(settings.ANTHROPIC_API_KEY), "gemini_configured": bool(settings.GEMINI_API_KEY), } @@ -252,6 +255,39 @@ async def analyze(request: AnalysisRequest): result = await ai_client.generate_trade_journal(trade_info, market_data) return {"analysis_type": analysis_type, "result": result} + elif analysis_type == "BATCH_STOCK_ANALYSIS": + stocks = data.get("stocks") + if not stocks or not isinstance(stocks, list): + raise HTTPException( + status_code=400, + detail="stocks (list) is required" + ) + result = await ai_client.analyze_batch_stocks(stocks) + return {"analysis_type": analysis_type, "result": result} + + elif analysis_type == "NEWS_SENTIMENT": + stock_code = data.get("stock_code") + stock_name = data.get("stock_name") + news_titles = data.get("news_titles") + if not stock_code or not stock_name or not news_titles: + raise HTTPException( + status_code=400, + detail="stock_code, stock_name, and news_titles are required" + ) + result = await ai_client.check_news_sentiment(stock_code, stock_name, news_titles) + return {"analysis_type": analysis_type, "result": result} + + elif analysis_type == "DAILY_REVIEW": + trades_summary = data.get("trades_summary") + market_summary = data.get("market_summary") + if not trades_summary or not market_summary: + raise HTTPException( + status_code=400, + detail="trades_summary and market_summary are required" + ) + result = await ai_client.review_daily_trades(trades_summary, market_summary) + return {"analysis_type": analysis_type, "result": result} + else: raise HTTPException( status_code=400,