58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
WBS-9.3: RSI14 자동 충전 절차
|
|
|
|
조건: rsi_14 IS NULL AND close_price IS NOT NULL
|
|
"""
|
|
|
|
def calculate_rsi14(closes: list[float], period: int = 14) -> float:
|
|
"""
|
|
Calculate RSI (Relative Strength Index) for 14 days.
|
|
|
|
입력: close 가격 리스트 (최소 14일)
|
|
출력: RSI14 (0-100)
|
|
"""
|
|
if len(closes) < period:
|
|
return None
|
|
|
|
# 가격 변화 계산
|
|
deltas = [closes[i] - closes[i - 1] for i in range(1, len(closes))]
|
|
|
|
# Gains/Losses 분리
|
|
gains = [d if d > 0 else 0 for d in deltas]
|
|
losses = [abs(d) if d < 0 else 0 for d in deltas]
|
|
|
|
# 평균 계산 (간단한 이동평균)
|
|
avg_gain = sum(gains[-period:]) / period if period <= len(gains) else sum(gains) / len(gains)
|
|
avg_loss = sum(losses[-period:]) / period if period <= len(losses) else sum(losses) / len(losses)
|
|
|
|
if avg_loss == 0:
|
|
return 100 if avg_gain > 0 else 50
|
|
|
|
rs = avg_gain / avg_loss
|
|
rsi = 100 - (100 / (1 + rs))
|
|
|
|
return round(rsi, 2)
|
|
|
|
|
|
def auto_fill_rsi14(row: dict, historical_closes: list[float] = None) -> dict:
|
|
"""
|
|
자동 충전: rsi_14 필드
|
|
|
|
입력:
|
|
row: 현재 행 (rsi_14 = None)
|
|
historical_closes: 과거 close 가격 (최소 14일)
|
|
|
|
출력:
|
|
row (rsi_14 채워짐) 또는 원본 (실패시)
|
|
"""
|
|
if not historical_closes or row.get("rsi_14") is not None:
|
|
return row
|
|
|
|
rsi14 = calculate_rsi14(historical_closes)
|
|
if rsi14 is not None:
|
|
row["rsi_14"] = rsi14
|
|
row["_fill_source"] = "auto_fill_rsi14_v1"
|
|
|
|
return row
|