"""스윙 트레이딩 진입 서비스 trade_executor.py의 execute_auto_trade_background() 로직을 기반으로 strategy_type='swing' 진입만 처리하는 독립 서비스. 진입 기준 (CLAUDE.md 섹션 6.2): - BULL: RSI(14) < 46, 가격이 BB 하단 부근, 거래량 > 20MA - BEAR: RSI(14) < 35, 가격 < BB 하단, 거래량 > 1.5x20MA, 반전 캔들 - SIDEWAYS: 진입 없음 (CLAUDE.md 6.1) - 포지션 사이징: 종목당 총자본 MAX_POSITION_SIZE(기본 4%, 절대 상한), 최대 MAX_HOLDINGS_COUNT 종목(기본 20) — 값은 config.py/§0 우선순위를 따른다 듀얼 전략 (CLAUDE.md 섹션 8.4): - 모멘텀 돌파: 20일 신고가 + 거래량 2x + RSI 50~70 (BULL만) - 눌림목 매수: MA20>MA60 + MA20 근처 조정 + RSI 반등 (BULL만) 진입 전 체크: 1. 장 오픈 여부 (정규장 09:00~15:30만 허용) 2. 스윙 포지션 한도 (MAX_SWING_POSITIONS) 3. MDD 신규 진입 가능 여부 4. 적응형 전략 Crisis 모드 차단 5. 플레이북 이벤트 차단 (FOMC, 옵션만기 등) 6. 시장 국면 SIDEWAYS 차단 """ import asyncio import logging import math from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, desc from datetime import date from app.core.config import settings 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 from app.services.swing.position_manager import count_swing_positions from app.services.market_playbook import is_entry_allowed_by_playbook, get_position_multiplier_from_playbook from app.core.log_throttle import warn_throttled # noqa: E402 logger = logging.getLogger(__name__) # 스윙 동시 보유 최대 종목 수 (CLAUDE.md 5.1: MAX_HOLDINGS_COUNT = 20) MAX_SWING_POSITIONS = 20 # RSI 반등 방향 확인용 과거 구간 (kosdaq_scanner.py RSI_DIRECTION_LOOKBACK과 동일) RSI_DIRECTION_LOOKBACK = 3 async def execute_swing_entries(recommendations_response: dict, db: AsyncSession) -> list[dict]: """배치 분석 추천 결과를 기반으로 스윙 매수 실행. 기존 execute_auto_trade_background()를 재활용하되 strategy_type='swing' 레이블을 주입하여 데이트레이딩과 분리. 처리 흐름: 1. 장 오픈 여부 체크 (정규장만 허용) 2. 포지션 한도 체크 (MAX_SWING_POSITIONS) 3. MDD 체크 (can_open_new_position) 4. 적응형 전략 Crisis 모드 차단 5. 플레이북 이벤트 차단 6. SIDEWAYS 국면 차단 7. 추천 목록에 strategy_type='swing' 주입 8. execute_auto_trade_background() 호출 Args: recommendations_response: Redis 또는 DB에서 가져온 추천 딕셔너리 - recommendations_by_rule: 규칙별 추천 종목 리스트 db: AsyncSession 인스턴스 Returns: list[dict]: 실행된 매수 결과 목록 """ # 장 마감 후 매수 차단 (15:30 이후, KIS 모의투자 24시간 허용 방지) from app.core.market_schedule import is_market_open if not is_market_open(): logger.info("[SwingEntry] 장 마감 — 스윙 매수 차단 (정규장만 매매)") return [] # 포지션 한도 체크 current_count = await count_swing_positions(db) if current_count >= MAX_SWING_POSITIONS: logger.info( f"[SwingEntry] 최대 포지션 수 도달 ({current_count}/{MAX_SWING_POSITIONS}) — 신규 진입 스킵" ) return [] # MDD 체크 can_open = await mdd_manager.can_open_new_position() if not can_open: logger.warning("[SwingEntry] MDD 초과 — 신규 진입 차단") return [] # 적응형 전략 체크 — Crisis 모드에서 스윙 진입 전면 차단 # [§4 2026-10-02] 판정은 try 밖. 조회 실패 = Crisis 여부 불명 → 진입 차단(fail-closed). _crisis_block = _adaptive_entry_block_reason_v2() if _crisis_block: logger.warning(f"[SwingEntry] {_crisis_block} — 스윙 진입 전면 차단") return [] # 플레이북 체크 — FOMC/옵션만기/서킷브레이커 등 이벤트 당일 차단 if not is_entry_allowed_by_playbook(): logger.info("[SwingEntry] 플레이북 차단 — 신규 진입 비활성화 (이벤트 대응)") return [] # SIDEWAYS 국면 체크 + [안전형 개선②] 지수 서킷브레이커 # [H3 2026-10-02] 국면 조회 실패·국면 키 누락은 신규 진입 차단(fail-closed). # 기존엔 조회 예외를 warning 후 통과시키고 키 누락을 "bull" 로 간주했다. try: regime = await check_market_regime() except Exception as e: logger.error(f"[SwingEntry] 시장 국면 조회 실패 — 신규 진입 차단(fail-closed): {e}") return [] regime_str = _regime_200d_or_none_v2(regime) if regime_str is None: logger.error( "[SwingEntry] 시장 국면 판정값(regime_200d) 없음 — 신규 진입 차단(fail-closed)" ) return [] if regime_str == "sideways": logger.info("[SwingEntry] SIDEWAYS 국면 — 진입 신호 무시 (CLAUDE.md 6.1)") return [] # [안전형 개선②] 지수 서킷브레이커 (검증 완료: 2024H2 CAGR -9.9%→+1.0%, MDD -7→-1.9) # KODEX200(069500) 일봉 종가 < MA20 → 신규 스윙 진입 전면 중단 (기존 보유 청산/관리는 정상 동작). # check_market_regime()이 이미 계산한 drop_from_ma20(=(현재가-MA20)/MA20*100)을 재사용하여 # 지수 데이터 중복 조회를 피한다. # 2026-09-17: 사용자 결정으로 기본 OFF (SWING_INDEX_BREAKER_ENABLED=true 로 재활성화 가능). # [§4 2026-10-02] 값 파싱만 try 안, 판정은 밖. 켜져 있는데 값을 해석할 수 없으면 차단. _breaker_block = _index_breaker_block_reason_v2(regime) if _breaker_block: logger.info(f"[SwingEntry] {_breaker_block} — 신규 진입 전면 중단") return [] # strategy_type='swing' 레이블 주입 rec = dict(recommendations_response) _inject_strategy_type(rec, "swing") # 기존 execute_auto_trade_background 재활용 (시그니처 변경 없음) # [2026-09-27] execute_auto_trade_background() 는 항상 list 를 반환한다(과거엔 # 암묵적 None). 이 list 에는 성공한 BUY 뿐 아니라 주문 실패(success=False)한 # BUY/SELL, 청산 우선 로직으로 섞인 SELL 도 함께 담기므로 len(result) 를 그대로 # "진입 n건"으로 로깅하지 않고 action/success 기준으로 분리 집계한다. try: result = await execute_auto_trade_background(rec, db) if isinstance(result, list): buy_success = sum( 1 for r in result if isinstance(r, dict) and r.get("action") == "BUY" and r.get("success") ) sell_count = sum( 1 for r in result if isinstance(r, dict) and r.get("action") == "SELL" ) fail_count = sum( 1 for r in result if isinstance(r, dict) and not r.get("success") ) if buy_success or sell_count or fail_count: logger.info( "[SwingEntry] 진입 %d건 / 청산 %d건 / 주문실패 %d건", buy_success, sell_count, fail_count, ) return result except Exception as e: logger.error(f"[SwingEntry] 진입 실행 오류: {e}", exc_info=True) return [] def _adaptive_entry_block_reason_v2() -> str: """적응형 Crisis 진입 차단 사유. 빈 문자열이면 진입 허용. 조회만 try 안에 두고, 판정은 밖에서 한다(CLAUDE.md §4: 가드를 try 안에 두지 말 것). 조회가 실패하면 Crisis 여부를 알 수 없으므로 신규 진입을 차단한다(fail-closed). fallback 파라미터(적응형 미산출)는 정상 상태로 간주한다 — 기존 동작 유지. """ _adp = None try: from app.services.adaptive_strategy import get_current_adaptive_params _adp = get_current_adaptive_params() except Exception as e: logger.warning(f"[SwingEntry] 적응형 파라미터 조회 실패 — Crisis 여부 불명: {e}") return "적응형 파라미터 조회 실패(fail-closed)" if _adp is None: logger.warning("[SwingEntry] 적응형 파라미터 None — Crisis 여부 불명") return "적응형 파라미터 없음(fail-closed)" if not getattr(_adp, "fallback_used", False) and getattr(_adp, "trading_mode", None) == "crisis": return "적응형 전략 CRISIS 모드" return "" def _index_breaker_block_reason_v2(regime) -> str: """지수 서킷브레이커 차단 사유. 빈 문자열이면 진입 허용. KODEX200 종가 < MA20 (check_market_regime() 의 drop_from_ma20 < 0) 이면 차단. settings.SWING_INDEX_BREAKER_ENABLED=False 면 항상 허용(2026-09-17 사용자 결정 기본 OFF). 값 파싱만 try 안 — 켜져 있는데 값을 해석할 수 없으면 차단(fail-closed, §4). 값이 아예 없으면(None) 기존 동작대로 통과한다. 평균회귀(execute_swing_entries)·듀얼(scan_dual_strategy_signals) 공용 — [H-1 2026-10-02 3회차] 듀얼에는 이 판정이 없었다. """ if not settings.SWING_INDEX_BREAKER_ENABLED: return "" _raw_drop = regime.get("drop_from_ma20") if isinstance(regime, dict) else None if _raw_drop is None: return "" try: _drop_from_ma20 = float(_raw_drop) except (TypeError, ValueError) as e: logger.warning(f"[SwingEntry] 지수 서킷브레이커 값 해석 실패({_raw_drop!r}) — 신규 진입 차단: {e}") return f"지수 서킷브레이커 값 해석 실패({_raw_drop!r}, fail-closed)" if _drop_from_ma20 < 0: return f"지수 서킷브레이커 — KODEX200 종가 "str | None": """check_market_regime() 결과에서 regime_200d 를 꺼낸다. 없거나 형식이 틀리면 None. 기존 `.get("regime_200d", "bull")` 은 판정 실패(키 누락)를 BULL 로 해석해 진입을 계속시켰다. 호출부는 None 을 진입 차단으로 처리한다. """ if not isinstance(regime, dict): return None val = regime.get("regime_200d") if not isinstance(val, str) or not val: return None return val def _inject_strategy_type(rec: dict, strategy_type: str) -> None: """추천 목록의 각 종목에 strategy_type 필드를 주입. recommendations_by_rule 구조가 리스트 또는 딕셔너리일 수 있어 두 가지 형태를 모두 지원. Args: rec: 추천 딕셔너리 (원본을 직접 수정) strategy_type: 주입할 전략 타입 (예: "swing") """ groups = rec.get("recommendations_by_rule", []) if isinstance(groups, list): for group in groups: if isinstance(group, dict): for stock in group.get("stocks", []): if isinstance(stock, dict): stock["strategy_type"] = strategy_type elif isinstance(groups, dict): for group in groups.values(): if isinstance(group, list): for stock in group: if isinstance(stock, dict): stock["strategy_type"] = strategy_type def _below_min_price_krw_v2(price) -> bool: """국내 가격 필터 — True 면 진입 거부 (CLAUDE.md §5.1 '국내 >5,000원' = 초과만 허용). [하드닝 R2 2026-10-09] 듀얼 스캔은 `<`(정확히 MIN_PRICE_KRW 허용), 박스권은 `<=`(거부)로 경로마다 경계가 달랐다. 진입 단계(trade_executor·manual_buy·portfolio_v2)는 `<=` 이므로 후보 단계에서 통과한 5,000원 종목이 뒤에서 조용히 탈락해 평가 슬롯을 소모했다. 해석 불가(None·문자열·NaN·inf)도 거부한다(fail-closed). """ try: p = float(price if price is not None else 0) except (TypeError, ValueError): return True if not math.isfinite(p): return True return p <= float(settings.MIN_PRICE_KRW) # ─── 듀얼·박스권 스캔용 일봉 (하드닝 R2 2026-10-09) ───────────────────────── # KIS get_daily_chart(FHKST01010400 inquire-daily-price)는 limit 과 무관하게 30행만 돌려준다 # (로컬 로그 'Fetched N chart data points' 전부 N=30). 눌림목 v2 는 60봉(MA60), 박스권은 40봉을 # 요구하므로 두 스캔은 구조적으로 한 번도 발화하지 못했다. 네이버 일봉(market_analyzer· # adaptive 와 같은 1차 소스) → KIS 기간 조회(FHKST03010100, 날짜 소급 페이지네이션) → # 기존 get_daily_chart 순으로 확보한다. DUAL_SCAN_MIN_BARS = 60 # check_pullback_buy_signal_v2: len(closes) < 60 이면 None RANGE_SCAN_MIN_BARS = 40 # scan_range_trading_signals 기존 하한 # get_naver_daily_chart(days=N) 은 캘린더 (N+30)일 창 — 120 → 150일 ≈ 100 거래일 # (market_analyzer 실측: 80→75~77봉, 200→약 155봉). _SCAN_NAVER_CHART_DAYS = 120 _SCAN_KIS_RANGE_CALENDAR_DAYS = 150 def _bars_end_today_v2(bars_oldest_first: list) -> bool: """마지막 봉이 오늘(KST) 봉인가. 스캔 로직은 `closes[-1]`/`volumes[-1]` 을 '오늘(장중)' 값으로 쓴다(모멘텀 v2·눌림목 v2 의 오늘 봉 제외 계산 포함). 오늘 봉이 빠진 소스를 그대로 쓰면 어제 종가로 신호·주문가가 정해지므로, 그런 소스는 채택하지 않고 다음 소스로 넘어간다. """ if not bars_oldest_first: return False from app.core.timeutil import today_kst last = str(bars_oldest_first[-1].get("stck_bsop_date") or "") return last == today_kst().strftime("%Y%m%d") async def _fetch_scan_daily_bars_v2(api, stock_code: str, min_bars: int) -> tuple[list, str]: """스캔용 일봉(오래된 → 최신, KIS 필드명). 반환: (bars, source). 1) 네이버 일봉 — 오늘 봉으로 끝나고 min_bars 이상이면 채택 2) KIS 기간 조회(get_daily_ohlcv_range) — 같은 조건 3) 둘 다 실패(빈 결과·오늘 봉 없음·예외)면 기존 get_daily_chart(limit=60) (≈30봉, 기존 동작) 1·2 가 오늘 봉으로 끝나지만 min_bars 미달이면(신규 상장 등) 그중 긴 쪽을 쓴다. 봉수 부족 경고는 호출부가 스캔 단위로 집계해 남긴다. """ from datetime import timedelta from app.core.timeutil import today_kst best: list = [] best_src = "" # 1) 네이버 try: from app.services.naver_finance import get_naver_daily_chart nv = await get_naver_daily_chart(stock_code, days=_SCAN_NAVER_CHART_DAYS) nv = [b for b in (nv or []) if isinstance(b, dict) and b.get("stck_bsop_date")] if _bars_end_today_v2(nv): if len(nv) >= min_bars: return nv, "naver" best, best_src = nv, "naver" except Exception as e: warn_throttled( logger, "entry_service:scan_bars_naver", f"[ScanBars] {stock_code} 네이버 일봉 조회 실패(KIS 폴백): {type(e).__name__}: {e}", ) # 2) KIS 기간 조회 (FHKST03010100) — 시세 실서버 위임 설정을 get_daily_chart 와 동일하게 따른다 try: src_api = api _delegate = getattr(api, "_quote_delegate", None) if callable(_delegate): _d = _delegate() if _d is not None: src_api = _d end = today_kst() start = end - timedelta(days=_SCAN_KIS_RANGE_CALENDAR_DAYS) rows = await src_api.get_daily_ohlcv_range( stock_code, start.strftime("%Y%m%d"), end.strftime("%Y%m%d"), ) rows = [r for r in (rows or []) if isinstance(r, dict) and r.get("stck_bsop_date")] # 최신 → 오래된 순으로 오므로 날짜 기준 정렬 + 중복 제거 kis = sorted({r["stck_bsop_date"]: r for r in rows}.values(), key=lambda r: r["stck_bsop_date"]) if _bars_end_today_v2(kis): if len(kis) >= min_bars: return kis, "kis_range" if len(kis) > len(best): best, best_src = kis, "kis_range" except Exception as e: warn_throttled( logger, "entry_service:scan_bars_kis_range", f"[ScanBars] {stock_code} KIS 기간 일봉 조회 실패: {type(e).__name__}: {e}", ) if best: return best, best_src # 3) 기존 경로 (≈30봉). 예외는 호출부의 종목 단위 except 로 전파 — 기존 동작 유지. chart = await api.get_daily_chart(stock_code, limit=60) output2 = (chart or {}).get("output2") or [] return list(reversed(output2)), "kis_daily_price" def _warn_short_scan_bars_v2(tag: str, short: list, min_bars: int) -> None: """스캔 1회분의 봉수 부족 종목을 경고 1줄로 가시화한다(조용한 미발화 방지).""" if not short: return sample = ", ".join(f"{c}:{n}봉({s})" for c, n, s in short[:5]) warn_throttled( logger, f"entry_service:short_bars:{tag}", f"[{tag}] 일봉 {min_bars}봉 미달 {len(short)}종목 — 해당 전략 판정 불가: {sample}", ) 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_momentum_breakout_signal_v2, check_pullback_buy_signal_v2, calculate_rsi, ) from app.services.signal_types import BuySignal if not is_market_open(): return [] # BULL 시장에서만 듀얼 전략 실행 try: regime = await check_market_regime() except Exception as e: logger.warning(f"[DualStrategy] 시장 국면 조회 실패 — 스킵(fail-closed): {e}") return [] # [H3 2026-10-02] 국면 키 누락을 "bull" 로 간주하지 않는다(fail-closed). regime_str = _regime_200d_or_none_v2(regime) if regime_str != "bull": if regime_str is None: logger.warning("[DualStrategy] 국면 판정값(regime_200d) 없음 — 스킵(fail-closed)") else: logger.debug("[DualStrategy] BULL이 아님 (%s) — 듀얼 전략 스킵", regime_str) 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 모드 체크 — [§4 2026-10-02] 판정 try 밖, 조회 실패 시 차단(fail-closed) _crisis_block_dual = _adaptive_entry_block_reason_v2() if _crisis_block_dual: logger.info(f"[DualStrategy] {_crisis_block_dual} — 듀얼 전략 스킵") return [] # [H-1 2026-10-02 3회차] 플레이북(FOMC·옵션만기 등 이벤트 당일)·지수 서킷브레이커 — # execute_swing_entries 와 같은 의미로 적용한다(이전엔 듀얼에 없어 이벤트 당일에도 진입). # 판정은 try 밖. 플레이북 position_multiplier 는 평균회귀 경로에서도 실제 사이징에 # 적용되는 곳이 없어(import 만 존재) 듀얼에도 적용하지 않는다 — 동일 의미 유지. if not is_entry_allowed_by_playbook(): logger.info("[DualStrategy] 플레이북 차단 — 신규 진입 비활성화 (이벤트 대응)") return [] _breaker_block_dual = _index_breaker_block_reason_v2(regime) if _breaker_block_dual: logger.info(f"[DualStrategy] {_breaker_block_dual} — 듀얼 전략 스킵") return [] # [H-2 2026-10-02 3회차] 듀얼은 MCP 리스크 평가(RISK_ASSESSMENT) 대상이 아니다 # (2026-10-02 사용자 결정 — 거래 확대, AI 필터 미적용). 하드 리스크 한도(4%·20종목· # MDD·밴·현금 95%)는 execute_auto_trade_background → _execute_buy 경로에서 그대로 적용된다. # [M-1 2026-10-02 3회차] 일일 매수 한도 선확인 — 이미 소진됐으면 종목별 일봉 조회 # (최대 10회 KIS 호출)를 하지 않는다. 한도 판정의 정본은 execute_auto_trade_background # 이고 여기는 같은 기준(스윙 소유 BUY distinct 종목 수, 국면 max_trades 반영)의 조기 생략일 뿐. # 조회 실패(None)는 그쪽과 같이 fail-closed. from app.services.trade_executor import _count_swing_buys_today_v2, _regime_limits_v2 _limit_regime = regime if settings.BEAR_MARKET_FILTER_ENABLED else None _regime_max, _ = _regime_limits_v2(_limit_regime) _daily_limit = min(int(settings.MAX_DAILY_TRADES), _regime_max) _bought_today = await _count_swing_buys_today_v2(db) if _bought_today is None: logger.warning("[DualStrategy] 당일 스윙 매수 건수 조회 실패 — 스킵(fail-closed)") return [] if _bought_today >= _daily_limit: logger.info( f"[DualStrategy] 일일 매수 한도 도달 (오늘 스윙 {_bought_today}종목 / 한도 " f"{_daily_limit}) — 듀얼 스캔 생략" ) return [] # 현재 보유 종목 (중복 진입 방지) result = await db.execute(select(Portfolio.stock_code).where(Portfolio.quantity > 0)) held_codes = {row[0] for row in result.fetchall()} # 최근 분석된 종목 중 상위 30개 조회 (score 기준) # [H5 주석 2026-10-02, 동작 무변경] 듀얼 전략은 독립 유니버스가 없다. 후보는 일일 배치가 # 저장한 StockAnalysis(당일분) 상위 30개(룰 점수순, 0~15 스케일)이고 그중 미보유· # 가격필터 통과 10개만 차트를 본다. 신고가 돌파·눌림목 종목이라도 배치 룰 점수 상위 # 30 에 들지 못하면 스캔되지 않는다. 배치 실패 시 0건(§6.4). 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 short_bars: list = [] for stock in candidates: if scan_count >= 10: break if stock.stock_code in held_codes: continue if _below_min_price_krw_v2(stock.current_price): continue try: # [하드닝 R2 2026-10-09] get_daily_chart 는 30봉 상한이라 눌림목(60봉)이 구조적으로 # 미발화였다 → 네이버/KIS 기간 조회로 60봉 이상 확보 (_fetch_scan_daily_bars_v2). bars, bars_src = await _fetch_scan_daily_bars_v2( korea_investment_api, stock.stock_code, DUAL_SCAN_MIN_BARS, ) if len(bars) < DUAL_SCAN_MIN_BARS: short_bars.append((stock.stock_code, len(bars), bars_src)) if not bars or len(bars) < 20: continue closes = [float(b.get("stck_clpr", 0) or 0) for b in bars] highs = [float(b.get("stck_hgpr", 0) or 0) for b in bars] volumes = [int(float(b.get("acml_vol", 0) or 0)) for b in bars] if not closes or closes[-1] <= 0: continue if _below_min_price_krw_v2(closes[-1]): continue current_close = closes[-1] current_volume = volumes[-1] if volumes else 0 rsi = calculate_rsi(closes, period=14) if rsi is None: continue rsi_prev = calculate_rsi(closes[:-RSI_DIRECTION_LOOKBACK], period=14) scan_count += 1 # 모멘텀 돌파 체크 # [C2 2026-10-02] v1 은 오늘 봉이 20일 고점 구간에 포함돼 구조적으로 미발동. # MOMENTUM_BREAKOUT_ENABLED(기본 True)면 오늘 봉 제외 v2, False 면 기존 v1. if settings.MOMENTUM_BREAKOUT_ENABLED: momentum_signal = check_momentum_breakout_signal_v2( closes=closes, highs=highs, volumes=volumes, current_close=current_close, current_volume=current_volume, rsi=rsi, ) else: 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, # [2026-09-27 FIX] signals(복수) 누락으로 trade_executor 의 # should_buy 판정(:1906)이 항상 False 였다 → ML 앙상블이 독립적으로 # BUY 를 낼 때만 진입되던 버그. 진입 근거와 **의미가 일치하는** # 키워드만 채운다(거짓 신호는 저널·ML 학습 데이터를 오염시킴). # · 신고가돌파 ← current_close > 20일 고점(v2: 직전 20봉) (StrongBuySignal) # · 거래량증가상승 ← vol_ratio >= 2.0 & 상승 돌파 "signals": [ BuySignal.신고가돌파.value, BuySignal.거래량증가상승.value, ], "signal": momentum_signal["description"], "strategy_type": "swing", # ActiveTrade.strategy_name 으로 저장돼 swing/exit_service 의 모멘텀 전용 # 청산(+10% 30% → +20% 30% → 트레일링, CLAUDE.md §8.4)을 타게 한다. "strategy_name": "MOMENTUM_BREAKOUT", "entry_reason": f"[MOMENTUM_BREAKOUT] {momentum_signal['description']}", }) logger.info( f"[DualStrategy] 모멘텀 돌파 시그널: {stock.stock_code} {stock.stock_name}" ) continue # 눌림목 매수 체크 # [C2 2026-10-02] 20일 평균거래량에서 오늘(장중) 봉 제외 — 모멘텀과 같은 원칙. pullback_signal = check_pullback_buy_signal_v2( closes=closes, current_close=current_close, current_volume=current_volume, rsi=rsi, rsi_prev=rsi_prev, volumes=volumes, exclude_today_volume=True, ) if pullback_signal: dual_stocks.append({ "stock_code": stock.stock_code, "stock_name": stock.stock_name, "current_price": int(current_close), "score": 0.6, # [2026-09-27 FIX] 모멘텀 돌파와 동일 사유. check_pullback_buy_signal 의 # 진입 근거와 1:1 대응하는 키워드만 채운다. # · 정배열눌림목 ← MA20>MA60(정배열) & MA20 ±3% 조정 (StrongBuySignal) # · 눌림목진입 ← RSI 40~55 반등 + 거래량 ≥1.2×20MA (StrongBuySignal) # · MA20상향 ← ma20 > ma60 ("20일선 > 60일선") "signals": [ BuySignal.정배열눌림목.value, BuySignal.눌림목진입.value, BuySignal.MA20상향.value, ], "signal": pullback_signal["description"], "strategy_type": "swing", "strategy_name": "PULLBACK_BUY", "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: # 키는 정적(메모리 증가 방지), 종목 구분은 메시지의 종목 코드로 한다. warn_throttled( logger, "entry_service:dual_scan", f"[DualStrategy] {stock.stock_code} 스캔 실패: {type(e).__name__}: {e}", ) continue _warn_short_scan_bars_v2("DualStrategy", short_bars, DUAL_SCAN_MIN_BARS) 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, }] } # [관측성] execute_auto_trade_background() 는 [2026-09-27 수정]으로 내부 # trade_results 를 항상 list 로 반환한다(과거엔 암묵적 None 을 돌려줘 아래 # isinstance(result, list) 가 항상 False 였다 — 지금은 정상 동작). # 다만 이 list 에는 성공한 BUY 뿐 아니라 **주문 실패(success=False)한 # BUY/SELL 과, 청산 우선 로직으로 섞여 들어온 SELL 도 함께 담긴다.** # 따라서 단순 len(result) 는 "진입 n건"으로 오인시킬 수 있어, 아래에서 # action/success 기준으로 분리 집계한다. 위임 직전 후보 로그는 관측성 # 확보를 위해 계속 남긴다. logger.info( "[DualStrategy] %d건 시그널 → 매매 파이프라인 위임: %s", len(dual_stocks), ", ".join( f"{s['stock_code']}({s.get('strategy_name')})" for s in dual_stocks ), ) try: result = await execute_auto_trade_background(rec, db) if isinstance(result, list): buy_success = sum( 1 for r in result if isinstance(r, dict) and r.get("action") == "BUY" and r.get("success") ) sell_count = sum( 1 for r in result if isinstance(r, dict) and r.get("action") == "SELL" ) fail_count = sum( 1 for r in result if isinstance(r, dict) and not r.get("success") ) logger.info( "[DualStrategy] 진입 %d건 / 청산 %d건 / 주문실패 %d건", buy_success, sell_count, fail_count, ) return result # [2026-09-27] execute_auto_trade_background() 는 이제 항상 list 를 반환하므로 # 이 분기는 정상 경로에서 도달하지 않는다. 다만 향후 시그니처 회귀(반환 타입 # 변경) 시 조용히 무시되지 않도록 isinstance 방어는 유지하고 경고로 남긴다. logger.warning( "[DualStrategy] 예상치 못한 반환 타입(list 아님, %r) — 방어 로직 발동", type(result), ) except Exception as e: logger.error(f"[DualStrategy] 진입 실행 오류: {e}", exc_info=True) return [] async def scan_range_trading_signals(db: AsyncSession) -> list[dict]: """박스권(SIDEWAYS) 레인지 매매 시그널 스캔 및 진입 실행. CLAUDE.md §8.5 항목 9 — SIDEWAYS 국면 담당 전략이 없어 코스피가 BULL 이 아닌 구간에서 거래가 전무한 문제를 메우기 위한 **신규 신호원**. ⚠️ **백테스트 미검증.** `settings.RANGE_TRADE_ENABLED` 가 False(기본값)이면 함수 진입 즉시 [] 를 반환한다 — 어떤 조회·판정·주문도 수행하지 않으므로 플래그가 꺼져 있는 동안 기존 동작은 완전히 불변이다. ┌─ SIDEWAYS 차단 우회에 관한 명시적 설계 결정 ────────────────────────────┐ │ 이 전략은 **SIDEWAYS 국면 전용**이므로, CLAUDE.md §6.1 의 "SIDEWAYS = │ │ 모든 진입 무시" 전면 차단을 의도적으로 우회한다. │ │ · market_analyzer.check_market_regime() 은 sideways 에서 │ │ position_multiplier=0, max_trades=0 을 반환한다. │ │ → execute_auto_trade_background() 는 max_trades==0 에서 즉시 []을 │ │ 반환하므로(2026-09-27 수정 이후에도 동일 — 반환 타입만 list로 정정 │ │ 됐을 뿐 조기 종료 자체는 그대로) 사용할 수 없다. 종목별로 │ │ execute_auto_trade() 를 직접 호출하고 │ │ _position_multiplier 를 RANGE_TRADE_POSITION_MULTIPLIER 로 주입한다. │ │ · trade_executor._execute_buy 는 _regime_str=="sideways" 를 하드 차단 │ │ 한다(방어코드). 따라서 자본 한도가 가장 보수적인 "bear" 라벨을 주입해 │ │ BEAR 전략 자본 상한(30%) 아래에서 동작시킨다. │ │ · 우회는 **RANGE_TRADE_ENABLED 뒤에서만** 일어난다. market_analyzer 와 │ │ mean_reversion_strategy 는 수정하지 않았으므로 다른 전략(평균회귀· │ │ 모멘텀·눌림목)의 SIDEWAYS 차단은 그대로 유지된다. │ └──────────────────────────────────────────────────────────────────────────┘ 리스크 게이트 (scan_dual_strategy_signals 와 동일 + 레인지 전용 추가): 1. RANGE_TRADE_ENABLED 플래그 2. 장 오픈 여부 (정규장만) 3. 시장 국면 == SIDEWAYS (조회 실패 시 진입 중단 — fail-closed) 4. 스윙 포지션 한도 (MAX_SWING_POSITIONS) 5. 레인지 전용 포지션 한도 (RANGE_TRADE_MAX_POSITIONS ∩ 자본 비율 유도 상한) 6. MDD (await mdd_manager.can_open_new_position()) 7. 적응형 Crisis 모드 (조회 실패 시 fail-closed) 8. 플레이북 이벤트 차단 9. 보유 종목 중복 진입 방지 10. 연속손실 밴 (loss_tracker.is_banned_safe) 11. 가격 필터 (MIN_PRICE_KRW 이하 거부 — _below_min_price_krw_v2) 12. 이후 execute_auto_trade() 내부에서 가격필터·MDD·밴·보유수·당일매도쿨다운, _execute_buy 내부에서 correlation_filter·레버리지ETF한도·뉴스필터· MIN_BUY_AMOUNT_KRW 가 다시 적용된다. Returns: list[dict]: 실행된 매수 결과 목록 """ # ── 게이트 1: 기능 플래그 (백테스트 미검증 — 기본 False) ────────────── if not settings.RANGE_TRADE_ENABLED: return [] from app.core.market_schedule import is_market_open from app.models.models import ActiveTrade from app.services.korea_investment_api import korea_investment_api from app.services.loss_tracker import loss_tracker from app.services.mean_reversion_strategy import ( calculate_adx, calculate_atr, calculate_bollinger, calculate_rsi, ) from app.services.range_trading_strategy import ( RANGE_TRADE_STRATEGY_NAME, calculate_range_stop_loss, check_range_buy_signal, ) from app.services.signal_types import BuySignal # ── 게이트 2: 정규장만 ──────────────────────────────────────────────── if not is_market_open(): return [] # ── 게이트 3: SIDEWAYS 국면에서만 동작 (fail-closed) ────────────────── try: regime = await check_market_regime() except Exception as e: logger.warning(f"[RangeTrade] 시장 국면 조회 실패 — 진입 중단(fail-closed): {e}") return [] regime_str = regime.get("regime_200d", "") if isinstance(regime, dict) else "" if regime_str != "sideways": logger.debug("[RangeTrade] SIDEWAYS 아님 (%s) — 레인지 전략 스킵", regime_str or "unknown") return [] # ── 게이트 4: 스윙 전체 포지션 한도 ─────────────────────────────────── current_count = await count_swing_positions(db) remaining_slots = MAX_SWING_POSITIONS - current_count if remaining_slots <= 0: logger.info("[RangeTrade] 스윙 포지션 한도 도달 (%d) — 스킵", current_count) return [] # ── 게이트 5: 레인지 전용 포지션/자본 한도 ──────────────────────────── # 자본 배분: 새 자본 몫을 만들지 않고 스윙 예산(SWING_MAX_CAPITAL_RATIO=0.70)을 # 공유한다. 레인지 몫은 스윙 몫 **내부** 비율(RANGE_TRADE_MAX_CAPITAL_RATIO). # 실효 상한 = 총자본 × 0.70 × 0.10 = 7% # 종목당 = MAX_POSITION_SIZE(4%) × RANGE_TRADE_POSITION_MULTIPLIER(0.5) = 2% # → 허용 종목 수 = floor(7% / 2%) = 3 _per_position_ratio = settings.MAX_POSITION_SIZE * settings.RANGE_TRADE_POSITION_MULTIPLIER _range_capital_ratio = settings.SWING_MAX_CAPITAL_RATIO * settings.RANGE_TRADE_MAX_CAPITAL_RATIO _ratio_cap = int(_range_capital_ratio / _per_position_ratio) if _per_position_ratio > 0 else 0 max_range_positions = max(0, min(settings.RANGE_TRADE_MAX_POSITIONS, _ratio_cap)) if max_range_positions <= 0: logger.info("[RangeTrade] 자본 배분 상한이 0 — 스킵") return [] _range_count_result = await db.execute( select(ActiveTrade.stock_code).where( ActiveTrade.status == "active", ActiveTrade.strategy_name == RANGE_TRADE_STRATEGY_NAME, ) ) _range_held = {row[0] for row in _range_count_result.fetchall()} range_slots = max_range_positions - len(_range_held) if range_slots <= 0: logger.info( "[RangeTrade] 레인지 포지션 한도 도달 (%d/%d) — 스킵", len(_range_held), max_range_positions, ) return [] remaining_slots = min(remaining_slots, range_slots) # ── 게이트 6: MDD ──────────────────────────────────────────────────── if not await mdd_manager.can_open_new_position(): logger.warning("[RangeTrade] MDD 초과 — 신규 진입 차단") return [] # ── 게이트 7: 적응형 Crisis 모드 (fail-closed) ──────────────────────── try: from app.services.adaptive_strategy import get_current_adaptive_params _adp = get_current_adaptive_params() _adp_crisis = (not _adp.fallback_used) and _adp.trading_mode == "crisis" except Exception as e: logger.warning(f"[RangeTrade] 적응형 파라미터 조회 실패 — 진입 중단(fail-closed): {e}") return [] if _adp_crisis: logger.warning("[RangeTrade] 적응형 CRISIS 모드 — 레인지 진입 차단") return [] # ── 게이트 8: 플레이북 ─────────────────────────────────────────────── if not is_entry_allowed_by_playbook(): logger.info("[RangeTrade] 플레이북 차단 — 신규 진입 비활성화") return [] # ── 후보 조회 ──────────────────────────────────────────────────────── held_result = await db.execute(select(Portfolio.stock_code).where(Portfolio.quantity > 0)) held_codes = {row[0] for row in held_result.fetchall()} 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("[RangeTrade] 분석 데이터 없음 — 스킵") return [] range_stocks: list[dict] = [] scan_count = 0 short_bars: list = [] for stock in candidates: if scan_count >= settings.RANGE_TRADE_SCAN_LIMIT: break if len(range_stocks) >= remaining_slots: break # 게이트 9: 중복 진입 방지 if stock.stock_code in held_codes: continue # 게이트 11: 가격 필터 if _below_min_price_krw_v2(stock.current_price): continue # 게이트 10: 연속손실 밴 (try 밖 — 실패를 삼키지 않는다) if await loss_tracker.is_banned_safe(stock.stock_code): logger.debug("[RangeTrade] %s 연속손실 밴 — 스킵", stock.stock_code) continue try: # [하드닝 R2 2026-10-09] get_daily_chart 30봉 상한 → 40봉 하한에 항상 걸렸다. bars, bars_src = await _fetch_scan_daily_bars_v2( korea_investment_api, stock.stock_code, RANGE_SCAN_MIN_BARS, ) if len(bars) < RANGE_SCAN_MIN_BARS: short_bars.append((stock.stock_code, len(bars), bars_src)) continue opens = [float(b.get("stck_oprc", 0) or 0) for b in bars] closes = [float(b.get("stck_clpr", 0) or 0) for b in bars] highs = [float(b.get("stck_hgpr", 0) or 0) for b in bars] lows = [float(b.get("stck_lwpr", 0) or 0) for b in bars] volumes = [int(float(b.get("acml_vol", 0) or 0)) for b in bars] if not closes or closes[-1] <= 0: continue if _below_min_price_krw_v2(closes[-1]): continue scan_count += 1 current_close = closes[-1] current_volume = volumes[-1] if volumes else 0 rsi = calculate_rsi(closes, period=14) if rsi is None: continue rsi_prev = calculate_rsi(closes[:-RSI_DIRECTION_LOOKBACK], period=14) bb = calculate_bollinger(closes, period=20) if bb is None: continue bb_upper, bb_middle, bb_lower = bb adx = calculate_adx(highs, lows, closes, period=14) atr = calculate_atr(highs, lows, closes, period=14) or 0.0 signal = check_range_buy_signal( highs=highs, lows=lows, closes=closes, volumes=volumes, current_close=current_close, current_volume=current_volume, rsi=rsi, bb_upper=bb_upper, bb_middle=bb_middle, bb_lower=bb_lower, adx=adx, rsi_prev=rsi_prev, opens=opens, ) if not signal: await asyncio.sleep(0.3) # KIS API 레이트 리밋 continue stops = calculate_range_stop_loss( entry_price=current_close, atr=atr, box_low=signal["box_low"], ) range_stocks.append({ "stock_code": stock.stock_code, "stock_name": stock.stock_name, "current_price": int(current_close), "score": 5.0, # signals: trade_executor 의 BUY_KEYWORDS/STRONG_BUY_KEYWORDS 매칭용. # 레인지 진입은 'BB 하단 터치 + RSI 바닥 탈출 반등' 과 동일 성격이다. "signals": [ BuySignal.볼린저하한선.value, BuySignal.RSI바닥탈출.value, BuySignal.횡보장평균회귀.value, ], "signal": signal["description"], "strategy_type": "swing", # 스윙 예산/청산 파이프라인 공유 "strategy_name": RANGE_TRADE_STRATEGY_NAME, # ActiveTrade.strategy_name "entry_reason": f"[{RANGE_TRADE_STRATEGY_NAME}] {signal['description']}", "stop_loss": stops["stop_loss_pct"], # 음수 보장 (-abs) "take_profit": max( 0.0, (signal["target_price"] - current_close) / current_close * 100.0, ), "atr": atr, "closes": closes, "rsi": rsi, "bb": (bb_upper, bb_middle, bb_lower), }) logger.info( "[RangeTrade] 박스권 진입 시그널: %s %s — %s", stock.stock_code, stock.stock_name, signal["description"], ) await asyncio.sleep(0.3) except Exception as e: warn_throttled(logger, "entry_service:682", f"[RangeTrade] {stock.stock_code} 스캔 실패: {e}") continue _warn_short_scan_bars_v2("RangeTrade", short_bars, RANGE_SCAN_MIN_BARS) if not range_stocks: logger.debug("[RangeTrade] 시그널 없음") return [] range_stocks = range_stocks[:remaining_slots] # ── 진입 실행 ──────────────────────────────────────────────────────── # execute_auto_trade_background() 는 SIDEWAYS 의 max_trades=0 에서 즉시 종료하므로 # 종목별 execute_auto_trade() 를 직접 호출한다(상단 설계 결정 주석 참조). # _regime_str="bear": _execute_buy 의 sideways 하드 차단을 피하면서 가장 보수적인 # 자본 한도(BEAR 30%)를 적용받는다. _position_multiplier 로 종목당 비중을 2%로 축소. results: list[dict] = [] for stock_dict in range_stocks: stock_dict["_regime_str"] = "bear" stock_dict["_position_multiplier"] = float(settings.RANGE_TRADE_POSITION_MULTIPLIER) try: result = await execute_auto_trade(stock_dict, db) except Exception as e: logger.error( f"[RangeTrade] {stock_dict.get('stock_code')} 진입 실행 오류: {e}", exc_info=True, ) continue if isinstance(result, dict) and result.get("action") == "BUY": results.append(result) else: logger.info( "[RangeTrade] %s 진입 보류: action=%s reason=%s", stock_dict.get("stock_code"), (result or {}).get("action") if isinstance(result, dict) else "?", (result or {}).get("reason", "미상") if isinstance(result, dict) else "?", ) await asyncio.sleep(0.5) # KIS API 레이트 리밋 if results: logger.info(f"[RangeTrade] {len(results)}건 진입 실행 완료") return results