diff --git a/signal_v2/momentum_classifier.py b/signal_v2/momentum_classifier.py new file mode 100644 index 0000000..3488f4d --- /dev/null +++ b/signal_v2/momentum_classifier.py @@ -0,0 +1,69 @@ +"""분봉 OHLCV → 5-level 모멘텀 분류.""" +from __future__ import annotations +from collections import deque + +# 분류 카테고리 +STRONG_UP = "strong_up" +WEAK_UP = "weak_up" +NEUTRAL = "neutral" +WEAK_DOWN = "weak_down" +STRONG_DOWN = "strong_down" + +_BARS_PER_5MIN = 5 +_LOOKBACK_5MIN_BARS = 5 +_VOLUME_AVG_WINDOW = 12 # 60분 = 5분봉 12개 + + +def aggregate_1min_to_5min(minute_bars: list[dict]) -> list[dict]: + """1분봉 N개 → 5분봉 floor(N/5) 개. 시간 오름차순. + + 각 5분봉: open=첫 1분봉 open, high=max, low=min, close=마지막 close, volume=sum. + """ + bars_5min = [] + chunks = len(minute_bars) // _BARS_PER_5MIN + for i in range(chunks): + chunk = minute_bars[i * _BARS_PER_5MIN : (i + 1) * _BARS_PER_5MIN] + bars_5min.append({ + "datetime": chunk[0]["datetime"], + "open": chunk[0]["open"], + "high": max(b["high"] for b in chunk), + "low": min(b["low"] for b in chunk), + "close": chunk[-1]["close"], + "volume": sum(b["volume"] for b in chunk), + }) + return bars_5min + + +def classify_minute_momentum(minute_bars: deque) -> str: + """1분봉 deque → 5-level 모멘텀 분류. + + Returns: STRONG_UP / WEAK_UP / NEUTRAL / WEAK_DOWN / STRONG_DOWN + """ + minute_list = list(minute_bars) + if len(minute_list) < _BARS_PER_5MIN * _LOOKBACK_5MIN_BARS: + return NEUTRAL # 데이터 부족 + + bars_5min = aggregate_1min_to_5min(minute_list) + if len(bars_5min) < _LOOKBACK_5MIN_BARS: + return NEUTRAL + + recent = bars_5min[-_LOOKBACK_5MIN_BARS:] + up_count = sum(1 for b in recent if b["close"] > b["open"]) + + # 거래량 multiplier: recent 5 avg vs 60분 avg + recent_vol_avg = sum(b["volume"] for b in recent) / len(recent) + long_window = bars_5min[-_VOLUME_AVG_WINDOW:] + long_vol_avg = sum(b["volume"] for b in long_window) / len(long_window) + vol_mult = recent_vol_avg / long_vol_avg if long_vol_avg > 0 else 1.0 + + # 5-level 분류 + if up_count == 5 and vol_mult >= 1.5: + return STRONG_UP + elif up_count >= 3 and vol_mult >= 1.0: + return WEAK_UP + elif up_count == 0 and vol_mult >= 1.5: + return STRONG_DOWN + elif up_count <= 2 and vol_mult < 1.0: + return WEAK_DOWN + else: + return NEUTRAL diff --git a/signal_v2/tests/test_momentum_classifier.py b/signal_v2/tests/test_momentum_classifier.py new file mode 100644 index 0000000..a23e9d2 --- /dev/null +++ b/signal_v2/tests/test_momentum_classifier.py @@ -0,0 +1,92 @@ +"""Tests for minute momentum classifier.""" +from collections import deque + +from signal_v2.momentum_classifier import ( + aggregate_1min_to_5min, classify_minute_momentum, + STRONG_UP, WEAK_UP, NEUTRAL, WEAK_DOWN, STRONG_DOWN, +) + + +def _bar(open_, high, low, close, volume): + return { + "datetime": "2026-05-18T09:00:00+09:00", + "open": open_, "high": high, "low": low, "close": close, "volume": volume, + } + + +def _make_chunks(num_chunks_up: int, num_chunks_total: int, base_vol: int = 1000): + """num_chunks_total 개의 5-bar 청크. num_chunks_up 청크는 양봉, 나머지는 음봉. + 각 청크는 5개 1분봉. 거래량 = base_vol per bar. + """ + bars = [] + for i in range(num_chunks_total): + is_up = i < num_chunks_up + o, c = (100, 110) if is_up else (110, 100) + for j in range(5): + bars.append(_bar(o, max(o, c) + 5, min(o, c) - 5, c, base_vol)) + return bars + + +def test_strong_up_5_consecutive_green_with_high_volume(): + """직전 5개 5분봉 모두 양봉 + 거래량 1.5x → STRONG_UP.""" + # 60분 (12 5분봉) 데이터: 7 normal + 5 high-vol up + older = _make_chunks(num_chunks_up=3, num_chunks_total=7, base_vol=1000) + recent = _make_chunks(num_chunks_up=5, num_chunks_total=5, base_vol=2500) + minute_bars = deque(older + recent, maxlen=60) + assert classify_minute_momentum(minute_bars) == STRONG_UP + + +def test_weak_up_3of5_green_normal_volume(): + """직전 5개 5분봉 중 3-4개 양봉 + 거래량 ≥ 1.0x → WEAK_UP.""" + older = _make_chunks(num_chunks_up=3, num_chunks_total=7, base_vol=1000) + # 5 chunks: 3 up + 2 down, normal vol + recent_up = _make_chunks(num_chunks_up=3, num_chunks_total=3, base_vol=1000) + recent_down = _make_chunks(num_chunks_up=0, num_chunks_total=2, base_vol=1000) + minute_bars = deque(older + recent_up + recent_down, maxlen=60) + assert classify_minute_momentum(minute_bars) == WEAK_UP + + +def test_neutral_mixed(): + """up_count=2, vol normal → NEUTRAL (rule 미해당).""" + older = _make_chunks(num_chunks_up=3, num_chunks_total=7, base_vol=1000) + recent_up = _make_chunks(num_chunks_up=2, num_chunks_total=2, base_vol=1000) + recent_down = _make_chunks(num_chunks_up=0, num_chunks_total=3, base_vol=1000) + minute_bars = deque(older + recent_up + recent_down, maxlen=60) + # up_count=2, vol_mult=1.0 → 어느 분기 조건도 만족 안 함 → NEUTRAL + assert classify_minute_momentum(minute_bars) == NEUTRAL + + +def test_weak_down_low_green_low_volume(): + """up_count <= 2 + vol < 1.0 → WEAK_DOWN.""" + older = _make_chunks(num_chunks_up=3, num_chunks_total=7, base_vol=1000) + recent_up = _make_chunks(num_chunks_up=1, num_chunks_total=1, base_vol=500) + recent_down = _make_chunks(num_chunks_up=0, num_chunks_total=4, base_vol=500) + minute_bars = deque(older + recent_up + recent_down, maxlen=60) + # recent 5 chunks avg vol = 500, long 12 avg ≈ (7*1000 + 5*500) / 12 ≈ 791 → vol_mult ≈ 0.63 + assert classify_minute_momentum(minute_bars) == WEAK_DOWN + + +def test_strong_down_5_consecutive_red_high_volume(): + """직전 5개 5분봉 모두 음봉 + 거래량 1.5x → STRONG_DOWN.""" + older = _make_chunks(num_chunks_up=3, num_chunks_total=7, base_vol=1000) + recent = _make_chunks(num_chunks_up=0, num_chunks_total=5, base_vol=2500) + minute_bars = deque(older + recent, maxlen=60) + assert classify_minute_momentum(minute_bars) == STRONG_DOWN + + +def test_aggregate_1min_to_5min_correctness(): + """5 1분봉 → 1개 5분봉 — open/close/high/low/volume 정확.""" + bars = [ + _bar(100, 105, 99, 102, 1000), + _bar(102, 108, 101, 107, 1500), + _bar(107, 110, 105, 106, 800), + _bar(106, 109, 104, 108, 1200), + _bar(108, 112, 107, 111, 900), + ] + result = aggregate_1min_to_5min(bars) + assert len(result) == 1 + assert result[0]["open"] == 100 # 첫 bar + assert result[0]["close"] == 111 # 마지막 bar + assert result[0]["high"] == 112 # max + assert result[0]["low"] == 99 # min + assert result[0]["volume"] == 5400 # sum