451 lines
16 KiB
Python
451 lines
16 KiB
Python
"""
|
||
RSI + 볼린저 밴드 평균회귀 전략 (CLAUDE.md 섹션 7.1-7.5)
|
||
|
||
일믐9 기준, 전체 자본의 엕시 기마 30% 이내에서만 운용.
|
||
|
||
[2026-02 수익 극대화 개선]
|
||
① R:R 개선: 손절 -8%→-6%, 1차 익절 RSI 45~50→55~65, 트레일링 스탑 도입
|
||
② 진입 기회 확대: BULL RSI ≤25 → ≤30 (2배 증가)
|
||
③ 동적 포지션 사이징: RSI/ADX 기반 4~8% 다이나믉 비중
|
||
"""
|
||
import logging
|
||
from typing import Optional, Dict, List, Tuple
|
||
from dataclasses import dataclass
|
||
from enum import Enum
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MarketRegime(Enum):
|
||
BULL = "bull"
|
||
BEAR = "bear"
|
||
|
||
|
||
@dataclass
|
||
class ExitSignal:
|
||
action: str # "SELL_PARTIAL", "SELL_ALL"
|
||
reason: str
|
||
sell_pct: float # 0.5 = 50%, 1.0 = 전량
|
||
|
||
|
||
def calculate_atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]:
|
||
"""ATR(Average True Range) 계산
|
||
|
||
TR = max(H-L, |H-Cprev|, |L-Cprev|)
|
||
ATR = SMA(TR, period)
|
||
"""
|
||
if len(highs) < period + 1 or len(lows) < period + 1 or len(closes) < period + 1:
|
||
return None
|
||
|
||
true_ranges = []
|
||
for i in range(1, len(highs)):
|
||
h, l, c_prev = highs[i], lows[i], closes[i - 1]
|
||
tr = max(h - l, abs(h - c_prev), abs(l - c_prev))
|
||
true_ranges.append(tr)
|
||
|
||
if len(true_ranges) < period:
|
||
return None
|
||
return sum(true_ranges[-period:]) / period
|
||
|
||
|
||
def calculate_adx(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]:
|
||
"""Wilder's ADX(Average Directional Index) 계산 — 추세 강도 지표
|
||
|
||
알고리즘:
|
||
1. TR = max(H-L, |H-Cprev|, |L-Cprev|)
|
||
2. +DM = H-Hprev if (H-Hprev) > (Lprev-L) and > 0, else 0
|
||
3. -DM = Lprev-L if (Lprev-L) > (H-Hprev) and > 0, else 0
|
||
4. Wilder 평활: ATR14 = ATR14_prev - ATR14_prev/14 + TR (초기값은 14일 합산)
|
||
5. +DI14 = 100 × (+DM14 / ATR14), -DI14 동일
|
||
6. DX = 100 × |+DI14 - -DI14| / (+DI14 + -DI14)
|
||
7. ADX = DX의 Wilder 평활
|
||
|
||
해석 (평균회귀 전략):
|
||
- ADX < 20: 횊보장 → 평균회귀 유리
|
||
- ADX 20~30: 중간 추세 → 조건부 허용
|
||
- ADX > 30: 강한 추세 → 평균회귀 위험 (진입 차단)
|
||
"""
|
||
n = len(highs)
|
||
# 최소 2*period + 1개 필요 (초기 평활 + ADX 평활)
|
||
if n < period * 2 + 1:
|
||
return None
|
||
|
||
trs: List[float] = []
|
||
plus_dms: List[float] = []
|
||
minus_dms: List[float] = []
|
||
|
||
for i in range(1, n):
|
||
h, l, c_prev = highs[i], lows[i], closes[i - 1]
|
||
h_prev, l_prev = highs[i - 1], lows[i - 1]
|
||
|
||
tr = max(h - l, abs(h - c_prev), abs(l - c_prev))
|
||
up_move = h - h_prev
|
||
down_move = l_prev - l
|
||
|
||
plus_dm = up_move if up_move > down_move and up_move > 0 else 0.0
|
||
minus_dm = down_move if down_move > up_move and down_move > 0 else 0.0
|
||
|
||
trs.append(tr)
|
||
plus_dms.append(plus_dm)
|
||
minus_dms.append(minus_dm)
|
||
|
||
def _wilder_smooth(values: List[float]) -> List[float]:
|
||
"""Wilder 평활: 초기값 = 첫 period개 합산, 이후 재귀 평활."""
|
||
smoothed = [sum(values[:period])]
|
||
for v in values[period:]:
|
||
smoothed.append(smoothed[-1] - smoothed[-1] / period + v)
|
||
return smoothed
|
||
|
||
atr_s = _wilder_smooth(trs)
|
||
plus_dm_s = _wilder_smooth(plus_dms)
|
||
minus_dm_s = _wilder_smooth(minus_dms)
|
||
|
||
dxs: List[float] = []
|
||
for atr_v, p_dm, m_dm in zip(atr_s, plus_dm_s, minus_dm_s):
|
||
if atr_v == 0:
|
||
dxs.append(0.0)
|
||
continue
|
||
plus_di = 100.0 * p_dm / atr_v
|
||
minus_di = 100.0 * m_dm / atr_v
|
||
di_sum = plus_di + minus_di
|
||
dxs.append(100.0 * abs(plus_di - minus_di) / di_sum if di_sum else 0.0)
|
||
|
||
if len(dxs) < period:
|
||
return None
|
||
|
||
adx_s = _wilder_smooth(dxs)
|
||
return round(adx_s[-1], 2)
|
||
|
||
|
||
def detect_market_regime(index_closes: List[float]) -> MarketRegime:
|
||
"""섹션 7.1: 200일 MA + 기울기로 상승장/하락장 판별
|
||
|
||
상승장(Bull): 지수 > MA200 AND MA200 기울기 양(+)
|
||
하락장(Bear): 그 외
|
||
"""
|
||
if len(index_closes) < 205:
|
||
return MarketRegime.BEAR # 데이터 부족 시 보수적
|
||
|
||
ma200_current = sum(index_closes[-200:]) / 200
|
||
ma200_5d_ago = sum(index_closes[-205:-5]) / 200
|
||
slope_positive = ma200_current > ma200_5d_ago
|
||
|
||
is_bull = index_closes[-1] > ma200_current and slope_positive
|
||
return MarketRegime.BULL if is_bull else MarketRegime.BEAR
|
||
|
||
|
||
def calculate_rsi(closes: List[float], period: int = 14) -> Optional[float]:
|
||
"""RSI(Relative Strength Index) 계산"""
|
||
if len(closes) < period + 1:
|
||
return None
|
||
|
||
gains, losses = [], []
|
||
for i in range(-period, 0):
|
||
diff = closes[i] - closes[i - 1]
|
||
gains.append(max(diff, 0))
|
||
losses.append(max(-diff, 0))
|
||
|
||
avg_gain = sum(gains) / period
|
||
avg_loss = sum(losses) / period
|
||
if avg_loss == 0:
|
||
return 100.0
|
||
rs = avg_gain / avg_loss
|
||
return 100 - (100 / (1 + rs))
|
||
|
||
|
||
def calculate_bollinger(closes: List[float], period: int = 20) -> Optional[Tuple[float, float, float]]:
|
||
"""볼린저 밴드 (20일, 2σ) 계산
|
||
|
||
Returns: (upper, middle, lower)
|
||
"""
|
||
if len(closes) < period:
|
||
return None
|
||
window = closes[-period:]
|
||
middle = sum(window) / period
|
||
variance = sum((p - middle) ** 2 for p in window) / period
|
||
std = variance ** 0.5
|
||
return (middle + 2 * std, middle, middle - 2 * std)
|
||
|
||
|
||
def is_reversal_candle(opens: List[float], highs: List[float],
|
||
lows: List[float], closes: List[float]) -> bool:
|
||
"""하락장 매수 조건용 되돌림 캔들 감지 (섹션 7.2)
|
||
|
||
양봉 또는 긴 하단 꼼리(하단 그림자 > 2× 몸통)
|
||
"""
|
||
if not closes:
|
||
return False
|
||
o, h, l, c = opens[-1], highs[-1], lows[-1], closes[-1]
|
||
body = abs(c - o)
|
||
lower_shadow = min(o, c) - l
|
||
is_bullish = c > o
|
||
has_long_lower_shadow = lower_shadow > 2 * max(body, 0.01)
|
||
return is_bullish or has_long_lower_shadow
|
||
|
||
|
||
def check_buy_signal(
|
||
regime: MarketRegime,
|
||
rsi: float,
|
||
close: float,
|
||
bb_upper: float,
|
||
bb_middle: float,
|
||
bb_lower: float,
|
||
avg_volume_20d: float,
|
||
current_volume: int,
|
||
prev_volume: int,
|
||
is_reversal: bool = False,
|
||
) -> Optional[Dict]:
|
||
"""섹안 7.2: 매수 조건 판별
|
||
|
||
[2026-02 개선] BULL RSI 조건 완화: ≤25 → ≤30 (진입 기회 ~2배 증가)
|
||
|
||
Bull: RSI(14) ≤ 30 (과매도) OR 거래량 ≥ 20일평균×200%, 종가가 BB 하단선 -3% ~ +1.5% 구간
|
||
Bear: RSI(14) ≤ 35 (과매도), 종가 ≤ BB 하단선, 되돌림 캔들 출현, 전일 대비 거래량 증가
|
||
|
||
공통: 거래대금 ≥ 20일 평균
|
||
"""
|
||
if current_volume < avg_volume_20d:
|
||
return None
|
||
|
||
if regime == MarketRegime.BULL:
|
||
rsi_ok = rsi <= 30 # ← 25 → 30으로 완화: 진입 기회 ~2배 증가
|
||
vol_ok = current_volume >= avg_volume_20d * 2 # 거래량 20일평균 200% 이상
|
||
if rsi_ok or (vol_ok and rsi <= 50): # vol_ok 단독 시 RSI 50 상한 (과열 진입 차단)
|
||
bb_low_minus = bb_lower * 0.97 # -3%
|
||
bb_low_plus = bb_lower * 1.015 # +1.5%
|
||
if bb_low_minus <= close <= bb_low_plus:
|
||
trigger = f"RSI({rsi:.1f})<=30" if rsi_ok else f"거래량({current_volume / avg_volume_20d * 100:.0f}%>=200%)"
|
||
return {
|
||
"signal": "BUY",
|
||
"strategy": "MEAN_REVERSION_BULL",
|
||
"split_plan": [0.50, 0.50], # 2회 분할 매수
|
||
"description": f"{trigger} + BB 하단 -3%~+1.5% (상승장 과매도)",
|
||
}
|
||
|
||
elif regime == MarketRegime.BEAR:
|
||
if rsi <= 35: # 하락장 조건 유지
|
||
# 하락장 진입 필터: 거래량 전일 대비 증가 확인 (매수세 유입)
|
||
if close <= bb_lower and is_reversal and current_volume > prev_volume:
|
||
return {
|
||
"signal": "BUY",
|
||
"strategy": "MEAN_REVERSION_BEAR",
|
||
"split_plan": [0.30, 0.30, 0.40], # 3회 분할 매수
|
||
"description": f"RSI({rsi:.1f})<=35 + BB 하단 + 되돌림 + 거래량↑ (하락장 과매도)",
|
||
}
|
||
|
||
return None
|
||
|
||
|
||
def check_exit_signal(
|
||
regime: MarketRegime,
|
||
rsi: float,
|
||
close: float,
|
||
bb_upper: float,
|
||
bb_middle: float,
|
||
bb_lower: float,
|
||
entry_price: float,
|
||
atr: float,
|
||
lowest_since_entry: float,
|
||
remaining_pct: float,
|
||
trailing_high: Optional[float] = None, # 진입 후 최고가 (트레일링 스탑용) — 기존 호출부 호환 유지
|
||
) -> Optional[ExitSignal]:
|
||
"""섹션 7.3 + 7.4: 청산 조건 (R:R 약 2:1 목표)
|
||
|
||
[2026-02 개선]
|
||
① 손절: -8% → -6% (타이트 손절로 단위 손실 축소)
|
||
② 1차 익절: RSI 45~50 → 55~65 (더 오래 홀드, 수익 극대화)
|
||
③ 2차 익절: 전량매도 → 50% 부분익절 + 나머지는 트레일링 스탑 전환
|
||
|
||
시햀리오:
|
||
1. 손절 체크 (최우선)
|
||
2. 1차 익절 (remaining_pct > threshold)
|
||
└─ RSI 55~65 회복 + BB중앙선 상회: Bull 30% / Bear 50% 매도
|
||
3. 2차 익절 (1차 완료 후)
|
||
├─ 트레일링 스탑: 취고가 대비 -3% 하락 시 잔여 전량 청산
|
||
└─ BB 상단 도달 시 50% 추가 익절
|
||
"""
|
||
# ── 1. 손절 (최우선) ──────────────────────────────────────────
|
||
atr_stop = lowest_since_entry - 2.5 * atr
|
||
pct_stop = entry_price * 0.94 # -6% (기존 -8% → -6% 타이트하게)
|
||
stop_price = max(atr_stop, pct_stop) # 높은 값 = 타이트한 손절
|
||
|
||
if close <= stop_price:
|
||
return ExitSignal(
|
||
action="SELL_ALL",
|
||
reason=f"손절 (close={close:.0f} <= stop={stop_price:.0f}, ATR={atr:.0f})",
|
||
sell_pct=1.0,
|
||
)
|
||
|
||
# 시장 상태별 찾은 익절 임계값
|
||
# Bull: 1차 30% 후 remaining=0.7, Bear: 1차 50% 후 remaining=0.5
|
||
threshold = 0.7 if regime == MarketRegime.BULL else 0.5
|
||
|
||
# ── 2. 1차 익절: 아직 1차 안 했을 때 ──────────────────────────
|
||
if remaining_pct > threshold:
|
||
if 55 <= rsi <= 65 and close > bb_middle:
|
||
if regime == MarketRegime.BULL:
|
||
return ExitSignal(
|
||
action="SELL_PARTIAL",
|
||
reason=f"1차 익절(Bull 30%): RSI({rsi:.1f}) 55~65 회복 + BB중앙({bb_middle:.0f}) 상회",
|
||
sell_pct=0.30,
|
||
)
|
||
else:
|
||
return ExitSignal(
|
||
action="SELL_PARTIAL",
|
||
reason=f"1차 익절(Bear 50%): RSI({rsi:.1f}) 55~65 회복 + BB중앙({bb_middle:.0f}) 상회",
|
||
sell_pct=0.50,
|
||
)
|
||
|
||
# ── 3. 2차 익절: 1차 완료 후 ─────────────────────────────────────
|
||
if remaining_pct <= threshold:
|
||
|
||
# 3a. 트레일링 스탑: 최고가 대비 -3% 하락 시 잔여 전량 청산
|
||
# close > entry_price 조건 제거 → dead zone 차단
|
||
# (entry 아래로 떨어져도 trailing_high가 entry보다 높으면 즉시 청산)
|
||
if trailing_high is not None and trailing_high > entry_price:
|
||
trail_stop = trailing_high * 0.97
|
||
if close < trail_stop:
|
||
return ExitSignal(
|
||
action="SELL_ALL",
|
||
reason=(
|
||
f"트레일링 청산: close={close:.0f} < 최고가×0.97={trail_stop:.0f} "
|
||
f"(최고가={trailing_high:.0f})"
|
||
),
|
||
sell_pct=1.0,
|
||
)
|
||
|
||
# 3b. BB 상단 도달 시 50% 추가 익절 (나머지는 트레일링으로)
|
||
if rsi >= 65 and close >= bb_upper * 0.99 and remaining_pct > 0.25:
|
||
return ExitSignal(
|
||
action="SELL_PARTIAL",
|
||
reason=f"2차 익절(50%): RSI({rsi:.1f})>=65 + BB상단({bb_upper:.0f}) 도달",
|
||
sell_pct=0.50,
|
||
)
|
||
|
||
return None
|
||
|
||
|
||
def calculate_position_size(
|
||
total_assets: float,
|
||
available_cash: float,
|
||
current_strategy_allocation: float,
|
||
regime: MarketRegime,
|
||
max_per_position: float = 0.04,
|
||
max_strategy_capital_bull: float = 0.70,
|
||
max_strategy_capital_bear: float = 0.30,
|
||
rsi: Optional[float] = None, # 낙음: RSI 낙을수록(신호 강할수록) 비중 증가
|
||
adx: Optional[float] = None, # 낙음: ADX 낙음 횡보장 확인 → 평균회귀 신뢰도 증가
|
||
) -> float:
|
||
"""섹션 6.1 + 7.5: 동적 포지션 사이징 (신호 강도 기반)
|
||
|
||
[2026-02 개선] RSI/ADX 매개변수 추가 (기존 호출부 호환 유지 — 모두 Optional)
|
||
|
||
신호 강도별 비중:
|
||
- RSI ≤ 20 (군도 과매도): 1.5× = 6% (진입 기회에서 가장 적체적)
|
||
- RSI ≤ 25 (강한 신호): 1.25× = 5%
|
||
- RSI ≤ 30 (표준 신호): 1.0× = 4% (default)
|
||
- ADX < 15 (강한 회황장): +0.25× 보너스 (최대 2.0× = 8%)
|
||
|
||
시장별 전략 자본 한도:
|
||
- 상승장(Bull): 전략 전체 최대 70% 운용
|
||
- 하락장(Bear): 전략 전체 최대 30% 운용
|
||
"""
|
||
max_strategy_capital = max_strategy_capital_bull if regime == MarketRegime.BULL else max_strategy_capital_bear
|
||
|
||
# RSI 기반 동적 변수 스케일링
|
||
if rsi is not None:
|
||
if rsi <= 20:
|
||
multiplier = 1.50 # 군도 과매도 → 6%
|
||
elif rsi <= 25:
|
||
multiplier = 1.25 # 강한 신호 → 5%
|
||
else:
|
||
multiplier = 1.00 # 표준 → 4%
|
||
else:
|
||
multiplier = 1.00
|
||
|
||
# ADX < 15: 특돉한 회황장 → +0.25× 비중 보너스 (최대 2.0× = 8%)
|
||
if adx is not None and adx < 15:
|
||
multiplier = min(multiplier + 0.25, 2.0)
|
||
|
||
max_by_position = total_assets * max_per_position * multiplier
|
||
remaining_budget = (total_assets * max_strategy_capital) - current_strategy_allocation
|
||
result = max(0.0, min(max_by_position, remaining_budget, available_cash))
|
||
|
||
if rsi is not None or adx is not None:
|
||
logger.debug(
|
||
"position_size: total=%.0f, multiplier=%.2f×, rsi=%s, adx=%s → %.0f원",
|
||
total_assets, multiplier,
|
||
f"{rsi:.1f}" if rsi is not None else "N/A",
|
||
f"{adx:.1f}" if adx is not None else "N/A",
|
||
result,
|
||
)
|
||
return result
|
||
|
||
|
||
def calculate_position_size_v2(
|
||
total_assets: float,
|
||
available_cash: float,
|
||
current_strategy_allocation: float,
|
||
regime: MarketRegime,
|
||
rsi: Optional[float] = None,
|
||
adx: Optional[float] = None,
|
||
max_per_position: float = 0.04,
|
||
max_strategy_capital_bull: float = 0.70,
|
||
max_strategy_capital_bear: float = 0.30,
|
||
) -> float:
|
||
"""CLAUDE.md 섹션 6.1 + RSI/ADX 동적 포지션 사이징
|
||
|
||
calculate_position_size 기반으로 RSI·ADX 보정 추가:
|
||
- RSI ≤ 20: ×1.2 (깊은 과매도 = 더 유리한 진입)
|
||
- RSI ≤ 25: ×1.1
|
||
- ADX > 30: ×0.8 (강추세 중 평균회귀 위험 = 포지션 축소)
|
||
보정 후에도 종목당 4% 상한 재적용 (CLAUDE.md 절대 룰)
|
||
"""
|
||
base = calculate_position_size(
|
||
total_assets, available_cash, current_strategy_allocation,
|
||
regime, max_per_position, max_strategy_capital_bull, max_strategy_capital_bear,
|
||
rsi=None, adx=None, # v2에서 자체 보정하므로 base는 배율 없이 계산 (이중 보정 방지)
|
||
)
|
||
if base <= 0:
|
||
return 0.0
|
||
|
||
mult = 1.0
|
||
if rsi is not None:
|
||
if rsi <= 20:
|
||
mult *= 1.2
|
||
elif rsi <= 25:
|
||
mult *= 1.1
|
||
|
||
if adx is not None and adx > 30:
|
||
mult *= 0.8
|
||
|
||
max_allowed = total_assets * max_per_position
|
||
return min(base * mult, max_allowed, available_cash)
|
||
|
||
|
||
def check_h1_breakout_with_ma5(
|
||
h1_closes: List[float],
|
||
h1_current_close: float,
|
||
) -> bool:
|
||
"""섹션 7.5: 60분봉 5일선 돌파 확인 (가짜 반등 필터)
|
||
|
||
60분봉 기준 5일(120시간 = 5일×24시간) 이동평균선 상향 돌파 여부 확인
|
||
실제로는 5일 분량의 60분봉 데이터가 필요: 5일 × 6.5시간(장 운영) = 약 30~40개 봉
|
||
|
||
Args:
|
||
h1_closes: 60분봉 종가 리스트 (최소 5개 이상)
|
||
h1_current_close: 현재 60분봉 종가
|
||
|
||
Returns:
|
||
True: 5일선 상향 돌파, False: 미돌파
|
||
|
||
Note:
|
||
- 60분봉 데이터가 충분하지 않으면 False 반환 (보수적)
|
||
- 복합 타임프레임 전략에서 선택적으로 사용
|
||
"""
|
||
if len(h1_closes) < 5:
|
||
return False # 데이터 부족 시 보수적 대응
|
||
|
||
ma5 = sum(h1_closes[-5:]) / 5
|
||
return h1_current_close > ma5
|