feat(trade-monitor): 순수 지표 모듈 (sma/rsi/highest_high)

services/trade-monitor/indicators.py

@@ -0,0 +1,38 @@
+"""순수 TA 지표 — sma / rsi_series / highest_high."""
+from __future__ import annotations
+
+
+def sma(values: list[float], period: int) -> float | None:
+    if period <= 0 or len(values) < period:
+        return None
+    return sum(values[-period:]) / period
+
+
+def highest_high(highs: list[float], period: int) -> float | None:
+    if period <= 0 or len(highs) < period:
+        return None
+    return max(highs[-period:])
+
+
+def rsi_series(closes: list[float], period: int = 14) -> list[float]:
+    """Wilder RSI. 반환 리스트는 closes[period:]에 1:1 정렬. 부족하면 []."""
+    if len(closes) <= period:
+        return []
+    deltas = [closes[i] - closes[i - 1] for i in range(1, len(closes))]
+    gains = [d if d > 0 else 0.0 for d in deltas]
+    losses = [-d if d < 0 else 0.0 for d in deltas]
+
+    def _rsi(ag: float, al: float) -> float:
+        if al == 0:
+            return 100.0
+        rs = ag / al
+        return 100.0 - 100.0 / (1.0 + rs)
+
+    avg_gain = sum(gains[:period]) / period
+    avg_loss = sum(losses[:period]) / period
+    out = [_rsi(avg_gain, avg_loss)]
+    for i in range(period, len(deltas)):
+        avg_gain = (avg_gain * (period - 1) + gains[i]) / period
+        avg_loss = (avg_loss * (period - 1) + losses[i]) / period
+        out.append(_rsi(avg_gain, avg_loss))
+    return out

services/trade-monitor/tests/test_indicators.py

@@ -0,0 +1,39 @@
+"""indicators — 순수 수치 검증."""
+from indicators import sma, rsi_series, highest_high
+
+
+def test_sma_basic():
+    assert sma([1, 2, 3, 4, 5], 5) == 3.0
+    assert sma([1, 2, 3, 4, 5], 2) == 4.5
+
+
+def test_sma_insufficient():
+    assert sma([1, 2], 5) is None
+    assert sma([], 3) is None
+
+
+def test_highest_high():
+    assert highest_high([1, 9, 3, 4], 3) == 9
+    assert highest_high([1, 2, 3], 3) == 3
+    assert highest_high([1, 2], 3) is None
+
+
+def test_rsi_all_gains_is_100():
+    # 단조 증가 → 손실 0 → RSI 100
+    closes = [float(i) for i in range(1, 20)]
+    rs = rsi_series(closes, 14)
+    assert rs, "series should not be empty"
+    assert rs[-1] == 100.0
+
+
+def test_rsi_insufficient():
+    assert rsi_series([1, 2, 3], 14) == []
+
+
+def test_rsi_known_range():
+    # 등락 섞인 시계열 → RSI는 0~100 사이
+    closes = [10, 11, 10.5, 11.5, 11, 12, 11.8, 12.5, 12, 13,
+              12.7, 13.2, 12.9, 13.5, 13.1, 13.8]
+    rs = rsi_series(closes, 14)
+    assert len(rs) == len(closes) - 14
+    assert all(0.0 <= v <= 100.0 for v in rs)