159 lines
4.0 KiB
Python
159 lines
4.0 KiB
Python
"""
|
|
Technical Indicators
|
|
|
|
보조지표 계산 함수들
|
|
"""
|
|
from typing import List, Optional, Tuple
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def sma(prices: List[float], period: int) -> Optional[float]:
|
|
"""단순 이동평균 (Simple Moving Average)"""
|
|
if len(prices) < period:
|
|
return None
|
|
return sum(prices[-period:]) / period
|
|
|
|
|
|
def ema(prices: List[float], period: int, prev_ema: Optional[float] = None) -> Optional[float]:
|
|
"""지수 이동평균 (Exponential Moving Average)"""
|
|
if len(prices) < period:
|
|
return None
|
|
|
|
k = 2 / (period + 1)
|
|
if prev_ema is None:
|
|
# 초기값은 SMA
|
|
prev_ema = sum(prices[-period:]) / period
|
|
|
|
return prices[-1] * k + prev_ema * (1 - k)
|
|
|
|
|
|
def rsi(prices: List[float], period: int = 14) -> Optional[float]:
|
|
"""
|
|
RSI (Relative Strength Index)
|
|
|
|
Args:
|
|
prices: 종가 리스트
|
|
period: 기간 (기본 14)
|
|
|
|
Returns:
|
|
0~100 사이 RSI 값
|
|
"""
|
|
if len(prices) < period + 1:
|
|
return None
|
|
|
|
gains, losses = [], []
|
|
for i in range(-period, 0):
|
|
delta = prices[i] - prices[i - 1]
|
|
gains.append(max(delta, 0))
|
|
losses.append(max(-delta, 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 bollinger_bands(prices: List[float], period: int = 20, std_mult: float = 2.0) -> Optional[Tuple[float, float, float]]:
|
|
"""
|
|
볼린저 밴드
|
|
|
|
Returns:
|
|
(upper, middle, lower) 또는 None
|
|
"""
|
|
if len(prices) < period:
|
|
return None
|
|
|
|
window = prices[-period:]
|
|
middle = sum(window) / period
|
|
variance = sum((p - middle) ** 2 for p in window) / period
|
|
std = variance ** 0.5
|
|
|
|
upper = middle + std_mult * std
|
|
lower = middle - std_mult * std
|
|
|
|
return (upper, middle, lower)
|
|
|
|
|
|
def macd(prices: List[float], fast: int = 12, slow: int = 26, signal: int = 9) -> Optional[Tuple[float, float, float]]:
|
|
"""
|
|
MACD (Moving Average Convergence Divergence)
|
|
|
|
Returns:
|
|
(macd_line, signal_line, histogram) 또는 None
|
|
"""
|
|
if len(prices) < slow:
|
|
return None
|
|
|
|
# EMA 계산 (재귀적으로 계산)
|
|
def calc_ema(data: List[float], period: int) -> float:
|
|
k = 2 / (period + 1)
|
|
ema_val = sum(data[:period]) / period
|
|
for p in data[period:]:
|
|
ema_val = p * k + ema_val * (1 - k)
|
|
return ema_val
|
|
|
|
ema_fast = calc_ema(prices, fast)
|
|
ema_slow = calc_ema(prices, slow)
|
|
macd_line = ema_fast - ema_slow
|
|
|
|
# Signal line = MACD의 EMA(9)
|
|
# 실제로는 과거 MACD 값들 필요 (간략화)
|
|
signal_line = macd_line # TODO: 정확한 구현 필요
|
|
histogram = macd_line - signal_line
|
|
|
|
return (macd_line, signal_line, histogram)
|
|
|
|
|
|
def atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]:
|
|
"""
|
|
ATR (Average True Range)
|
|
|
|
Args:
|
|
highs: 고가 리스트
|
|
lows: 저가 리스트
|
|
closes: 종가 리스트
|
|
period: 기간
|
|
|
|
Returns:
|
|
ATR 값
|
|
"""
|
|
if len(highs) < period + 1 or len(lows) < period + 1 or len(closes) < period + 1:
|
|
return None
|
|
|
|
trs = []
|
|
for i in range(1, len(highs)):
|
|
h = highs[i]
|
|
l = lows[i]
|
|
cp = closes[i - 1]
|
|
tr = max(h - l, abs(h - cp), abs(l - cp))
|
|
trs.append(tr)
|
|
|
|
if len(trs) < period:
|
|
return None
|
|
|
|
return sum(trs[-period:]) / period
|
|
|
|
|
|
def vwap(prices: List[float], volumes: List[float]) -> Optional[float]:
|
|
"""
|
|
VWAP (Volume Weighted Average Price)
|
|
|
|
Returns:
|
|
VWAP 값
|
|
"""
|
|
if len(prices) != len(volumes) or not prices:
|
|
return None
|
|
|
|
total_volume = sum(volumes)
|
|
if total_volume == 0:
|
|
return None
|
|
|
|
total_pv = sum(p * v for p, v in zip(prices, volumes))
|
|
return total_pv / total_volume
|