"""데이트레이딩 인메모리 포지션 상태 관리 기존 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") 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 # 당일 청산 종목 (재진입 방지) self.closed_today: 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() 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.loss_streak_today = 0 self.last_reset_date = today 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 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.portfolio_cache_dirty = True return pos 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 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연패 시 신규 진입 차단. Returns: bool: 쿨다운 활성화 여부 """ 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), # [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 # [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 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 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}" ) self.positions[trade.stock_code] = { "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, } 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 # 싱글톤 인스턴스 — daytrading 패키지 전체에서 공유 # entry_service, exit_service, scheduler에서 동일 인스턴스 참조 day_trading_state = DayTradingState()