Coverage for src / mysingle / dsl / stdlib.py: 0%

1"""DSL 표준 라이브러리 - 공통 함수""" 

2 

3from typing import Any, Callable, Literal 

4 

5import pandas as pd 

6 

7 

8def SMA(series: pd.Series, window: int) -> pd.Series: 

9 """ 

10 Simple Moving Average 

11 

12 Args: 

13 series: 입력 시리즈 (보통 close 가격) 

14 window: 이동평균 기간 

15 

16 Returns: 

17 pd.Series: SMA 값 

18 """ 

19 return series.rolling(window=window).mean() 

20 

21 

22def EMA(series: pd.Series, span: int) -> pd.Series: 

23 """ 

24 Exponential Moving Average 

25 

26 Args: 

27 series: 입력 시리즈 

28 span: EMA 기간 

29 

30 Returns: 

31 pd.Series: EMA 값 

32 """ 

33 return series.ewm(span=span, adjust=False).mean() 

34 

35 

36def WMA(series: pd.Series, window: int) -> pd.Series: 

37 """ 

38 Weighted Moving Average 

39 

40 Args: 

41 series: 입력 시리즈 

42 window: WMA 기간 

43 

44 Returns: 

45 pd.Series: WMA 값 

46 """ 

47 weights = pd.Series(range(1, window + 1)) 

48 return series.rolling(window).apply(lambda x: (x * weights).sum() / weights.sum()) 

49 

50 

51def crossover(series1: pd.Series, series2: pd.Series) -> pd.Series: 

52 """ 

53 상향 돌파 검사 

54 

55 series1이 series2를 아래에서 위로 돌파하는 시점 탐지 

56 

57 Args: 

58 series1: 비교 대상 시리즈 1 

59 series2: 비교 대상 시리즈 2 

60 

61 Returns: 

62 pd.Series: 상향 돌파 시 True 

63 """ 

64 return (series1 > series2) & (series1.shift(1) <= series2.shift(1)) 

65 

66 

67def crossunder(series1: pd.Series, series2: pd.Series) -> pd.Series: 

68 """ 

69 하향 돌파 검사 

70 

71 series1이 series2를 위에서 아래로 돌파하는 시점 탐지 

72 

73 Args: 

74 series1: 비교 대상 시리즈 1 

75 series2: 비교 대상 시리즈 2 

76 

77 Returns: 

78 pd.Series: 하향 돌파 시 True 

79 """ 

80 return (series1 < series2) & (series1.shift(1) >= series2.shift(1)) 

81 

82 

83def highest(series: pd.Series, window: int) -> pd.Series: 

84 """ 

85 N일 최고값 

86 

87 Args: 

88 series: 입력 시리즈 

89 window: 기간 

90 

91 Returns: 

92 pd.Series: N일 최고값 

93 """ 

94 return series.rolling(window=window).max() 

95 

96 

97def lowest(series: pd.Series, window: int) -> pd.Series: 

98 """ 

99 N일 최저값 

100 

101 Args: 

102 series: 입력 시리즈 

103 window: 기간 

104 

105 Returns: 

106 pd.Series: N일 최저값 

107 """ 

108 return series.rolling(window=window).min() 

109 

110 

111def change(series: pd.Series, periods: int = 1) -> pd.Series: 

112 """ 

113 절대 변화량 

114 

115 Args: 

116 series: 입력 시리즈 

117 periods: 기간 (기본 1) 

118 

119 Returns: 

120 pd.Series: 절대 변화량 

121 """ 

122 return series.diff(periods) 

123 

124 

125def pct_change(series: pd.Series, periods: int = 1) -> pd.Series: 

126 """ 

127 백분율 변화량 

128 

129 Args: 

130 series: 입력 시리즈 

131 periods: 기간 (기본 1) 

132 

133 Returns: 

134 pd.Series: 백분율 변화량 

135 """ 

136 return series.pct_change(periods) 

137 

138 

139def stdev(series: pd.Series, window: int) -> pd.Series: 

140 """ 

141 표준편차 

142 

143 Args: 

144 series: 입력 시리즈 

145 window: 기간 

146 

147 Returns: 

148 pd.Series: 표준편차 

149 """ 

150 return series.rolling(window=window).std() 

151 

152 

153def bbands(series: pd.Series, window: int = 20, num_std: float = 2.0) -> pd.DataFrame: 

154 """ 

155 Bollinger Bands 

156 

157 Args: 

158 series: 입력 시리즈 (보통 close 가격) 

159 window: 이동평균 기간 (기본 20) 

160 num_std: 표준편차 배수 (기본 2.0) 

161 

162 Returns: 

163 pd.DataFrame: upper, middle, lower 컬럼 

164 """ 

165 middle = SMA(series, window) 

166 std = stdev(series, window) 

167 

168 return pd.DataFrame( 

169 { 

170 "upper": middle + (std * num_std), 

171 "middle": middle, 

172 "lower": middle - (std * num_std), 

173 } 

174 ) 

175 

176 

177def RSI(series: pd.Series, period: int = 14) -> pd.Series: 

178 """ 

179 Relative Strength Index 

180 

181 Args: 

182 series: 입력 시리즈 (보통 close 가격) 

183 period: RSI 기간 (기본 14) 

184 

185 Returns: 

186 pd.Series: RSI 값 (0-100 범위) 

187 """ 

188 if period <= 0: 

189 raise ValueError("Period must be greater than 0") 

190 

191 # 가격 변화 계산 

192 delta = series.diff().astype(float) 

193 

194 # 상승/하락 분리 

195 gain = delta.where(delta > 0, 0.0) 

196 loss = -delta.where(delta < 0, 0.0) 

197 

198 # 평균 상승/하락 (EMA 사용) 

199 avg_gain = gain.ewm(span=period, adjust=False).mean() 

200 avg_loss = loss.ewm(span=period, adjust=False).mean() 

201 

202 # RS 계산 

203 rs = avg_gain / avg_loss 

204 

205 # RSI 계산 

206 rsi = 100 - (100 / (1 + rs)) 

207 

208 return rsi 

209 

210 

211def atr( 

212 high: pd.Series, low: pd.Series, close: pd.Series, window: int = 14 

213) -> pd.Series: 

214 """ 

215 Average True Range 

216 

217 Args: 

218 high: 고가 시리즈 

219 low: 저가 시리즈 

220 close: 종가 시리즈 

221 window: ATR 기간 (기본 14) 

222 

223 Returns: 

224 pd.Series: ATR 값 

225 """ 

226 # True Range 계산 

227 high_low = high - low 

228 high_close = abs(high - close.shift(1)) 

229 low_close = abs(low - close.shift(1)) 

230 

231 tr = pd.DataFrame({"hl": high_low, "hc": high_close, "lc": low_close}).max(axis=1) 

232 

233 # ATR (EMA of TR) 

234 return EMA(tr, window) 

235 

236 

237# ============================================================================ 

238# 전략 특화 함수 (Strategy Service 전용) 

239# ============================================================================ 

240 

241 

242def generate_signal( 

243 condition: pd.Series, signal_type: Literal["long", "short"] = "long" 

244) -> pd.Series: 

245 """ 

246 조건을 명시적 boolean 시그널로 변환 

247 

248 Args: 

249 condition: 조건 Series (True/False) 

250 signal_type: 시그널 방향 ("long" or "short") 

251 

252 Returns: 

253 pd.Series[bool]: 시그널 (True = 진입, False = 보유) 

254 

255 Example: 

256 >>> oversold = data['RSI'] < 30 

257 >>> buy_signal = generate_signal(oversold, signal_type="long") 

258 """ 

259 # 명시적 boolean 변환 

260 return condition.astype(bool) 

261 

262 

263def entry_exit_signals( 

264 entry_condition: pd.Series, exit_condition: pd.Series 

265) -> pd.DataFrame: 

266 """ 

267 진입 조건과 청산 조건을 페어로 생성 

268 

269 Args: 

270 entry_condition: 진입 조건 Series 

271 exit_condition: 청산 조건 Series 

272 

273 Returns: 

274 pd.DataFrame: {'entry': pd.Series[bool], 'exit': pd.Series[bool]} 

275 

276 Example: 

277 >>> entry = crossover(data['SMA_50'], data['SMA_200']) 

278 >>> exit = crossunder(data['SMA_50'], data['SMA_200']) 

279 >>> signals = entry_exit_signals(entry, exit) 

280 >>> result = signals['entry'] # 진입 시그널만 반환 

281 """ 

282 return pd.DataFrame( 

283 {"entry": entry_condition.astype(bool), "exit": exit_condition.astype(bool)} 

284 ) 

285 

286 

287def signal_filter(signals: pd.Series, filter_condition: pd.Series) -> pd.Series: 

288 """ 

289 시그널을 필터 조건으로 필터링 

290 

291 Args: 

292 signals: 시그널 Series 

293 filter_condition: 필터 조건 Series 

294 

295 Returns: 

296 pd.Series[bool]: 필터링된 시그널 

297 

298 Example: 

299 >>> # RSI 과매도 시그널 

300 >>> oversold = data['RSI'] < 30 

301 >>> 

302 >>> # 거래량 필터 (평균 대비 1.5배 이상) 

303 >>> high_volume = data['volume'] > data['volume'].rolling(20).mean() * 1.5 

304 >>> 

305 >>> # 필터링된 시그널 

306 >>> filtered = signal_filter(oversold, high_volume) 

307 """ 

308 return (signals & filter_condition).astype(bool) 

309 

310 

311def get_stdlib_functions() -> dict[str, Callable[..., Any]]: 

312 """ 

313 표준 라이브러리 함수 딕셔너리 반환 

314 

315 Returns: 

316 dict: 함수명 -> 함수 매핑 

317 """ 

318 return { 

319 # 이동평균 

320 "SMA": SMA, 

321 "EMA": EMA, 

322 "WMA": WMA, 

323 # 기술적 지표 

324 "RSI": RSI, 

325 # 크로스오버 

326 "crossover": crossover, 

327 "crossunder": crossunder, 

328 # 최고/최저 

329 "highest": highest, 

330 "lowest": lowest, 

331 # 변화율 

332 "change": change, 

333 "pct_change": pct_change, 

334 # 변동성 

335 "stdev": stdev, 

336 "bbands": bbands, 

337 "atr": atr, 

338 # 전략 특화 함수 

339 "generate_signal": generate_signal, 

340 "entry_exit_signals": entry_exit_signals, 

341 "signal_filter": signal_filter, 

342 }