Coverage for strategies / volume_breakout_strategy.py: 92%

1from __future__ import annotations 

2 

3from dataclasses import dataclass 

4from typing import Any, Dict, List, Optional, Literal, Callable 

5from strategies.base_strategy_config import BaseStrategyConfig 

6 

7# =============================== 

8# 거래량 돌파 전략 설정 클래스 

9# =============================== 

10@dataclass 

11class VolumeBreakoutConfig(BaseStrategyConfig): 

12 """거래량 돌파 전략 및 백테스트용 설정""" 

13 trigger_pct: float = 10.0 # 시가 대비 +10% 도달 시 매수 트리거 

14 entry_push_pct: float = 2.0 # 신호 발생 후 추가 상승폭 (라이브 전략용) 

15 trailing_stop_pct: float = 8.0 # 고가 대비 -8% 하락 시 익절 

16 stop_loss_pct: float = -5.0 # 매수가 대비 -5% 손절 

17 avg_vol_lookback_days: int = 20 # 평균 거래량 계산 기간 (거래일 기준 약 1개월) 

18 avg_vol_multiplier: float = 2.0 # 평균 거래량 대비 최소 배수 (≥2배) 

19 session: Literal["REGULAR", "EXTENDED"] = "REGULAR" # 거래 세션 구분 

20 allow_reentry: bool = True # 당일 재진입 허용 여부 (False면 종목당 하루 1회만 매수) 

21 

22# =============================== 

23# 거래량 돌파 전략 클래스 

24# =============================== 

25class VolumeBreakoutStrategy: 

26 """ 

27 거래량 돌파 전략 및 단일일자 분봉 백테스트 지원: 

28 - 시가 대비 trigger_pct 도달 시 매수 

29 - 이후 고가 대비 trailing_stop_pct 하락 시 익절, 매수가 대비 stop_loss_pct 하락 시 손절 

30 - 둘 다 도달하지 않으면 장 마감가로 청산 

31 """ 

32 

33 def __init__( 

34 self, 

35 *, 

36 stock_query_service: Any, 

37 market_clock: Any, 

38 logger: Optional[Any] = None, 

39 config: Optional[VolumeBreakoutConfig] = None, 

40 ) -> None: 

41 self.svc = stock_query_service # 분봉 데이터를 가져오는 서비스 

42 self.market_clock = market_clock # 시간 포맷 변환 등 유틸 

43 self.log = logger 

44 self.cfg = config or VolumeBreakoutConfig() 

45 

46 # ------------------------- 

47 # 내부 유틸 함수 

48 # ------------------------- 

49 @staticmethod 

50 def _get_first_available(d: Dict[str, Any], keys: List[str], default: Any = None) -> Any: 

51 """여러 키 중 첫 번째 유효한 값을 반환""" 

52 for k in keys: 52 ↛ 56line 52 didn't jump to line 56 because the loop on line 52 didn't complete

53 v = d.get(k) 

54 if v is not None and v != "-": 54 ↛ 52line 54 didn't jump to line 52 because the condition on line 54 was always true

55 return v 

56 return default 

57 

58 def _sort_key(self, r: Dict[str, Any]) -> tuple: 

59 """분봉 데이터를 날짜+시간 순으로 정렬하기 위한 키""" 

60 d = str(self._get_first_available(r, ["stck_bsop_date", "bsop_date", "date"], "")) 

61 t = str(self._get_first_available(r, ["stck_cntg_hour", "cntg_hour", "time"], "")) 

62 return d, self.market_clock.to_hhmmss(t) 

63 

64 # ------------------------- 

65 # 공개 메서드: 분봉 백테스트 

66 # ------------------------- 

67 async def backtest_open_threshold_intraday( 

68 self, 

69 stock_code: str, 

70 *, 

71 date_ymd: Optional[str] = None, 

72 session: Optional[Literal["REGULAR", "EXTENDED"]] = None, 

73 trigger_pct: Optional[float] = None, 

74 trailing_stop_pct: Optional[float] = None, 

75 sl_pct: Optional[float] = None, 

76 price_getter: Optional[Callable[[Dict[str, Any]], Optional[float]]] = None, 

77 ) -> Dict[str, Any]: 

78 """하루치 분봉 데이터를 이용한 단일일자 백테스트 수행""" 

79 session = session or self.cfg.session 

80 trigger = self.cfg.trigger_pct if trigger_pct is None else trigger_pct 

81 ts_pct = self.cfg.trailing_stop_pct if trailing_stop_pct is None else trailing_stop_pct 

82 sl = self.cfg.stop_loss_pct if sl_pct is None else sl_pct 

83 

84 # 1) 분봉 데이터 로드 

85 rows: List[Dict[str, Any]] = await self.svc.get_day_intraday_minutes_list( 

86 stock_code=stock_code, 

87 date_ymd=date_ymd, 

88 session=session, 

89 ) 

90 day_label = date_ymd or self.market_clock.get_current_kst_time().strftime("%Y%m%d") 

91 if not rows: 

92 return {"ok": False, "message": "분봉 데이터 없음", "stock_code": stock_code, "date": day_label, "trades": []} 

93 

94 rows = sorted(rows, key=self._sort_key) 

95 

96 # 2) 시가 설정 (첫 분봉의 시가 사용, 없으면 종가/가격 사용) 

97 def default_price_getter(r: Dict[str, Any]) -> Optional[float]: 

98 v = self._get_first_available(r, ["stck_prpr", "prpr", "close", "price"]) 

99 return float(v) if v not in (None, "", "-") else None 

100 

101 pg = price_getter or default_price_getter 

102 open0_raw = self._get_first_available(rows[0], ["stck_oprc", "oprc", "open"]) or \ 

103 self._get_first_available(rows[0], ["stck_prpr", "prpr", "close", "price"]) 

104 try: 

105 open0 = float(open0_raw) 

106 except Exception: 

107 return {"ok": False, "message": f"시가 파싱 실패(open0={open0_raw!r})", "stock_code": stock_code, "date": day_label, "trades": []} 

108 

109 # 3) 매수 트리거 찾기 (+trigger_pct 도달 시점) 

110 entry_idx = None 

111 entry_px = None 

112 for i, r in enumerate(rows): 

113 p = pg(r) 

114 if p is None: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true

115 continue 

116 change = (p / open0 - 1.0) * 100.0 

117 if change >= trigger: 

118 entry_idx, entry_px = i, p 

119 break 

120 

121 if entry_idx is None: 

122 return {"ok": True, "message": f"트리거 {trigger}% 미발생", "stock_code": stock_code, "date": day_label, "trades": [], "equity": 1.0} 

123 

124 # 4) 익절/손절 조건 확인 

125 exit_idx = None 

126 exit_px = None 

127 outcome = "close_exit" 

128 curr_high = entry_px 

129 for j in range(entry_idx + 1, len(rows)): 

130 p = pg(rows[j]) 

131 if p is None: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 continue 

133 curr_high = max(curr_high, p) 

134 drop_from_high = (p / curr_high - 1.0) * 100.0 

135 if drop_from_high <= -ts_pct: 

136 exit_idx, exit_px, outcome = j, p, "trailing_stop" 

137 break 

138 pnl = (p / entry_px - 1.0) * 100.0 

139 if pnl <= sl: 

140 exit_idx, exit_px, outcome = j, p, "stop_loss" 

141 break 

142 

143 if exit_idx is None: 

144 last_price = pg(rows[-1]) 

145 if last_price is None: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true

146 return {"ok": False, "message": "종가 가격 파싱 실패", "stock_code": stock_code, "date": day_label, "trades": []} 

147 exit_idx, exit_px = len(rows) - 1, last_price 

148 

149 ret = (exit_px / entry_px) - 1.0 

150 

151 def fmt_ts(row: Dict[str, Any]) -> str: 

152 d = str(self._get_first_available(row, ["stck_bsop_date", "bsop_date", "date"], "")) 

153 t = str(self._get_first_available(row, ["stck_cntg_hour", "cntg_hour", "time"], "")) 

154 return f"{d} {self.market_clock.to_hhmmss(t)}" 

155 

156 trade = { 

157 "entry_time": fmt_ts(rows[entry_idx]), 

158 "entry_px": float(entry_px), 

159 "exit_time": fmt_ts(rows[exit_idx]), 

160 "exit_px": float(exit_px), 

161 "outcome": outcome, 

162 "ret": ret, 

163 "ret_pct": round(ret * 100.0, 3), 

164 "open0": float(open0), 

165 "trigger_pct": float(trigger), 

166 "trailing_stop_pct": float(ts_pct), 

167 "sl_pct": float(sl), 

168 } 

169 

170 return {"ok": True, "message": "success", "stock_code": stock_code, "date": day_label, "equity": 1.0 + ret, "trades": [trade]}