""" 변동성 기반 리스크 관리 모듈 - 예수금 비율 기반 비중 조절 - ATR/변동성 역가중 (Volatility Inverse Weighting) - 종목별 안전 매수 금액 계산 """ import logging import pandas as pd import numpy as np from typing import List, Dict logger = logging.getLogger("RiskManager") class RiskManager: """ [예수금 연동 리스크 관리 모듈] - 고정 금액이 아닌 '현재 예수금' 대비 퍼센트로 투자금 계산 - 변동성 큰 종목 = 적게 매수, 변동성 작은 종목 = 많이 매수 """ def __init__( self, risk_pct_per_trade: float = 0.02, # 1회 거래 시 허용 손실 비율 (2%) max_position_pct: float = 0.20, # 종목당 최대 비중 (20%) min_position_amount: int = 50000, # 최소 매수 금액 (5만원) use_kelly: bool = False, # 켈리 공식 사용 여부 kelly_multiplier: float = 0.5 # 켈리 배율 (0.5 = 하프 켈리) ): """ Args: risk_pct_per_trade: 1회 매매 시 허용할 손실 비율 (기본 0.02 = 2%) max_position_pct: 종목당 최대 투자 비율 (기본 0.20 = 20%) min_position_amount: 최소 매수 금액 (너무 작은 주문 방지) use_kelly: 켈리 공식 활성화 여부 kelly_multiplier: 켈리 배율 (0.5 = 하프켈리, 0.25 = 쿼터켈리) """ self.risk_pct = risk_pct_per_trade self.max_pos_pct = max_position_pct self.min_amount = min_position_amount self.use_kelly = use_kelly self.kelly_mult = kelly_multiplier logger.info( f"💰 RiskManager 초기화: 리스크{self.risk_pct*100}% | " f"최대비중{self.max_pos_pct*100}% | 켈리{'ON' if use_kelly else 'OFF'}" ) def calculate_volatility_atr(self, df: pd.DataFrame, period: int = 14) -> float: """ ATR(Average True Range) 기반 변동성 계산 Args: df: OHLC 데이터프레임 (컬럼: open, high, low, close) period: ATR 계산 기간 (기본 14) Returns: 현재가 대비 ATR 비율 (%) """ try: if df is None or len(df) < period: logger.warning(f"⚠️ ATR 계산: 데이터 부족 -> 기본값 3.0% 리턴") return 3.0 # True Range 계산 df = df.copy() df['tr'] = np.maximum( df['high'] - df['low'], np.maximum( np.abs(df['high'] - df['close'].shift()), np.abs(df['low'] - df['close'].shift()) ) ) # ATR = TR의 이동평균 atr = df['tr'].rolling(window=period).mean().iloc[-1] current_price = df['close'].iloc[-1] if current_price == 0: return 3.0 # 현재가 대비 ATR 비율 (%) volatility_pct = (atr / current_price) * 100 # 최소값 보정 (0.5% 미만은 0.5%로) return max(round(volatility_pct, 2), 0.5) except Exception as e: logger.error(f"❌ ATR 계산 에러: {e}") return 3.0 def calculate_volatility_simple(self, candles: List[Dict], period: int = 10) -> float: """ 간단한 변동성 계산 (고저차 비율 평균) Args: candles: 캔들 데이터 리스트 [{'high': , 'low': , 'close': }, ...] period: 계산 기간 (기본 10일) Returns: 변동성 (%) """ try: if not candles or len(candles) < period: return 3.0 recent = candles[-period:] volatility_sum = 0.0 for candle in recent: high = float(candle.get('high', 0)) low = float(candle.get('low', 0)) close = float(candle.get('close', 1)) if close == 0: continue # (고가 - 저가) / 종가 volatility_sum += (high - low) / close avg_vol = (volatility_sum / len(recent)) * 100 return max(round(avg_vol, 2), 0.5) except Exception as e: logger.error(f"❌ 변동성 계산 에러: {e}") return 3.0 def get_position_size( self, stock_name: str, current_balance: float, volatility_pct: float = None, df: pd.DataFrame = None, kelly_fraction: float = None, size_class: str = None, ) -> int: """ [메인 함수] 종목별 안전 매수 금액 계산 Args: stock_name: 종목명 (로깅용) current_balance: 현재 예수금 volatility_pct: 변동성 (직접 제공 시) df: OHLC 데이터프레임 (ATR 계산용) kelly_fraction: 켈리 비율 (DB에서 계산한 값) size_class: 대/중/소형 구분 ("대", "중", "소") Returns: 추천 매수 금액 (원) """ try: if current_balance <= 0: logger.warning(f"⚠️ [{stock_name}] 예수금 0원 이하") return 0 # 1. 변동성 계산 if volatility_pct is None: if df is not None and not df.empty: volatility_pct = self.calculate_volatility_atr(df) else: logger.warning(f"⚠️ [{stock_name}] 변동성 데이터 없음 -> 기본값 3.0%") volatility_pct = 3.0 # 2. 기본 리스크 금액 계산 (예수금 * 리스크 비율) # 예: 1,000만 원 * 2% = 20만 원 (이 종목에서 최대 20만원까지 잃을 수 있음) allowable_risk = current_balance * self.risk_pct # 3. 켈리 공식 적용 (쿼터 켈리/하프 켈리) if self.use_kelly: if kelly_fraction is not None and kelly_fraction > 0: # DB에서 계산한 켈리 비율 사용 (하프 켈리) base_position_pct = kelly_fraction * self.kelly_mult allowable_risk = current_balance * base_position_pct logger.info(f"🎲 [{stock_name}] 켈리 적용(DB): {base_position_pct*100:.1f}% (켈리={kelly_fraction*100:.1f}% × 배율={self.kelly_mult})") else: # 켈리 비율이 없으면 쿼터 켈리 배율을 기본값으로 사용 # 예: 리스크 1% × 쿼터 켈리 0.25 = 0.25% (더 보수적) base_position_pct = self.risk_pct * self.kelly_mult allowable_risk = current_balance * base_position_pct logger.info(f"🎲 [{stock_name}] 켈리 적용(기본): {base_position_pct*100:.1f}% (리스크={self.risk_pct*100}% × 쿼터켈리={self.kelly_mult})") # 4. 변동성 역가중으로 매수 금액 계산 # 공식: (허용 손실액) / (변동성 비율) # 변동성 5% -> 20만 / 0.05 = 400만원 매수 # 변동성 10% -> 20만 / 0.10 = 200만원 매수 risk_ratio = volatility_pct / 100.0 calculated_amount = allowable_risk / risk_ratio # 5. 상한선 체크 (종목당 최대 비중) max_limit = current_balance * self.max_pos_pct if calculated_amount > max_limit: final_amount = max_limit note = f"(최대{self.max_pos_pct*100:.0f}% 제한)" else: final_amount = calculated_amount note = "(변동성기반)" # 5-2. 대형/소형 구간별 조정 (소형주는 포지션 축소) if size_class == "소": final_amount = int(final_amount * 0.7) note = "(소형주 70%)" elif size_class == "중": final_amount = int(final_amount * 0.85) note = "(중형주 85%)" elif size_class == "대": pass # 대형주는 기존과 동일 # 6. 하한선 체크 (너무 작은 금액은 거래 안함) if final_amount < self.min_amount: logger.info( f"🚫 [{stock_name}] 계산 금액({final_amount:,.0f}원) < " f"최소금액({self.min_amount:,.0f}원) -> 매수 보류" ) return 0 logger.info( f"💰 [{stock_name}] 예수금:{current_balance:,.0f}원 | " f"변동성:{volatility_pct:.2f}% | 추천:{int(final_amount):,.0f}원 {note}" ) return int(final_amount) except Exception as e: logger.error(f"❌ [{stock_name}] 포지션 사이즈 계산 실패: {e}") return 0 def calculate_quantity( self, current_price: float, target_amount: int, fee_rate: float = 0.00015 ) -> int: """ 목표 금액을 현재가와 수수료 고려하여 수량으로 변환 Args: current_price: 현재 주가 target_amount: 목표 투자 금액 fee_rate: 수수료율 (기본 0.015%) Returns: 매수 가능 수량 (주) """ try: if current_price <= 0 or target_amount <= 0: return 0 # 수수료 고려 실제 매수 가능 금액 # 총비용 = 가격 * 수량 * (1 + 수수료율) # 수량 = 목표금액 / (가격 * (1 + 수수료율)) max_buy_amount = target_amount / (1 + fee_rate) quantity = int(max_buy_amount / current_price) if quantity < 1: return 0 # 예상 총 비용 계산 (확인용) expected_cost = quantity * current_price * (1 + fee_rate) logger.debug( f"💵 수량 계산: {current_price:,.0f}원 × {quantity}주 = " f"예상비용 {expected_cost:,.0f}원 (목표: {target_amount:,.0f}원)" ) return quantity except Exception as e: logger.error(f"❌ 수량 계산 에러: {e}") return 0