투자 분석
[모멘텀] 투자 전략 검증
SKY-STONE
2024. 9. 20. 01:01
Overview
- 모멘텀 투자 전략이 정말로 효과가 있을지에 관해서 탐구하는 시간
- 5개의 모멘텀 지표를 자동 서치로 최적화 시킨 후 Back-Test 진행
- ADX는 시장을 능가하는 결과를 보이나 약한 추세(4)에서 최적화가 된걸 보면 추세는 큰 의미가 없음
- S&P500 시장 수익률
- Cumulative Market Return: 405.05%
- Annualized Market Return: 11.64%
- Moving Average Search 결과
- Best Moving Average Parameters: {'long_window': 50, 'short_window': 11}
- Best Moving Average Cumulative Return: 20.71%
- Annualized Moving Average Return: 1.29%
- Bollinger Bands Search 결과
- Best Bollinger Bands Parameters: {'std_dev': 1.5, 'window': 11}
- Best Bollinger Bands Cumulative Return: 9.94%
- Annualized Bollinger Bands Return: 0.65%
- RSI Search 결과
- Best RSI Parameters: {'window': 2}
- Best RSI Cumulative Return: 5.97%
- Annualized RSI Return: 0.39%
- MACD Search 결과
- Best MACD Parameters: {'long_window': 31, 'short_window': 14, 'signal_window': 18}
- Best MACD Cumulative Return: 37.86%
- Annualized MACD Return: 2.21%
- ADX Search 결과
- Best ADX Parameters: {'window': 4}
- Best ADX Cumulative Return: 504.06%
- Annualized ADX Return: 13.01%
1. 이동 평균 (Moving Average, MA)
- 설명: 일정 기간 동안의 주가 평균을 계산하여 추세를 파악하는 지표입니다.
- 활용: 단기와 장기 이동 평균을 비교하여 매수/매도 신호를 생성합니다. 단기 MA가 장기 MA를 상향 돌파할 때 매수 신호, 하향 돌파할 때 매도 신호로 해석합니다.
2. 볼린저 밴드 (Bollinger Bands)
- 설명: 가격의 변동성을 측정하는 지표로, 중간선(이동 평균)과 그 위아래의 두 개의 밴드(표준편차)를 포함합니다.
- 활용: 가격이 하단 밴드 아래로 떨어지면 매수 신호, 상단 밴드를 초과하면 매도 신호로 해석합니다. 변동성이 클 때 유용합니다.
3. 상대 강도 지수 (Relative Strength Index, RSI)
- 설명: 가격의 상승과 하락의 속도를 비교하여 과매수 또는 과매도 상태를 나타내는 모멘텀 지표입니다. 0에서 100 사이의 값을 가집니다.
- 활용: RSI가 70 이상이면 과매수 상태로 판단하여 매도 신호, 30 이하이면 과매도 상태로 판단하여 매수 신호로 해석합니다.
4. 이동 평균 수렴 확산 지표 (Moving Average Convergence Divergence, MACD)
- 설명: 두 개의 이동 평균 간의 관계를 나타내는 지표로, 추세의 방향과 강도를 평가합니다.
- 활용: MACD 라인이 신호선을 상향 돌파하면 매수 신호, 하향 돌파하면 매도 신호로 해석합니다.
5. 평균 방향 지수 (Average Directional Index, ADX)
- 설명: 추세의 강도를 측정하는 지표로, 0에서 100 사이의 값을 가집니다. 25 이상이면 강한 추세, 20 이하이면 약한 추세를 나타냅니다.
- 활용: ADX가 25를 초과하면 강한 추세가 형성되었다고 보고, 매수 또는 매도 신호를 확인하는 데 사용합니다.
import yfinance as yf
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from sklearn.model_selection import ParameterGrid
import warnings
warnings.filterwarnings("ignore")
# 주식 데이터 다운로드
def get_data(ticker, start_date, end_date):
data = yf.download(ticker, start=start_date, end=end_date)
return data
# 이동 평균 계산
def calculate_moving_averages(data, short_window=20, long_window=50):
data['Short MA'] = data['Close'].rolling(window=short_window).mean()
data['Long MA'] = data['Close'].rolling(window=long_window).mean()
return data
# 볼린저 밴드 계산
def calculate_bollinger_bands(data, window=20, std_dev=2):
data['Middle Band'] = data['Close'].rolling(window=window).mean()
data['Upper Band'] = data['Middle Band'] + (data['Close'].rolling(window=window).std() * std_dev)
data['Lower Band'] = data['Middle Band'] - (data['Close'].rolling(window=window).std() * std_dev)
return data
# RSI 계산
def calculate_rsi(data, window=14):
delta = data['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
rs = gain / loss
data['RSI'] = 100 - (100 / (1 + rs))
return data
# MACD 계산
def calculate_macd(data, short_window=12, long_window=26, signal_window=9):
data['MACD'] = data['Close'].ewm(span=short_window, adjust=False).mean() - data['Close'].ewm(span=long_window, adjust=False).mean()
data['Signal Line'] = data['MACD'].ewm(span=signal_window, adjust=False).mean()
return data
# ADX 계산
def calculate_adx(data, window=14):
high = data['High']
low = data['Low']
close = data['Close']
plus_dm = (high.diff() > low.diff()).astype(int) * (high.diff()).clip(lower=0)
minus_dm = (low.diff() > high.diff()).astype(int) * (low.diff()).clip(lower=0)
tr = pd.DataFrame({'high-low': high - low, 'high-close': (high - close.shift()).abs(), 'low-close': (low - close.shift()).abs()})
tr['true_range'] = tr.max(axis=1)
atr = tr['true_range'].rolling(window).mean()
plus_di = 100 * (plus_dm.rolling(window).sum() / atr)
minus_di = 100 * (minus_dm.rolling(window).sum() / atr)
data['ADX'] = 100 * ((plus_di - minus_di).abs() / (plus_di + minus_di)).rolling(window).mean()
return data
# 매수 및 매도 신호 생성 (이동 평균)
def generate_ma_signals(data):
data['Signal'] = 0
data['Signal'][(data['Short MA'] > data['Long MA']) & (data['Short MA'].shift(1) <= data['Long MA'].shift(1))] = 1
data['Signal'][(data['Short MA'] < data['Long MA']) & (data['Short MA'].shift(1) >= data['Long MA'].shift(1))] = -1
data['Position'] = data['Signal'].diff()
data['Signal'] = data['Signal'].where(data['Close'] > data['Long MA'], 0) # 필터링
return data
# 매수 및 매도 신호 생성 (볼린저 밴드)
def generate_bb_signals(data):
data['Signal'] = 0
data['Signal'][(data['Close'] < data['Lower Band'])] = 1 # 매수
data['Signal'][(data['Close'] > data['Upper Band'])] = -1 # 매도
data['Position'] = data['Signal'].diff()
data['Signal'] = data['Signal'].where(data['Close'] > data['Middle Band'], 0) # 필터링
return data
# 매수 및 매도 신호 생성 (RSI)
def generate_rsi_signals(data):
data['Signal'] = 0
data['Signal'][data['RSI'] < 30] = 1 # 매수
data['Signal'][data['RSI'] > 70] = -1 # 매도
data['Position'] = data['Signal'].diff()
data['Signal'] = data['Signal'].where(data['Close'] > data['Close'].rolling(window=20).mean(), 0) # 필터링
return data
# 매수 및 매도 신호 생성 (MACD)
def generate_macd_signals(data):
data['Signal'] = 0
data['Signal'][(data['MACD'] > data['Signal Line']) & (data['MACD'].shift(1) <= data['Signal Line'].shift(1))] = 1 # 매수
data['Signal'][(data['MACD'] < data['Signal Line']) & (data['MACD'].shift(1) >= data['Signal Line'].shift(1))] = -1 # 매도
data['Position'] = data['Signal'].diff()
return data
# 매수 및 매도 신호 생성 (ADX)
def generate_adx_signals(data):
data['Signal'] = 0
data['Signal'][data['ADX'] > 25] = 1 # 강한 추세에 매수
data['Signal'][data['ADX'] < 20] = -1 # 약한 추세에 매도
data['Position'] = data['Signal'].diff()
return data
# 수익률 계산
def calculate_returns(data):
# 1. 시장 수익률 계산
data['Market Return'] = data['Close'].pct_change()
# 2. 전략 수익률 계산
data['Strategy Return'] = data['Market Return'] * data['Signal'].shift(1)
# 3. 누적 시장 수익률 계산 (기하 평균)
data['Cumulative Market Return'] = (1 + data['Market Return']).cumprod() - 1
# 4. 누적 전략 수익률 계산 (기하 평균)
data['Cumulative Strategy Return'] = (1 + data['Strategy Return']).cumprod() - 1
# 5. 기하 평균 수익률 계산
data['Geometric Market Return'] = np.exp(np.log1p(data['Market Return']).cumsum()) - 1
data['Geometric Strategy Return'] = np.exp(np.log1p(data['Strategy Return']).cumsum()) - 1
return data
# 최적화 함수
def optimize_moving_average(data, param_grid):
best_return = -float('inf')
best_params = {}
for params in ParameterGrid(param_grid):
temp_data = calculate_moving_averages(data.copy(), params['short_window'], params['long_window'])
temp_data = generate_ma_signals(temp_data)
temp_data = calculate_returns(temp_data)
if temp_data['Cumulative Strategy Return'].iloc[-1] > best_return:
best_return = temp_data['Cumulative Strategy Return'].iloc[-1]
best_params = params
return best_params, best_return
def optimize_bollinger_bands(data, param_grid):
best_return = -float('inf')
best_params = {}
for params in ParameterGrid(param_grid):
temp_data = calculate_bollinger_bands(data.copy(), params['window'], params['std_dev'])
temp_data = generate_bb_signals(temp_data)
temp_data = calculate_returns(temp_data)
if temp_data['Cumulative Strategy Return'].iloc[-1] > best_return:
best_return = temp_data['Cumulative Strategy Return'].iloc[-1]
best_params = params
return best_params, best_return
def optimize_rsi(data, param_grid):
best_return = -float('inf')
best_params = {}
for params in ParameterGrid(param_grid):
temp_data = calculate_rsi(data.copy(), params['window'])
temp_data = generate_rsi_signals(temp_data)
temp_data = calculate_returns(temp_data)
if temp_data['Cumulative Strategy Return'].iloc[-1] > best_return:
best_return = temp_data['Cumulative Strategy Return'].iloc[-1]
best_params = params
return best_params, best_return
def optimize_macd(data, param_grid):
best_return = -float('inf')
best_params = {}
for params in ParameterGrid(param_grid):
temp_data = calculate_macd(data.copy(), params['short_window'], params['long_window'], params['signal_window'])
temp_data = generate_macd_signals(temp_data)
temp_data = calculate_returns(temp_data)
if temp_data['Cumulative Strategy Return'].iloc[-1] > best_return:
best_return = temp_data['Cumulative Strategy Return'].iloc[-1]
best_params = params
return best_params, best_return
def optimize_adx(data, param_grid):
best_return = -float('inf')
best_params = {}
for params in ParameterGrid(param_grid):
temp_data = calculate_adx(data.copy(), params['window'])
temp_data = generate_adx_signals(temp_data)
temp_data = calculate_returns(temp_data)
if temp_data['Cumulative Strategy Return'].iloc[-1] > best_return:
best_return = temp_data['Cumulative Strategy Return'].iloc[-1]
best_params = params
return best_params, best_return
# 데이터 시각화 (서브플롯)
def plot_subplots(data, ticker, strategy_name):
fig, axs = plt.subplots(2, 1, figsize=(12, 12))
axs[0].plot(data['Close'], label=f'{ticker} Price', alpha=0.5)
if strategy_name == "Moving Average":
axs[0].plot(data['Short MA'], label='Short MA', color='orange', alpha=0.7)
axs[0].plot(data['Long MA'], label='Long MA', color='red', alpha=0.7)
elif strategy_name == "Bollinger Bands":
axs[0].plot(data['Upper Band'], label='Upper Band', color='red', linestyle='--')
axs[0].plot(data['Middle Band'], label='Middle Band', color='orange')
axs[0].plot(data['Lower Band'], label='Lower Band', color='red', linestyle='--')
elif strategy_name == "RSI":
axs[0].plot(data['RSI'], label='RSI', color='purple')
axs[0].axhline(70, linestyle='--', color='red')
axs[0].axhline(30, linestyle='--', color='green')
elif strategy_name == "MACD":
axs[0].plot(data['MACD'], label='MACD', color='blue')
axs[0].plot(data['Signal Line'], label='Signal Line', color='orange')
elif strategy_name == "ADX":
axs[0].plot(data['ADX'], label='ADX', color='purple')
axs[0].axhline(25, linestyle='--', color='red')
axs[0].axhline(20, linestyle='--', color='green')
axs[0].plot(data[data['Position'] == 1].index, data['Close'][data['Position'] == 1], '^', markersize=10, color='g', label='Buy Signal')
axs[0].plot(data[data['Position'] == -1].index, data['Close'][data['Position'] == -1], 'v', markersize=10, color='r', label='Sell Signal')
axs[0].set_title(f'{ticker} Price and {strategy_name} with Buy/Sell Signals')
axs[0].legend()
axs[1].plot(data['Cumulative Market Return'], label='Cumulative Market Return', color='blue')
axs[1].plot(data['Cumulative Strategy Return'], label='Cumulative Strategy Return', color='orange')
axs[1].set_title(f'{ticker} Cumulative Returns ({strategy_name})')
axs[1].legend()
plt.tight_layout()
plt.show()
# 매개변수 설정
ticker = '^GSPC' # 주식 티커
start_date = '2010-01-01'
end_date = '2024-09-20'
# 파라미터 탐색 범위 설정
ma_param_grid = {
'short_window': range(5, 51, 1), # 5에서 50까지 5씩 증가
'long_window': range(50, 201, 10), # 50에서 200까지 10씩 증가
}
bb_param_grid = {
'window': range(1, 61, 1), # 1에서 60까지 5씩 증가
'std_dev': [1, 1.5, 2, 2.5, 3, 4], # 다양한 표준편차 값
}
rsi_param_grid = {
'window': range(1, 31), # 5에서 30까지
}
macd_param_grid = {
'short_window': range(5, 21, 1), # 5에서 20까지 1씩 증가
'long_window': range(21, 51, 1), # 21에서 50까지 1씩 증가
'signal_window': range(5, 21, 1) # 5에서 20까지 1씩 증가
}
adx_param_grid = {
'window': range(1, 51), # 1에서 50까지
}
data = get_data(ticker, start_date, end_date)
# 시장 수익률 계산
def calculate_market_returns(data):
data['Market Return'] = data['Close'].pct_change()
data['Cumulative Market Return'] = (1 + data['Market Return']).cumprod() - 1
return data['Cumulative Market Return'].iloc[-1]
# 연간 수익률 계산 함수
def annualized_return(cumulative_return, total_periods):
return (1 + cumulative_return) ** (1 / total_periods) - 1
# 시장 수익률 계산
market_return = calculate_market_returns(data)
total_periods = (data.index[-1] - data.index[0]).days / 365.25 # 투자 기간을 연수로 변환
# 연간 시장 수익률
annual_market_return = annualized_return(market_return, total_periods)
print("Cumulative Market Return: {:.2f}%".format(market_return * 100))
print("Annualized Market Return: {:.2f}%".format(annual_market_return * 100))
# 각 전략에 대한 파라미터 최적화
best_ma_params, best_ma_return = optimize_moving_average(data, ma_param_grid)
annual_ma_return = annualized_return(best_ma_return, total_periods)
print("Best Moving Average Parameters:", best_ma_params)
print("Best Moving Average Cumulative Return: {:.2f}%".format(best_ma_return * 100))
print("Annualized Moving Average Return: {:.2f}%".format(annual_ma_return * 100))
best_bb_params, best_bb_return = optimize_bollinger_bands(data, bb_param_grid)
annual_bb_return = annualized_return(best_bb_return, total_periods)
print("Best Bollinger Bands Parameters:", best_bb_params)
print("Best Bollinger Bands Cumulative Return: {:.2f}%".format(best_bb_return * 100))
print("Annualized Bollinger Bands Return: {:.2f}%".format(annual_bb_return * 100))
best_rsi_params, best_rsi_return = optimize_rsi(data, rsi_param_grid)
annual_rsi_return = annualized_return(best_rsi_return, total_periods)
print("Best RSI Parameters:", best_rsi_params)
print("Best RSI Cumulative Return: {:.2f}%".format(best_rsi_return * 100))
print("Annualized RSI Return: {:.2f}%".format(annual_rsi_return * 100))
best_macd_params, best_macd_return = optimize_macd(data, macd_param_grid)
annual_macd_return = annualized_return(best_macd_return, total_periods)
print("Best MACD Parameters:", best_macd_params)
print("Best MACD Cumulative Return: {:.2f}%".format(best_macd_return * 100))
print("Annualized MACD Return: {:.2f}%".format(annual_macd_return * 100))
best_adx_params, best_adx_return = optimize_adx(data, adx_param_grid)
annual_adx_return = annualized_return(best_adx_return, total_periods)
print("Best ADX Parameters:", best_adx_params)
print("Best ADX Cumulative Return: {:.2f}%".format(best_adx_return * 100))
print("Annualized ADX Return: {:.2f}%".format(annual_adx_return * 100))
# 최적화된 파라미터로 최종 결과 확인
final_ma_data = calculate_moving_averages(data.copy(), best_ma_params['short_window'], best_ma_params['long_window'])
final_ma_data = generate_ma_signals(final_ma_data)
final_ma_data = calculate_returns(final_ma_data)
plot_subplots(final_ma_data, ticker, "Moving Average")
final_bb_data = calculate_bollinger_bands(data.copy(), best_bb_params['window'], best_bb_params['std_dev'])
final_bb_data = generate_bb_signals(final_bb_data)
final_bb_data = calculate_returns(final_bb_data)
plot_subplots(final_bb_data, ticker, "Bollinger Bands")
final_rsi_data = calculate_rsi(data.copy(), best_rsi_params['window'])
final_rsi_data = generate_rsi_signals(final_rsi_data)
final_rsi_data = calculate_returns(final_rsi_data)
plot_subplots(final_rsi_data, ticker, "RSI")
final_macd_data = calculate_macd(data.copy(), best_macd_params['short_window'], best_macd_params['long_window'], best_macd_params['signal_window'])
final_macd_data = generate_macd_signals(final_macd_data)
final_macd_data = calculate_returns(final_macd_data)
plot_subplots(final_macd_data, ticker, "MACD")
final_adx_data = calculate_adx(data.copy(), best_adx_params['window'])
final_adx_data = generate_adx_signals(final_adx_data)
final_adx_data = calculate_returns(final_adx_data)
plot_subplots(final_adx_data, ticker, "ADX")