파이썬으로 연습해보는 퀀트 투자 - 듀얼 모멘텀 중 상대 모멘텀 전략

절대 모멘텀 전략은 단일 종목에도 백테스트를 할 수 있다. 그러나 상대 모멘텀 전략은 다수 종목으로 투자 대상군을 형성해야 이용할 수 있다. 

상대 모멘텀 전략에서는 모멘텀 지수를 계산하기 위해 과거 1개월 종가의 수익률을 계산한다. 지난달 마지막 일자를 기준으로 전체 투자 대상 종목에서 상대적으로 높은 수익률을 보인 상위 종목에 매수 신호를 발생시킨다. 

먼저 사용할 라이브러리들을 읽어들인다.

import pandas as pd 
import numpy as np
import FinanceDataReader as fdr

그 다음 파이낸스데이타리더 라이브러리를 사용해 주식 데이터를 읽어들인다.

# 주식을 먼저 선정한다. 여기서는 10개의 종목을 선정한다. 
# 애플 AAPL, 아마존 AMZN, 메타 META, 월마트 WMT, 넷플릭스 NFLX, 제너럴 모터스 GM, 마이크로소프트MSFT, 테슬라 TSLA, 엔비디아 NVDA, 코카콜라 KO
# 주식 기간은 2015.1.1~2023.08.31
ticker = ['AAPL', 'AMZN', 'META', 'WMT', 'NFLX', 'GM', 'MSFT', 'TSLA', 'NVDA', 'KO']
all_df = pd.DataFrame()
for f in ticker: 
    temp = fdr.DataReader(f, '2015-01-01', '2023-08-31')
    temp['CODE'] = f
    temp.reset_index(inplace = True)
    all_df = pd.concat([all_df, temp], axis = 0, ignore_index = True) #pandas 2.0이상에서는 append 함수가 없어지고, concat함수를 사용해야 함
all_df.head()

읽어드린 주식 데이터에서 필요한 컬럼만 추출해 사용할 수 있도록 데이터를 전처리하는 함수를 아래처럼 만든다.

def data_preprocessing(sample, t_code, base_date):
    sample = sample[sample['Date'] >= base_date][['Date','CODE','Adj Close']].copy()
    sample.reset_index(inplace = True, drop = True)
    sample['STD_YM'] = sample['Date'].dt.to_period(freq='M') 
    sample['1M_RET'] = 0.0
    ym_keys = list(sample['STD_YM'].unique())
    return sample, ym_keys

data_preprocessing함수로 데이터를 전처리해서 price_df, ym_keys 변수에 저장한다. 
이때 기준 날짜를 정할 수 있다. 이번에는 2019-01-01로 기준 날짜를 정했다.

price_df, ym_keys = data_preprocessing(all_df, ticker, base_date ='2019-01-01')
price_df.head()

월별 데이터를 저장할 month_last_df라는 데이터프레임을 만들고, 각 주식별 1개월간 수익률을 계산하여 '1M_RET'이라는 컬럼에 넣도록 한다.

month_last_df = pd.DataFrame(columns = ['Date', 'CODE', '1M_RET'])

for f in ticker: 
    temp_df = price_df[price_df['CODE'] == f][['Date','CODE','STD_YM','Adj Close','1M_RET']].copy()
    for ym in ym_keys:
        m_ret = temp_df.loc[temp_df[temp_df['STD_YM'] == ym].index[-1],'Adj Close'] \
        / temp_df.loc[temp_df[temp_df['STD_YM'] == ym].index[0],'Adj Close']
        temp_df.loc[temp_df['STD_YM'] == ym, ['1M_RET']] = m_ret
        month_last_df = month_last_df._append(temp_df.loc[temp_df[temp_df['STD_YM'] == ym].index[-1], ['Date', 'CODE', '1M_RET']])

그 다음으로 종목별 종가를 저장할 stock_df라는 데이터프레임도 만들도록 한다.

stock_df = pd.DataFrame(columns = ['Date', 'CODE', 'Adj Close'])
stock_df = pd.concat([stock_df, price_df.loc[:,['Date', 'CODE', 'Adj Close']]], sort=False)
stock_df.sort_values(by = 'Date', ascending = True).head(10)

월별 수익률이 행으로 쌓여있는 데이터를 피벗해 일자별로 종목별 수익률 데이터로 만든다. 종목 코드를 컬럼으로 올리고, DataFrame의 rank()함수를 사용해 월말 수익률의 순위를 퍼센트 순위로 계산하고 상위 40% 종목만 선별한 다음 나머지 값에는 np.nan값을 채운다.

month_ret_df = month_last_df.pivot_table(index='Date', columns='CODE', values='1M_RET').copy()
month_ret_df = month_ret_df.rank(axis=1, ascending=False, method='max', pct=True)
month_ret_df = month_ret_df.where(month_ret_df < 0.4, np.nan)
month_ret_df.fillna(0, inplace=True)
month_ret_df[month_ret_df !=0] = 1
stock_codes = list(stock_df['CODE'].unique())

month_ret_df.head()

월말일자로 1로 표시된 종목코드와 0으로 표시된 종목코드를 확인할 수 있다. 다음 날짜로 넘어갈 때 0에서 1로 되면 제로 포지션에서 매수 포지션으로 변경되고 1에서 0으로 변경되면 청산한다. 1에서 1로 변함없는 종목은 매수후 보유상태를 유지한다.

반응형

다음에 진행할 순서는 월말에 거래 신호가 나타난 종목을 대상으로 포지셔닝을 처리하는 것이다.

sig_dict = dict()
# 신호가 포착된 종목 코드만 읽어오기
for date in month_ret_df.index: 
    ticker_list = list(month_ret_df.loc[date, month_ret_df.loc[date,:] >= 1.0].index)
    sig_dict[date] = ticker_list
stock_c_matrix = stock_df.pivot_table(index='Date', columns='CODE', values='Adj Close').copy()
book = create_trade_book(stock_c_matrix, list(stock_df['CODE'].unique()))

for date, values in sig_dict.items(): # 포지셔닝
    for stock in values:
        book.loc[date,'p '+ stock] = 'ready ' + stock
        
book = tradings(book, stock_codes) # 트레이딩

신호가 발생한 종목 리스트를 만들고 stock_df 변수를 피벗해 만든 stock_c_matrix 변수를 전달해 새로운 거래 장부 역할 변수를 만든다. 

거래 장부 역할 변수는 다음의 코드로 작성된 create_trade_book() 함수로 만들어진다. 

그 다음 for 문을 사용해 월초 일자별 신호가 발생한 종목에 포지션을 기록한다. 이렇게 포지셔닝을 해놓고 두번째로 만든 tradings() 함수를 통해 트레이딩을 진행하게 된다.

- create_trade_book() 함수

def create_trade_book(sample, sample_codes):
    book = pd.DataFrame()
    book = sample[sample_codes].copy()
    book['STD_YM'] = book.index.to_period(freq='M')
    
    for c in sample_codes:
        book['p '+c] = ''
        book['r '+c] = ''
    return book

- tradings() 함수

def tradings(book, s_codes):
    std_ym = ''
    buy_phase = False
    for s in s_codes:
        print(s)
        
        for i in book.index:
            if book.loc[i, 'p '+ s] == '' and book.shift(1).loc[i, 'p '+s] == 'ready ' +s:
                std_ym = book.loc[i, 'STD_YM']
                buy_phase = True
                
            if book.loc[i, 'p '+ s] == '' and book.loc[i, 'STD_YM'] == std_ym and buy_phase == True:
                book.loc[i, 'p '+ s] = 'buy ' + s
            if book.loc[i, 'p '+ s] == '':
                std_ym = None
                buy_phase = False
    return book

마지막으로 거래 장부 book에 있는 거래를 가지고 상대 모멘텀 전략의 거래 수익률을 계산할 multi_returns() 함수를 아래와 같이 만든다.

def multi_returns(book, s_codes):
    # 손익 계산
    rtn = 1.0
    buy_dict = {}
    num = len(s_codes)
    sell_dict = {}
    
    for i in book.index:
        for s in s_codes:
            if book.loc[i, 'p ' + s] == 'buy '+ s and \
            book.shift(1).loc[i, 'p '+s] == 'ready '+s and \
            book.shift(2).loc[i, 'p '+s] == '' :     # long 진입
                buy_dict[s] = book.loc[i, s]
#                 print('진입일 : ',i, '종목코드 : ',s ,' long 진입가격 : ', buy_dict[s])
            elif book.loc[i, 'p '+ s] == '' and book.shift(1).loc[i, 'p '+s] == 'buy '+ s:     # long 청산
                sell_dict[s] = book.loc[i, s]
                # 손익 계산
                rtn = (sell_dict[s] / buy_dict[s]) -1
                book.loc[i, 'r '+s] = rtn
                print('개별 청산일 : ',i,' 종목코드 : ', s , 'long 진입가격 : ', buy_dict[s], ' |  long 청산가격 : ',\
                      sell_dict[s],' | return:', round(rtn * 100, 2),'%') # 수익률 계산.
            if book.loc[i, 'p '+ s] == '':     # zero position || long 청산.
                buy_dict[s] = 0.0
                sell_dict[s] = 0.0


    acc_rtn = 1.0        
    for i in book.index:
        rtn  = 0.0
        count = 0
        for s in s_codes:
            if book.loc[i, 'p '+ s] == '' and book.shift(1).loc[i,'p '+ s] == 'buy '+ s: 
                # 청산 수익률계산.
                count += 1
                rtn += book.loc[i, 'r '+s]
        if (rtn != 0.0) & (count != 0) :
            acc_rtn *= (rtn /count )  + 1
            print('누적 청산일 : ',i,'청산 종목수 : ',count, \
                  '청산 수익률 : ',round((rtn /count),4),'누적 수익률 : ' ,round(acc_rtn, 4)) # 수익률 계산.
        book.loc[i,'acc_rtn'] = acc_rtn
    print ('누적 수익률 :', round(acc_rtn, 4))

상대 모멘텀 전략의 수익률은 다수 종목을 가지고 계산하므로 단일 종목으로 계산한 절대 모멘텀 전략의 수익률 계산과는 다른 것을 알 수 있다. 하지만, 절대 모멘텀 전략이나 상대 모멘텀 전략 모두 전에 보유하고 있던 포지션 여부를 검사한 후 그에 따라 진입과 청산, 유지를 결정하는 것이라는 기본적인 개념은 동일하다. 

마지막 수익률 계산은 다음 코드로 실행된다.

multi_returns(book, stock_codes) # 수익률 계산

 

<참고 서적> 퀀트 전략을 위한 인공지능 트레이딩