"""스윙 트레이딩 독립 스케줄러 main.py의 lifespan에서 asyncio.create_task()로 시작되는 백그라운드 태스크. 기존 auto_trade_scheduler()를 대체. 60초 주기 실행 흐름: 1. 장 오픈 체크 (장외 시 30분 대기) 2. 스윙 포지션 손절/익절 체크 (check_and_execute_exit_v2) 3. MDD 체크 + 강제 축소 실행 (10% → 30% 축소, 20% → 50% 축소) 4. AUTO_TRADE_ENABLED 체크 5. Redis 캐시 또는 DB에서 추천 조회 6. MCP RISK_ASSESSMENT → 승인 시 스윙 매수 실행 MDD 축소 (CLAUDE.md 5.3): - reduction_applied == "10pct": 30% 축소 (REDUCE_AND_PAUSE) - reduction_applied == "20pct": 50% 축소 (FORCE_REDUCE, 레버리지 ETF 우선) 추천 데이터 소스 (우선순위): 1. Redis: "recommendations:latest" 키 (배치 분석 결과 캐시) 2. DB: get_rule_based_recommendations_from_db() (Redis 미스 시 폴백) """ import asyncio import logging import pytz from datetime import timedelta as _td from app.core.config import settings 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, scan_dual_strategy_signals logger = logging.getLogger(__name__) _KST = pytz.timezone("Asia/Seoul") # [M2] MDD 갱신 실패 감시 — 연속 실패가 이 횟수를 넘으면 error 로 승격 MDD_UPDATE_FAILURE_ALERT = 5 # MDD 갱신 상태 (모니터링/디버깅용 인메모리 상태) _mdd_health: dict = {"consecutive_failures": 0, "last_error": None, "empty_account": False} def get_mdd_update_health() -> dict: """MDD 갱신 상태 스냅샷 반환 (연속 실패 횟수/마지막 오류/진입 차단 여부). swing_scheduler 의 MDD 갱신이 조용히 실패 중인지 외부에서 확인하기 위한 읽기 전용 헬퍼. [항목7 2026-09-27] 헬스의 원본(source of truth)은 mdd_manager 로 이동했다 (can_open_new_position() 이 순환 import 없이 참조하려면 mdd_manager 가 소유해야 함). 이 함수는 시그니처를 유지한 읽기 전용 파사드이며, mdd_manager 스냅샷을 병합해 blind_block_active/threshold 까지 함께 노출한다. """ snapshot = dict(_mdd_health) snapshot.update(mdd_manager.get_update_health()) return snapshot # ───────────────────────────────────────────────────────────────────────────── # [하드닝 R1 2026-10-09] 단계 타임아웃 · 청산 판정 하트비트 · 텔레그램 경보 # # (1) 단계 상한: DT 의 _guard(2026-09-22 KIS 무응답 1시간 49분 정지 → EOD 청산 누락)와 # 같은 목적. 단, 취소 정책을 단계별로 나눈다(과다매도 방지가 우선). # - 체결가 보정·잔고 조회(주문 없음): wait_for 로 **취소**한다. # - 청산 판정·MDD 축소·진입(평균회귀+듀얼) — 주문 경로: **취소하지 않는다**. # 상한 초과 시 루프만 풀어 다음 단계로 넘어가고, 그 작업이 끝날 때까지 같은 단계를 # 새로 시작하지 않는다. 청산은 내부 매도가 shield 라 취소해도 매도 자체는 계속되는데, # 그 상태로 새 판정을 시작하면 같은 종목 분할 매도가 이중 실행될 수 있다. MDD 축소· # 진입은 shield 가 없어 취소 시 "브로커 체결·장부 미반영" 후 재주문 위험이 있다. # (2) 청산 판정 하트비트: 장중에 청산 판정이 마지막으로 정상 완료된 뒤 # SWING_EXIT_STALE_ALERT_SEC 를 넘기면 error + 텔레그램. 개장 직후는 개장 시각을 기준으로 # 계산해 밤사이 공백을 오경보하지 않는다. 사이클 하트비트(heartbeat.beat("swing"))도 # 매 반복 시작마다 찍어 main.py 감시 루프가 루프 정지를 잡게 한다. # (3) 경보: MDD fail-safe 진입 차단 진입/해제, 청산 체크 연속 실패, 단계 타임아웃. # 같은 key 는 SWING_ALERT_REPEAT_SEC 간격으로만 재전송한다. 전송은 백그라운드 태스크라 # 텔레그램 지연이 손절 루프를 막지 않는다. 텔레그램 미설정이면 기동 시 error 1회 + # get_swing_scheduler_health_v2()["telegram_enabled"]=False. # ───────────────────────────────────────────────────────────────────────────── _DEFAULT_EXIT_STEP_TIMEOUT_SEC = 180.0 _DEFAULT_FILL_SWEEP_TIMEOUT_SEC = 60.0 _DEFAULT_BALANCE_TIMEOUT_SEC = 60.0 _DEFAULT_MDD_REDUCTION_TIMEOUT_SEC = 180.0 _DEFAULT_ENTRY_PHASE_TIMEOUT_SEC = 300.0 _DEFAULT_EXIT_FAIL_ALERT_COUNT = 3 _DEFAULT_EXIT_STALE_ALERT_SEC = 600.0 _DEFAULT_ALERT_REPEAT_SEC = 1800.0 _STEP_TIMED_OUT = object() # 상한 초과(작업은 취소 또는 백그라운드 계속) _STEP_SKIPPED = object() # 이전 같은 단계 작업이 아직 진행 중이라 이번 사이클 생략 _exit_health: dict = { "last_success_at": None, # monotonic — 청산 판정 정상 완료 시각 "market_open_since": None, # monotonic — 이번 장중 구간을 처음 관측한 시각 "consecutive_failures": 0, # 예외 + 타임아웃 "last_error": None, "timeouts": 0, } _step_timeouts: dict = {} # step -> 누적 타임아웃 횟수 _inflight_steps: dict = {} # step -> asyncio.Task (취소하지 않는 단계의 진행 중 작업) _alert_sent_at: dict = {} # key -> monotonic 마지막 전송 시각 _alert_tasks: set = set() # 텔레그램 백그라운드 전송 태스크 참조 유지 _blind_prev: dict = {"active": False} _telegram_state: dict = {"enabled": None} # [하드닝 R3 2026-10-09] 비취소형 단계 하드 데드라인 — 상한(timeout)의 N배를 넘겨도 끝나지 않는 # 작업은 critical 경보 + /health degraded. 주문이 shield 로 분리된 '청산 판정'만, 진행 중인 # shield 주문 태스크가 없을 때 강제 취소 후 다음 사이클에 재시작한다(아래 _handle_hard_deadline_v2). _DEFAULT_STEP_HARD_DEADLINE_MULT = 5.0 _DEFAULT_STEP_CANCEL_GRACE_SEC = 10.0 _HARD_CANCEL_STEPS = frozenset({"청산 판정"}) # swing/exit_service 가 asyncio.shield 로 감싸는 "주문 전송 ~ 장부 반영" 코루틴 이름. # 이 태스크가 살아 있으면 청산 판정을 재시작하지 않는다(같은 종목 분할매도 이중 실행 방지). _SWING_SHIELDED_ORDER_CORO_NAMES = frozenset({"_sell_and_record_v2"}) _inflight_meta: dict = {} # step -> {"started": monotonic, "hard": 하드 데드라인(초)} _hard_deadline_stats: dict = {"cancels": 0, "exceeded": {}} # exceeded: step -> 누적 초과 감지 횟수 # [하드닝 R3 2026-10-09] 텔레그램 경보 전송 결과 추적 — 실패를 '전송됨'으로 두지 않는다. _DEFAULT_ALERT_SEND_RETRIES = 2 # 첫 시도 외 추가 시도 횟수(백그라운드) _DEFAULT_ALERT_RETRY_BACKOFF_SEC = 5.0 # 시도 간 대기(선형 증가) _DEFAULT_ALERT_FAIL_RETRY_SEC = 60.0 # 최종 실패 시 같은 key 재경보 허용까지의 간격 _telegram_send_health: dict = { "sent": 0, "failures": 0, # 재시도까지 모두 실패한 경보 수(누계) "consecutive_failures": 0, "last_success_at": None, # monotonic "last_failure_at": None, # monotonic "last_error": None, } # [하드닝 R3 2026-10-09] MDD 단계 임계 — mdd_manager.update() 가 실제로 전이하는 값과 같아야 한다. _MDD_MANAGER_PAUSE_PCT = -10.0 # mdd_manager.update: `drawdown_pct <= -10` 리터럴 _MDD_MANAGER_FORCE_PCT = -20.0 # mdd_manager.update: `drawdown_pct <= -20` 리터럴 _mdd_knob_state: dict = {"mismatches": []} # [하드닝 R4 2026-10-09] 스윙 튜닝 키(SWING_*_TIMEOUT_SEC, MDD_STAGE_CONFIRM_CYCLES 등)는 # config.Settings 에 선언돼 있지 않다. Settings 는 `extra="ignore"` 라 .env/compose 에 넣어도 # 속성이 생기지 않아 getattr 가 늘 모듈 기본값을 돌려줬다(운영자 오버라이드가 조용히 무시됨). # 선언 전까지는 settings 속성 → 프로세스 환경변수 → Settings 의 env_file 순으로 직접 읽는다. # config.py 에 선언되면 1순위(settings 속성)가 그대로 이긴다. _env_file_cache: dict = {} _undeclared_env_warned: set = set() def _env_file_values_v2() -> dict: """Settings 가 읽는 env_file(.env) 의 키/값. 실패하면 빈 dict(경고 1회).""" try: cfg = getattr(type(settings), "model_config", None) or {} path = cfg.get("env_file") if isinstance(cfg, dict) else None except Exception as e: logger.warning(f"[Swing] settings env_file 경로 확인 실패: {e}") return {} if not path or isinstance(path, (list, tuple)): return {} key = str(path) if key in _env_file_cache: return _env_file_cache[key] vals: dict = {} try: import os as _os if _os.path.isfile(key): from dotenv import dotenv_values vals = {k: v for k, v in dotenv_values(key).items() if v is not None} except Exception as e: logger.warning(f"[Swing] env_file({key}) 읽기 실패 — 스윙 튜닝 키는 기본값 사용: {e}") vals = {} _env_file_cache[key] = vals return vals def _raw_setting_v2(name: str): """튜닝 키 원시값. settings 속성 > os.environ > env_file. 없으면 None.""" raw = getattr(settings, name, None) if raw is not None: return raw import os as _os src = None if name in _os.environ: raw, src = _os.environ.get(name), "환경변수" else: file_vals = _env_file_values_v2() if name in file_vals: raw, src = file_vals[name], "env_file" if src is not None and name not in _undeclared_env_warned: _undeclared_env_warned.add(name) logger.warning( f"[Swing] 설정 {name}={raw!r} 는 config.Settings 에 미선언 — {src}에서 직접 적용" f"(Settings 에 선언 필요)" ) return raw def _cfg_float_v2(name: str, default: float) -> float: """settings 값을 양수 float 로 읽는다. 누락·비정상 값이면 기본값(상한을 끄지 않는다).""" raw = _raw_setting_v2(name) try: val = float(raw) if raw is not None else float(default) except (TypeError, ValueError): logger.warning(f"[Swing] 설정 {name}={raw!r} 비정상 — 기본값 {default} 사용") return float(default) if not val > 0: logger.warning(f"[Swing] 설정 {name}={raw!r} 는 양수가 아님 — 기본값 {default} 사용") return float(default) return val def _now_mono_v2() -> float: import time as _time return _time.monotonic() def _telegram_enabled_v2() -> bool: try: from app.services.telegram_notifier import _is_enabled return bool(_is_enabled()) except Exception as e: logger.warning(f"[Swing] 텔레그램 설정 확인 실패: {type(e).__name__}: {e}") return False def _check_telegram_configured_v2() -> bool: enabled = _telegram_enabled_v2() _telegram_state["enabled"] = enabled if not enabled: logger.error( "[Swing] 🚨 텔레그램 미설정(TELEGRAM_BOT_TOKEN/CHAT_ID) — 청산 정지·MDD 진입 차단·" "단계 타임아웃 경보가 로그로만 남는다" ) return enabled async def _send_system_warning_checked_v2(category: str, message: str) -> "bool | None": """텔레그램 시스템 경고를 보내고 **전송 결과**를 돌려준다. True=전송 성공, False=전송 실패(비200·전송 오류), None=텔레그램 미설정(전송 시도 안 함). telegram_notifier.notify_system_warning 은 _send_message 의 bool 을 버리므로 같은 포맷으로 공개 API send_message(bool 반환)를 직접 호출한다. """ from app.services.telegram_notifier import telegram_notifier, _is_enabled, _escape_markdown_v2 if not _is_enabled(): return None esc = _escape_markdown_v2 text = "\n".join([ f"\U0001F6A8 *{esc('시스템 경고')}*", "", f"\U0001F3F7 {esc('유형')}: {esc(category)}", f"\U0001F4DD {esc(message)}", ]) return bool(await telegram_notifier.send_message(text)) def _note_telegram_send_result_v2(ok: bool, err: str | None = None) -> None: h = _telegram_send_health now = _now_mono_v2() if ok: h["sent"] += 1 h["consecutive_failures"] = 0 h["last_success_at"] = now return h["failures"] += 1 h["consecutive_failures"] += 1 h["last_failure_at"] = now h["last_error"] = err def _notify_throttled_v2(key: str, category: str, message: str, repeat_sec: float | None = None) -> bool: """같은 key 는 repeat_sec 간격으로만 텔레그램 시스템 경고를 보낸다(백그라운드, 비차단). 반환: 이번 호출에서 전송을 예약했으면 True. 로그는 호출부가 남긴다. [하드닝 R3] 전송 결과를 확인한다. 실패(비200·예외)면 백그라운드에서 재시도하고, 끝내 실패하면 실패 누계를 올리고(/health) 이 key 의 억제 시각을 되돌려 짧은 간격 (SWING_ALERT_FAIL_RETRY_SEC) 뒤 다음 호출이 다시 보내게 한다 — 실패를 '전송됨'으로 두지 않는다. """ interval = repeat_sec if repeat_sec is not None else _cfg_float_v2( "SWING_ALERT_REPEAT_SEC", _DEFAULT_ALERT_REPEAT_SEC ) now = _now_mono_v2() last = _alert_sent_at.get(key) if last is not None and now - last < interval: return False _alert_sent_at[key] = now stamp = now try: retries = _cfg_int_v2("SWING_ALERT_SEND_RETRIES", _DEFAULT_ALERT_SEND_RETRIES, minimum=0) backoff = _cfg_float_v2("SWING_ALERT_RETRY_BACKOFF_SEC", _DEFAULT_ALERT_RETRY_BACKOFF_SEC) fail_retry = _cfg_float_v2("SWING_ALERT_FAIL_RETRY_SEC", _DEFAULT_ALERT_FAIL_RETRY_SEC) async def _send(): err = None for attempt in range(retries + 1): try: ok = await _send_system_warning_checked_v2(category=category, message=message) except Exception as e: ok = False err = f"{type(e).__name__}: {e}" if ok is None: return # 미설정 — 실패가 아니다(기동 시 error + /health telegram_enabled) if ok: _note_telegram_send_result_v2(True) return if err is None: err = "sendMessage 실패(비200 응답 또는 전송 오류 — telegram_notifier warning 로그 참조)" if attempt < retries: await asyncio.sleep(backoff * (attempt + 1)) _note_telegram_send_result_v2(False, err) # 억제 롤백 — 다른 전송이 그 사이 key 를 갱신하지 않았을 때만 if _alert_sent_at.get(key) == stamp: _alert_sent_at[key] = stamp - max(0.0, interval - fail_retry) logger.error( f"[Swing] 🚨 텔레그램 경보 전송 실패({key}, {retries + 1}회 시도): {err} — " f"{min(interval, fail_retry):.0f}초 뒤 재경보 허용. 원문: {message}" ) task = asyncio.create_task(_send()) _alert_tasks.add(task) task.add_done_callback(_alert_tasks.discard) return True except Exception as e: if _alert_sent_at.get(key) == stamp: if last is None: _alert_sent_at.pop(key, None) else: _alert_sent_at[key] = last logger.warning(f"[Swing] 텔레그램 경보 예약 실패({key}): {type(e).__name__}: {e}") return False def _record_step_timeout_v2(step: str, timeout: float, cancelled: bool) -> None: _step_timeouts[step] = _step_timeouts.get(step, 0) + 1 how = "취소" if cancelled else "백그라운드 계속(취소 안 함 — 주문 중복 방지), 끝날 때까지 이 단계 생략" msg = f"스윙 '{step}' {timeout:.0f}초 초과 — {how} (KIS 무응답·락 대기 가능성)" logger.error(f"[Swing] ⏱️ {msg}") _notify_throttled_v2(f"timeout:{step}", "SWING_SCHEDULER", msg) async def _guard_cancel_v2(coro, step: str, timeout: float): """취소형 상한(DT _guard 패턴). 초과 시 작업을 취소하고 _STEP_TIMED_OUT 반환. 작업이 낸 예외는 그대로 전파한다(호출부의 기존 처리 유지). """ try: return await asyncio.wait_for(coro, timeout=timeout) except asyncio.TimeoutError: _record_step_timeout_v2(step, timeout, cancelled=True) return _STEP_TIMED_OUT def _log_detached_result_v2(step: str, task: "asyncio.Task") -> None: if task.cancelled(): logger.warning(f"[Swing] 상한 초과 후 백그라운드 '{step}' 취소됨") return exc = task.exception() if exc is not None: logger.error(f"[Swing] 상한 초과 후 백그라운드 '{step}' 예외: {type(exc).__name__}: {exc}") else: logger.warning(f"[Swing] 상한 초과 후 백그라운드 '{step}' 완료") def _hard_deadline_sec_v2(timeout: float) -> float: """비취소형 단계의 하드 데드라인(초) = 상한 × SWING_STEP_HARD_DEADLINE_MULT(최소 1배).""" mult = _cfg_float_v2("SWING_STEP_HARD_DEADLINE_MULT", _DEFAULT_STEP_HARD_DEADLINE_MULT) if mult < 1.0: logger.warning(f"[Swing] SWING_STEP_HARD_DEADLINE_MULT={mult} < 1 — 기본값 {_DEFAULT_STEP_HARD_DEADLINE_MULT} 사용") mult = _DEFAULT_STEP_HARD_DEADLINE_MULT return float(timeout) * mult def _pending_shielded_order_tasks_v2() -> list: """살아 있는 스윙 shield 주문 태스크(_sell_and_record_v2) — 있으면 청산 판정 재시작 금지.""" out = [] for t in asyncio.all_tasks(): if t.done(): continue coro = t.get_coro() if getattr(coro, "__name__", "") in _SWING_SHIELDED_ORDER_CORO_NAMES: out.append(t) return out async def _handle_hard_deadline_v2(step: str, prev: "asyncio.Task") -> bool: """진행 중인 비취소형 단계가 하드 데드라인을 넘겼는지 판정·처리. 반환 True = 이전 작업을 취소·종료시켰으므로 이번 사이클에 새로 시작해도 된다. 반환 False = 그대로 생략(_STEP_SKIPPED). 정책(과다매도·손절 무력화 양쪽을 막는 쪽): - 하드 데드라인 초과 시 모든 단계에 critical 로그 + 텔레그램 + /health degraded. - 강제 취소는 _HARD_CANCEL_STEPS('청산 판정')만. 이 단계의 주문 전송~장부 반영은 swing/exit_service 가 asyncio.shield(_sell_and_record_v2) 로 분리해 두었으므로 바깥 작업을 취소해도 '전송됨·장부 미반영'이 생기지 않는다. 단 그 shield 태스크가 살아 있으면 (=주문 전송 구간 진행 중) 취소하지 않는다 — 재시작한 판정이 같은 종목을 이중 분할매도할 수 있다. - 'MDD 축소'·'진입' 은 trade_executor/mdd_manager 주문이 shield 되어 있지 않아 취소 시 중복 주문 위험이 있으므로 경보만 한다(진입 정지는 fail-closed 방향). """ meta = _inflight_meta.get(step) or {} started = meta.get("started") hard = meta.get("hard") if started is None or hard is None: return False age = _now_mono_v2() - float(started) if age <= float(hard): return False _hard_deadline_stats["exceeded"][step] = _hard_deadline_stats["exceeded"].get(step, 0) + 1 can_cancel = step in _HARD_CANCEL_STEPS shielded = _pending_shielded_order_tasks_v2() if can_cancel else [] if can_cancel and not shielded: prev.cancel() grace = _cfg_float_v2("SWING_STEP_CANCEL_GRACE_SEC", _DEFAULT_STEP_CANCEL_GRACE_SEC) await asyncio.wait({prev}, timeout=grace) if prev.done(): _inflight_steps.pop(step, None) _inflight_meta.pop(step, None) _hard_deadline_stats["cancels"] += 1 msg = ( f"스윙 '{step}' 이 하드 데드라인 {hard:.0f}초를 넘겨 {age:.0f}초 정체 — 강제 취소 후 " f"이번 사이클에 재시작(진행 중 주문 없음 확인). 정체 원인(락·외부 호출) 확인 필요" ) logger.critical(f"[Swing] 🚨 {msg}") _notify_throttled_v2(f"hard_deadline:{step}", "SWING_SCHEDULER_STUCK", msg) return True how = f"취소 요청이 {grace:.0f}초 안에 처리되지 않음(취소 무시)" elif can_cancel: how = f"shield 주문 태스크 {len(shielded)}건 진행 중 — 이중 매도 방지로 취소하지 않음" else: how = "주문 경로(비 shield)라 취소하지 않음 — 중복 주문 방지" msg = ( f"스윙 '{step}' 이 하드 데드라인 {hard:.0f}초를 넘겨 {age:.0f}초 정체 — {how}. " f"재기동 등 수동 조치 필요" ) logger.critical(f"[Swing] 🚨 {msg}") _notify_throttled_v2(f"hard_deadline:{step}", "SWING_SCHEDULER_STUCK", msg) return False async def _guard_detached_v2(factory, step: str, timeout: float): """비취소형 상한 — 주문 경로(MDD 축소·진입)용. - 같은 step 의 이전 작업이 진행 중이면 새로 시작하지 않고 _STEP_SKIPPED. - 상한 안에 끝나면 결과 반환(예외는 전파). - 초과하면 작업은 취소하지 않고 남겨 둔 채 _STEP_TIMED_OUT 반환 → 루프는 다음 사이클(청산 우선)로 진행한다. 주문 전송 중 취소로 "브로커 체결·장부 미반영"이 생겨 같은 물량을 다시 주문하는 것을 막기 위함이다. - 스케줄러 자신이 취소되면(종료) 작업은 취소하지 않는다(shield 와 같은 의미). """ prev = _inflight_steps.get(step) if prev is not None and not prev.done(): # [하드닝 R3] 하드 데드라인 판정은 이 분기 맨 앞, try 밖(가드)에서 한다. if not await _handle_hard_deadline_v2(step, prev): msg = f"스윙 '{step}' 이전 작업이 아직 진행 중 — 이번 사이클 생략" logger.error(f"[Swing] ⏱️ {msg}") _notify_throttled_v2(f"inflight:{step}", "SWING_SCHEDULER", msg) return _STEP_SKIPPED task = asyncio.ensure_future(factory()) _inflight_steps[step] = task _inflight_meta[step] = {"started": _now_mono_v2(), "hard": _hard_deadline_sec_v2(timeout)} done, _pending = await asyncio.wait({task}, timeout=timeout) if task in done: _inflight_steps.pop(step, None) _inflight_meta.pop(step, None) return task.result() task.add_done_callback(lambda t, _s=step: _log_detached_result_v2(_s, t)) _record_step_timeout_v2(step, timeout, cancelled=False) return _STEP_TIMED_OUT def _note_market_open_v2(now: float) -> None: if _exit_health["market_open_since"] is None: _exit_health["market_open_since"] = now def _note_market_closed_v2() -> None: _exit_health["market_open_since"] = None def _record_exit_success_v2(now: float) -> None: if _exit_health["consecutive_failures"]: logger.warning( f"[Swing] 청산 판정 복구 (연속 실패 {_exit_health['consecutive_failures']}회 후 정상화)" ) _had_fail_alert = _alert_sent_at.pop("exit_failures", None) is not None _had_stale_alert = _alert_sent_at.pop("exit_stale", None) is not None if _had_fail_alert or _had_stale_alert: _notify_throttled_v2( "exit_recovered", "SWING_EXIT", f"스윙 청산 판정 복구 (연속 실패 {_exit_health['consecutive_failures']}회 후)", repeat_sec=0.0, ) _exit_health["last_success_at"] = now _exit_health["consecutive_failures"] = 0 _exit_health["last_error"] = None try: from app.core import heartbeat as _hb _hb.beat("swing_exit", "exit_ok") except Exception as e: logger.warning(f"[Swing] 청산 하트비트 기록 실패: {type(e).__name__}: {e}") def _record_exit_failure_v2(err: str, timed_out: bool = False) -> None: _exit_health["consecutive_failures"] += 1 _exit_health["last_error"] = err if timed_out: _exit_health["timeouts"] += 1 n = _exit_health["consecutive_failures"] threshold = int(_cfg_float_v2("SWING_EXIT_FAIL_ALERT_COUNT", _DEFAULT_EXIT_FAIL_ALERT_COUNT)) if n >= threshold: msg = f"스윙 청산 판정 연속 실패 {n}회 — 손절·익절이 멈춰 있을 수 있음: {err}" logger.error(f"[Swing] 🚨 {msg}") _notify_throttled_v2("exit_failures", "SWING_EXIT", msg) def _check_exit_staleness_v2(now: float) -> bool: """장중 청산 판정 무완료 시간이 상한을 넘으면 error + 텔레그램. 넘었으면 True. 기준 시각 = max(마지막 정상 완료, 이번 장중 구간 첫 관측) — 밤사이 공백은 세지 않는다. """ open_since = _exit_health["market_open_since"] if open_since is None: return False last_ok = _exit_health["last_success_at"] ref = open_since if last_ok is None else max(last_ok, open_since) age = now - ref limit = _cfg_float_v2("SWING_EXIT_STALE_ALERT_SEC", _DEFAULT_EXIT_STALE_ALERT_SEC) if age <= limit: return False msg = ( f"스윙 청산 판정이 장중 {age:.0f}초 동안 완료되지 않음(상한 {limit:.0f}초) — " f"보유 종목 손절이 방치될 수 있음. 마지막 오류: {_exit_health['last_error']}" ) logger.error(f"[Swing] 🚨 {msg}") _notify_throttled_v2("exit_stale", "SWING_EXIT", msg) return True def _check_blind_block_transition_v2() -> None: """MDD fail-safe 진입 차단(blind_block_active) 진입·해제 시점에 경보.""" try: h = mdd_manager.get_update_health() or {} except Exception as e: logger.warning(f"[Swing] MDD 헬스 조회 실패: {type(e).__name__}: {e}") return active = bool(h.get("blind_block_active")) prev = _blind_prev["active"] _blind_prev["active"] = active if active and not prev: msg = ( f"MDD 갱신 연속 실패 {h.get('consecutive_failures')}회 — 신규 매수 전면 차단(fail-safe). " f"청산은 계속. 마지막 오류: {h.get('last_error')}" ) logger.error(f"[Swing] 🚨 {msg}") _notify_throttled_v2("mdd_blind_on", "MDD_FAILSAFE", msg) elif prev and not active: logger.warning("[Swing] MDD fail-safe 진입 차단 해제(갱신 정상화)") _notify_throttled_v2("mdd_blind_off", "MDD_FAILSAFE", "MDD 갱신 정상화 — 신규 매수 차단 해제") def get_swing_scheduler_health_v2() -> dict: """스윙 스케줄러 감시 상태(읽기 전용) — /health 노출용. 청산 판정 경과·연속 실패·단계 타임아웃·진행 중 단계·텔레그램 설정 여부. """ now = _now_mono_v2() last_ok = _exit_health["last_success_at"] open_since = _exit_health["market_open_since"] ref = None if open_since is not None: ref = open_since if last_ok is None else max(last_ok, open_since) stale_limit = _cfg_float_v2("SWING_EXIT_STALE_ALERT_SEC", _DEFAULT_EXIT_STALE_ALERT_SEC) exit_age = None if ref is None else round(now - ref, 1) inflight_ages: dict = {} stuck: list = [] for k, t in _inflight_steps.items(): if t.done(): continue meta = _inflight_meta.get(k) or {} if meta.get("started") is None: continue a = now - float(meta["started"]) inflight_ages[k] = round(a, 1) if meta.get("hard") is not None and a > float(meta["hard"]): stuck.append(k) stuck.sort() tg = _telegram_send_health reasons: list = [] if stuck: reasons.append("stuck_steps") if exit_age is not None and exit_age > stale_limit: reasons.append("exit_stale") if tg["consecutive_failures"] > 0: reasons.append("telegram_send_failing") if _mdd_knob_state["mismatches"]: reasons.append("mdd_threshold_knob_mismatch") # [하드닝 R4] 청산 판정 실행 실패·종목별 연속 예외·매도 거부(계좌 단위 포함) try: from app.services.swing.exit_service import get_swing_exit_health_v2 _exh = get_swing_exit_health_v2() except Exception as e: logger.warning(f"[Swing] 청산 감시 상태 조회 실패: {type(e).__name__}: {e}") _exh = {"degraded_reasons": ["exit_health_unavailable"]} for _r in _exh.get("degraded_reasons") or (): if _r not in reasons: reasons.append(_r) return { "exit_last_success_age": None if last_ok is None else round(now - last_ok, 1), "exit_unjudged_market_sec": exit_age, "exit_stale": bool(exit_age is not None and exit_age > stale_limit), "exit_consecutive_failures": _exit_health["consecutive_failures"], "exit_timeouts": _exit_health["timeouts"], "exit_last_error": _exit_health["last_error"], "step_timeouts": dict(_step_timeouts), "inflight_steps": sorted(k for k, t in _inflight_steps.items() if not t.done()), "inflight_step_ages": inflight_ages, "stuck_steps": stuck, "hard_deadline_cancels": _hard_deadline_stats["cancels"], "hard_deadline_exceeded": dict(_hard_deadline_stats["exceeded"]), "mdd_blind_block_active": _blind_prev["active"], "mdd_threshold_knob_mismatch": list(_mdd_knob_state["mismatches"]), "telegram_enabled": ( _telegram_state["enabled"] if _telegram_state["enabled"] is not None else _telegram_enabled_v2() ), "telegram_sent": tg["sent"], "telegram_send_failures": tg["failures"], "telegram_consecutive_failures": tg["consecutive_failures"], "telegram_last_success_age": ( None if tg["last_success_at"] is None else round(now - tg["last_success_at"], 1) ), "telegram_last_error": tg["last_error"], "exit_symbol_error_counts": _exh.get("symbol_error_counts", {}), "exit_sell_reject_counts": _exh.get("sell_reject_counts", {}), "exit_sell_reject_last": _exh.get("sell_reject_last", {}), "exit_last_run": _exh.get("last_run"), "status": "degraded" if reasons else "ok", "degraded_reasons": reasons, } # ───────────────────────────────────────────────────────────────────────────── # [하드닝 R2 2026-10-09] MDD 갱신 입력 가드 — 입출금·부분응답을 낙폭으로 오인하지 않게 # # 결함: tot_evlu_amt 원값을 그대로 mdd_manager.update() 에 넣어, 출금(또는 KIS 가 일부 종목 # 평가를 0 으로 준 응답) 한 번으로 drawdown<=-10% → is_paused + 전 보유 30% 시장가 축소 # (되돌릴 수 없는 주문)가 실행됐다. 입금은 peak 를 부풀리거나 진행 중인 MDD 정지를 해제했다. # # (1) 현금흐름 보정: KIS 국내잔고 항등식 tot_evlu_amt = scts_evlu_amt + prvs_rcdl_excc_amt # (portfolio.py 참조)에서 현금 성분 = tot − scts. 직전 관측 대비 현금 성분이 총자산의 # MDD_CASH_FLOW_MIN_RATIO(기본 2%) 이상 움직였는데 유가평가(scts)는 그 변화의 # MDD_CASH_FLOW_SCTS_TOLERANCE(기본 25%) 이내로만 움직였으면 외부 입출금으로 본다. # 매수·매도는 현금↔유가가 같이 움직이고 시세 손실은 유가 쪽에서만 나므로 이 판정에 # 걸리지 않는다(실손실을 입출금으로 숨기는 방향의 오판을 구조적으로 피한다). # 입출금이면 peak 를 (직전총액+흐름)/직전총액 비율로 스케일해 낙폭 %를 보존한다. # (2) 단계 전이 확인: 이번 값이 새 MDD 단계(-10%/-20%)를 유발하면 즉시 반영하지 않는다. # (a) 보유 종목 평가 교차검증 — hldg_qty>0 인데 evlu_amt/prpr 가 0 이하이거나, output1 # 평가합이 scts_evlu_amt 와 어긋나면 부분응답으로 보고 갱신 실패 처리(fail-safe 계수). # MDD_STAGE_SUSPECT_MAX_HOLDS(기본 5)회 연속이면 그대로 반영한다 — 실제 낙폭을 # 무기한 억제하지 않기 위해서다(§5.3). # (b) MDD_STAGE_CONFIRM_CYCLES(기본 2) 사이클 연속 전이 조건이어야 반영한다. 대기 중인 # 사이클은 신규 진입을 차단한다(실제 낙폭일 수 있으므로 보수적으로). # 전이가 아닌 갱신(peak 갱신·단계 내 변동)은 이전과 동일하게 즉시 반영한다. # 한계(추정): 입출금과 매매가 같은 사이클에 겹치면 (1)이 걸리지 않고 (2)만 적용된다. # ───────────────────────────────────────────────────────────────────────────── _DEFAULT_MDD_STAGE_CONFIRM_CYCLES = 2 _DEFAULT_MDD_STAGE_SUSPECT_MAX_HOLDS = 5 _DEFAULT_MDD_CASH_FLOW_MIN_RATIO = 0.02 _DEFAULT_MDD_CASH_FLOW_SCTS_TOLERANCE = 0.25 _DEFAULT_MDD_EVAL_SUM_TOLERANCE = 0.05 _mdd_gate_state: dict = { "prev": None, # 직전 정상 관측 {"total": float, "scts": float|None} "pending": 0, # 단계 전이 조건 연속 관측 횟수 "suspect": 0, # 부분응답 의심으로 보류한 연속 횟수 "last_flow": None, # 마지막으로 보정한 외부 현금흐름(원) } def _cfg_int_v2(name: str, default: int, minimum: int = 1) -> int: raw = _raw_setting_v2(name) if isinstance(raw, str): raw = raw.strip() try: val = int(raw) if raw is not None else int(default) except (TypeError, ValueError): logger.warning(f"[Swing] 설정 {name}={raw!r} 비정상 — 기본값 {default} 사용") return int(default) if val < minimum: logger.warning(f"[Swing] 설정 {name}={raw!r} < {minimum} — 기본값 {default} 사용") return int(default) return val def _num_or_none_v2(raw): if raw is None or str(raw).strip() == "": return None try: return float(raw) except (TypeError, ValueError): return None def _mdd_stage_thresholds_v2() -> tuple: """(pause, force) 임계(음수 %) — mdd_manager.update() 가 **실제로 전이하는** 값. [하드닝 R3 2026-10-09] 예전에는 settings.MDD_PAUSE/FORCE_THRESHOLD 중 얕은 쪽을 썼다. 그래서 설정이 -8 이면 -8%~-10% 구간에서 전이를 '예측'해 확인 사이클(신규 진입 차단)을 채운 뒤 update() 를 불러도 mdd_manager(-10 리터럴)는 전이하지 않고, pending 이 리셋됐다가 다시 쌓여 축소 없이 진입만 막혔다. 예측은 실제 전이와 같아야 하므로 mdd_manager 리터럴을 쓴다. 설정/DB 값이 이와 다르면 _check_mdd_threshold_knobs_v2 가 '효과 없는 손잡이'로 경보한다. """ return _MDD_MANAGER_PAUSE_PCT, _MDD_MANAGER_FORCE_PCT def _mdd_threshold_mismatches_v2(sources: dict) -> list: """sources = {"settings": {key: raw}, "strategy_cfg": {key: raw}} → 실효 임계와 다른 항목 목록.""" effective = {"MDD_PAUSE_THRESHOLD": _MDD_MANAGER_PAUSE_PCT, "MDD_FORCE_THRESHOLD": _MDD_MANAGER_FORCE_PCT} out = [] for src, vals in sources.items(): for key, eff in effective.items(): v = _num_or_none_v2((vals or {}).get(key)) if v is None: continue if abs(-abs(v) - eff) > 1e-9: out.append({"source": src, "key": key, "value": v, "effective": eff}) return out async def _check_mdd_threshold_knobs_v2() -> list: """MDD 임계 '조절 손잡이'(settings·strategy_configs DB)가 실효값과 다르면 error + 텔레그램. mdd_manager.update() 는 -10/-20 리터럴로만 전이하므로 다른 값은 아무 효과가 없다 (운영자가 UI 에서 바꿔도 반영되지 않는다). 읽기 전용 — strategy_cfg 캐시 조회(SELECT)만. """ sources: dict = {"settings": { "MDD_PAUSE_THRESHOLD": getattr(settings, "MDD_PAUSE_THRESHOLD", None), "MDD_FORCE_THRESHOLD": getattr(settings, "MDD_FORCE_THRESHOLD", None), }} try: from app.services.strategy_config_service import strategy_cfg db_vals = {} for key in ("MDD_PAUSE_THRESHOLD", "MDD_FORCE_THRESHOLD"): db_vals[key] = await asyncio.wait_for(strategy_cfg.aget(key, None), timeout=10.0) sources["strategy_cfg"] = db_vals except Exception as e: logger.warning(f"[Swing] MDD 임계 DB 파라미터 조회 실패(대조 생략): {type(e).__name__}: {e}") mism = _mdd_threshold_mismatches_v2(sources) _mdd_knob_state["mismatches"] = mism if mism: detail = ", ".join(f"{m['source']}.{m['key']}={m['value']}" for m in mism) msg = ( f"MDD 임계 설정이 실효값(진입중단 {_MDD_MANAGER_PAUSE_PCT}% / 강제축소 {_MDD_MANAGER_FORCE_PCT}%)과 " f"다름 — 효과 없음(mdd_manager 는 고정 임계로 전이): {detail}" ) logger.error(f"[Swing] 🚨 {msg}") _notify_throttled_v2("mdd_threshold_knob", "MDD_CONFIG", msg) return mism def _detect_cash_flow_v2(prev, total: float, scts) -> float | None: """직전 관측 대비 외부 입출금 추정액(원, 출금 음수). 판정 불가·해당 없음이면 None.""" if not prev or scts is None or prev.get("scts") is None: return None prev_total = float(prev.get("total") or 0.0) if prev_total <= 0: return None d_cash = (total - scts) - (prev_total - float(prev["scts"])) d_scts = scts - float(prev["scts"]) min_ratio = _cfg_float_v2("MDD_CASH_FLOW_MIN_RATIO", _DEFAULT_MDD_CASH_FLOW_MIN_RATIO) tol = _cfg_float_v2("MDD_CASH_FLOW_SCTS_TOLERANCE", _DEFAULT_MDD_CASH_FLOW_SCTS_TOLERANCE) if abs(d_cash) < prev_total * min_ratio: return None if abs(d_scts) > abs(d_cash) * tol: return None # 매매(현금↔유가 동반 이동) 또는 시세 변동 — 입출금으로 보지 않음 return d_cash def _holdings_eval_suspect_v2(balance_resp, scts) -> str | None: """보유 평가 교차검증 — 부분응답 의심 사유 문자열, 정상/판정불가면 None. 필드가 아예 없는 경우는 의심으로 보지 않는다(필드명 불일치로 실제 MDD 를 막지 않게). """ holdings = (balance_resp or {}).get("output1") if not isinstance(holdings, list): return None evals = [] for h in holdings: if not isinstance(h, dict): continue qty = _num_or_none_v2(h.get("hldg_qty")) if not qty or qty <= 0: continue ev = _num_or_none_v2(h.get("evlu_amt")) px = _num_or_none_v2(h.get("prpr")) code = h.get("pdno") or "?" if (ev is not None and ev <= 0) or (px is not None and px <= 0): return f"보유 {code} 평가 0 (evlu_amt={h.get('evlu_amt')!r}, prpr={h.get('prpr')!r})" evals.append(ev) if scts is None or not evals or any(e is None for e in evals): return None tol = _cfg_float_v2("MDD_EVAL_SUM_TOLERANCE", _DEFAULT_MDD_EVAL_SUM_TOLERANCE) s = sum(evals) if s > scts * (1 + tol) + 1: return f"보유 평가합 {s:,.0f} > 유가평가 {scts:,.0f}" from app.services.mdd_manager import _balance_page_full_v2 if not _balance_page_full_v2(holdings) and s < scts * (1 - tol) - 1: return f"보유 평가합 {s:,.0f} < 유가평가 {scts:,.0f} (페이지 미만)" return None def _predict_mdd_transition_v2(total: float) -> str | None: """이 값으로 update() 하면 새 단계로 전이하는지 (I/O 없음, 판정만).""" peak = float(getattr(mdd_manager, "peak_balance", 0.0) or 0.0) if getattr(mdd_manager, "_rebase_peak_pending", False) or peak <= 0 or total >= peak: return None dd = (total - peak) / peak * 100.0 pause, force = _mdd_stage_thresholds_v2() applied = getattr(mdd_manager, "reduction_applied", None) if dd <= force and applied != "20pct": return "20pct" if dd <= pause and applied is None: return "10pct" return None async def _rescale_peak_for_cash_flow_v2(prev_total: float, flow: float) -> None: """외부 입출금만큼 peak 를 비율 스케일(낙폭 % 보존) + Redis 영속화(락 밖).""" after = prev_total + flow if prev_total <= 0 or after <= 0: return ratio = after / prev_total async with mdd_manager._lock: old = float(mdd_manager.peak_balance or 0.0) if old <= 0: return mdd_manager.peak_balance = old * ratio _mdd_gate_state["last_flow"] = flow logger.warning( f"[Swing] MDD 입출금 감지 추정 {flow:+,.0f}원 — peak {old:,.0f} → " f"{old * ratio:,.0f} 보정(낙폭 % 보존). 수동 확인 권장" ) _notify_throttled_v2( "mdd_cash_flow", "MDD_CASH_FLOW", f"계좌 입출금 추정 {flow:+,.0f}원 — MDD peak 를 비율 보정했습니다(낙폭 오인 방지). 수동 확인 권장", ) await mdd_manager.save_to_redis() def _mdd_has_known_balance_v2() -> bool: """[하드닝 R1-라운드1] MDD 관리자가 이미 양(+)의 자산을 관측했는가 (I/O 없음). peak_balance 또는 마지막 갱신 잔고가 >0 이면 '운용 중 계좌'로 본다. 이 상태의 tot_evlu_amt<=0 응답은 빈 계좌가 아니라 부분/오류 응답으로 취급해야 한다. 값 해석이 불가하면 True(=빈 계좌로 인정하지 않음 — fail-safe 쪽). """ try: peak = float(getattr(mdd_manager, "peak_balance", 0.0) or 0.0) last = float(getattr(mdd_manager, "_last_balance", 0.0) or 0.0) except (TypeError, ValueError) as e: logger.warning(f"[Swing] MDD 기존 잔고 해석 실패 — 빈 계좌 판정 불허: {type(e).__name__}: {e}") return True return peak > 0 or last > 0 async def _mdd_update_guarded_v2(total: float, balance_resp) -> dict: """mdd_manager.update(total) 앞단 가드. 반환 {"status": "applied"|"pending", "result": dict|None}. 부분응답 의심은 ValueError 로 올려 호출부의 갱신 실패 경로(fail-safe 계수)를 탄다. """ st = _mdd_gate_state out2 = ((balance_resp or {}).get("output2") or [{}])[0] or {} scts = _num_or_none_v2(out2.get("scts_evlu_amt")) prev = st["prev"] flow = _detect_cash_flow_v2(prev, total, scts) if flow is not None: await _rescale_peak_for_cash_flow_v2(float(prev["total"]), flow) stage = _predict_mdd_transition_v2(total) if stage is None: st["pending"] = 0 st["suspect"] = 0 st["prev"] = {"total": total, "scts": scts} return {"status": "applied", "result": await mdd_manager.update(total)} suspect = _holdings_eval_suspect_v2(balance_resp, scts) if suspect: st["suspect"] += 1 max_holds = _cfg_int_v2("MDD_STAGE_SUSPECT_MAX_HOLDS", _DEFAULT_MDD_STAGE_SUSPECT_MAX_HOLDS) if st["suspect"] < max_holds: # 직전 정상 관측(prev)은 유지 — 부분응답을 흐름 판정 기준으로 쓰지 않는다 raise ValueError( f"MDD {stage} 전이 보류 — 잔고 부분응답 의심 {st['suspect']}/{max_holds}: {suspect}" ) logger.error( f"[Swing] 🚨 MDD {stage} 전이 — 부분응답 의심이 {st['suspect']}회 연속이라 그대로 반영: {suspect}" ) st["suspect"] = 0 st["prev"] = {"total": total, "scts": scts} st["pending"] += 1 need = _cfg_int_v2("MDD_STAGE_CONFIRM_CYCLES", _DEFAULT_MDD_STAGE_CONFIRM_CYCLES) if st["pending"] < need: logger.warning( f"[Swing] MDD {stage} 전이 조건 관측 {st['pending']}/{need} — 확인 대기" f"(총평가 {total:,.0f}, peak {mdd_manager.peak_balance:,.0f}). 이번 사이클 신규 진입 차단" ) return {"status": "pending", "result": None} st["pending"] = 0 return {"status": "applied", "result": await mdd_manager.update(total)} async def _run_exit_phase_v2(): """청산 판정 1회. 청산 판정 성공/실패를 직접 기록한다. 비취소형 상한(_guard_detached_v2) 아래에서 돌기 때문에 세션을 스스로 연다 — 상한 초과로 루프가 먼저 진행해도 세션이 작업보다 먼저 닫히지 않게 하기 위함이다. [하드닝 R8 2026-10-10] 체결가 후속 보정(sweep)은 여기서 분리해 _run_fill_sweep_phase_v2 (취소형 별도 단계)로 옮겼다. 이전엔 sweep(최대 SWING_FILL_SWEEP_TIMEOUT_SEC)이 청산 판정과 같은 비취소형 상한(SWING_EXIT_STEP_TIMEOUT_SEC)을 공유해, 청산이 끝났는데도 상한 초과로 '청산 실패' 집계·다음 사이클 _STEP_SKIPPED·가짜 연속 실패 경보가 날 수 있었다. 반환: check_and_execute_exit_v2 결과(list) 또는 예외 시 None. """ exits = None # DB-READ-EXCEPTION: 청산 판정용 SELECT 전용 세션(commit 없음, 쓰기는 db_task_queue 경유) — CLAUDE.md §3.2 ③-b 스윙 스케줄러 SELECT, 시간손절 TradeHistory SELECT 포함 async with AsyncSessionLocal() as exit_session: try: # [하드닝 R4 2026-10-09] check_and_execute_exit_v2 는 '전체 오류'·'보유 종목 전부 예외'를 # 삼키고 정상 반환한다. 실행 일련번호로 이번 실행분 결과를 확인해 실패로 집계한다 # (이전엔 무조건 성공 기록 → 연속 실패·무완료 경보 우회). from app.services.swing import exit_service as _es_run _seq0 = _es_run.get_last_exit_run_status_v2()["seq"] exits = await check_and_execute_exit_v2(exit_session) _run_st = _es_run.get_last_exit_run_status_v2() if _run_st["seq"] != _seq0 and not _run_st["ok"]: logger.error(f"[Swing] 청산 판정 실패(예외 삼킴 경로): {_run_st['error']}") _record_exit_failure_v2(str(_run_st["error"])) else: _record_exit_success_v2(_now_mono_v2()) if exits: logger.info(f"[Swing] 청산 실행: {len(exits)}건") except Exception as e: logger.error(f"[Swing] 청산 체크 오류: {e}", exc_info=True) _record_exit_failure_v2(f"{type(e).__name__}: {e}") return exits async def _run_fill_sweep_phase_v2(): """체결가 미확인 TradeHistory 후속 보정(swing/fill_sweep) — 청산 판정과 분리된 취소형 단계. 주문이 없는 조회·보정 경로라 취소형 상한(SWING_FILL_SWEEP_TIMEOUT_SEC)을 쓴다. 갱신은 db_task_queue 경유(swing/fill_sweep). 청산 판정의 성공/실패 집계에 관여하지 않는다. 청산 판정 작업이 아직 살아 있으면(상한 초과 후 백그라운드 계속) 이번 사이클은 생략한다 — 진행 중인 매도와 같은 행을 동시에 보정하지 않기 위함(이전 동작: sweep 은 청산 판정 뒤에만 실행). 반환: 보정 건수, _STEP_TIMED_OUT, 또는 생략·예외 시 None. """ _prev_exit = _inflight_steps.get("청산 판정") if _prev_exit is not None and not _prev_exit.done(): logger.warning("[Swing] 청산 판정 진행 중 — 체결가 후속 보정 이번 사이클 생략") return None try: from app.services.swing.fill_sweep import sweep_unconfirmed_fills_v2 # DB-READ-EXCEPTION: 체결가 후속 보정 대상 SELECT 전용 세션(commit 없음, 갱신은 db_task_queue 경유) — CLAUDE.md §3.2 ③-b 스윙 스케줄러 SELECT async with AsyncSessionLocal() as sweep_session: _swept = await _guard_cancel_v2( sweep_unconfirmed_fills_v2(sweep_session), "체결가 후속 보정", _cfg_float_v2("SWING_FILL_SWEEP_TIMEOUT_SEC", _DEFAULT_FILL_SWEEP_TIMEOUT_SEC), ) if _swept and _swept is not _STEP_TIMED_OUT: logger.info(f"[Swing] 체결가 후속 보정: {_swept}건") return _swept except Exception as e: logger.warning(f"[Swing] 체결가 후속 보정 실패: {type(e).__name__}: {e}") return None async def _run_entry_phase_v2( get_json, get_rule_based_recommendations_from_db, request_mcp_risk_assessment, ) -> None: """평균회귀 진입 → 듀얼 스캔(순서 유지, M-1). 하나의 비취소형 단계로 묶어 상한을 건다. 평균회귀 단계 예외는 듀얼 스캔을 막지 않는다(H5). """ try: await _run_recommendation_entry_v2( get_json, get_rule_based_recommendations_from_db, request_mcp_risk_assessment, ) except Exception as e: logger.error(f"[Swing] 평균회귀 진입 단계 오류 — 듀얼 스캔은 계속: {e}", exc_info=True) # 듀얼 전략(모멘텀 돌파·눌림목) — MCP 평가 대상 아님(2026-10-02 사용자 결정) await _run_dual_strategy_scan_v2() def _normalize_recommendations(rec: dict) -> dict: """recommendations_by_rule 키 정규화. Redis 캐시에 "recommendations" 키로 저장된 경우 "recommendations_by_rule"로 변환하여 execute_swing_entries()와 호환. Args: rec: 추천 딕셔너리 Returns: dict: 정규화된 추천 딕셔너리 """ if not isinstance(rec, dict): return rec if "recommendations_by_rule" in rec: return rec if "recommendations" in rec: rec["recommendations_by_rule"] = rec["recommendations"] 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 _run_dual_strategy_scan_v2() -> None: """듀얼 전략(모멘텀 돌파·눌림목) 스캔 1회 — 추천/MCP 경로와 독립 (H5 2026-10-02). 게이트(장 시간·BULL 전용·MDD·Crisis)는 scan_dual_strategy_signals 내부에 있고, 주문은 execute_auto_trade_background 의 일일 한도·리스크 가드를 그대로 탄다. 예외는 이 사이클의 평균회귀 경로를 막지 않도록 warning 으로 흡수한다. [3회차] 평균회귀 진입 다음에 호출된다(M-1). 루프의 AUTO_TRADE_ENABLED 게이트 뒤에서만 불리지만 단독 호출에도 안전하도록 여기서 한 번 더 확인한다(try 밖). 플레이북·지수 서킷브레이커·일일 한도 선확인은 scan_dual_strategy_signals 내부(H-1/M-1). MCP 리스크 평가 대상 아님(2026-10-02 사용자 결정, H-2). """ if not settings.AUTO_TRADE_ENABLED: logger.info("[Swing] AUTO_TRADE_ENABLED=False — 듀얼 전략 스캔 스킵") return try: # DB-READ-EXCEPTION: 듀얼 전략 후보 SELECT 전용(commit 없음) — CLAUDE.md §3.2 ③-b 스윙 스케줄러 SELECT async with AsyncSessionLocal() as dual_session: dual_results = await scan_dual_strategy_signals(dual_session) if dual_results: # [2026-09-27] scan_dual_strategy_signals() 가 위임하는 # execute_auto_trade_background() 는 항상 list 를 반환하며, # 그 list 에는 성공 BUY 뿐 아니라 주문 실패(success=False)한 # BUY/SELL, 청산 우선 로직으로 섞인 SELL 도 포함된다. # 따라서 len(dual_results) 를 그대로 "진입 n건"으로 쓰지 않고 # action/success 기준으로 분리 집계한다(상세 집계는 # entry_service.scan_dual_strategy_signals() 내부 로그 참조). buy_success = sum( 1 for r in dual_results if isinstance(r, dict) and r.get("action") == "BUY" and r.get("success") ) sell_count = sum( 1 for r in dual_results if isinstance(r, dict) and r.get("action") == "SELL" ) fail_count = sum( 1 for r in dual_results if isinstance(r, dict) and not r.get("success") ) logger.info( f"[Swing] 듀얼 전략 — 진입 {buy_success}건 / " f"청산 {sell_count}건 / 주문실패 {fail_count}건" ) except Exception as e: logger.warning(f"[Swing] 듀얼 전략 스캔 오류: {e}") async def _run_recommendation_entry_v2( get_json, get_rule_based_recommendations_from_db, request_mcp_risk_assessment, ) -> None: """평균회귀(추천 기반) 진입 1회 — 추천 조회 → AI 장전 필터 → MCP 평가 → 진입. [M-1 2026-10-02 3회차] swing_scheduler 루프 본문에서 분리했다. 조기 종료(추천 없음· 필터 후 0종목·MCP 거부)는 `continue` 가 아니라 `return` 이라, 호출부가 이어서 듀얼 스캔(_run_dual_strategy_scan_v2)을 항상 실행할 수 있다(H5 의도 유지). 인자는 순환 import 를 피하려고 루프에서 지연 import 한 함수들을 그대로 받는다. """ # ── Phase 3: 추천 조회 + 진입 실행 (별도 세션) ── # 4. Redis 캐시 확인 → cache miss 시 DB 조회 rec = await get_json("recommendations:latest") # [FIX] Redis 캐시가 존재해도 recommendations_by_rule이 비어있으면(배치가 빈 # 추천을 캐싱한 경우) 기존엔 '히트'로 처리해 DB 폴백을 건너뛰고 스킵했다. # 빈 캐시가 30분 TTL 동안 DB의 유효 후보를 가려 매매를 막으므로, 비어있으면 # 캐시 미스와 동일하게 DB 조회로 폴백한다. if rec and rec.get("recommendations_by_rule"): rec = _normalize_recommendations(rec) logger.info(f"[Swing] Redis 캐시 히트 (추천 {len(rec.get('recommendations_by_rule', []))}그룹)") else: logger.info("[Swing] Redis 캐시 미스/빈값 → DB 조회") # DB-READ-EXCEPTION: 추천 목록 SELECT 전용(commit 없음) — CLAUDE.md §3.2 ③-b 스윙 스케줄러 SELECT async with AsyncSessionLocal() as rec_session: rec = await get_rule_based_recommendations_from_db(rec_session) if not rec or not rec.get("recommendations_by_rule"): logger.info("[Swing] 추천 데이터 없음 — 스킵") return # 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 as _swallowed: logger.warning("[Swing] AI 장 전 분석 필터 적용 실패: %s", _swallowed) # 추천 목록에서 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()) top_stocks = [ item.get("stock_code", "") for items in recommendations.values() for item in (items[:3] if isinstance(items, list) else []) ][:10] else: rec_count = sum( len(g.get("stocks", [])) for g in recommendations if isinstance(g, dict) ) top_stocks = [ s.get("stock_code", "") for g in recommendations if isinstance(g, dict) for s in g.get("stocks", [])[:3] ][:10] if not top_stocks: logger.info("[Swing] 필터링 후 추천 종목 없음 — 스킵") return # MCP 리스크 평가 (Haiku — 실제 진입 직전 1회) trade_plan = { "action": "BUY", "strategy": "swing", "total_recommendations": rec_count, "top_stocks": top_stocks, } logger.info(f"[Swing] MCP 리스크 평가 요청: {len(top_stocks)}종목") mcp_approved, mcp_reason = await request_mcp_risk_assessment( trade_plan, {"strategy": "swing"}, {"top_stocks": top_stocks} ) logger.info(f"[Swing] MCP 결과: approved={mcp_approved}, reason={mcp_reason}") if not mcp_approved: logger.warning(f"[Swing] MCP 거부: {mcp_reason}") from app.services.db_worker import enqueue_db_write await enqueue_db_write({ "type": "log_mcp_deny", "reason": mcp_reason, "trade_plan": trade_plan, }) return # 6. 스윙 진입 실행 (별도 세션) # DB-READ-EXCEPTION: 진입 후보 조회 SELECT 전용(commit 없음, 주문 기록은 db_task_queue 경유) — CLAUDE.md §3.2 ③-b 스윙 스케줄러 SELECT async with AsyncSessionLocal() as entry_session: logger.info(f"[Swing] 진입 실행 시작: {top_stocks[:5]}") # execute_swing_entries() 내부에서 진입/청산/실패 건수를 이미 # 분리 집계해 로깅하므로(entry_service.py 참조), 여기서는 결과를 # 그대로 소비하지 않고 반환값(list, 2026-09-27 이후 항상 list)만 # 확인한다. await execute_swing_entries(rec, entry_session) # ── [하드닝 R9 2026-10-12] 청산 미종결 사이클의 MDD 축소·진입 생략 (redteam #8) ───────────── # 청산 판정이 상한 초과(TIMED_OUT)로 백그라운드에서 계속 돌거나, 이전 판정이 아직 진행 중이라 # 생략(SKIPPED)됐거나, 예외로 끝났으면 그 사이클의 MDD 축소·신규 진입을 하지 않는다. detached 청산과 # MDD 축소가 같은 종목을 동시에 매도하면 분할 단계 판정이 이중 적용될 수 있고, 청산 판정을 끝내지 # 못한 상태에서 매수를 내는 것은 '청산 우선'(§8.1-11) 위반이다. 청산 자체는 계속 실행된다 — # 다음 사이클이 청산 판정을 정상 완료하면 자동 재개(_inflight_steps 의 '청산 판정' 태스크 기준). _EXIT_STEP_NAME = "청산 판정" # [R9 재수정] 예외로 끝난 청산 단계는 태스크가 종료돼 동시 매도가 없으므로 **진입만** 생략하고 # MDD 축소는 실행한다. TIMED_OUT·SKIPPED·백그라운드 진행 중만 축소·진입 둘 다 생략. _DEFAULT_SWING_EXIT_UNSETTLED_ESCALATE_CYCLES = 3 # MDD 축소가 '청산 진행 중'으로 연속 생략된 사이클 수(정상 청산 사이클에서 리셋) _mdd_unsettled_skip_state: dict = {"count": 0, "escalated": False} def _exit_phase_unsettled_reason_v2(exit_res, raised: bool) -> str | None: """진입 생략 사유(예외 포함). 정상이면 None (순수 판정 + 인메모리 조회).""" if raised: return "청산 단계 예외" return _exit_inflight_reason_v2(exit_res) def _exit_inflight_reason_v2(exit_res) -> str | None: """청산 판정이 아직 돌고 있을 수 있는 사유(MDD 축소·진입 모두 생략). 아니면 None.""" if exit_res is _STEP_TIMED_OUT: return "청산 판정 상한 초과(백그라운드 진행 중)" if exit_res is _STEP_SKIPPED: return "이전 청산 판정이 아직 진행 중" _t = _inflight_steps.get(_EXIT_STEP_NAME) if _t is not None and not _t.done(): return "청산 판정 백그라운드 진행 중" return None def _skip_for_unsettled_exit_v2(reason: str | None, stage: str) -> bool: """reason 이 있으면 warning + (스로틀) 텔레그램 후 True — 호출부가 해당 단계를 생략한다.""" if not reason: return False msg = (f"[Swing] {stage} 생략 — {reason}. 같은 종목 이중 매도·청산 전 매수 방지, " f"청산 판정이 정상 완료되는 사이클에 재개") logger.warning(msg) _notify_throttled_v2(f"exit_unsettled:{stage}", "SWING_SCHEDULER", msg) return True def _note_mdd_reduction_unsettled_v2(skipped: bool) -> None: """MDD 축소 연속 생략 에스컬레이션. skipped=False 면 리셋. 축소를 강제 실행하지는 않는다 — 청산 판정이 백그라운드에서 같은 종목을 매도 중일 수 있어 이중 매도 위험이 축소 지연보다 크다. 대신 N사이클 연속이면 error + 텔레그램(스트릭당 1회). """ if not skipped: _mdd_unsettled_skip_state["count"] = 0 _mdd_unsettled_skip_state["escalated"] = False return _mdd_unsettled_skip_state["count"] += 1 n = _mdd_unsettled_skip_state["count"] limit = _cfg_int_v2( "SWING_EXIT_UNSETTLED_ESCALATE_CYCLES", _DEFAULT_SWING_EXIT_UNSETTLED_ESCALATE_CYCLES, ) if n >= limit and not _mdd_unsettled_skip_state["escalated"]: _mdd_unsettled_skip_state["escalated"] = True msg = (f"MDD 축소 {n}사이클 연속 미실행(청산 판정 미종결, 기준 {limit}사이클) — 수동 확인 필요. " f"이중 매도 방지로 축소를 강제하지 않음") logger.error(f"[Swing] 🚨 {msg}") _notify_throttled_v2("exit_unsettled:mdd_escalate", "SWING_SCHEDULER", msg) async def swing_scheduler() -> None: """60초 주기 스윙 트레이딩 스케줄러. 무한 루프로 실행되며 asyncio.CancelledError로만 종료. 각 사이클에서 예외 발생 시 로그 기록 후 다음 사이클 계속. DB 세션은 사이클마다 AsyncSessionLocal 로 새로 생성하여 장시간 세션 유지로 인한 커넥션 풀 고갈 방지. """ await asyncio.sleep(2) from app.api.endpoints.recommendations import ( get_rule_based_recommendations_from_db, ) from app.api.endpoints.mcp import request_mcp_risk_assessment from app.core.redis import get_json logger.info("📈 [Swing] Scheduler started (every 60s)") _check_telegram_configured_v2() try: await _check_mdd_threshold_knobs_v2() except Exception as e: logger.warning(f"[Swing] MDD 임계 설정 대조 실패: {type(e).__name__}: {e}") _mdd_update_failures = 0 _mdd_last_error: str | None = None # [M3] '정상 응답이지만 총평가금액 0'(빈/미입금 계좌) 누적 횟수 — 장애 카운터와 분리 _mdd_empty_account_seen = 0 try: while True: try: logger.info("[Swing] ── 사이클 시작 ──") # [하드닝 R1] 사이클 하트비트 — 반복 시작마다 찍는다(어느 단계에서 멈추든 # main.py 감시 루프가 장중 정체로 잡도록). 기록 실패가 루프를 막지 않게 흡수. try: from app.core import heartbeat as _hb _hb.beat("swing", "cycle_start") except Exception as _hb_e: logger.warning(f"[Swing] 사이클 하트비트 기록 실패: {type(_hb_e).__name__}: {_hb_e}") if not is_market_open(): _note_market_closed_v2() logger.info("[Swing] 장 미오픈 — 개장까지 대기") # 개장을 넘겨 자지 않도록 상한 (2026-09-23) from app.core.market_schedule import seconds_until_market_open _until_open = seconds_until_market_open() _closed_sleep = 1800.0 if _until_open is None else max(15.0, min(1800.0, _until_open)) # [하드닝 R4 2026-10-09] 종가 동시호가(15:20~15:30) 주문은 15:30 에 체결된다 — # 장 마감 직후 창에서 체결가 후속 보정(swing/fill_sweep). 창 안이면 대기를 줄인다. try: from app.services.swing.fill_sweep import run_post_close_fill_sweep_v2 _pc_sleep = await run_post_close_fill_sweep_v2() if _pc_sleep is not None: _closed_sleep = min(_closed_sleep, float(_pc_sleep)) except Exception as _pc_e: logger.warning(f"[Swing] 장 마감 후 체결가 보정 실패: {type(_pc_e).__name__}: {_pc_e}") await asyncio.sleep(_closed_sleep) continue # ── Phase 1: 청산 체크 (별도 세션, _run_exit_phase_v2) ── # [하드닝 R1] 비취소형 상한(SWING_EXIT_STEP_TIMEOUT_SEC). 초과 시 판정 작업은 # 백그라운드로 계속되고 루프는 MDD 단계로 진행한다. 끝날 때까지 새 청산 판정을 # 시작하지 않는다 — 취소 후 재시작하면 shield 로 계속 진행 중인 매도와 새 판정의 # 매도가 같은 종목에 겹쳐 분할 매도가 이중 실행될 수 있다(과다매도). _note_market_open_v2(_now_mono_v2()) _exit_timeout = _cfg_float_v2("SWING_EXIT_STEP_TIMEOUT_SEC", _DEFAULT_EXIT_STEP_TIMEOUT_SEC) _exit_res = None _exit_raised = False try: _exit_res = await _guard_detached_v2(_run_exit_phase_v2, "청산 판정", _exit_timeout) if _exit_res is _STEP_TIMED_OUT: _record_exit_failure_v2(f"청산 판정 {_exit_timeout:.0f}초 초과", timed_out=True) elif _exit_res is _STEP_SKIPPED: _record_exit_failure_v2("이전 청산 판정이 아직 진행 중(상한 초과 상태)") except Exception as e: _exit_raised = True logger.error(f"[Swing] 청산 단계 오류: {e}", exc_info=True) _record_exit_failure_v2(f"{type(e).__name__}: {e}") # [하드닝 R9] 청산 미종결이면 이번 사이클 MDD 축소·진입 생략(판정은 try 밖 — 가드) _exit_unsettled = _exit_phase_unsettled_reason_v2(_exit_res, _exit_raised) _exit_inflight = _exit_inflight_reason_v2(_exit_res) if _exit_inflight is None: _note_mdd_reduction_unsettled_v2(False) # 청산 판정 종결(정상/예외) → 리셋 try: from app.core import heartbeat as _hb _hb.beat("swing", "exit_phase_done") except Exception as _hb_e: logger.warning(f"[Swing] 사이클 하트비트 기록 실패: {type(_hb_e).__name__}: {_hb_e}") # 청산 판정 무완료 감시(장중, 개장 시각 기준) — try 밖(가드) _check_exit_staleness_v2(_now_mono_v2()) # [하드닝 R8] 체결가 후속 보정 — 청산 판정 상한과 분리된 취소형 단계(내부에서 예외 흡수) await _run_fill_sweep_phase_v2() # ── Phase 2: MDD 업데이트 + 체크 + 강제 축소 ── # [FIX] mdd_manager.update()를 매 사이클 호출하여 MDD 감지 활성화 # [M2 2026-09-27] 여기가 mdd_manager.update() 의 유일한 실효 호출부다. # 실패를 debug 로 삼키면 KIS 장애 시 peak/drawdown 이 갱신되지 않아 # -10%/-20% 게이트가 무증상으로 영구 미발동한다 → 경고 + 연속 실패 카운트. # [M3 2026-09-27] "응답 없음/파싱 실패"(진짜 장애)와 "정상 응답 + 총평가금액 # 0"(빈/미입금 계좌)을 구분한다. 후자는 정상 응답이므로 실패로 계수하지 않고 # MDD 갱신만 스킵한다 — 그러지 않으면 신규 모의계좌에서 매 사이클 실패가 # 누적돼 🚨 error 로그가 계속 찍혀 실제 KIS 장애와 구분되지 않는다. _empty_account = False _mdd_stage_pending = False try: from app.services.korea_investment_api import korea_investment_api # [하드닝 R1] 읽기 전용 → 취소형 상한. 초과는 TimeoutError 로 아래 실패 경로를 탄다. balance_resp = await asyncio.wait_for( korea_investment_api.get_balance(), timeout=_cfg_float_v2("SWING_BALANCE_TIMEOUT_SEC", _DEFAULT_BALANCE_TIMEOUT_SEC), ) if not balance_resp: raise ValueError("잔고 조회 응답 없음 (resp=없음)") _out2 = balance_resp.get("output2") if not isinstance(_out2, list) or not _out2: raise ValueError("잔고 응답 파싱 실패 (output2 누락/형식 이상)") _raw_total = (_out2[0] or {}).get("tot_evlu_amt") if _raw_total is None or str(_raw_total).strip() == "": raise ValueError("잔고 응답 파싱 실패 (tot_evlu_amt 누락)") try: _total = float(_raw_total) except (TypeError, ValueError): raise ValueError(f"잔고 응답 파싱 실패 (tot_evlu_amt={_raw_total!r})") if _total <= 0 and _mdd_has_known_balance_v2(): # [하드닝 R1-라운드1] 이미 자산을 관측한(peak/last>0) 운용 계좌에서 0 은 # '빈 계좌'가 아니라 부분/오류 응답으로 본다 → 실패 경로(fail-safe 계수). # 빈 계좌로 분류하면 blind 차단·staleness 가 모두 해제돼 실제 -10%/-20% # 구간에서도 신규 진입이 열린다. (실제 전액 출금이면 자산 0 이라 매수 # 자체가 불가하므로 차단 비용이 없다.) raise ValueError( f"잔고 응답 이상 (tot_evlu_amt={_total:,.0f} 이나 기존 peak=" f"{float(getattr(mdd_manager, 'peak_balance', 0.0) or 0.0):,.0f} — " f"빈 계좌로 보지 않고 갱신 실패 처리)" ) if _total <= 0: # 정상 응답이지만 자산 0 — 빈 계좌/미입금(자산 관측 이력 없음). 장애 아님. _empty_account = True _mdd_empty_account_seen += 1 if _mdd_empty_account_seen == 1 or _mdd_empty_account_seen % 60 == 0: logger.info( f"[Swing] 잔고 총평가금액 0 (빈/미입금 계좌) — MDD 갱신 스킵 " f"({_mdd_empty_account_seen}회 누적)" ) else: _mdd_empty_account_seen = 0 if _empty_account: # 정상 응답(장애 아님) — 실패 카운터 리셋 + 빈 계좌 표시. # 빈 계좌는 진입 차단 사유가 아니다(항목7). await mdd_manager.record_update_success(empty_account=True) if _total > 0: # [하드닝 R2] 입출금 보정 + 단계 전이 확인(부분응답 의심은 ValueError → 실패 경로) _guarded = await _mdd_update_guarded_v2(_total, balance_resp) _mdd_stage_pending = _guarded["status"] == "pending" mdd_result = _guarded["result"] or {"action": "PENDING_CONFIRM"} # [항목7] 갱신 성공 → fail-safe 차단 해제 await mdd_manager.record_update_success(empty_account=False) if _mdd_update_failures: logger.warning( f"[Swing] MDD 갱신 복구 (연속 실패 {_mdd_update_failures}회 후 정상화)" ) _mdd_update_failures = 0 _mdd_last_error = None if mdd_result.get("action") != "NONE": logger.warning(f"[Swing] MDD 상태 변경: {mdd_result}") except Exception as e: _mdd_update_failures += 1 _mdd_last_error = str(e) # [항목7] 실패를 mdd_manager 에 보고 → 임계 도달 시 # can_open_new_position() 이 모든 진입 경로를 차단한다(청산은 무영향). await mdd_manager.record_update_failure(_mdd_last_error) if _mdd_update_failures >= MDD_UPDATE_FAILURE_ALERT: logger.error( f"[Swing] 🚨 MDD 갱신 연속 실패 {_mdd_update_failures}회 — " f"peak/drawdown 미갱신 상태, -10%/-20% 게이트가 작동하지 않을 수 있음: {e}", exc_info=True, ) else: logger.warning( f"[Swing] MDD 갱신 실패 ({_mdd_update_failures}회 연속): {e}" ) # 상태 노출용 (모니터링/헬스체크에서 참조 가능) _mdd_health["consecutive_failures"] = _mdd_update_failures _mdd_health["last_error"] = _mdd_last_error else: _mdd_health["consecutive_failures"] = _mdd_update_failures _mdd_health["last_error"] = _mdd_last_error _mdd_health["empty_account"] = _empty_account can_open_new = await mdd_manager.can_open_new_position() # [하드닝 R2] MDD 단계 전이 확인 대기 중엔 신규 진입 차단 (try 밖 — 가드) if _mdd_stage_pending: can_open_new = False logger.info(f"[Swing] MDD 체크: can_open={can_open_new}") # [하드닝 R1] fail-safe 진입 차단 진입/해제 경보 (판정 자체는 위 메서드가 소유) _check_blind_block_transition_v2() if not can_open_new: if mdd_manager.get_update_health().get("blind_block_active"): logger.error( f"[Swing] 🚨 MDD 갱신 장애로 신규 진입 차단(fail-safe, 연속 실패 " f"{_mdd_update_failures}회, 마지막 오류: {_mdd_last_error}) — " f"청산/축소는 계속 수행" ) else: logger.warning("[Swing] MDD 초과 — 신규 진입 차단") _mdd_skip = bool(mdd_manager.reduction_applied) and _skip_for_unsettled_exit_v2( _exit_inflight, "MDD 축소", ) if _mdd_skip: _note_mdd_reduction_unsettled_v2(True) if mdd_manager.reduction_applied and not _mdd_skip: try: reduce_pct = 0.50 if mdd_manager.reduction_applied == "20pct" else 0.30 mdd_action = { "action": "FORCE_REDUCE" if mdd_manager.reduction_applied == "20pct" else "REDUCE_AND_PAUSE", "reduce_pct": reduce_pct, "prioritize_leverage": mdd_manager.reduction_applied == "20pct", } # [하드닝 R1] 주문 경로라 비취소형 상한 — 초과 시 백그라운드로 남기고 # 끝날 때까지 새 축소를 시작하지 않는다(이중 축소 방지). results = await _guard_detached_v2( lambda: mdd_manager.execute_reduction(mdd_action), "MDD 축소", _cfg_float_v2("SWING_MDD_REDUCTION_TIMEOUT_SEC", _DEFAULT_MDD_REDUCTION_TIMEOUT_SEC), ) if results is _STEP_TIMED_OUT or results is _STEP_SKIPPED: results = None if results: logger.warning(f"[Swing] MDD 강제 축소 완료: {len(results)}종목") except Exception as e: logger.error(f"[Swing] MDD 강제 축소 실패: {e}", exc_info=True) await asyncio.sleep(60) continue # 3. 자동매매 활성화 여부 if not settings.AUTO_TRADE_ENABLED: logger.info("[Swing] AUTO_TRADE_ENABLED=False — 스킵") await asyncio.sleep(60) continue # ── Phase 3: 평균회귀(추천 기반) 진입 → 듀얼 전략 스캔 ── # [M-1 2026-10-02 3회차] 듀얼 스캔을 평균회귀 진입 **다음**으로 옮겼다. # 일일 매수 한도(MAX_DAILY_TRADES)를 두 경로가 공유하므로, 듀얼이 먼저 돌면 # 평균회귀 후보가 슬롯을 잃었다. 평균회귀 단계의 조기 종료(추천 없음·필터 후 # 0종목·MCP 거부)·예외는 듀얼을 건너뛰지 않는다(H5 의도 유지) — 헬퍼가 # continue 대신 return 하고, 예외는 여기서 흡수한다. 한도가 이미 찼으면 # scan_dual_strategy_signals 가 일봉 조회 전에 생략한다. # [하드닝 R1] 진입(평균회귀→듀얼)은 주문 경로라 비취소형 상한. 초과 시 루프는 # 다음 사이클(청산 우선)로 진행하고, 끝날 때까지 새 진입을 시작하지 않는다. if _skip_for_unsettled_exit_v2(_exit_unsettled, "진입"): await asyncio.sleep(60) continue await _guard_detached_v2( lambda: _run_entry_phase_v2( get_json, get_rule_based_recommendations_from_db, request_mcp_risk_assessment, ), "진입", _cfg_float_v2("SWING_ENTRY_PHASE_TIMEOUT_SEC", _DEFAULT_ENTRY_PHASE_TIMEOUT_SEC), ) logger.info("[Swing] ── 사이클 완료 ──") except Exception as e: logger.error(f"[Swing] 스케줄러 사이클 오류: {e}", exc_info=True) await asyncio.sleep(60) except asyncio.CancelledError: logger.info("[Swing] Scheduler stopped") raise