271 lines
11 KiB
Python
271 lines
11 KiB
Python
"""
|
||
변동성 기반 리스크 관리 모듈
|
||
- 예수금 비율 기반 비중 조절
|
||
- 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
|