Coverage for strategies / momentum_strategy.py: 90%

1# strategies/momentum_strategy.py 

2import logging 

3from interfaces.strategy import Strategy 

4from typing import List, Dict, Optional, Callable 

5from common.types import ErrorCode, ResCommonResponse 

6import inspect 

7 

8 

9class MomentumStrategy(Strategy): 

10 def __init__( 

11 self, 

12 broker, 

13 min_change_rate: float = 10.0, 

14 min_follow_through: float = 3.0, 

15 min_follow_through_time: int = 5, # 몇 분 후 기준인지 

16 mode: str = "live", # 'live' or 'backtest' 

17 backtest_lookup: Optional[Callable[[str, Dict, int], int]] = None, 

18 logger: Optional[logging.Logger] = None 

19 ): 

20 self.broker = broker # ✅ 통합 wrapper (API + CSV) 

21 self.min_change_rate = min_change_rate 

22 self.min_follow_through = min_follow_through 

23 self.min_follow_through_time = min_follow_through_time # 추가 

24 self.mode = mode 

25 self.backtest_lookup = backtest_lookup 

26 self.logger = logger or logging.getLogger(__name__) 

27 

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

29 results = [] 

30 

31 for code in stock_codes: 

32 summary : ResCommonResponse = await self.broker.get_price_summary(code) # ✅ wrapper 통해 조회 

33 

34 if not summary: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 self.logger.warning(f"종목 {code}: 가격 요약 데이터를 받지 못했습니다. 건너뜁니다.") 

36 continue 

37 

38 if hasattr(summary, 'data') and summary.data is None: 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 self.logger.warning(f"종목 {code}: 가격 요약 데이터가 비어있습니다. 건너뜁니다.") 

40 continue 

41 

42 if self.mode == "backtest": 

43 if not self.backtest_lookup: 

44 raise ValueError("Backtest 모드에서는 backtest_lookup 함수가 필요합니다.") 

45 

46 result = self.backtest_lookup( 

47 code, 

48 summary.data, 

49 self.min_follow_through_time 

50 ) 

51 after_price = await result if inspect.isawaitable(result) else result 

52 else: 

53 price_data : ResCommonResponse = await self.broker.get_current_price(code, caller="MomentumStrategy") # ✅ wrapper 통해 조회 

54 after_price = int(price_data.data.get("stck_prpr", "0") or 0) if price_data and price_data.data else 0 

55 

56 if after_price is None: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 self.logger.warning(f"종목 {code}: after_price를 받지 못했습니다. 0으로 대체합니다.") 

58 after_price = 0 

59 

60 summary.data["after"] = after_price 

61 current_price = summary.data.get("current", 0) or 0 

62 summary.data["after_rate"] = ( 

63 (after_price - current_price) / current_price * 100 

64 if current_price else 0 

65 ) 

66 results.append(summary.data) 

67 

68 follow_through = [] 

69 not_follow_through = [] 

70 

71 

72 for s in results: 

73 code = s["symbol"] 

74 name : str = await self.broker.get_name_by_code(code) 

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

76 

77 # ▼▼▼▼▼ 핵심 로직 및 로그 수정 ▼▼▼▼▼ 

78 # 1. 초기 모멘텀 (시가 대비 현재가) 조건 확인 

79 initial_momentum_ok = s["change_rate"] >= self.min_change_rate 

80 # 2. 추세 지속 (현재가 대비 N분 후 가격) 조건 확인 

81 follow_through_ok = s["after_rate"] >= self.min_follow_through 

82 

83 is_success = initial_momentum_ok and follow_through_ok 

84 

85 # 이제 로그에 각 단계의 실제값과 기준을 명확히 기록합니다. 

86 log_initial_rate = f"초기 등락률: {s['change_rate']:.2f}% (기준: {self.min_change_rate}%)" 

87 # ▼▼▼▼▼ 여기가 핵심 수정 부분입니다 ▼▼▼▼▼ 

88 # N분 전/후 가격을 로그에 포함시킵니다. 

89 before_price = s['current'] 

90 after_price = s['after'] 

91 log_follow_rate = ( 

92 f"{self.min_follow_through_time}분 후 상승률: {s['after_rate']:.2f}% " 

93 f"({before_price:,}원 → {after_price:,}원, 기준: {self.min_follow_through}%)" 

94 ) 

95 # ▲▲▲▲▲ 여기가 핵심 수정 부분입니다 ▲▲▲▲▲ 

96 if is_success: 

97 follow_through.append({"code": code, "name": name}) 

98 self.logger.info( 

99 f"[성공] 종목: {display} | {log_initial_rate} | {log_follow_rate}" 

100 ) 

101 else: 

102 failure_reasons = [] 

103 if not initial_momentum_ok: 

104 failure_reasons.append("초기 등락률 미달") 

105 if not follow_through_ok: 

106 failure_reasons.append("추세 지속 실패") 

107 

108 reason_str = " & ".join(failure_reasons) 

109 not_follow_through.append({"code": code, "name": name}) 

110 self.logger.info( 

111 f"[실패] 종목: {display} | 사유: {reason_str} | {log_initial_rate} | {log_follow_rate}" 

112 ) 

113 # ▲▲▲▲▲ 핵심 로직 및 로그 수정 ▲▲▲▲▲ 

114 

115 total = len(results) 

116 success = len(follow_through) 

117 fail = len(not_follow_through) 

118 success_rate = (success / total * 100) if total > 0 else 0 

119 

120 self.logger.info( 

121 f"[결과 요약] 총 종목: {total}, 성공: {success}, 실패: {fail}, 성공률: {success_rate:.2f}%, 모드: {self.mode}" 

122 ) 

123 

124 return { 

125 "follow_through": follow_through, # [{"code": "005930", "name": "삼성전자"}, ...] 

126 "not_follow_through": not_follow_through 

127 }