406 lines
18 KiB
Python
406 lines
18 KiB
Python
# 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() |