Coverage for strategies / GapUpPullback_strategy.py: 100%

1# strategies/GapUpPullback_strategy.py 

2 

3import logging 

4from interfaces.strategy import Strategy 

5from typing import List, Dict, Optional 

6from core.logger import get_strategy_logger 

7 

8class GapUpPullbackStrategy(Strategy): 

9 def __init__( 

10 self, 

11 broker, 

12 min_gap_rate: float = 5.0, 

13 max_pullback_rate: float = 2.0, 

14 rebound_rate: float = 2.0, 

15 mode: str = "live", 

16 logger: Optional[logging.Logger] = None 

17 ): 

18 self.broker = broker 

19 self.min_gap_rate = min_gap_rate 

20 self.max_pullback_rate = max_pullback_rate 

21 self.rebound_rate = rebound_rate 

22 self.mode = mode 

23 if logger: 

24 self.logger = logger 

25 else: 

26 self.logger = get_strategy_logger("GapUpPullback") 

27 

28 async def run(self, stock_codes: List[str]) -> Dict: 

29 selected = [] 

30 rejected = [] 

31 

32 for code in stock_codes: 

33 summary = await self.broker.get_price_summary(code) 

34 name = await self.broker.get_name_by_code(code) 

35 display = f"{name}({code})" if name else code 

36 

37 previous_close = summary.get("prev_close") 

38 open_price = summary.get("open") 

39 low = summary.get("low") 

40 current = summary.get("current") 

41 

42 if not all([previous_close, open_price, low, current]): 

43 self.logger.warning(f"[데이터 누락] {display} - 필수 가격 정보 없음") 

44 continue 

45 

46 gap_up = (open_price - previous_close) / previous_close * 100 

47 pullback = (open_price - low) / open_price * 100 

48 rebound = (current - low) / low * 100 

49 

50 is_candidate = ( 

51 gap_up >= self.min_gap_rate and 

52 pullback >= self.max_pullback_rate and 

53 rebound >= self.rebound_rate 

54 ) 

55 

56 if is_candidate: 

57 selected.append({"code": code, "name": name}) 

58 self.logger.info( 

59 f"[후보 선정] {display} | 전일종가: {previous_close} | 시가: {open_price} | 저가: {low} | 종가: {current} | " 

60 f"갭상승: {gap_up:.2f}% | 눌림: {pullback:.2f}% | 반등: {rebound:.2f}%" 

61 ) 

62 else: 

63 rejected.append({"code": code, "name": name}) 

64 self.logger.info( 

65 f"[제외] {display} | 전일종가: {previous_close} | 시가: {open_price} | 저가: {low} | 종가: {current} | " 

66 f"갭상승: {gap_up:.2f}% | 눌림: {pullback:.2f}% | 반등: {rebound:.2f}%" 

67 ) 

68 

69 return { 

70 "gapup_pullback_selected": selected, 

71 "gapup_pullback_rejected": rejected 

72 }