VVA(Variable Asset Allocation) 전략은 모멘텀 기반으로 자산을 동적 배분하는 투자 전략입니다. 자산별 모멘텀을 평가하여 성과가 좋은 자산(공격적 자산)을 선택하고, 성과가 저조할 경우 안전한 자산(방어적 자산)으로 전환합니다.
이번 글에서는 VVA 전략을 파이썬으로 구현하고, 백테스트를 통해 실제로 성능을 검증해보는 과정을 다룹니다.
# VVA 전략 구현
# 해당 코드는 다양한 ETF의 과거 데이터를 이용해 VVA 전략(모멘텀 기반 자산배분 전략)을 백테스트 하는 예제입니다.
import pandas as pd
from dateutil.relativedelta import *
import math
import numpy as np
import yfinance as yf
def merge_file(etflist):
"""
여러 ETF의 데이터를 불러와 하나의 DataFrame으로 병합하는 함수.
매개변수:
etflist: ETF 티커명 리스트 (예: ['spy', 'vea', 'eem', ...])
과정:
- 각 ETF에 대해 load_and_arrange_data 함수를 호출하여 데이터를 불러옴.
- 각 ETF의 데이터를 날짜(index)를 기준으로 outer merge하여 병합.
- 결측치가 있는 행들을 제거하여 최종 DataFrame 반환.
"""
benchmark_df = pd.DataFrame()
for number in range(0, len(etflist)):
each_etf_df = load_and_arrange_data(etflist[number])
# 인덱스(날짜)를 기준으로 데이터를 병합함
benchmark_df = pd.merge(benchmark_df, each_etf_df, left_index=True, right_index=True, how='outer')
benchmark_final = benchmark_df.dropna() # 결측치 제거
return benchmark_final
def load_and_arrange_data(etfname):
"""
개별 ETF의 데이터를 yfinance를 이용해 불러오고, 필요한 컬럼만 추출하여 정리하는 함수.
매개변수:
etfname: ETF 티커명 (소문자 또는 대문자 상관 없음)
과정:
- 티커명을 대문자로 변환하여 yfinance Ticker 객체 생성.
- 'max' 기간의 역사적 데이터를 불러옴.
- 'Adj Close' 컬럼이 존재하면 해당 컬럼을 사용, 없으면 'Close' 컬럼 사용.
- ETF 티커명을 컬럼명으로 변경 후, 중복된 날짜(index)는 제거.
"""
load_key = etfname.upper()
item_df = yf.Ticker(load_key).history(period='max')
# 'Adj Close'가 있으면 조정 종가 사용, 없으면 종가 사용
if 'Adj Close' in item_df.columns:
item_df = item_df.rename(columns={'Adj Close': etfname})
elif 'Adj Close' not in item_df.columns:
item_df = item_df.rename(columns={'Close': etfname})
# 해당 ETF의 가격 데이터만 남김
item_df = item_df[[etfname]]
# 중복된 날짜가 있다면 첫번째 값만 사용
item_df = item_df[~item_df.index.duplicated(keep='first')]
return item_df
def find_rebalancing_period(data, start_date, end_date, byperiod):
"""
리밸런싱(재조정) 날짜를 결정하는 함수.
매개변수:
data: 전체 가격 데이터 DataFrame (날짜를 인덱스로 가짐)
start_date: 리밸런싱 시작 날짜
end_date: 리밸런싱 종료 날짜
byperiod: 몇 개월 간격으로 리밸런싱할지 (예: 1이면 매월)
과정:
- start_date부터 end_date까지의 날짜 인덱스를 DataFrame으로 변환.
- pd.Grouper를 사용하여 지정한 개월 단위로 그룹화한 후, 각 그룹의 마지막 날짜(최대 날짜)를 선택.
- 백테스트 종료일(backtest_end_date) 이후의 날짜는 제외하고 최종 리밸런싱 날짜 리스트 반환.
"""
# start_date ~ end_date 기간의 날짜 인덱스를 DataFrame으로 변환
period_df = pd.DataFrame(data[start_date:end_date].index, columns=['Date'])
# 지정한 주기(byperiod개월)별로 그룹화 후, 각 그룹의 마지막 날짜 선택
select_date = period_df.loc[period_df.groupby(pd.Grouper(key='Date', freq=f'{byperiod}M')).Date.idxmax()]
# 백테스트 종료 날짜를 조정 (마지막 달의 데이터를 사용할 수 있도록 조정)
backtest_end_date = end_date + relativedelta(months=-1) + relativedelta(day=+40)
# 백테스트 종료 날짜 이전의 날짜들만 선택
rebalancing_day = [intime for intime in select_date['Date'] if intime <= backtest_end_date]
return rebalancing_day
def get_backtest_result(rebalancing_period, benchmark_data, offensive_asset, defensive_asset, invest_share=1):
"""
VVA 전략에 따라 각 리밸런싱 기간별로 백테스트를 수행하는 함수.
매개변수:
rebalancing_period: 리밸런싱 날짜 리스트
benchmark_data: 전체 ETF 가격 데이터 DataFrame
offensive_asset: 공격적 자산 리스트 (예: 모멘텀이 좋은 자산들)
defensive_asset: 방어적 자산 리스트 (예: 모멘텀이 좋지 않을 때 선택할 자산들)
invest_share: 투자 비중 (기본값 1: 전액 투자)
과정:
- 초기 자산을 1000으로 설정.
- 각 리밸런싱 구간마다,
1. 현재까지의 데이터를 이용해 모멘텀(vaa_momentum) 계산.
2. check_condition 함수를 통해 최종 투자 자산(final_asset) 결정.
3. 해당 구간의 가격 데이터(sub_benchmark_data) 선택.
4. 자산 변경이 있을 경우 수수료(0.23% 차감)를 반영.
5. do_backtest 함수를 이용해 구간별 백테스트 수행.
6. 구간별 결과를 누적하여 total_backtest_result에 저장.
"""
initial_money = 1000
total_backtest_result = pd.DataFrame()
for index in range(len(rebalancing_period) - 1):
# 현재 리밸런싱 이전까지의 데이터를 사용하여 모멘텀 계산
vaa_momentum = calculate_vaa_momentum(benchmark_data[:rebalancing_period[index]])
# 계산된 모멘텀에 따라 투자할 자산 결정
final_asset = check_condition(vaa_momentum, offensive_asset, defensive_asset)
# 해당 리밸런싱 기간(현재 날짜부터 다음 리밸런싱 날짜까지)의 데이터 선택
sub_benchmark_data = pd.DataFrame(
benchmark_data.loc[rebalancing_period[index]:rebalancing_period[index + 1], final_asset])
# 만약 이전 구간과 자산이 달라졌다면 매매 수수료(0.23% 차감) 반영
if index > 0 and save_asset != final_asset:
initial_money = initial_money * 0.9977
else:
pass
save_asset = final_asset # 현재 선택된 자산 저장
# 선택된 자산에 대해 백테스트 실행
backtest_result = do_backtest(initial_money, sub_benchmark_data, invest_share)
# 다음 구간의 초기 자산은 현재 구간의 최종 자산으로 설정
initial_money = backtest_result['total_asset'].iloc[-1]
# 구간별 결과를 누적 (마지막 행이 중복되지 않도록 조정)
total_backtest_result = pd.concat([total_backtest_result[:-1], backtest_result])
return total_backtest_result
def calculate_vaa_momentum(benchmark_data):
"""
여러 기간의 수익률을 이용해 각 자산의 모멘텀 점수를 계산하는 함수.
매개변수:
benchmark_data: 가격 데이터 DataFrame (날짜를 인덱스로 가짐)
과정:
- 각 자산에 대해 1, 3, 6, 12개월의 수익률을 계산하고, 연환산하여 합산.
- 결과적으로 각 자산별 모멘텀 점수를 dictionary 형태로 반환.
"""
vaa_momentum = {}
monthlist = [1, 3, 6, 12] # 모멘텀 계산에 사용할 기간 (월 단위)
current_day = benchmark_data.index[-1] # 최신 날짜
for name in benchmark_data.columns:
rate = 0
# 각 기간마다 수익률 계산 후 가중치(연환산) 적용
for pr in monthlist:
base_day = current_day - relativedelta(months=pr)
sub_data = pd.DataFrame(benchmark_data.loc[base_day:current_day, name])
# 기간 수익률을 계산하고 연환산 (12/pr 배)
rate += (sub_data.iloc[-1][name] / sub_data.iloc[0][name] - 1) * (12 / pr) * 100
vaa_momentum[name] = rate
return vaa_momentum
def check_condition(vaa_momentum, offensive_asset, defensive_asset):
"""
계산된 모멘텀 점수를 바탕으로 최종 투자할 자산을 선택하는 함수.
매개변수:
vaa_momentum: 각 자산의 모멘텀 점수를 담은 dict
offensive_asset: 공격적 자산 리스트
defensive_asset: 방어적 자산 리스트
과정:
- 모멘텀 점수를 공격적/방어적 자산으로 분리하여 각각 내림차순 정렬.
- 모든 공격적 자산의 모멘텀이 양수이면, 가장 높은 모멘텀의 공격적 자산 선택.
- 그렇지 않으면, 방어적 자산 중 가장 높은 모멘텀을 가진 자산 선택.
"""
offensive_asset_m = {}
defensive_asset_m = {}
# 각 자산을 공격적, 방어적 그룹으로 분류
for name, momentum in vaa_momentum.items():
if name in offensive_asset:
offensive_asset_m[name] = momentum
elif name in defensive_asset:
defensive_asset_m[name] = momentum
# 모멘텀 점수 내림차순으로 정렬
offensive_asset_m = dict(sorted(offensive_asset_m.items(), key=lambda item: item[1], reverse=True))
defensive_asset_m = dict(sorted(defensive_asset_m.items(), key=lambda item: item[1], reverse=True))
# 공격적 자산 중 모멘텀이 양수인 자산의 개수 계산
satisfied_asset_number = sum(x > 0 for x in offensive_asset_m.values())
# 모든 공격적 자산이 양수이면 공격적 자산 중 최고 모멘텀 선택, 아니면 방어적 자산 선택
if len(offensive_asset) == satisfied_asset_number:
final_asset = list(offensive_asset_m.keys())[0]
else:
final_asset = list(defensive_asset_m.keys())[0]
return final_asset
def do_backtest(initial_money, sub_benchmark_data, invest_share):
"""
특정 기간 동안 선택된 자산에 대해 백테스트를 실행하는 함수.
매개변수:
initial_money: 해당 기간 시작 시점의 자산 금액
sub_benchmark_data: 해당 기간의 가격 데이터 (하나 이상의 자산)
invest_share: 투자 비중 (예: 1이면 전액 투자)
과정:
- invest_money = initial_money * invest_share 만큼 투자.
- 각 자산에 동일한 비중으로 투자하여 구매 가능한 주식 수 계산 (소수점 버림).
- 투자 자산의 가치는 매일의 가격에 따라 계산.
- 남은 현금과 투자 자산 가치를 합산해 총 자산 계산.
- 일일 수익률(daily_return)도 계산하여 DataFrame 반환.
"""
invest_money = initial_money * invest_share
backtest_result = pd.DataFrame(index=sub_benchmark_data.index)
backtest_result['invest_asset'] = 0
# 각 자산별로 투자 금액을 동일 분할하여 계산
for col in sub_benchmark_data.columns:
each_money = invest_money / len(sub_benchmark_data.columns)
# 구매 가능한 주식 수 (정수) 계산
initial_buy = math.trunc(each_money / sub_benchmark_data[col][0])
# 각 날짜별 자산 가치를 누적 (구매한 주식 수 * 해당 날짜 가격)
backtest_result['invest_asset'] = backtest_result['invest_asset'] + initial_buy * sub_benchmark_data[col]
# 첫 날에 투자 후 남은 현금 계산
backtest_result['cash_asset'] = initial_money - backtest_result['invest_asset'].iloc[0]
# 총 자산 = 투자 자산 + 현금
backtest_result['total_asset'] = backtest_result['invest_asset'] + backtest_result['cash_asset']
# 일별 수익률 계산 (백분율)
backtest_result['daily_return'] = backtest_result['total_asset'].pct_change() * 100
return backtest_result
def calculate_cagr(total_backtest_result):
"""
백테스트 기간 동안의 CAGR (연평균 복리 성장률)을 계산하는 함수.
매개변수:
total_backtest_result: 전체 백테스트 결과 DataFrame
과정:
- 시작과 종료 날짜 사이의 기간(년 단위) 계산.
- 초기 자산과 최종 자산의 비율로부터 CAGR 계산.
- 소수점 둘째자리까지 반올림 후 반환.
"""
year_period = total_backtest_result.index[-1].year - total_backtest_result.index[0].year
month_period = (total_backtest_result.index[-1].month - total_backtest_result.index[0].month) / 12
final_period = year_period + month_period
CAGR = round(((total_backtest_result.iloc[-1]['total_asset'] / total_backtest_result.iloc[0]['total_asset']) ** (
1 / final_period) - 1) * 100, 2)
return CAGR
def calculate_mdd(total_backtest_result):
"""
백테스트 기간 동안의 최대 손실률(MDD; Maximum DrawDown)을 계산하는 함수.
매개변수:
total_backtest_result: 전체 백테스트 결과 DataFrame
과정:
- 누적 최대 자산 값을 계산.
- 현재 자산 대비 누적 최대 자산의 하락 비율 계산.
- 이 중 가장 큰 손실(최소값)을 MDD로 산출 후, 소수점 둘째자리까지 반올림.
"""
max_value = np.maximum.accumulate(total_backtest_result['total_asset'])
rate_value = (total_backtest_result['total_asset'] - max_value) / max_value
mdd = round(rate_value.min() * 100, 2)
return mdd
# ----------------- 메인 실행부 -----------------
# 분석에 사용할 ETF 리스트
choice_asset = ['spy', 'vea', 'eem', 'agg', 'shy', 'ief', 'lqd']
# 공격적 자산(모멘텀이 좋은 자산)
offensive_asset = ['spy', 'vea', 'eem', 'agg']
# 방어적 자산(모멘텀이 좋지 않을 때 선택할 자산)
defensive_asset = ['ief', 'shy', 'lqd']
# ETF 데이터를 불러와서 병합
benchmark_data = merge_file(choice_asset)
# 리밸런싱 시작 날짜는 데이터 시작 후 12개월 뒤, 종료 날짜는 마지막 날짜로 설정
start_date = benchmark_data.index[0] + relativedelta(months=12)
end_date = benchmark_data.index[-1]
# 리밸런싱 날짜를 매월 단위로 결정
rebalancing_period = find_rebalancing_period(benchmark_data, start_date, end_date, 1)
# VVA 전략에 따른 백테스트 실행
total_backtest_result = get_backtest_result(rebalancing_period, benchmark_data, offensive_asset, defensive_asset)
# 백테스트 결과를 바탕으로 CAGR 및 MDD 계산
cagr = calculate_cagr(total_backtest_result)
mdd = calculate_mdd(total_backtest_result)2. 주요 파이썬 코드 분석
2.1. ETF 데이터 병합 (merge_file)
yfinance를 사용해 각 ETF 데이터를 불러와 하나의 DataFrame으로 병합합니다. 각 ETF의 종가 데이터를 기준으로 결측치를 제거하여 안정성을 확보했습니다.
<python>def merge_file(etflist):
benchmark_df = pd.DataFrame()
for etf in etflist:
each_etf_df = load_and_arrange_data(etf)
benchmark_df = pd.merge(benchmark_df, each_etf_df, left_index=True, right_index=True, how='outer')
return benchmark_df.dropna()
2.2. 모멘텀 계산 (calculate_vaa_momentum)
1, 3, 6, 12개월 단위로 각 자산의 수익률을 연환산하여 모멘텀 점수를 부여합니다. 이 점수를 바탕으로 공격적 자산과 방어적 자산을 구분합니다.
<python>for pr in [1, 3, 6, 12]:
rate += (sub_data.iloc[-1][name] / sub_data.iloc[0][name] - 1) * (12 / pr) * 100
2.3. 자산 선택 (check_condition)
공격적 자산의 모멘텀이 모두 양수일 경우, 모멘텀이 가장 높은 자산을 선택하고, 하나라도 음수이면 방어적 자산 중 최고 모멘텀을 가진 자산을 선택합니다.
3. 리밸런싱과 백테스트
- 리밸런싱 주기 설정:
find_rebalancing_period함수는 매월 말일을 리밸런싱 시점으로 설정합니다. - 백테스트 수행: 각 기간별로 투자 자산을 선택하고, 매매 수수료(0.23%)까지 반영한
get_backtest_result함수를 통해 백테스트를 수행합니다.
4. 백테스트 결과 분석
CAGR (연평균 복리 성장률)
- 구현된
calculate_cagr함수를 통해 12개월 이상 데이터를 기반으로 CAGR을 계산합니다. - 예:
연평균 수익률 12.5%
MDD (최대 손실률)
calculate_mdd로 누적 최고 자산 대비 최대 하락폭을 계산해 리스크 관리에 활용합니다.- 예:
최대 손실 -10.7%
5. 결론
이번 포스팅에서는 VVA 전략을 파이썬으로 구현하고 백테스트를 수행하는 과정을 살펴봤습니다. 모멘텀 기반의 동적 자산배분 전략은 단순한 포트폴리오보다 안정적이고 효율적인 수익을 추구할 수 있습니다. 직접 파이썬 코드로 구현하며 투자 전략을 테스트해보세요! 🚀