# backtester.py # ----------------------------------------------------------------------------- # Finder - Backtesting Engine # ----------------------------------------------------------------------------- import config import data_manager import strategy import pandas as pd import numpy as np from datetime import datetime import matplotlib.pyplot as plt from typing import Dict, List, Tuple def run_backtest() -> None: """ 설정(config)에 따라 백테스트를 실행합니다. Returns: None Raises: ValueError: 설정 파일에 필수 항목이 누락된 경우 """ print("백테스트를 시작합니다...") print(f"초기 자본: {config.BACKTEST_CAPITAL:,.0f} 원") print(f"테스트 기간: {config.BACKTEST_START_DATE} ~ {config.BACKTEST_END_DATE}") # --- 1. 데이터 준비 --- print("데이터 로드 및 전처리 중...") # [수정] 대상 티커 로드 (list -> dict) tickers_dict = {} # <-- tickers = [] 에서 변경 if config.TARGET_MARKET in ["KR", "BOTH"]: tickers_dict.update(data_manager.get_kr_tickers(config.KR_MARKET_CAP_START, config.KR_MARKET_CAP_END)) if config.TARGET_MARKET in ["US", "BOTH"]: tickers_dict.update(data_manager.get_us_tickers(config.US_MARKET_CAP_START, config.US_MARKET_CAP_END)) print(f"총 {len(tickers_dict)}개 종목 대상 분석 시작.") # 전체 기간 데이터 로드 및 보조지표 계산 # (이평선 계산을 위해 실제 백테스트 시작일보다 훨씬 이전부터 로드) # 예: 448일 이평선을 위해 최소 2~3년 전 데이터 필요 analysis_start_date = (pd.to_datetime(config.BACKTEST_START_DATE) - \ pd.DateOffset(years=config.DATA_HISTORY_YEARS)).strftime('%Y-%m-%d') stock_data_history = {} # [v3.0 수정] 신호 사전 계산 추가 buy_signals_history = {} sell_signals_by_ticker_buydate = {} # {ticker: {buy_date: sell_signals_df}} for ticker, name in tickers_dict.items(): df = data_manager.get_stock_data(ticker, analysis_start_date, config.BACKTEST_END_DATE) if not df.empty: # --- Precompute last peak series to avoid repeated find_peaks calls --- try: last_peak_series = strategy.compute_last_peak_series(df, config.STOPLOSS_PEAK_FIND_PERIOD) df['LAST_PEAK'] = last_peak_series except (ValueError, KeyError) as e: print(f"[Warning] {ticker} LAST_PEAK precompute 실패 (데이터 부족): {e}") continue except Exception as e: print(f"[Error] {ticker} LAST_PEAK 예상치 못한 오류: {type(e).__name__}: {e}") raise df_clean = df.dropna() stock_data_history[ticker] = df_clean # --- [v3.0] 매수 신호 사전 계산 --- try: buy_signals = strategy.compute_buy_signals_vectorized(df_clean) buy_signals_history[ticker] = buy_signals except (ValueError, KeyError) as e: print(f"[Warning] {ticker} 신호 계산 실패 (데이터 부족): {e}") except Exception as e: print(f"[Error] {ticker} 신호 계산 예상치 못한 오류: {type(e).__name__}: {e}") raise print("데이터 준비 완료.") # --- 2. 시뮬레이션 변수 초기화 --- capital = config.BACKTEST_CAPITAL portfolio = {} # e.g., {'AAPL': {'buy_price': 150, 'shares': 10, ...}} trade_log = [] daily_portfolio_value = [] # 실제 백테스트가 진행될 날짜 범위 simulation_dates = pd.date_range(config.BACKTEST_START_DATE, config.BACKTEST_END_DATE) # --- 3. 일일 시뮬레이션 루프 --- for today_date in simulation_dates: today_str = today_date.strftime('%Y-%m-%d') # (EOD - End of Day 기준 백테스트: 오늘 종가(Close)로 모든 것을 판단하고 거래) # (더 정교한 백테스트는 어제 종가로 판단 -> 오늘 시가(Open)에 거래) # 여기서는 사용자님의 요청(당일 돌파)을 반영하기 위해 EOD 기준으로 구현 # --- 3-1. 보유 포트폴리오 매도 신호 점검 --- tickers_to_sell = list(portfolio.keys()) for ticker in tickers_to_sell: if ticker not in stock_data_history: continue df = stock_data_history[ticker] try: pos = df.index.get_loc(today_date) except KeyError: continue if pos == 0: continue today_data = df.iloc[pos] position = portfolio[ticker] position['highest_price'] = max(position['highest_price'], today_data['High']) # [v3.1 수정] 매도 신호 체크 로직을 strategy.py의 함수로 위임 yesterday_data = df.iloc[pos - 1] sell_signal, sell_ratio = strategy.check_sell_signal(today_data, yesterday_data, position) # 백테스트 마지막 날 강제 청산 if today_date.strftime('%Y-%m-%d') == config.BACKTEST_END_DATE: sell_signal = 'FORCED_EXIT' sell_ratio = 1.0 if sell_signal: # 매도 실행 sell_price = today_data['Close'] shares_to_sell = position['shares'] * sell_ratio if sell_ratio < 1.0: shares_to_sell = np.floor(shares_to_sell) if shares_to_sell <= 0: continue sell_value = shares_to_sell * sell_price * (1 - config.TRANSACTION_FEE_PCT) capital += sell_value # 포지션 업데이트 position['shares'] -= shares_to_sell profit_pct = (sell_price - position['buy_price']) / position['buy_price'] trade_log.append({ 'ticker': ticker, 'name': position['name'], 'entry_date': position['buy_date'], 'exit_date': today_date, 'entry_price': position['buy_price'], 'exit_price': sell_price, 'shares_sold': shares_to_sell, 'profit_pct': profit_pct, 'signal_type': position.get('signal_type', 'N/A'), 'exit_reason': sell_signal }) log_msg = f"[{today_str}] SELL ({sell_signal}): {position['name']}({ticker}) @ {sell_price:,.0f} / {shares_to_sell:,.0f}주 ({profit_pct*100:,.2f}%)" print(log_msg) if sell_signal == 'PROFIT_TAKE': position['partial_sold'] = True if position['shares'] < 0.01: del portfolio[ticker] else: # 매도 안 할 시, 현재가로 자산 가치 합산 pass # --- 3-2. 신규 매수 신호 점검 --- # (현금 있고, 최대 포트폴리오 개수 미만일 때만) if len(portfolio) < config.MAX_PORTFOLIO_SIZE and capital > 0: # [수정] 투자 금액을 초기 자본이 아닌 현재 자산을 기준으로 계산 (복리 효과) if daily_portfolio_value: current_equity = daily_portfolio_value[-1][1] else: current_equity = config.BACKTEST_CAPITAL investment_per_stock = current_equity * config.INVESTMENT_PER_STOCK_PCT # 매수 가능한 금액이 최소 투자금액보다 많아야 함 if capital >= investment_per_stock: for ticker in tickers_dict.keys(): # 이미 보유 중이거나 최대 개수 도달 시 스킵 if ticker in portfolio or len(portfolio) >= config.MAX_PORTFOLIO_SIZE: continue if ticker not in stock_data_history: continue # 데이터 없음 df = stock_data_history[ticker] try: pos = df.index.get_loc(today_date) except KeyError: continue # 오늘 거래일이 아니거나 데이터 없음 if pos == 0: continue # 오늘이 이 종목의 첫 데이터이므로 비교/매수 불가 # [v3.0] 사전 계산된 신호에서 lookup if ticker not in buy_signals_history: continue buy_signals_df = buy_signals_history[ticker] try: signal_row = buy_signals_df.loc[today_date] except KeyError: continue buy_signal = signal_row['BUY_SIGNAL'] stop_loss_price = signal_row['STOP_LOSS_PRICE'] signal_type = signal_row['SIGNAL_TYPE'] # 매매 금지일 처리 if today_str in config.TRADING_BLACKOUT_DATES: buy_signal = False stop_loss_price = None signal_type = None if buy_signal: today_data = df.iloc[pos] buy_price = today_data['Close'] # 손절가 무결성 검증: NaN 또는 현재가보다 높으면 매수 불가 if pd.isna(stop_loss_price) or stop_loss_price >= buy_price: continue shares_to_buy = (investment_per_stock * (1 - config.TRANSACTION_FEE_PCT)) / buy_price shares_to_buy = np.floor(shares_to_buy) # 정수 단위 if shares_to_buy <= 0: continue buy_value = shares_to_buy * buy_price * (1 + config.TRANSACTION_FEE_PCT) if capital < buy_value: continue # 포트폴리오 및 자본 업데이트 capital -= buy_value ticker_name = tickers_dict.get(ticker, f"Unknown_{ticker}") portfolio[ticker] = { 'name': ticker_name, 'signal_type': signal_type, 'buy_price': buy_price, 'shares': shares_to_buy, 'stop_loss_price': stop_loss_price, 'highest_price': buy_price, 'partial_sold': False, 'buy_date': today_date } log_msg = f"[{today_str}] BUY: {ticker_name}({ticker}) @ {buy_price:,.0f} / {shares_to_buy:,.0f}주 (SL: {stop_loss_price:,.0f})" print(log_msg) # (v1.0 업그레이드: 최대 슬롯 도달 시 내부 루프 탈출) if len(portfolio) >= config.MAX_PORTFOLIO_SIZE: break # --- 3-3. 일일 자산 기록 --- # (루프 마지막에 보유 종목 가치 재계산) final_daily_value = capital for ticker, position in portfolio.items(): if ticker in stock_data_history: df = stock_data_history[ticker] # 오늘 데이터가 있으면 오늘 종가 사용 if today_str in df.index: close_price = df.loc[today_str]['Close'] else: # 오늘 데이터가 없으면(휴장일 등) 가장 최근 거래일의 종가 사용 available_dates = df.index[df.index <= today_date] if len(available_dates) > 0: last_available_date = available_dates[-1] close_price = df.loc[last_available_date]['Close'] else: # 매수 이후 한 번도 거래일이 없는 경우 (거의 없겠지만) 매수가 사용 close_price = position['buy_price'] final_daily_value += position['shares'] * close_price daily_portfolio_value.append((today_date, final_daily_value)) # --- 4. 백테스트 결과 리포트 --- print("\n" + "="*50) print(" Portfolio Backtesting Result Report (v3.0)") print("="*50) df_trades = pd.DataFrame(trade_log) if not daily_portfolio_value: print(" -> 발생한 거래가 없어 리포트를 생성할 수 없습니다.") return df_equity = pd.DataFrame(daily_portfolio_value, columns=['date', 'equity']).set_index('date') # 4-2. v1.0 성과 지표 계산 df_equity['return'] = df_equity['equity'].pct_change() df_equity['cumulative_return'] = (1 + df_equity['return']).cumprod() total_days = (df_equity.index[-1] - df_equity.index[0]).days years = max(total_days / 365.25, 1/365.25) # 0으로 나누기 방지 total_return = (df_equity['equity'].iloc[-1] / config.BACKTEST_CAPITAL) - 1 cagr = ((1 + total_return) ** (1/years)) - 1 peak = df_equity['cumulative_return'].cummax() drawdown = (df_equity['cumulative_return'] - peak) / peak max_drawdown = drawdown.min() daily_return = df_equity['return'].dropna() sharpe_ratio = 0.0 daily_std = daily_return.std() if daily_std > 1e-8 and not pd.isna(daily_std): sharpe_ratio = (daily_return.mean() / daily_std) * np.sqrt(252) else: sharpe_ratio = 0.0 # 변동성이 없으면 Sharpe Ratio 계산 불가 # 4-3. [v1.0 업그레이드] 상세 거래 성과 계산 total_trades = 0 wins = 0 win_rate = 0 avg_profit_pct = 0 avg_win_pct = 0 avg_loss_pct = 0 if not df_trades.empty: total_trades = len(df_trades) # 매도(청산) 기준 wins = (df_trades['profit_pct'] > 0).sum() win_rate = (wins / total_trades) * 100 avg_profit_pct = df_trades['profit_pct'].mean() * 100 # 승리/패배 건이 0일 경우 NaN이 아닌 0을 반환하도록 수정 win_profits = df_trades[df_trades['profit_pct'] > 0]['profit_pct'] avg_win_pct = win_profits.mean() * 100 if not win_profits.empty else 0 loss_profits = df_trades[df_trades['profit_pct'] <= 0]['profit_pct'] avg_loss_pct = loss_profits.mean() * 100 if not loss_profits.empty else 0 # 4-4. v1.0 스타일로 리포트 출력 final_capital = df_equity['equity'].iloc[-1] print("\n### Portfolio Performance Summary ###") print(f" - 기간: {config.BACKTEST_START_DATE} ~ {config.BACKTEST_END_DATE} ({years:.2f} 년)") print(f" - 초기 자본: {config.BACKTEST_CAPITAL:,.0f} 원") print(f" - 최종 자산: {final_capital:,.0f} 원") print(f" - 총 수익률 (Total Return): {total_return * 100:,.2f} %") print(f" - 연평균 복리 수익률 (CAGR): {cagr * 100:,.2f} %") print(f" - 최대 낙폭 (MDD): {max_drawdown * 100:,.2f} %") print(f" - 샤프 비율 (Sharpe Ratio): {sharpe_ratio:.2f}") print("\n### Trade Performance Summary (v3.0) ###") print(f" - 총 거래 횟수: {total_trades} 회 (매도 기준)") print(f" - 승률 (Win Rate): {win_rate:.2f} %") print(f" - 평균 수익/손실률: {avg_profit_pct:.2f} %") print(f" - 평균 승리 수익률: {avg_win_pct:.2f} %") print(f" - 평균 패배 손실률: {avg_loss_pct:.2f} %") # 4-5. [v1.0 업그레이드] 청산 사유별/신호 유형별 통계 if not df_trades.empty: print("\n### Exit Reason Statistics ###") try: # (부분 매도(PROFIT_TAKE)와 최종 청산을 구분하기 위해 집계) print(df_trades.groupby('exit_reason')['profit_pct'].agg( count='count', mean_profit=lambda x: f"{x.mean()*100:.2f}%" ).sort_values(by='count', ascending=False)) except KeyError as e: print(f" (청산 사유 통계 생성 실패 - 컬럼 없음: {e})") except Exception as e: print(f" (청산 사유 통계 생성 예상치 못한 오류: {type(e).__name__}: {e})") print("\n### Signal Type Performance ###") try: print(df_trades.groupby('signal_type')['profit_pct'].agg( count='count', mean_profit=lambda x: f"{x.mean()*100:.2f}%" ).sort_values(by='count', ascending=False)) except KeyError as e: print(f" (신호 유형 통계 생성 실패 - 컬럼 없음: {e})") except Exception as e: print(f" (신호 유형 통계 생성 예상치 못한 오류: {type(e).__name__}: {e})") # 4-6. v1.0 그래프 출력 if config.ENABLE_PLOT: try: plt.figure(figsize=(14, 7)) df_equity['equity_pct'] = df_equity['equity'] / config.BACKTEST_CAPITAL plt.plot(df_equity['equity_pct'], label='Portfolio Equity Curve (1.0 = Initial Capital)') plt.title(f'Portfolio Backtest Result ({years:.2f} years) - CAGR: {cagr*100:.2f}%') plt.xlabel('Date') plt.ylabel('Equity (Normalized)') plt.legend() plt.grid(True) plt.show() except ImportError as e: print(f"\n[Warning] matplotlib 미설치: {e}") except Exception as e: print(f"\n[Warning] 그래프 생성 예상치 못한 오류: {type(e).__name__}: {e}") if __name__ == "__main__": # 이 파일을 직접 실행하면 백테스트가 시작됩니다. run_backtest()