add checking volatility breaking out

smtm/strategy/strategy_hey.py

@@ -1,4 +1,6 @@
-"""이동 평균선, 변동성 돌파 전략등 주요 이벤트 발생시 알림을 전달하는 전략 클래스"""
+"""이동 평균선, 변동성 돌파 전략등 주요 이벤트 발생시 알림을 전달하는 전략 클래스
+    변동성 돌파 이벤트를 분봉을 기준으로 했을때, 너무 자주 발생됨 -> 주석 처리
+"""
 
 import copy
 from .strategy_sas import StrategySas
@@ -26,6 +28,8 @@ class StrategyHey(StrategySas):
     SMA_MID = 40
     SMA_LONG = 120
     MIN_MARGIN = 0.002
+    ATR_PERIOD = 30
+    VOLATILITY_BREAKOUT = 1.5
 
     def __init__(self):
         self.is_intialized = False
@@ -38,11 +42,16 @@ def __init__(self):
         self.min_price = 0
         self.result = []
         self.current_process = "ready"
+        self.loss_cut_alerted = False
         self.closing_price_list = []
         self.buy_price = 0
         self.logger = LogManager.get_logger(__class__.__name__)
         self.waiting_requests = {}
         self.add_spot_callback = None
+        self.add_line_callback = None
+        self.true_ranges = []
+        self.atr = None
+        self.prev_close = None
 
     def initialize(
         self,
@@ -64,6 +73,7 @@ def initialize(
         self.min_price = min_price
         self.add_spot_callback = add_spot_callback
         self.alert_callback = alert_callback
+        self.add_line_callback = add_line_callback
 
     def update_trading_info(self, info):
         """새로운 거래 정보를 업데이트
@@ -93,6 +103,8 @@ def update_trading_info(self, info):
 
         self.data.append(copy.deepcopy(target))
         self._checking_sma(target)
+        # 변동성 돌파 이벤트를 분봉을 기준으로 했을때, 너무 자주 발생됨
+        # self._checking_volitility_breakout(target)
 
     def _checking_sma(self, info):
         current_price = info["closing_price"]
@@ -118,13 +130,16 @@ def _checking_sma(self, info):
         if (sma_short > sma_long > sma_mid) and self.current_process != "buy":
             self.current_process = "buy"
             self.buy_price = current_price
+            self.loss_cut_alerted = False
             self.logger.debug(f"[HEY] BUY #{current_idx} {sma_short} : {sma_mid} : {sma_long}")
-        elif (
-            sma_short < sma_mid < sma_long or self._is_loss_cut_entered(current_price)
-        ) and self.current_process != "sell":
+        elif (sma_short < sma_mid < sma_long) and self.current_process != "sell":
             self.current_process = "sell"
             self.buy_price = 0
+            self.loss_cut_alerted = False
             self.logger.debug(f"[HEY] SELL #{current_idx} {sma_short} : {sma_mid} : {sma_long}")
+        elif self._is_loss_cut_entered(current_price) and self.current_process != "sell" and not self.loss_cut_alerted:
+            self.loss_cut_alerted = True
+            self.logger.debug(f"[HEY] LOSS CUT #{current_idx} {sma_short} : {sma_mid} : {sma_long}")
         else:
             return
 
@@ -134,6 +149,61 @@ def _checking_sma(self, info):
             f"[HEY] SMA #{current_idx} {self.current_process} : {current_price}",
         )
 
+    def _checking_volitility_breakout(self, info):
+        self.update_atr_info(info)
+        breakout_buy_signal, breakout_sell_signal = self.detect_breakout_signals()
+        if breakout_buy_signal or breakout_sell_signal:
+            self.logger.debug(f"[HEY] BREAKOUT {info['date_time']} {breakout_buy_signal} {breakout_sell_signal}")
+            self.logger.debug(f"-- {self.atr} {info['closing_price']} {self.prev_close}")
+            self._make_alert(
+                info["date_time"],
+                info["closing_price"],
+                f"[HEY] BREAKOUT {info['date_time']} {breakout_buy_signal} {breakout_sell_signal}",
+            )
+
+    def update_atr_info(self, new_price_info):
+        """
+        새로운 거래 정보를 받아서 변동성 돌파 이벤트 정보를 업데이트
+        """
+
+        if len(self.data) > 1:
+            # 이전 거래일 종가
+            self.prev_close = self.data[-2]['closing_price']
+
+            # 새로운 True Range 계산
+            current_high = new_price_info['high_price']
+            current_low = new_price_info['low_price']
+            prev_close = self.data[-2]['closing_price']
+
+            high_low = current_high - current_low
+            high_close = abs(current_high - prev_close)
+            low_close = abs(current_low - prev_close)
+
+            true_range = max(high_low, high_close, low_close)
+            self.true_ranges.append(true_range)
+
+            # 최신 True Range 기반으로 ATR 업데이트
+            if len(self.true_ranges) > self.ATR_PERIOD:
+                self.true_ranges.pop(0)  # 가장 오래된 True Range 제거
+
+            self.atr = np.mean(self.true_ranges)
+
+    def detect_breakout_signals(self):
+        """
+        변동성 돌파 신호 감지
+        """
+        if len(self.data) < 2:
+            return False, False
+
+        current_price = self.data[-1]
+        current_high = current_price['high_price']
+        current_low = current_price['low_price']
+
+        breakout_buy_signal = current_high > self.prev_close + self.VOLATILITY_BREAKOUT * self.atr
+        breakout_sell_signal = current_low < self.prev_close - self.VOLATILITY_BREAKOUT * self.atr
+
+        return breakout_buy_signal, breakout_sell_signal
+
     def _is_loss_cut_entered(self, current_price):
         if self.buy_price * (1 - self.MIN_MARGIN) > current_price:
             self.logger.info(f"[loss cut] loss! {current_price}")

tests/strategy_hey_test.py

@@ -1,7 +1,7 @@
 import unittest
 from smtm import StrategyHey
 from unittest.mock import *
-
+import numpy as np
 
 class StrategyHeyTests(unittest.TestCase):
     def test_initialize_update_initial_balance(self):
@@ -152,3 +152,80 @@ def test_get_request_return_None(self):
         sas.initialize(100, 10)
         requests = sas.get_request()
         self.assertEqual(requests, None)
+
+    def test_update_atr_info_should_update_correctly(self):
+        detector = StrategyHey()
+        detector.ATR_PERIOD = 3
+
+        # 초기 상태 확인
+        self.assertEqual(len(detector.data), 0)
+        self.assertIsNone(detector.prev_close)
+        self.assertIsNone(detector.atr)
+
+        # 새로운 거래 정보 추가 및 업데이트
+        new_price_info_1 = {"date_time": "2024-04-25T09:00:00", "opening_price": 10000, "high_price": 10500, "low_price": 9800, "closing_price": 10300}
+        detector.data.append(new_price_info_1)
+        detector.update_atr_info(new_price_info_1)
+
+        self.assertEqual(len(detector.data), 1)
+        self.assertIsNone(detector.prev_close)
+        self.assertIsNone(detector.atr)  # 최소 2개의 거래 정보가 필요함
+
+        new_price_info_2 = {"date_time": "2024-04-25T09:30:00", "opening_price": 10300, "high_price": 10600, "low_price": 10000, "closing_price": 10450}
+        detector.data.append(new_price_info_2)
+        detector.update_atr_info(new_price_info_2)
+
+        self.assertEqual(len(detector.data), 2)
+        self.assertEqual(detector.prev_close, 10300)  # 이전 거래일 종가 업데이트 확인
+        self.assertEqual(detector.atr, 600.0)
+
+        new_price_info_3 = {"date_time": "2024-04-25T10:00:00", "opening_price": 10450, "high_price": 10700, "low_price": 10200, "closing_price": 10580}
+        detector.data.append(new_price_info_3)
+        detector.update_atr_info(new_price_info_3)
+
+        self.assertEqual(len(detector.data), 3)
+        self.assertEqual(detector.prev_close, 10450)
+        self.assertEqual(detector.atr, 550.0)
+
+    def test_detect_breakout_signals_should_return_correct_value(self):
+        detector = StrategyHey()
+        detector.ATR_PERIOD = 3
+        detector.VOLATILITY_BREAKOUT = 1.2
+
+        # 변동성 돌파 신호 감지 테스트
+        new_price_info_1 = {"date_time": "2024-04-25T09:00:00", "opening_price": 10000, "high_price": 10500, "low_price": 9800, "closing_price": 10300}
+        new_price_info_2 = {"date_time": "2024-04-25T09:30:00", "opening_price": 10300, "high_price": 10600, "low_price": 10000, "closing_price": 10450}
+
+        detector.data.append(new_price_info_1)
+        detector.update_atr_info(new_price_info_1)
+
+        detector.data.append(new_price_info_2)
+        detector.update_atr_info(new_price_info_2)
+
+        breakout_buy_signal, breakout_sell_signal = detector.detect_breakout_signals()
+
+        # 최소 2개의 거래 정보가 필요하므로 초기값은 False 여야 함
+        self.assertFalse(breakout_buy_signal)
+        self.assertFalse(breakout_sell_signal)
+
+        # 추가 정보로 변동성 돌파 신호 확인
+        new_price_info_3 = {"date_time": "2024-04-25T10:00:00", "opening_price": 10450, "high_price": 12700, "low_price": 10200, "closing_price": 10580}
+        detector.data.append(new_price_info_3)
+        detector.update_atr_info(new_price_info_3)
+
+        breakout_buy_signal, breakout_sell_signal = detector.detect_breakout_signals()
+
+        # 변동성 돌파 신호 예상 확인
+        self.assertTrue(breakout_buy_signal)
+        self.assertFalse(breakout_sell_signal)
+
+        # 추가 정보로 변동성 돌파 신호 확인
+        new_price_info_4 = {"date_time": "2024-04-25T10:30:00", "opening_price": 10580, "high_price": 10800, "low_price": 8200, "closing_price": 10650}
+        detector.data.append(new_price_info_4)
+        detector.update_atr_info(new_price_info_4)
+
+        breakout_buy_signal, breakout_sell_signal = detector.detect_breakout_signals()
+
+        # 변동성 돌파 신호 예상 확인
+        self.assertFalse(breakout_buy_signal)
+        self.assertTrue(breakout_sell_signal)