hyowons
84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
"""기술지표 — 순수 함수. 일봉 리스트(시간 오름차순)나 종가 리스트를 받아 계산만 한다.
|
|
|
|
입력 candle dict 키: open/high/low/close/volume (daily_candles_cache.get_candles 형식).
|
|
모든 함수는 데이터 부족 시 None 을 반환한다 (raise X).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
|
|
def sma(values: list[float], period: int) -> float | None:
|
|
"""최근 period 개 단순이동평균."""
|
|
if period <= 0 or len(values) < period:
|
|
return None
|
|
return sum(values[-period:]) / period
|
|
|
|
|
|
def rsi(closes: list[float], period: int = 14) -> float | None:
|
|
"""Wilder RSI. closes 오름차순, period+1 개 이상 필요."""
|
|
if len(closes) < period + 1:
|
|
return None
|
|
gains = losses = 0.0
|
|
for i in range(1, period + 1):
|
|
diff = closes[i] - closes[i - 1]
|
|
if diff >= 0:
|
|
gains += diff
|
|
else:
|
|
losses -= diff
|
|
avg_gain = gains / period
|
|
avg_loss = losses / period
|
|
for i in range(period + 1, len(closes)):
|
|
diff = closes[i] - closes[i - 1]
|
|
gain = diff if diff > 0 else 0.0
|
|
loss = -diff if diff < 0 else 0.0
|
|
avg_gain = (avg_gain * (period - 1) + gain) / period
|
|
avg_loss = (avg_loss * (period - 1) + loss) / period
|
|
if avg_loss == 0:
|
|
return 100.0
|
|
rs = avg_gain / avg_loss
|
|
return 100.0 - (100.0 / (1.0 + rs))
|
|
|
|
|
|
def atr(candles: list[dict], period: int = 14) -> float | None:
|
|
"""Average True Range (Wilder). candles 오름차순, period+1 개 이상 필요."""
|
|
if len(candles) < period + 1:
|
|
return None
|
|
trs: list[float] = []
|
|
for i in range(1, len(candles)):
|
|
h = candles[i]['high']
|
|
lo = candles[i]['low']
|
|
prev_close = candles[i - 1]['close']
|
|
trs.append(max(h - lo, abs(h - prev_close), abs(lo - prev_close)))
|
|
if len(trs) < period:
|
|
return None
|
|
atr_val = sum(trs[:period]) / period
|
|
for tr in trs[period:]:
|
|
atr_val = (atr_val * (period - 1) + tr) / period
|
|
return atr_val
|
|
|
|
|
|
def recent_high(candles: list[dict], period: int, exclude_last: bool = True) -> float | None:
|
|
"""최근 period 봉 최고가. exclude_last 면 마지막(오늘) 봉 제외 → 돌파 판정용."""
|
|
series = candles[:-1] if exclude_last else candles
|
|
window = series[-period:]
|
|
if not window:
|
|
return None
|
|
return max(c['high'] for c in window)
|
|
|
|
|
|
def recent_low(candles: list[dict], period: int, exclude_last: bool = True) -> float | None:
|
|
"""최근 period 봉 최저가 — 손절선 후보."""
|
|
series = candles[:-1] if exclude_last else candles
|
|
window = series[-period:]
|
|
if not window:
|
|
return None
|
|
return min(c['low'] for c in window)
|
|
|
|
|
|
def volume_avg(candles: list[dict], period: int, exclude_last: bool = True) -> float | None:
|
|
"""최근 period 봉 평균 거래량. exclude_last 면 오늘 제외."""
|
|
series = candles[:-1] if exclude_last else candles
|
|
window = series[-period:]
|
|
if len(window) < period:
|
|
return None
|
|
return sum(c['volume'] for c in window) / period
|