"""데이트레이딩 인메모리 포지션 상태 관리 기존 day_trading.py의 전역 변수들을 DayTradingState 클래스로 캡슐화. 싱글톤 인스턴스 day_trading_state 를 패키지 전체에서 공유. CLAUDE.md 3.1: 실시간 레이어에서 DB I/O 절대 금지. 포지션 상태는 인메모리 유지, 체결 확정 시에만 db_task_queue 경유. 사용 패턴: from app.services.daytrading.position_manager import day_trading_state # 날짜 초기화 (스케줄러 사이클 시작 시 호출) day_trading_state.reset_if_new_day() # 포지션 추가/제거 시 반드시 trade_lock 획득 async with day_trading_state.trade_lock: day_trading_state.add_position(code, data) 주의사항: - [동시성] positions, daily_pnl, closed_today 수정 시 반드시 trade_lock 획득 - [영속성] 앱 재시작 시 restore_from_db()/load_orb_cache() 로 복원한다. 복원 직전에 반드시 begin_restore() 를 호출할 것 — 호출하지 않으면 첫 reset_if_new_day() 가 복원분을 전부 지운다(2026-09-28 수정). - [단일 프로세스] 다중 워커 환경에서는 상태 불일치 발생 가능 """ import asyncio import logging from datetime import date, datetime from typing import Any import pytz from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # 한국 표준시(KST) 타임존 객체 — 날짜 비교 및 시간 계산에 사용 _KST = pytz.timezone("Asia/Seoul") # [R2 2026-10-09] 일일 안전장치 원장 재구성 (Redis 복원 실패 시) — 운영 상수(전략 파라미터 아님) _DAILY_STATE_REBUILD_DB_TIMEOUT_SEC = 15.0 _DAILY_STATE_REBUILD_RETRY_SEC = 120.0 _DAILY_STATE_NOTIFY_TIMEOUT_SEC = 10.0 # 전일·고아 청산의 exit_reason 접두사(daytrading/exit_service.close_stale_positions). # 이 청산 손익은 당일 실현손익·연패에 넣지 않는다(§7, count_daily_pnl=False) — 재구성도 동일. # 문자열이 바뀌면 test_dt_daily_state_rebuild_r2_20261009 의 동기화 테스트가 깨진다. _STALE_EXIT_REASON_PREFIXES = ("전일 미청산 포지션 강제 청산", "DT 고아 잔고 청산") _DT_STRATEGY_TYPE = "daytrading" _VOID_STATUSES = frozenset({"CANCELLED", "FAILED", "REJECTED"}) _daily_state_retry_task: "asyncio.Task | None" = None def _today_kst() -> date: """현재 KST 기준 날짜 반환. Returns: date: 오늘 날짜 (KST 기준) """ return datetime.now(_KST).date() class DayTradingState: """데이트레이딩 일일 상태 컨테이너. 날짜가 바뀌면 reset_if_new_day()로 자동 초기화. trade_lock 획득 후 positions / daily_pnl / closed_today 수정. Attributes: positions: 활성 포지션 딕셔너리 {종목코드: 포지션정보} orb_cache: 시초가 레인지(ORB) 캐시 {종목코드: OpeningRange} daily_pnl: 당일 누적 실현 손익 (원) last_total_assets: 마지막 조회된 총자산 (포지션 사이징 기준) closed_today: 당일 청산 완료된 종목 코드 집합 (재진입 방지) loss_streak_today: 당일 연속 손실 횟수 (3회 이상 시 쿨다운) trade_lock: 포지션 수정 동시성 보호용 asyncio.Lock """ def __init__(self): """DayTradingState 초기화. 모든 상태를 기본값으로 설정. 실제 초기화는 reset_if_new_day()에서 수행. """ # 활성 포지션: {stock_code: {...entry info...}} self.positions: dict[str, dict[str, Any]] = {} # ORB 캐시: {stock_code: OpeningRange} self.orb_cache: dict[str, Any] = {} # 당일 실현 손익 self.daily_pnl: float = 0.0 # 마지막으로 조회된 총자산 self.last_total_assets: float = 0.0 # 당일 청산 종목 (재진입 방지) # ⚠️ 스윙 손절(swing/exit_service)·MDD 등 **타 전략 청산도** 전략 간 당일 쿨다운으로 # 여기에 들어온다 → "DT 청산 수"로 세면 안 된다(아래 dt_exited_today 사용). self.closed_today: set[str] = set() # [2026-10-08] 당일 **DT 포지션이 전량 청산된** 종목 (표시·집계 전용, 재진입 판정에 쓰지 않음) # DT 엔진 카드 '청산 N' 에 스윙 손절 종목이 섞이던 문제(10/08: DT 6 + 스윙 2 = 8). self.dt_exited_today: set[str] = set() # [R2 하드닝 2026-10-09] 당일 DT 전량 청산(dt_exited_today) 뒤 **보유가 재개된** 종목. # 전량 매도가 0체결·부분체결로 종결돼 exit_service._on_unfilled 가 포지션을 되살린 경우 # (add_position 이 dt_exited_today 종목을 받는 유일한 경로 — DT 는 당일 재진입 금지)에 # 표시한다. 기동 복원 검증·장중 복원이 이 종목을 '청산 완료'로 보지 않게 한다. # remove_position(다시 전량 청산)에서 해제된다. 재진입 차단(closed_today)과 무관. self.dt_reheld_today: set[str] = set() # [R2 하드닝] 구버전 Redis 페이로드(dt_exited_today 키 없음)로 복원된 closed_today — # DT 청산과 재진입 차단을 구분할 수 없으므로 복원 판정에서만 '청산 완료'로 간주(보수적). self.restore_block_legacy: set[str] = set() # 당일 연속 손실 횟수 (쿨다운 트리거) self.loss_streak_today: int = 0 # 마지막 초기화 날짜 self.last_reset_date: date | None = None # 포트폴리오 요약 캐시 (총자산, 현금 등) self.portfolio_summary_cache: dict[str, float] = {} # 포트폴리오 캐시 타임스탬프 self.portfolio_summary_ts: datetime | None = None # 포트폴리오 캐시 무효화 플래그 self.portfolio_cache_dirty: bool = True # 매매 실행 락 — positions 수정 시 반드시 획득 self.trade_lock = asyncio.Lock() # [H-B 2026-10-02] 청산 주문 진행 중 종목. exit_service._execute_exit_impl 이 # trade_lock 안에서 등록하고 shield 된 주문 작업의 finally 에서 해제한다. # 청산 판정 경로(손절 선판정·분봉 판정·폴백 손절·force_close_all)는 이 종목을 건너뛴다 # — 주문은 진행 중인데 인메모리 수량이 아직 그대로라 같은 물량을 다시 파는 것을 막는다. # 날짜 초기화(reset_if_new_day)로 지우지 않는다(진행 중 주문의 보호가 풀린다). self.exits_in_progress: set[str] = set() # [R2 2026-10-09] 일일 안전장치 **미검증** 상태 — Redis 복원 실패 + TradeHistory 재구성 # 실패(또는 손실 상태인데 총자산 분모 미확보)면 True. is_cooldown_active() 가 True 를 # 반환해 **DT 신규 진입만** 막는다(청산 경로는 이 값을 보지 않는다 — §8.1-11). # 재구성 재시도가 성공하거나 날짜가 바뀌면(reset_if_new_day) 해제된다. self.daily_state_unverified: bool = False self.daily_state_unverified_reason: str = "" # 일일 안전장치의 출처: "" | "redis" | "trade_history" (표시·진단용) self.daily_state_source: str = "" def reset_if_new_day(self) -> bool: """날짜가 바뀌면 일일 상태 초기화. 스케줄러 사이클 시작 시 호출하여 전일 데이터가 당일 매매에 영향을 주지 않도록 보장. Returns: bool: 초기화가 수행되었으면 True, 아니면 False """ today = _today_kst() if self.last_reset_date != today: self.positions = {} self.orb_cache = {} self.daily_pnl = 0.0 self.last_total_assets = 0.0 self.closed_today = set() self.dt_exited_today = set() self.dt_reheld_today = set() self.restore_block_legacy = set() self.loss_streak_today = 0 self.last_reset_date = today # [R2] 미검증 차단은 **그날의** 상태에 대한 것 — 새 날은 0 에서 시작하는 것이 정상 self.daily_state_unverified = False self.daily_state_unverified_reason = "" self.daily_state_source = "" self.portfolio_summary_cache = {} self.portfolio_summary_ts = None self.portfolio_cache_dirty = True logger.info(f"[DayTrading] 일일 상태 초기화: {today}") return True return False def begin_restore(self) -> bool: """기동 시 복원(load_orb_cache/restore_from_db) **직전에 1회** 호출. [2026-09-28 재기동 내구성 수정] 새 프로세스는 ``last_reset_date is None`` 이고 복원 경로 3개 (load_orb_cache / restore_from_db / close_stale_positions) 중 어느 것도 이 필드를 설정하지 않았다. 그래서 스케줄러 첫 루프의 ``reset_if_new_day()`` 가 무조건 참이 되어 **방금 복원한 positions/orb_cache 를 전부 지웠다**(복원 성공 로그는 그대로 출력되므로 조용히 실패). 손절·트레일링·시간손절·15:20 EOD 청산이 인메모리 positions 만 순회하므로 재기동한 당일의 포지션이 고아가 됐다. 복원 **전에** 날짜 경계를 확정하면 두 불변식을 동시에 만족한다: 1) 같은 날 첫 루프의 reset 은 no-op → 복원분 생존 2) 자정을 넘기면 ``last_reset_date != today`` → 정상 초기화되어 전일 closed_today / daily_pnl / loss_streak_today 가 새지 않는다 (§7 일일 -3% 한도·3연패 쿨다운·재진입 차단 보호) Returns: bool: 날짜 경계 확정 과정에서 초기화가 수행되었으면 True """ return self.reset_if_new_day() def add_position(self, stock_code: str, data: dict) -> None: """포지션 추가. 반드시 trade_lock 획득 후 호출해야 함. Args: stock_code: 종목 코드 (예: "005930") data: 포지션 정보 딕셔너리 - stock_name: 종목명 - strategy: 진입 전략 (ORB/VWAP/RSI_BB) - entry_price: 진입가 - stop_loss: 손절가 - target_price: 목표가 - quantity: 수량 - highest_price: 최고가 (트레일링스탑용) - entry_time: 진입 시간 - order_id: KIS 주문 ID """ self.positions[stock_code] = data self.portfolio_cache_dirty = True # [R2 하드닝] 당일 DT 전량 청산 기록이 있는 종목이 다시 들어오면 = 매도 미체결 종결 복원 # (closed_today 로 신규 진입은 막혀 있다). 재기동 복원 검증이 '청산 완료'로 오판하지 않게 표시. if stock_code in self.dt_exited_today or stock_code in self.restore_block_legacy: self.dt_reheld_today.add(stock_code) self.restore_block_legacy.discard(stock_code) def remove_position(self, stock_code: str) -> dict | None: """포지션 제거 및 반환. 반드시 trade_lock 획득 후 호출해야 함. 제거된 종목은 closed_today에 추가되어 당일 재진입 방지. Args: stock_code: 종목 코드 Returns: dict | None: 제거된 포지션 정보, 없으면 None """ pos = self.positions.pop(stock_code, None) if pos is not None: self.closed_today.add(stock_code) self.dt_exited_today.add(stock_code) self.dt_reheld_today.discard(stock_code) self.portfolio_cache_dirty = True return pos def restore_block_reason_v2(self, stock_code: str) -> str | None: """[R2 하드닝 2026-10-09] 당일 DT active 행을 인메모리로 **되살리면 안 되는** 사유(없으면 None). closed_today 는 **재진입 차단** 집합이다 — 불명 매수 등록(entry_service)·스윙 손절(전략 간 쿨다운)·매도 0체결 종결 뒤에도 남아 있어 '청산 완료'의 근거가 될 수 없다. 예전에는 closed_today 만으로 복원을 막아 실제 보유 중인 당일 DT 포지션이 손절·EOD 관리 밖으로 빠졌다(오버나이트). 이제 '청산 완료'는 DT 전량 청산(dt_exited_today) 기준이고, 그 뒤 보유가 재개된 종목(dt_reheld_today)은 제외한다. 나머지는 호출부가 KIS 실보유 − 타전략 수량(§3.5) 규칙으로 판정한다(이 메서드는 그 판정을 대신하지 않는다). """ if stock_code in self.exits_in_progress: return "청산 주문 진행/결과 불명 보류 중(exits_in_progress)" if stock_code in self.dt_reheld_today: return None if stock_code in self.dt_exited_today: return "당일 DT 전량 청산 기록(dt_exited_today) 있음" if stock_code in self.restore_block_legacy: return "구버전 일일상태(dt_exited_today 없음) — closed_today 를 청산 근거로 간주" return None def update_pnl(self, realized: float, is_full_close: bool = True) -> None: """실현 손익 누적 및 연속 손실 카운트 업데이트. 손실 시 loss_streak_today 증가, 이익 시 리셋. 3연패 이상이면 is_cooldown_active()가 True 반환. 연속 손실 카운트는 전량 매도(is_full_close=True)일 때만 업데이트. 분할 매도(is_full_close=False)나 손익분기(realized==0)는 카운트에 영향 없음. Args: realized: 실현 손익 (원). 양수=이익, 음수=손실 is_full_close: 전량 매도 여부. False이면 streak 카운트 미변경 """ self.daily_pnl += realized if not is_full_close: return if realized < 0: self.loss_streak_today += 1 elif realized > 0: # 이익 발생 시 연속 손실 카운트 리셋 (손익분기 realized==0은 중립 — 리셋하지 않음) self.loss_streak_today = 0 def adjust_realized_pnl(self, delta: float) -> None: """[2026-10-01] 실현손익 사후 보정 (체결가 확인으로 기록 손익이 바뀐 경우). 반드시 trade_lock 획득 후 호출해야 함. daily_pnl 만 보정하고 loss_streak_today 는 건드리지 않는다 — 연패 판정은 청산 시점에 한 번만 한다(같은 청산을 두 번 세지 않기 위함). """ self.daily_pnl += float(delta) def is_exit_in_progress(self, stock_code: str) -> bool: """[H-B] 해당 종목의 청산 주문이 진행 중(또는 결과 불명 보류 중)인지.""" return stock_code in self.exits_in_progress def get_position(self, stock_code: str) -> dict | None: """특정 종목의 활성 포지션 정보 반환. Args: stock_code: 종목 코드 Returns: dict | None: 포지션 정보 딕셔너리, 없으면 None """ return self.positions.get(stock_code) def is_closed_today(self, stock_code: str) -> bool: """해당 종목이 당일 이미 청산되었는지 확인. 데이트레이딩은 동일 종목 당일 재진입을 허용하지 않음. Args: stock_code: 종목 코드 Returns: bool: 당일 청산 완료 여부 """ return stock_code in self.closed_today def position_count(self) -> int: """현재 활성 포지션 수 반환. MAX_POSITIONS(3) 한도 체크에 사용. Returns: int: 활성 포지션 수 """ return len(self.positions) def is_cooldown_active(self) -> bool: """당일 3연패 이상이면 쿨다운 (신규 진입 차단). CLAUDE.md 7.3: 당일 3연패 시 신규 진입 차단. [R2 2026-10-09] 일일 안전장치가 미검증(daily_state_unverified)이어도 True — Redis·TradeHistory 어느 쪽으로도 당일 손익·연패·재진입 차단을 복원하지 못한 상태에서 0 부터 매매를 재개하면 -3% 한도·3연패 쿨다운이 풀린다(fail-open). 진입 게이트 (entry_service.scan_and_enter / scheduler._entry_blocked_by_exit_results)가 이 메서드를 쓰므로 시그니처를 바꾸지 않고 여기서 함께 막는다. 청산 경로는 무관. Returns: bool: 쿨다운 활성화 여부 """ if self.daily_state_unverified: return True return self.loss_streak_today >= 3 def has_orb(self, stock_code: str) -> bool: """해당 종목의 ORB(시초가 레인지)가 수집되었는지 확인. Args: stock_code: 종목 코드 Returns: bool: ORB 데이터 존재 여부 """ return stock_code in self.orb_cache async def save_orb_cache(self) -> bool: """ORB 캐시를 Redis 에 저장 (당일 키, 자정까지 TTL). 컨테이너 재기동 시 인메모리 캐시가 사라져 당일 ORB 전략이 불가능해지는 것을 막는다. """ try: from app.core.redis import setex_json payload = { code: { "high": float(getattr(o, "high", 0) or 0), "low": float(getattr(o, "low", 0) or 0), "total_volume": int(getattr(o, "total_volume", 0) or 0), "avg_vol_per_min": float(getattr(o, "avg_vol_per_min", 0) or 0), "open_price": float(getattr(o, "open_price", 0) or 0), } for code, o in self.orb_cache.items() } if not payload: return False return await setex_json(f"daytrading:orb:{_today_kst()}", 86400, payload) except Exception as e: logger.warning(f"[DayTrading] ORB 캐시 저장 실패: {e}") return False async def load_orb_cache(self) -> int: """Redis 에서 당일 ORB 캐시 복원. 복원한 종목 수 반환.""" try: from app.core.redis import get_json from app.services.day_trading_strategy import OpeningRange data = await get_json(f"daytrading:orb:{_today_kst()}") if not data: return 0 restored = 0 for code, o in data.items(): if code in self.orb_cache: continue self.orb_cache[code] = OpeningRange( high=float(o.get("high", 0) or 0), low=float(o.get("low", 0) or 0), total_volume=int(o.get("total_volume", 0) or 0), avg_vol_per_min=float(o.get("avg_vol_per_min", 0) or 0), open_price=float(o.get("open_price", 0) or 0), ) restored += 1 if restored: logger.info(f"[DayTrading] ORB 캐시 복원: {restored}개 종목") return restored except Exception as e: logger.warning(f"[DayTrading] ORB 캐시 복원 실패: {e}") return 0 async def save_daily_state(self) -> bool: """[F9 2026-09-28] 일일 안전장치 3종을 Redis 에 저장 (당일 키, 자정까지 TTL). ``daily_pnl`` / ``loss_streak_today`` / ``closed_today`` 는 어디에도 영속화되지 않아 **장중 재기동 시 전부 0/빈 집합으로 돌아갔다.** 이것은 단순 상태 유실이 아니라 **안전장치가 풀리는 방향**이다: - ``daily_pnl=0`` → §7 일일 -3% 손실한도(DAY_TRADE_DAILY_LOSS_LIMIT)가 리셋되어, 이미 한도를 맞고 중단됐어야 할 날에 매매가 재개된다 - ``loss_streak_today=0`` → 당일 3연패 쿨다운이 풀린다 - ``closed_today=∅`` → 손절로 나온 종목을 같은 날 다시 매수한다 저장 포맷에 ``date`` 를 실어 ``load_daily_state()`` 가 **다른 날짜면 버린다** (전일 값이 다음 날로 새면 F1 불변식 ②를 정면으로 깨뜨린다). ORB 캐시(save_orb_cache)와 동일한 Redis 경로를 그대로 확장한 것이다. """ try: from app.core.redis import setex_json today = _today_kst() payload = { "date": today.isoformat(), "daily_pnl": float(self.daily_pnl), "loss_streak_today": int(self.loss_streak_today), "closed_today": sorted(self.closed_today), # [2026-10-08] DT 전용 청산 집합(표시용) — 없으면 구버전 페이로드 "dt_exited_today": sorted(self.dt_exited_today), # [R2 하드닝] 청산 뒤 보유 재개(매도 미체결 종결 복원) — 재기동 복원 검증용 "dt_reheld_today": sorted(self.dt_reheld_today), "restore_block_legacy": sorted(self.restore_block_legacy), # [C2 2026-09-28] 손실률 분모. 이 값이 0 이면 # is_daily_loss_exceeded() 가 total_assets<=0 에서 **무조건 False** 라 # 재기동 직후 첫 scan_and_enter() 가 일일 -3% 한도를 통과해버린다 # (scan_and_enter 는 잔고 조회 전에 이 필드를 읽는다). "last_total_assets": float(self.last_total_assets), } ok = await setex_json(f"daytrading:daily:{today}", 86400, payload) if not ok: logger.warning( "[DayTrading] ⚠️ 일일 안전장치 저장 실패(Redis) — 재기동 시 " "일일 손실한도·연패 쿨다운·재진입 차단이 초기화된다" ) return ok except Exception as e: logger.warning(f"[DayTrading] ⚠️ 일일 안전장치 저장 실패: {e}") return False async def load_daily_state(self) -> bool: """[F9] Redis 에서 당일 안전장치 3종을 복원. 복원했으면 True. **날짜 게이트**: 저장된 ``date`` 가 오늘이 아니면 복원하지 않고 버린다. 키 자체에도 날짜가 들어 있지만(자정 TTL), 페이로드 날짜를 한 번 더 확인해 키 충돌·시계 오차·수동 주입으로 전일 값이 새는 경로를 이중으로 막는다. 복원값은 현재 인메모리 값과 **병합하지 않고 더 보수적인 쪽을 취한다** (손익은 합산이 아니라 저장값 채택, closed_today 는 합집합, 연패는 최대값) — 재기동 직후 인메모리는 비어 있는 것이 정상이고, 만약 비어 있지 않다면 안전장치를 약화시키지 않는 방향을 골라야 한다. """ try: from app.core.redis import get_json today = _today_kst() data = await get_json(f"daytrading:daily:{today}") if not data: return False saved_date = str(data.get("date") or "") if saved_date != today.isoformat(): logger.warning( f"[DayTrading] 일일 안전장치 복원 건너뜀: 저장 날짜 {saved_date!r} " f"≠ 오늘 {today.isoformat()} — 전일 값 누수 방지(폐기)" ) return False _pnl = float(data.get("daily_pnl", 0) or 0) _streak = int(data.get("loss_streak_today", 0) or 0) _closed = {str(c) for c in (data.get("closed_today") or []) if c} _assets = float(data.get("last_total_assets", 0) or 0) self.daily_pnl = _pnl if not self.daily_pnl else min(self.daily_pnl, _pnl) self.loss_streak_today = max(self.loss_streak_today, _streak) self.closed_today |= _closed # [2026-10-08] 구버전 페이로드(키 없음)면 표시용 집합은 비워 둔다 — 스윙 쿨다운 # 종목이 섞인 closed_today 로 대체하면 같은 오집계가 되살아난다. self.dt_exited_today |= { str(c) for c in (data.get("dt_exited_today") or []) if c } self.dt_reheld_today |= { str(c) for c in (data.get("dt_reheld_today") or []) if c } self.restore_block_legacy |= { str(c) for c in (data.get("restore_block_legacy") or []) if c } if "dt_exited_today" not in data: # [R2 하드닝] 구버전 페이로드 — DT 청산과 재진입 차단을 구분할 수 없다. 복원 판정에서만 # closed_today 를 청산 근거로 쓴다(재매도 방지 쪽, 구 동작과 동일). self.restore_block_legacy |= _closed # [C2] 분모는 **작은 쪽이 보수적**이다: pnl<0 일 때 (pnl/assets) 가 더 음수가 # 되어 -3% 한도가 더 빨리 걸린다. 단 0/음수는 게이트를 통째로 무력화하므로 # 후보에서 제외하고, 양수가 하나도 없으면 현재값(0.0)을 그대로 둔다. _cands = [v for v in (self.last_total_assets, _assets) if v > 0] if _cands: self.last_total_assets = min(_cands) logger.info( f"[DayTrading] 일일 안전장치 복원: 실현손익={self.daily_pnl:+,.0f}원, " f"연패={self.loss_streak_today}, 당일청산={len(self.closed_today)}종목, " f"총자산기준={self.last_total_assets:,.0f}원" ) return True except Exception as e: logger.warning( f"[DayTrading] ⚠️ 일일 안전장치 복원 실패: {e} — 일일 손실한도·연패 " f"쿨다운·재진입 차단이 초기화된 상태로 시작한다" ) return False async def rebuild_daily_state_after_redis_miss_v2(self, session_factory=None) -> bool: """[R2 2026-10-09] Redis 복원 실패 시 당일 DT 원장(TradeHistory)으로 일일 안전장치 재구성. ``load_daily_state()`` 는 Redis 단일 소스라, 장중에 Redis 재기동(RDB 간격 내 유실)· ``volatile-lru`` 축출(이 키는 setex TTL 을 가진다)·저장 실패 뒤 백엔드를 재기동하면 daily_pnl=0 / 연패 0 / closed_today=∅ 로 시작해 **이미 -3% 한도로 중단된 날에 매매가 재개**됐다(fail-open, warning 만). 당일 DT SELL 원장(net_profit_loss)은 DB 에 있으므로 그것으로 재구성한다. - 성공(행 0건 포함 — 당일 매매가 없었다는 검증된 사실): 보수적 병합 후 Redis 재저장, True. - 실패: ``daily_state_unverified=True`` → is_cooldown_active() 가 DT **신규 진입만** 차단, error 로그 + 텔레그램, 백그라운드 재시도 예약. False. - 손실 상태인데 총자산 분모(last_total_assets)를 확보하지 못해도 미검증으로 둔다 — 분모 0 이면 is_daily_loss_exceeded() 가 무조건 False 라 -3% 한도가 무력화된다. [§3.1 / §3.2 ③ — 기동 시 1회(+실패 시 재시도), 읽기 전용] SELECT 만 수행하고 commit()/add() 가 없다. 이 메서드는 쓰기를 하지 않는다 — 추가하지 말 것. 이 메서드는 예외를 던지지 않는다(차단 판정은 try 밖에서 한다, §4). Args: session_factory: 테스트 주입용 세션 팩토리(기본 AsyncSessionLocal). """ ok, err = await self._rebuild_daily_state_attempt_impl(session_factory) if ok: return True await self._mark_daily_state_unverified_impl(err) self._schedule_daily_state_rebuild_retry_impl(session_factory) return False async def _rebuild_daily_state_attempt_impl(self, session_factory=None) -> tuple[bool, str]: """재구성 1회 시도. (검증완료 여부, 실패 사유) — 예외를 던지지 않는다.""" today = _today_kst() rows = None err = "" try: rows = await asyncio.wait_for( _fetch_today_trade_rows_impl(today, session_factory), timeout=_DAILY_STATE_REBUILD_DB_TIMEOUT_SEC, ) except asyncio.TimeoutError: err = f"TradeHistory 조회 {_DAILY_STATE_REBUILD_DB_TIMEOUT_SEC:.0f}초 초과" except Exception as e: err = f"TradeHistory 조회 실패: {type(e).__name__}: {e}" if rows is None: logger.warning(f"[DayTrading] ⚠️ 일일 안전장치 원장 재구성 실패 — {err}") return False, err or "TradeHistory 조회 결과 없음" rebuilt = _rebuild_daily_state_from_rows(rows) async with self.trade_lock: # 같은 날 재시도(진입 차단 중에도 청산은 돌아 인메모리 손익이 쌓인다) 시 이중 집계를 # 피하려고 합산하지 않는다 — 원장 합계가 이미 그 청산을 포함한다. 둘 중 **더 보수적인** # 쪽(손익 min, 연패 max, 집합 합집합)을 취한다(load_daily_state 와 같은 규칙). _pnl = float(rebuilt["daily_pnl"]) self.daily_pnl = _pnl if not self.daily_pnl else min(self.daily_pnl, _pnl) self.loss_streak_today = max(self.loss_streak_today, int(rebuilt["loss_streak_today"])) self.closed_today |= set(rebuilt["closed_today"]) self.dt_exited_today |= set(rebuilt["dt_exited_today"]) self.dt_reheld_today |= set(rebuilt.get("dt_reheld_today") or ()) _need_assets = self.daily_pnl < 0 and float(self.last_total_assets or 0) <= 0 if _need_assets: # 분모 확보(청산 경로와 같은 함수 — KIS 잔고 → 마지막 값). 실패해도 예외를 던지지 않는다. try: from app.services.daytrading.exit_service import _ensure_total_assets_for_exit_impl await asyncio.wait_for( _ensure_total_assets_for_exit_impl(), timeout=_DAILY_STATE_REBUILD_DB_TIMEOUT_SEC, ) except asyncio.TimeoutError: logger.warning("[DayTrading] 일일 안전장치 재구성 후 총자산 조회 시간 초과") except Exception as e: logger.warning(f"[DayTrading] 일일 안전장치 재구성 후 총자산 조회 실패: {e}") # 판정은 try 밖 (§4) if self.daily_pnl < 0 and float(self.last_total_assets or 0) <= 0: err = ( f"당일 실현손익 {self.daily_pnl:+,.0f}원(손실)인데 총자산 분모 미확보 — " f"일일 손실한도 판정 불가" ) logger.warning(f"[DayTrading] ⚠️ {err}") return False, err self.daily_state_source = "trade_history" _was_blocked = self.daily_state_unverified self.daily_state_unverified = False self.daily_state_unverified_reason = "" logger.warning( f"[DayTrading] 일일 안전장치 원장(TradeHistory) 재구성: 실현손익={self.daily_pnl:+,.0f}원 " f"(전일·고아 청산 {rebuilt['stale_excluded']}건 제외), 연패={self.loss_streak_today}, " f"당일청산={len(self.closed_today)}종목, 원장 {rebuilt['dt_sell_rows']}건" + (" — 신규 진입 차단 해제" if _was_blocked else "") ) try: await self.save_daily_state() except Exception as e: logger.warning(f"[DayTrading] 재구성 상태 Redis 재저장 실패: {e}") if _was_blocked: await _notify_daily_state_impl( "DT 일일 안전장치 원장 재구성 성공 — 신규 진입 차단 해제 " f"(실현손익 {self.daily_pnl:+,.0f}원, 연패 {self.loss_streak_today})" ) return True, "" async def _mark_daily_state_unverified_impl(self, reason: str) -> None: """미검증 상태 표시 + error 로그 + 텔레그램 (예외를 던지지 않는다).""" _first = not self.daily_state_unverified self.daily_state_unverified = True self.daily_state_unverified_reason = reason logger.error( f"[DayTrading] 🚨 일일 안전장치 복원 불가(Redis·TradeHistory) — DT 신규 진입 차단 " f"(청산은 계속). 사유: {reason}" ) if _first: await _notify_daily_state_impl( "DT 일일 안전장치(일일 손실한도·연패·재진입 차단) 복원 불가 — 당일 DT 신규 진입 차단. " f"사유: {reason}. 원장 재구성을 {_DAILY_STATE_REBUILD_RETRY_SEC:.0f}초 간격으로 재시도한다." ) def _schedule_daily_state_rebuild_retry_impl(self, session_factory=None) -> None: """미검증 상태에서 재구성을 주기적으로 재시도(같은 날 한정). 중복 예약하지 않는다.""" global _daily_state_retry_task if _daily_state_retry_task is not None and not _daily_state_retry_task.done(): return try: _daily_state_retry_task = asyncio.get_running_loop().create_task( self._daily_state_rebuild_retry_loop_impl(_today_kst(), session_factory) ) except RuntimeError as e: logger.warning(f"[DayTrading] 일일 안전장치 재구성 재시도 예약 실패: {e}") async def _daily_state_rebuild_retry_loop_impl(self, day: date, session_factory=None) -> None: attempt = 0 while self.daily_state_unverified: await asyncio.sleep(_DAILY_STATE_REBUILD_RETRY_SEC) if _today_kst() != day or self.last_reset_date not in (None, day): # 날짜가 바뀌면 reset_if_new_day 가 플래그를 내린다 — 전일 재구성은 의미 없음 return if not self.daily_state_unverified: return attempt += 1 ok, err = await self._rebuild_daily_state_attempt_impl(session_factory) if ok: return self.daily_state_unverified_reason = err logger.warning( f"[DayTrading] 일일 안전장치 재구성 재시도 {attempt}회 실패 — 신규 진입 차단 유지: {err}" ) def set_orb(self, stock_code: str, orb: Any) -> None: """ORB(시초가 레인지) 데이터 저장. 09:00~09:30 사이에 collect_opening_ranges()에서 호출. Args: stock_code: 종목 코드 orb: OpeningRange 객체 (고가, 저가, VWAP 포함) """ self.orb_cache[stock_code] = orb def get_orb(self, stock_code: str) -> Any | None: """해당 종목의 ORB(시초가 레인지) 데이터 반환. Args: stock_code: 종목 코드 Returns: Any | None: OpeningRange 객체, 없으면 None """ return self.orb_cache.get(stock_code) async def restore_from_db(self, db: AsyncSession) -> int: """앱 재시작 시 ActiveTrade 테이블에서 당일 데이트레이딩 포지션 복원. strategy_name='day_trade' AND status='active' AND created_at이 오늘인 레코드를 인메모리 positions 딕셔너리에 복원합니다. 이미 positions에 존재하는 종목은 건너뜁니다. Args: db: AsyncSession 인스턴스 Returns: int: 복원된 포지션 수 """ from app.models.models import ActiveTrade from app.core.config import settings today = _today_kst() restored = 0 try: result = await db.execute( select(ActiveTrade).where( ActiveTrade.strategy_name == "day_trade", ActiveTrade.status == "active", ) ) active_trades = result.scalars().all() async with self.trade_lock: for trade in active_trades: # 당일 생성된 포지션만 복원 (데이트레이딩은 당일 청산 원칙) trade_date = trade.created_at.date() if trade.created_at else None if trade_date != today: logger.warning( f"[DayTrading] 전일 미청산 포지션 발견: {trade.stock_code} " f"(created_at={trade.created_at}) — EOD 청산 대상" ) # 전일 포지션도 복원하여 EOD 청산 가능하게 함 if trade.stock_code in self.positions: continue _restored = _active_trade_row_to_position(trade) self.positions[trade.stock_code] = _restored restored += 1 logger.info( f"[DayTrading] DB에서 포지션 복원: {trade.stock_code} " f"{trade.stock_name} qty={trade.quantity} " f"entry={trade.entry_price}" ) if restored: logger.info(f"[DayTrading] 총 {restored}개 포지션 DB에서 복원 완료") else: logger.debug("[DayTrading] 복원할 데이트레이딩 포지션 없음") except Exception as e: logger.error(f"[DayTrading] DB 포지션 복원 실패: {e}", exc_info=True) return restored async def verify_restored_today_positions_v2( self, session_factory=None, holdings_fetcher=None, ) -> dict: """[R3 2026-10-09] 기동 시 restore_from_db 가 되살린 **당일** DT 포지션 검증. restore_from_db 는 closed_today·KIS 실보유를 보지 않고 active 행 수량으로 포지션을 되살린다. 전량 청산 직후 close_active_trade 큐 작업이 반영되지 못한 채(DLQ·drain 초과) 재기동하거나 분할 매도 뒤 잔량 upsert 가 유실되면, 이미 판 물량을 일반 청산 경로가 다시 시장가로 판다(같은 종목 스윙 보유분 매도 또는 반복 거부 — §3.5). 장중 복원 경로(exit_service._restore_lost_today_positions_impl)와 같은 규칙을 적용한다: - 당일 DT 전량 청산 종목(restore_block_reason_v2: dt_exited_today − dt_reheld_today) → 복원 보류 [R2 하드닝] closed_today(재진입 차단)만으로는 보류하지 않는다 — 불명 매수 접수 확정· 매도 0체결 종결 복원·스윙 손절 쿨다운 종목도 들어 있어 실제 보유 포지션이 관리에서 빠졌다. - KIS 보유 − 비DT active 수량 < 행 수량(또는 수량 ≤ 0) → 복원 보류 - 판정 불가(잔고·DB 조회 실패) → DAY_TRADE_ORPHAN_DETECT_ENABLED 면 보류 (close_stale_positions 가 매 사이클 같은 규칙으로 재검증·복원한다), 꺼져 있으면 복원 유지(아무도 되살리지 않아 손절 공백이 되므로) + error. 보류는 인메모리에서만 뺀다(closed_today 에 넣지 않음). 수량·판정불가 보류는 잔고가 맞으면 장중 복원이 다시 잡는다. 청산 기록 보류는 장중 복원도 같은 사유로 막으므로 수동 확인 대상이다. 전일 진입분은 대상이 아니다(close_stale_positions 가 실보유·타전략 차감으로 처리). Returns: {"checked": n, "held": [codes], "unverified": bool} """ from app.core.config import settings today = _today_kst() held: list[str] = [] reasons: list[str] = [] mismatch = False # 실제 불일치(청산 기록·수량 부족) — 판정 불가와 구분해 경보 수준을 정한다 exit_held = False # 청산 기록으로 보류 — 장중 재복원 경로도 같은 사유로 막는다 async with self.trade_lock: cands = { c: p for c, p in self.positions.items() if p.get("restored_from_db") and _entry_date(p.get("entry_time")) == today } for c in list(cands): _why = self.restore_block_reason_v2(c) if _why is not None: if self.positions.get(c) is cands[c]: self.positions.pop(c, None) held.append(c) reasons.append(f"{c}: {_why}") mismatch = True exit_held = True cands.pop(c) out = {"checked": len(cands) + len(held), "held": held, "unverified": False} if cands: other_qty = None try: other_qty = await asyncio.wait_for( _fetch_non_dt_active_qty_by_code_impl(list(cands), session_factory), timeout=_DAILY_STATE_REBUILD_DB_TIMEOUT_SEC, ) except asyncio.TimeoutError: logger.warning("[DayTrading] 복원 검증: 비DT 보유 조회 시간 초과") except Exception as e: logger.warning(f"[DayTrading] 복원 검증: 비DT 보유 조회 실패: {type(e).__name__}: {e}") holdings = None if other_qty is not None: try: if holdings_fetcher is None: from app.services.daytrading.exit_service import ( _fetch_kis_holdings_v2_impl as holdings_fetcher, ) holdings = await holdings_fetcher() except Exception as e: logger.warning(f"[DayTrading] 복원 검증: KIS 잔고 조회 실패: {type(e).__name__}: {e}") holdings = None # 판정·보류는 try 밖 (§4 — 가드를 try 안에 두지 않는다) unverified = other_qty is None or holdings is None out["unverified"] = unverified if unverified and not settings.DAY_TRADE_ORPHAN_DETECT_ENABLED: logger.error( f"[DayTrading] 🚨 기동 복원 검증 불가(실보유·타전략 미확인) — 장중 재검증 경로" f"(DAY_TRADE_ORPHAN_DETECT_ENABLED=False)가 없어 복원 유지: {sorted(cands)}. " f"이미 매도된 물량이면 재매도 위험 — 수동 확인 필요" ) else: async with self.trade_lock: for c, p in cands.items(): qty = int(p.get("quantity", 0) or 0) if unverified: why = "실보유·타전략 수량 판정 불가 → 장중 close_stale_positions 재검증 대기" else: hldg = int((holdings.get(c) or {}).get("hldg", 0) or 0) oth = int(other_qty.get(c, 0) or 0) if qty > 0 and hldg - oth >= qty: continue why = f"실보유 {hldg}주 − 타전략 {oth}주 = {hldg - oth}주 < 행 수량 {qty}주" if self.positions.get(c) is p: self.positions.pop(c, None) held.append(c) reasons.append(f"{c}: {why}") mismatch = mismatch or not unverified if held: msg = ( f"[DayTrading] 🚨 기동 시 당일 DT 포지션 복원 보류 {len(held)}건 — " + "; ".join(reasons) + " (이미 매도된 물량 재매도 방지. 수량·판정불가 보류는 잔고가 맞으면 장중 " "close_stale_positions 가 재복원" + ("; 청산 기록 보류는 장중에도 재복원하지 않음 — 실보유가 남아 있으면 수동 청산 필요)" if exit_held else ")") ) if mismatch: logger.error(msg) await _notify_daily_state_impl(msg) else: logger.warning(msg) return out async def verify_restored_today_positions_v3( self, session_factory=None, holdings_fetcher=None, today_rows_fetcher=None, ) -> dict: """[C-2 r9] v2 + 주문가능수량·당일 PENDING DT 매도 반영. v2 는 KIS **보유수량(hldg)** 만 봤다. 재기동 직전 접수된 미체결 DT 매도(VI 단일가 등)가 있으면 hldg 는 그대로라 포지션이 복원되고, 일반 청산이 다시 팔아 같은 종목 스윙 보유분이 팔릴 수 있다(§3.5). 이제: - 판정 수량 = min(주문가능, 보유) − 비DT active (주문가능에서는 접수된 매도가 이미 빠진다) - 당일 TradeHistory 에 DT SELL PENDING 이 있는 종목은 복원 보류(warning). ⚠️ **현재 DT 경로는 PENDING SELL 행을 쓰지 않는다** — 매도 TradeHistory 는 접수 확정 뒤 COMPLETED 로만 들어가고, 결과 불명 매도는 행을 남기지 않는다. 따라서 이 분기는 현재 발동하지 않는 방어 코드다(PENDING 을 쓰는 경로가 생길 때 대비). 재기동 직전 미체결 매도의 **실효 방어는 위의 min(주문가능, 보유)** 이다. 장중 close_stale_positions 도 같은 규칙 (_restore_lost_today_positions_v2_impl)으로 재검증한다. - 원장 조회 실패: DAY_TRADE_ORPHAN_DETECT_ENABLED 면 전부 보류(v2 의 판정 불가 규칙과 동일). 이 보류는 실효 동작이다(조회 실패 시 기동 복원을 장중 재검증에 맡김). DB 는 SELECT 전용(§3.2 ③ restore 경로, commit 없음). """ from app.core.config import settings today = _today_kst() cand_codes = [ c for c, p in self.positions.items() if p.get("restored_from_db") and _entry_date(p.get("entry_time")) == today ] pre_held: list[str] = [] if cand_codes: rows = None try: _f = today_rows_fetcher or (lambda: _fetch_today_trade_rows_impl(today, session_factory)) rows = await asyncio.wait_for(_f(), timeout=_DAILY_STATE_REBUILD_DB_TIMEOUT_SEC) except asyncio.TimeoutError: logger.warning("[DayTrading] 복원 검증: 당일 원장(PENDING 매도) 조회 시간 초과") except Exception as e: logger.warning(f"[DayTrading] 복원 검증: 당일 원장(PENDING 매도) 조회 실패: {type(e).__name__}: {e}") # 보류 판정은 try 밖 (§4) if rows is None: hold_codes = set(cand_codes) if settings.DAY_TRADE_ORPHAN_DETECT_ENABLED else set() _why = "당일 PENDING 매도 확인 불가" else: hold_codes = { getattr(r, "stock_code", None) for r in rows if str(getattr(r, "strategy_type", "") or "") == "daytrading" and str(getattr(r, "order_type", "") or "").upper() == "SELL" and str(getattr(r, "status", "") or "").upper() == "PENDING" } & set(cand_codes) _why = "당일 DT 매도 PENDING(결과 미확정)" if hold_codes: async with self.trade_lock: for c in sorted(hold_codes): if self.positions.pop(c, None) is not None: pre_held.append(c) logger.warning( f"[DayTrading] ⚠️ 기동 복원 보류 {pre_held} — {_why}. 미체결 매도 물량 재매도 방지, " f"장중 close_stale_positions 가 다음 사이클에 재검증" ) async def _sellable_capped_holdings(): if holdings_fetcher is None: from app.services.daytrading.exit_service import _fetch_kis_holdings_v2_impl as _hf else: _hf = holdings_fetcher h = await _hf() if h is None: return None return {c: _cap_hldg_by_sellable_v2(v) for c, v in h.items()} out = await self.verify_restored_today_positions_v2( session_factory=session_factory, holdings_fetcher=_sellable_capped_holdings, ) out["held"] = pre_held + list(out.get("held") or []) out["checked"] = int(out.get("checked", 0)) + len(pre_held) return out def _active_trade_row_to_position(trade) -> dict: """ActiveTrade(day_trade) 행 → 인메모리 포지션 dict (restore_from_db 와 고아 복원 공용). [2026-10-08] restore_from_db 의 행 변환을 그대로 옮겼다(동작 무변경). """ from app.core.config import settings entry_price = float(trade.entry_price) if trade.entry_price else 0 stop_loss = float(trade.stop_loss) if trade.stop_loss else 0 target_price = float(trade.take_profit) if trade.take_profit else 0 # [FIX] stop_loss/target_price가 0/None으로 복원되면 entry_price 기반 # 기본값으로 대체 — day_trading_strategy.py의 target_price<=0 즉시매도 방지 if entry_price > 0 and stop_loss <= 0: stop_loss = entry_price * (1 - settings.DAY_TRADE_STOP_LOSS_PCT) logger.warning( f"[DayTrading] {trade.stock_code} stop_loss 복원값 없음 → " f"기본값 적용: {stop_loss:.0f}" ) if entry_price > 0 and target_price <= 0: risk = entry_price - stop_loss target_price = entry_price + risk * settings.DAY_TRADE_TARGET_MULTIPLIER logger.warning( f"[DayTrading] {trade.stock_code} target_price 복원값 없음 → " f"기본값 적용: {target_price:.0f}" ) _restored = { "stock_name": trade.stock_name or trade.stock_code, "strategy": trade.strategy_name or "day_trade", "entry_price": entry_price, "stop_loss": stop_loss, "target_price": target_price, "quantity": trade.quantity or 0, "highest_price": float(trade.highest_since_entry) if trade.highest_since_entry else ( float(trade.entry_price) if trade.entry_price else 0 ), "entry_time": trade.created_at, "reason": trade.entry_reason or "", "order_id": trade.order_id or "", "restored_from_db": True, } # [M1 2026-10-02] 분할매도 이력 복원. exit_service 의 분할매도가 # remaining_pct(<1.0) 를 ActiveTrade 에 남긴다. 복원 시 이를 보지 않으면 # target_price 가 원래 목표가로 되살아나 재기동 직후 같은 50% 분할이 # 재발동한다(잔량의 절반을 또 판다). 인메모리 분할 후 상태와 동일하게 # 목표가를 제거(inf)하고 partial_sold 를 세운다 → 트레일링만 적용. if _is_partial_sold_row(trade): _restored["target_price"] = float("inf") _restored["partial_sold"] = True logger.warning( f"[DayTrading] {trade.stock_code} 분할매도 이력 복원 " f"(remaining_pct={trade.remaining_pct}) — 목표가 분할 재발동 차단, " f"트레일링 고점={_restored['highest_price']:.0f}" ) return _restored def _entry_date(entry_time) -> date | None: """entry_time(datetime|None) → date. restore_from_db 의 created_at.date() 비교와 같은 기준.""" if isinstance(entry_time, datetime): return entry_time.date() return None def _cap_hldg_by_sellable_v2(v) -> dict: """[C-2 r9] {"hldg","sellable"} → hldg 를 min(주문가능, 보유) 로 낮춘 사본. sellable 없으면 그대로.""" d = dict(v or {}) if d.get("sellable") is not None: try: d["hldg"] = min(int(d.get("hldg", 0) or 0), int(d.get("sellable", 0) or 0)) except (TypeError, ValueError): d["hldg"] = 0 # 파싱 불가 → 복원 근거 없음(보류 쪽) return d async def _fetch_non_dt_active_qty_by_code_impl(codes: list, session_factory=None) -> dict[str, int]: """비-DT(스윙·유입·legacy NULL) active ActiveTrade 종목별 수량 합 — SELECT 전용(§3.2 ③, commit 없음). close_stale_positions 와 같은 필터(db_worker.not_dt_strategy_clause). 실패는 예외로 전파한다. """ from app.models.models import ActiveTrade from app.services.db_worker import not_dt_strategy_clause if session_factory is None: from app.core.database import AsyncSessionLocal as session_factory # noqa: N813 out: dict[str, int] = {} if not codes: return out # DB-READ-EXCEPTION: 기동 시 DT 복원 검증(§3.5 타전략 차감), SELECT 전용·commit 없음 async with session_factory() as db: rows = (await db.execute( select(ActiveTrade).where( ActiveTrade.stock_code.in_(list(codes)), ActiveTrade.status == "active", not_dt_strategy_clause(ActiveTrade.strategy_name), ) )).scalars().all() for r in rows or []: out[r.stock_code] = out.get(r.stock_code, 0) + int(r.quantity or 0) return out def _is_partial_sold_row(trade) -> bool: """[M1] ActiveTrade 행이 분할매도 이후 상태인지 (remaining_pct < 1.0).""" try: rp = getattr(trade, "remaining_pct", None) return rp is not None and 0.0 < float(rp) < 1.0 except (TypeError, ValueError): return False async def _fetch_today_trade_rows_impl(today: date, session_factory=None) -> list: """당일(KST) TradeHistory 행 조회 — SELECT 전용(§3.2 ③, commit 없음).""" from datetime import timedelta from app.models.models import TradeHistory if session_factory is None: from app.core.database import AsyncSessionLocal as session_factory # noqa: N813 start = datetime.combine(today, datetime.min.time()) end = start + timedelta(days=1) # DB-READ-EXCEPTION: DT 일일 안전장치 원장 재구성(기동 시/미검증 재시도), SELECT 전용·commit 없음 async with session_factory() as db: result = await db.execute( select(TradeHistory).where( TradeHistory.created_at >= start, TradeHistory.created_at < end, TradeHistory.order_type.in_(["BUY", "SELL"]), ).order_by(TradeHistory.created_at, TradeHistory.id) ) return list(result.scalars().all()) def _rebuild_daily_state_from_rows(rows) -> dict: """당일 TradeHistory 행 → 일일 안전장치 값 (순수 함수, 입력은 created_at·id 순 정렬 가정). 규칙 (exit_service._finalize_exit_impl 과 동일 기준): - daily_pnl: DT SELL 의 net_profit_loss(비용 차감 순손익) 합. 취소·실패 행 제외. **전일·고아 청산은 제외** — exit_reason 접두사, 또는 당일 DT BUY 가 없는 종목의 SELL. - loss_streak_today: 종목별 **마지막** 당일 SELL 시점을 전량 청산으로 보고, 그 종목의 당일 SELL net 합(분할 레그 포함 = 왕복 순손익, exit_service._apply_round_trip_streak_v2)으로 시간순 판정한다. 손실 +1, 이익 0 리셋, 0 중립. 시간손절도 센다(§5.1 — loss_tracker 와 달리 당일 연패는 시간손절 포함). - closed_today: DT SELL 종목 전부(취소 포함 — 전일 청산은 주문 실패해도 재진입 금지) + 타 전략 체결 SELL 종목(전략 간 당일 쿨다운, swing/exit_service). 보수적 상위집합. - dt_exited_today: 취소·실패가 아닌 DT SELL 종목(표시용). - dt_reheld_today: [R2 하드닝] dt_exited_today 중 당일 DT 원장 순체결(BUY COMPLETED 체결 − SELL 체결, exit_service._dt_net_qty_from_history 와 같은 규칙)이 양수인 종목 — 전량 매도가 부분체결로 종결돼 보유가 남은 경우. 기동 복원 검증이 '청산 완료'로 오판하지 않게 한다. """ dt_rows = [r for r in rows if str(getattr(r, "strategy_type", "") or "") == _DT_STRATEGY_TYPE] buy_codes = { str(r.stock_code) for r in dt_rows if str(getattr(r, "order_type", "")).upper() == "BUY" and str(getattr(r, "status", "") or "").upper() not in _VOID_STATUSES } closed: set[str] = set() dt_exited: set[str] = set() counted: list = [] stale_excluded = 0 dt_sell_rows = 0 for r in rows: if str(getattr(r, "order_type", "")).upper() != "SELL": continue code = str(r.stock_code) status = str(getattr(r, "status", "") or "").upper() is_dt = str(getattr(r, "strategy_type", "") or "") == _DT_STRATEGY_TYPE if not is_dt: if status not in _VOID_STATUSES: closed.add(code) continue dt_sell_rows += 1 closed.add(code) if status in _VOID_STATUSES: continue dt_exited.add(code) reason = str(getattr(r, "exit_reason", "") or "") if reason.startswith(_STALE_EXIT_REASON_PREFIXES) or code not in buy_codes: stale_excluded += 1 continue counted.append(r) pnl = 0.0 last_idx: dict[str, int] = {} for i, r in enumerate(counted): net = getattr(r, "net_profit_loss", None) if net is None: logger.warning( f"[DayTrading] 원장 재구성: {r.stock_code} SELL(id={getattr(r, 'id', '?')}) " f"net_profit_loss 없음 — 0 으로 간주" ) pnl += float(net or 0) last_idx[str(r.stock_code)] = i # [R3 2026-10-09] 연패는 **왕복 순손익**(분할 레그 포함 종목별 당일 SELL net 합)으로 판정한다. # 실시간 경로(exit_service._apply_round_trip_streak_v2, R2-4)와 같은 기준. 예전에는 마지막 # 레그만 봐 +1.5R 분할 익절(+3,000) 뒤 잔량 -300 청산을 손실로 셌다 → Redis 미스 재기동만으로 # 3연패 쿨다운이 걸릴 수 있었다. DT 는 당일 재진입이 금지(closed_today)라 종목당 왕복은 1회다. streak = 0 round_trip: dict[str, float] = {} for i, r in enumerate(counted): code = str(r.stock_code) round_trip[code] = round_trip.get(code, 0.0) + float(getattr(r, "net_profit_loss", None) or 0) if last_idx.get(code) != i: continue # 분할 매도 — 왕복 합에만 누적 net = round_trip[code] if net < 0: streak += 1 elif net > 0: streak = 0 # [R2 하드닝] 당일 DT 순체결 수량 — 양수면 전량 매도가 일부만 체결돼 보유가 남았다. net_qty: dict[str, int] = {} for r in dt_rows: code = str(r.stock_code) ot = str(getattr(r, "order_type", "") or "").upper() st = str(getattr(r, "status", "") or "").upper() eq = getattr(r, "executed_quantity", None) try: q = int(eq if eq is not None else (getattr(r, "order_quantity", 0) or 0)) except (TypeError, ValueError): logger.warning(f"[DayTrading] 원장 재구성: {code} 수량 파싱 실패 — 순체결 판정 제외") continue if ot == "BUY" and st == "COMPLETED": net_qty[code] = net_qty.get(code, 0) + q elif ot == "SELL" and st not in ("FAILED", "REJECTED"): net_qty[code] = net_qty.get(code, 0) - q reheld = {c for c in dt_exited if net_qty.get(c, 0) > 0} return { "daily_pnl": pnl, "loss_streak_today": streak, "closed_today": closed, "dt_exited_today": dt_exited, "dt_reheld_today": reheld, "stale_excluded": stale_excluded, "dt_sell_rows": dt_sell_rows, } async def _notify_daily_state_impl(message: str) -> None: """텔레그램 시스템 경고 (실패해도 예외를 던지지 않는다).""" try: from app.services.telegram_notifier import telegram_notifier await asyncio.wait_for( telegram_notifier.notify_system_warning("DAYTRADING", message), timeout=_DAILY_STATE_NOTIFY_TIMEOUT_SEC, ) except asyncio.TimeoutError: logger.warning("[DayTrading] 일일 안전장치 텔레그램 알림 시간 초과") except Exception as e: logger.warning(f"[DayTrading] 일일 안전장치 텔레그램 알림 실패: {e}") # 싱글톤 인스턴스 — daytrading 패키지 전체에서 공유 # entry_service, exit_service, scheduler에서 동일 인스턴스 참조 day_trading_state = DayTradingState() # [M5 2026-10-02] 데이트레이딩 사이클 전역 락 — 스케줄러와 수동 API(/api/daytrading/cycle)가 # 공유한다. 같은 프로세스에서 청산 체크·진입 스캔이 두 경로로 동시에 돌면 같은 포지션을 # 두 번 매도하거나 슬롯·예산 검사를 동시에 통과해 한도를 넘겨 매수한다. # trade_lock(인메모리 dict 보호, 짧게 잡음)과 다르다 — 이 락은 '한 사이클 단계' 전체를 감싼다. dt_cycle_lock = asyncio.Lock()