From 598eda3c06a71108ad53f23386d485662bbd7509 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=B0=95=EC=B0=AC=EC=9A=B0?= Date: Wed, 1 Oct 2025 16:47:41 +0900 Subject: [PATCH] feat: Add Bollinger Bands strategy optimizer script --- .../bollinger_bands_optimizer.py | 385 ++++++++++++++++++ 1 file changed, 385 insertions(+) create mode 100644 portfolio_strategies/bollinger_bands_optimizer.py diff --git a/portfolio_strategies/bollinger_bands_optimizer.py b/portfolio_strategies/bollinger_bands_optimizer.py new file mode 100644 index 0000000..314b347 --- /dev/null +++ b/portfolio_strategies/bollinger_bands_optimizer.py @@ -0,0 +1,385 @@ +import warnings +# --- 경고 무시 설정: 깔끔한 출력을 위해 모든 경고를 무시합니다. --- +warnings.filterwarnings("ignore") +# -------------------- + +import yfinance as yf +import pandas as pd +import numpy as np +import talib as ta +import matplotlib.pyplot as plt +from itertools import product + +# Matplotlib 한글 폰트 설정 +try: + plt.rcParams['font.family'] = 'Malgun Gothic' # Windows 기본 폰트 + plt.rcParams['axes.unicode_minus'] = False # 마이너스 기호 깨짐 방지 +except: + print("\n[Matplotlib 폰트 경고] 한글 폰트를 찾을 수 없습니다. 그래프의 한글 텍스트가 깨질 수 있습니다.") + plt.rcParams['font.family'] = 'DejaVu Sans' + +# --- 설정 상수 (Bollinger Bands 기준) --- +TICKER = 'OKLO' +START_DATE = '2024-05-09' +END_DATE = None +# --- 최적화 파라미터 범위 정의 --- +N_PERIODS = range(10, 31, 5) # N (기간): 10, 15, 20, 25, 30 +K_MULTIPLIERS = [1.5, 2.0, 2.5, 3.0] # K (표준편차 배수): 1.5, 2.0, 2.5, 3.0 + + +# --- 1. 데이터 다운로드 및 준비 (CLOSE, HIGH, LOW 필요) --- +def download_data(): + """Bollinger Bands 계산에 필요한 'CLOSE', 'HIGH', 'LOW' 컬럼을 다운로드하고 아웃라이어를 제거합니다.""" + print(f"--- 1. {TICKER} 주식 데이터 다운로드 시작 (기간: {START_DATE} ~ {END_DATE if END_DATE else '최신일'}) ---") + try: + data = yf.download(TICKER, start=START_DATE, end=END_DATE) + + if data.empty: + print("데이터를 다운로드했으나, 데이터프레임이 비어있습니다. 티커를 확인해 주세요.") + return pd.DataFrame() + + if isinstance(data.columns, pd.MultiIndex): + data.columns = data.columns.get_level_values(-1) + + data.columns = [col.upper().strip() for col in data.columns] + + # OKLO 비표준 컬럼 이슈 처리 (Adjusted Close를 최종 거래 가격으로 사용) + if len(data.columns) >= 5 and all(col == TICKER.upper() for col in data.columns.tolist()): + standard_names = ['OPEN', 'HIGH', 'LOW', 'CLOSE', 'ADJ CLOSE', 'VOLUME'] + data.columns = standard_names[:len(data.columns)] + print(f"\n[데이터 클렌징 알림]: 비표준 컬럼 이름 ({TICKER})이 감지되어, 표준 순서로 강제 재정의되었습니다.") + + data['CLOSE'] = data.get('ADJ CLOSE', data.get('CLOSE')) + + required_cols = ['CLOSE', 'HIGH', 'LOW'] + + if not all(col in data.columns for col in required_cols): + missing_cols = [col for col in required_cols if col not in data.columns] + print(f"다운로드된 데이터에 {required_cols} 중 부족한 컬럼이 있습니다. (누락: {missing_cols})") + return pd.DataFrame() + + data_cols = data[required_cols].copy() + + for col in required_cols: + if data_cols[col].dtype != np.float64: + try: + data_cols[col] = data_cols[col].astype(float) + except ValueError: + print(f"데이터 경고: '{col}' 컬럼을 float으로 변환할 수 없습니다. 데이터를 건너뜁니다.") + return pd.DataFrame() + + # 아웃라이어 제거 + quantile_999 = data_cols['CLOSE'].quantile(0.999) + initial_count = len(data_cols) + data_cols = data_cols[data_cols['CLOSE'] <= quantile_999] + removed_count = initial_count - len(data_cols) + + if removed_count > 0: + print(f"\n[데이터 클렌징 알림]: 극단적인 아웃라이어 {removed_count}개를 제거했습니다.") + + print(f"다운로드 완료. 총 데이터 개수: {data_cols.shape[0]}일") + return data_cols + + except Exception as e: + print(f"치명적 오류: 데이터 다운로드 중 오류 발생: {e}") + return pd.DataFrame() + + +# --- 2. 백테스팅 로직 (수익률과 거래 횟수 반환) --- +def backtest_strategy(data, n_period, k_multiplier): + """ + Bollinger Bands 역추세(밴드 재진입) 전략을 백테스팅하고 + 누적 수익률(복리)과 거래 횟수를 반환합니다. + """ + if data.empty: + return 0, 0 + + df = data.copy() + + # Bollinger Bands 계산 + df['Upper'], df['Middle'], df['Lower'] = ta.BBANDS( + df['CLOSE'].values, + timeperiod=n_period, + nbdevup=k_multiplier, + nbdevdn=k_multiplier, + matype=0 # Simple Moving Average + ) + + df.dropna(inplace=True) + if df.empty: + return 0, 0 + + # BB 역추세(밴드 재진입) 매매 신호 로직 + # 매수: 종가가 하한선 아래 -> 오늘 하한선 위로 진입 + df['Buy_Signal'] = (df['CLOSE'].shift(1) < df['Lower'].shift(1)) & (df['CLOSE'] > df['Lower']) + # 매도: 종가가 상한선 위 -> 오늘 상한선 아래로 진입 (포지션 정리) + df['Sell_Signal'] = (df['CLOSE'].shift(1) > df['Upper'].shift(1)) & (df['CLOSE'] < df['Upper']) + + # 포지션 관리 및 수익률 계산 + is_holding = False + buy_price = 0 + trades = [] # 실현된 거래의 수익률을 저장 + + for i in range(len(df)): + current_close = df['CLOSE'].iloc[i] + + # 매수 신호 발생 및 현재 포지션이 없는 경우 + if df['Buy_Signal'].iloc[i] and not is_holding: + buy_price = current_close + is_holding = True + + # 매도 신호 발생 및 현재 포지션이 있는 경우 + elif df['Sell_Signal'].iloc[i] and is_holding: + profit = (current_close - buy_price) / buy_price + trades.append(profit) + is_holding = False + + # 마지막 날까지 포지션을 정리하지 못하고 종료된 경우, 마지막 종가로 정리 + if is_holding: + final_price = df['CLOSE'].iloc[-1] + profit = (final_price - buy_price) / buy_price + trades.append(profit) + + if not trades: + return 0, 0 + + cumulative_profit = (np.prod([(1 + t) for t in trades]) - 1) * 100 + total_trades = len(trades) # 매수/매도 한 쌍의 거래 횟수 + + return cumulative_profit, total_trades + +# --- 4. 그래프 생성을 위한 일별 수익률 계산 및 플로팅 함수 --- +def calculate_daily_strategy_cumulative_returns(data, n_period, k_multiplier): + """최적화된 파라미터로 전략의 일별 누적 수익률을 계산하여 반환합니다.""" + df = data.copy() + + # Bollinger Bands 계산 + df['Upper'], df['Middle'], df['Lower'] = ta.BBANDS( + df['CLOSE'].values, + timeperiod=n_period, + nbdevup=k_multiplier, + nbdevdn=k_multiplier, + matype=0 + ) + df.dropna(inplace=True) + if df.empty: + return pd.Series(1.0, index=data.index) + + # BB 역추세 포지션 생성 로직 (일별 누적 수익률 계산을 위해 포지션 유지 여부를 기록) + df['Buy_Signal'] = (df['CLOSE'].shift(1) < df['Lower'].shift(1)) & (df['CLOSE'] > df['Lower']) + df['Sell_Signal'] = (df['CLOSE'].shift(1) > df['Upper'].shift(1)) & (df['CLOSE'] < df['Upper']) + + df['Position'] = 0 # 1: 매수 포지션, 0: 포지션 없음 + is_holding = 0 + + for i in range(len(df)): + if df['Buy_Signal'].iloc[i]: + is_holding = 1 # 매수 신호 발생 -> 포지션 진입 + elif df['Sell_Signal'].iloc[i]: + is_holding = 0 # 매도 신호 발생 -> 포지션 종료 + + df.loc[df.index[i], 'Position'] = is_holding + + df.loc[:, 'Position'] = df['Position'].shift(1).fillna(0) + + # 일별 수익률 계산: 전일 대비 수익률 * 전일 포지션 + df.loc[:, 'Strategy_Daily_Return'] = df['CLOSE'].pct_change() * df['Position'] + df.loc[:, 'Strategy_Daily_Return'].fillna(0, inplace=True) + + # 누적 수익률 (Cumulative Product) + df.loc[:, 'Strategy_Cumulative'] = (1 + df['Strategy_Daily_Return']).cumprod() + df.loc[:, 'Strategy_Cumulative'].iloc[0] = 1.0 # 첫 날은 1.0으로 시작 + + # Buy & Hold 수익률 계산 + df.loc[:, 'Buy_Hold_Daily_Return'] = df['CLOSE'].pct_change() + df.loc[:, 'Buy_Hold_Daily_Return'].fillna(0, inplace=True) + df.loc[:, 'Buy_Hold_Cumulative'] = (1 + df['Buy_Hold_Daily_Return']).cumprod() + df.loc[:, 'Buy_Hold_Cumulative'].iloc[0] = 1.0 + + return df[['Strategy_Cumulative', 'Buy_Hold_Cumulative']] + +def plot_optimization_results(ohlc_data, all_results, best_params): + """최적화 결과 및 최고 전략의 누적 수익률 그래프를 생성하고 화면에 표시합니다.""" + + results_df = pd.DataFrame(all_results) + + # 최적 파라미터 (N, K, Trades) 추출 + best_n, best_k, best_trades = best_params + + # 두 개의 서브플롯 생성 + fig, axes = plt.subplots(1, 2, figsize=(18, 7)) + + # ----------------------------------- + # Plot 1: 파라미터 조합 vs. 수익률 (Heatmap/Scatter) + # ----------------------------------- + ax1 = axes[0] + + scatter = ax1.scatter( + results_df['N_Period'], + results_df['K_Multiplier'], + c=results_df['Profit'], + cmap='viridis', + s=400, # 마커 크기 + alpha=0.8, + edgecolor='k' + ) + + # 최적 파라미터 강조 + ax1.scatter(best_n, best_k, s=800, color='red', marker='*', label=f'최적 파라미터 N={best_n}, K={best_k}', zorder=5) + + ax1.set_title(f'Bollinger Bands 파라미터별 누적 수익률 분포 (역추세 전략)', fontsize=14, pad=15) + ax1.set_xlabel('N (기간)', fontsize=12) + ax1.set_ylabel('K (표준편차 배수)', fontsize=12) + ax1.set_xticks(list(N_PERIODS)) + ax1.set_yticks(K_MULTIPLIERS) + ax1.grid(True, linestyle='--', alpha=0.6) + + # 컬러바 추가 + cbar = fig.colorbar(scatter, ax=ax1, orientation='vertical', pad=0.05) + cbar.set_label('누적 수익률 (%)', rotation=270, labelpad=15) + ax1.legend() + + + # ----------------------------------- + # Plot 2: 누적 수익률 비교 (Equity Curve) + # ----------------------------------- + + # 최적 전략의 일별 누적 수익률 계산 + cumulative_returns_df = calculate_daily_strategy_cumulative_returns(ohlc_data, best_n, best_k) + + ax2 = axes[1] + + # 최적 전략 수익률 + ax2.plot(cumulative_returns_df.index, cumulative_returns_df['Strategy_Cumulative'], + label=f'최적 BB 전략 (N={best_n}, K={best_k})', color='darkblue', linewidth=2) + + # Buy & Hold 수익률 + ax2.plot(cumulative_returns_df.index, cumulative_returns_df['Buy_Hold_Cumulative'], + label='Buy & Hold', color='red', linestyle='--', linewidth=2) + + ax2.set_title(f'최적 BB 전략 vs. Buy & Hold 누적 수익률 비교 ({TICKER})', fontsize=14, pad=15) + ax2.set_xlabel('날짜', fontsize=12) + ax2.set_ylabel('누적 수익률 (기준일 100%)', fontsize=12) + + # Y축 포맷팅 (기준 1.0 = 100%) + from matplotlib.ticker import FuncFormatter + formatter = FuncFormatter(lambda y, pos: f"{y*100:.0f}%") + ax2.yaxis.set_major_formatter(formatter) + + ax2.grid(True, linestyle='--', alpha=0.7) + ax2.legend(loc='upper left') + + plt.tight_layout() + + # 파일로 저장하는 대신, 바로 화면에 표시 (plt.show() 사용) + print("\n--- 4. 시각화 결과 표시 ---") + print("[시각화 알림] 최적화 결과 그래프를 화면에 표시합니다. 창을 닫으면 다음 단계로 진행됩니다.") + plt.show() + + +# --- 3. 최적화 실행 (모든 결과 저장) --- +def optimize_bollinger_bands_parameters(data): + """ + 주어진 파라미터 범위 내에서 Bollinger Bands 조합을 테스트하고 모든 결과를 반환합니다. + """ + if data.empty: + # 수익률, 거래횟수, Buy&Hold, 최적 파라미터 + return [], 0, 0, (0, 0, 0) + + all_results = [] + best_profit = -float('inf') + best_params = (0, 0, 0) # N, K, Trades + + total_combinations = len(N_PERIODS) * len(K_MULTIPLIERS) + + print(f"\n--- 2. Bollinger Bands 파라미터 최적화 시작 ({total_combinations}개 조합 테스트) ---") + + # 모든 N과 K 조합을 생성 + for n_period, k_multiplier in product(N_PERIODS, K_MULTIPLIERS): + # 수익률과 거래 횟수를 동시에 반환받음 + profit, trade_count = backtest_strategy(data, n_period, k_multiplier) + + all_results.append({ + 'N_Period': n_period, + 'K_Multiplier': k_multiplier, + 'Profit': profit, + 'Trades': trade_count # 거래 횟수 저장 + }) + + if profit > best_profit: + best_profit = profit + # 최적 파라미터와 거래 횟수를 튜플로 저장 + best_params = (n_period, k_multiplier, trade_count) + + # Buy & Hold 수익률 계산 + if not data.empty: + buy_and_hold_return = ((data['CLOSE'].iloc[-1] - data['CLOSE'].iloc[0]) / data['CLOSE'].iloc[0] * 100).item() + else: + buy_and_hold_return = 0 + + return all_results, best_profit, buy_and_hold_return, best_params + + +# --- 메인 실행 블록 --- +if __name__ == "__main__": + + ohlc_data = download_data() + + if not ohlc_data.empty: + # 최적화 실행 + all_results, best_profit, buy_and_hold_return, best_params = optimize_bollinger_bands_parameters(ohlc_data) + + # 결과 정렬 (수익률 내림차순) + sorted_results = sorted(all_results, key=lambda x: x['Profit'], reverse=True) + + # 상위 10개 결과 + top_10 = sorted_results[:10] + + # 하위 10개 결과 추출 (수익률 오름차순) + worst_sorted = sorted(all_results, key=lambda x: x['Profit']) + worst_10 = worst_sorted[:10] + + # 최적 파라미터 언패킹 + best_n, best_k, best_trades = best_params + + # 결과 출력 (Markdown 형식으로 보고서 생성, LaTeX 적용) + print("\n--- 3. Bollinger Bands 최적화 최종 결과 보고 ---") + + # LaTeX 포맷팅 + print(f"| 키: $\\text{{N Period}}$ (기간), $\\text{{K Multiplier}}$ (배수), $\\text{{Profit}}$ (수익률), $\\text{{Trades}}$ (거래 횟수) |") + print("-" * 85) + print(f"**종목:** {TICKER}") + print(f"**기간:** {START_DATE} ~ {ohlc_data.index[-1].strftime('%Y-%m-%d')}") + print(f"**검증 조합 수:** {len(all_results)}개") + + # 테스트 범위 LaTeX 포맷팅 + latex_part_n = r"\{" + str(list(N_PERIODS)[0]) + r", \dots, " + str(list(N_PERIODS)[-1]) + r"\}" + latex_part_k = r"\{" + str(list(K_MULTIPLIERS)[0]) + r", \dots, " + str(list(K_MULTIPLIERS)[-1]) + r"\}" + print(f"**테스트 범위:** $\\text{{N Period}} \in {latex_part_n}$, $\\text{{K Multiplier}} \in {latex_part_k}$") + print("-" * 85) + + print(f"**💰 동일 기간 Buy & Hold 수익률:** $\mathbf{{ {buy_and_hold_return:.2f}\\% }}$") + print(f"**🏆 최적 BB 전략 누적 수익률:** $\mathbf{{ {best_profit:.2f}\\% }}$") + print(f"**최적 파라미터 (N/K):** $\mathbf{{ {best_n} / {best_k} }}$") + print(f"**최적 전략 거래 횟수:** $\mathbf{{ {best_trades} }}$회") + print("-" * 85) + + # --- 상위 10개 출력 --- + print("\n## 상위 최적 파라미터 조합 (Top Results)") + print("| 순위 | N Period | K Multiplier | 누적 수익률 (%) | 거래 횟수 |") + print("|:----:|:--------:|:------------:|:--------------:|:--------:|") + for i, res in enumerate(top_10): + print(f"| {i+1} | {res['N_Period']:^8} | {res['K_Multiplier']:^12} | {res['Profit']:.2f}% | {res['Trades']:^7} |") + + # --- 하위 10개 출력 --- + print("\n## 하위 최악 파라미터 조합 (Worst Results)") + print("| 순위 | N Period | K Multiplier | 누적 수익률 (%) | 거래 횟수 |") + print("|:----:|:--------:|:------------:|:--------------:|:--------:|") + for i, res in enumerate(worst_10): + print(f"| {i+1} | {res['N_Period']:^8} | {res['K_Multiplier']:^12} | {res['Profit']:.2f}% | {res['Trades']:^7} |") + + # --- 그래프 생성 및 표시 --- + plot_optimization_results(ohlc_data, all_results, best_params) + + else: + print("\n데이터 문제로 인해 최적화를 실행할 수 없습니다. 티커를 확인해 주세요.")