"""지표 계산 코어 (순수 계산 — I/O·DB·네트워크·설정 의존성 없음) CLAUDE.md §8.5-1 "RSI/BB/ATR 계산 3벌 중복" 통합용 단일 출처(single source of truth). ### 설계 원칙 1. **알고리즘 이름을 함수명에 명시한다.** 스무딩 방식이 다르면 값도 다르므로 `rsi_wilder` / `rsi_sma` 처럼 분리해 유지한다. 억지로 하나로 합치지 않는다. 2. **순수 함수.** settings / DB / 네트워크를 참조하지 않는다. period·계수는 인자로 받는다. 3. **기존 값 보존.** 이 모듈의 함수는 기존 3벌 구현의 수치를 **비트 단위로 재현**하는 것이 목적이다. "더 정확한 계산"으로의 개선은 백테스트를 동반한 별도 작업이다. ### 입력 규약 - 모든 리스트는 **오래된 → 최신** 순서(시계열 오름차순). `[-1]`이 최신 봉. - 데이터 부족 시 예외 대신 `None` 반환. ### 대응 관계 (통합 전 구현 → 이 모듈) | 통합 전 | 이 모듈 | |---------|---------| | `mean_reversion_strategy.calculate_rsi` (Wilder 평활) | `rsi_wilder` | | `technical_indicators.analyze_rsi` 인라인 (period 구간 단순평균) | `rsi_sma` | | `mean_reversion_strategy.calculate_atr` (TR의 SMA) | `atr_sma` | | `mean_reversion_strategy.calculate_bollinger` (2σ 고정) | `bollinger_sma(..., num_std=2.0)` | | `technical_indicators.analyze_bollinger` 인라인 (σ배수 설정값) | `bollinger_sma(..., num_std=<설정>)` | | `technical_indicators.analyze_rsi` 인라인 BB (20봉·2σ 하드코딩) | `bollinger_sma(..., 20, 2.0)` | | `mean_reversion_strategy.calculate_adx` (Wilder) | `adx_wilder` | | `day_trading_strategy._rsi/_bollinger/_atr` (이미 위임 상태) | 동일 경로 재사용 | """ from typing import List, Optional, Tuple __all__ = [ "true_ranges", "rsi_wilder", "rsi_sma", "atr_sma", "bollinger_sma", "adx_wilder", ] # ── 공통 헬퍼 ──────────────────────────────────────────────────────────────── def true_ranges(highs: List[float], lows: List[float], closes: List[float]) -> List[float]: """TR 시계열 계산. TR = max(H-L, |H-Cprev|, |L-Cprev|) 길이는 `len(highs) - 1` (첫 봉은 이전 종가가 없어 제외). """ out: List[float] = [] for i in range(1, len(highs)): h, l, c_prev = highs[i], lows[i], closes[i - 1] out.append(max(h - l, abs(h - c_prev), abs(l - c_prev))) return out # ── RSI ────────────────────────────────────────────────────────────────────── def rsi_wilder(closes: List[float], period: int = 14) -> Optional[float]: """RSI — Wilder 평활법 (mean_reversion_strategy.calculate_rsi 와 동일) - 시드: 첫 `period`개 diff 의 단순평균 - 이후: avg = (prev_avg * (period-1) + current) / period - avg_loss == 0 → 100.0 - len(closes) < period+1 → None 주의: `len(closes) == period + 1` 인 경우 평활 루프가 돌지 않으므로 `rsi_sma()` 와 값이 **정확히 동일**하다. 두 함수가 갈라지는 것은 입력이 period+1 보다 길 때뿐이다. """ if len(closes) < period + 1: return None diffs = [closes[i] - closes[i - 1] for i in range(1, len(closes))] avg_gain = sum(max(d, 0) for d in diffs[:period]) / period avg_loss = sum(max(-d, 0) for d in diffs[:period]) / period for i in range(period, len(diffs)): avg_gain = (avg_gain * (period - 1) + max(diffs[i], 0)) / period avg_loss = (avg_loss * (period - 1) + max(-diffs[i], 0)) / period if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) def rsi_sma(closes: List[float], period: int = 14) -> Optional[float]: """RSI — 단순평균(SMA) 방식 (technical_indicators.analyze_rsi 인라인 구현과 동일) - **마지막 `period`개 diff** 만 단순평균 (평활 없음, Wilder 아님) - avg_loss == 0 → 100.0 - len(closes) < period+1 → None 금융공학적 표준(Wilder)과 다르지만, 추천/분석 엔진의 기존 신호 수치를 보존하기 위해 별도 함수로 유지한다. """ if len(closes) < period + 1: return None gains: List[float] = [] losses: List[float] = [] for i in range(1, len(closes)): change = closes[i] - closes[i - 1] if change > 0: gains.append(change) losses.append(0) else: gains.append(0) losses.append(abs(change)) avg_gain = sum(gains[-period:]) / period avg_loss = sum(losses[-period:]) / period if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) # ── ATR ────────────────────────────────────────────────────────────────────── def atr_sma(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]: """ATR — TR 의 단순이동평균 (mean_reversion_strategy.calculate_atr 와 동일) - ATR = mean(TR[-period:]) (Wilder 평활 아님) - highs/lows/closes 중 하나라도 len < period+1 → None - TR 개수 < period → None """ if len(highs) < period + 1 or len(lows) < period + 1 or len(closes) < period + 1: return None trs = true_ranges(highs, lows, closes) if len(trs) < period: return None return sum(trs[-period:]) / period # ── 볼린저 밴드 ────────────────────────────────────────────────────────────── def bollinger_sma(closes: List[float], period: int = 20, num_std: float = 2.0) -> Optional[Tuple[float, float, float]]: """볼린저 밴드 — SMA 중앙선 + 모표준편차(ddof=0) Returns: (upper, middle, lower) — 기존 `calculate_bollinger` 반환 순서 유지 - middle = mean(closes[-period:]) - std = sqrt( mean((p - middle)^2) ) ← 모분산(ddof=0), 표본분산 아님 - len(closes) < period → None """ 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 + num_std * std, middle, middle - num_std * std) # ── ADX ────────────────────────────────────────────────────────────────────── def adx_wilder(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]: """Wilder's ADX (mean_reversion_strategy.calculate_adx 와 동일) 1. TR / +DM / -DM 계산 2. TR·DM 평활: 초기값 = 첫 period 개 **합산**, 이후 `s - s/period + v` 3. +DI = 100*(+DM_s / TR_s), -DI 동일 4. DX = 100*|+DI - -DI| / (+DI + -DI) 5. ADX 평활: 초기값 = mean(DX[:period]), 이후 `a - a/period + dx/period` (DM/TR 과 달리 **평균** 시드 + 정규화된 갱신) - len < period*2 + 1 → None / DX 개수 < period → None - 반환값은 `round(..., 2)` """ n = len(highs) 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_val = sum(dxs[:period]) / period # 초기값 = 평균 for dx_v in dxs[period:]: adx_val = adx_val - adx_val / period + dx_v / period return round(adx_val, 2)