[Invest Class] 4. Absolute Momentum 전략 함수 생성

1. Absolute Momentum 전략

자산의 절대적인 수익률을 기준으로 투자 여부를 결정하는 방식으로 특정 자산의 과거 수익률이 양이면 매수하고, 음이면 매수하지 않는다.

여기에서는 '전월의 수정종가 / 전년도의 수정종가) - 1'를 인덱스로 설정하고 0보다 클때 매수하는 전략

 

*try-except 구조

test_df = pd.DataFrame()
for year in range(1980, 2026): # 년도를 반복
    for month in range(1, 13): # 월을 반복
        try:
            idx = f"{year}-{month}"
            df2 = df.loc[idx].tail(1)
            test_df = pd.concat( [test_df, df2], axis=0 )
        except Exception as e:
            print(e)
            continue

해당되는 연 혹은 월 데이터가 없을때 반복문이 돌아가지 않는 상황을 방지하기 위해 try-except구문을 사용해준다. Exception은 모든 종류의 에러를 의미하고 as e는 에러 내용을 변수 e에 저장하는 것이다. 에러가 발생할 경우 해당 연/월 데이터를 건너뀌도 다음으로 넘어가라는 명령을 continue로 해준다.

 

* 월말 데이터 뽑는 방법 

1) 위의 코드 참고2) shift 함수 이용

test2_df = df.loc[df['STD-YM'] != df['STD-YM'].shift(-1)]

-> shift 함수 사용:shift(1)은 아래로 한칸 데이터를 밀어버리는 것이고(-> 현재 행에 과거 값이 들어옴 ,shift(-1)위로 한칸 올려보내는 것(-> 현재 행에 미래 값이 들어옴)이다. 

3) list이용

ym_list =df['STD-YM'].unique()
test3_df = pd.DataFrame()
for ym in ym_list:
    data = df.loc[ym].tail(1)
    test3_df = pd.concat([test3_df, data], axis=0)
    
test3_df

 

 

2. Absolute Momentum 전략 함수 생성

 

def create_ym(df, col = 'Adj Close'):
    df2= df.copy()
    if 'Date' in df2.columns:
        df2.set_index('Date')
        
    df2.index = pd.to_datetime(df2.index, utc=True)
    
    flag = df2.isin([np.nan, np.inf, -np.inf]).any(axis=1)
    df2 = df2.loc[~flag, [col]]
    
    df2['STD-YM'] = df2.index.strftime('%Y-%m')
    return df2

 

 

def create_month(_df, _start = "2010-01-01", _end = datetime.now(), _momentum =12, _select =1):
    if _select == 1:
        flag = _df['STD-YM'] !=_df.shift(-1)['STD-YM']
        
    elif _select == 0:
        flag = _df['STD-YM'] != _df.shift(1)['STD-YM']
    else:
        return "_select 의 값은 0 아니면 1입니다"
    result = _df.loc[flag]
    
    col = result.columns[0]
    result['BF1'] = result.shift(1)[col]
    result['BF2'] = result.shift(_momentum)[col]
    try:
        result.index = result.index.tz_localize(None)
    except Exception as e:
        pritn(e)
    result = result.loc[_start:_end]
    return result

 

def momentumfunc(df1, df2, _score = 1):
    result = df1.copy()
    result['trade'] = ''
    
    try:
        result.index = result.index.tz_localize(None)
    except Exception as e:
        print(e)
    
    for idx in df2.index:
        momentum_index = df2.loc[idx,'BF1']/df2.loc[idx,'BF2'] - _score
        signal = ''
        if (momentum_index > 0) & (momentum_index != np.inf):
            signal = 'buy'
        result.loc[idx:, 'trade'] = signal
    return result

 

def create_rtn(df):
        result = df.copy()
        col = result.columns[0]
        result['rtn'] = 1
        for idx in result.index: 
            if (result.shift().loc[idx, 'trade'] == '') & (result.loc[idx, 'trade'] == 'buy'):
                buy = result.loc[idx, col]
            elif (result.shift().loc[idx, 'trade'] == 'buy') & (result.loc[idx, 'trade'] == ''):
                sell = result.loc[idx, col]
        
                rtn = sell / buy
                result.loc[idx, 'rtn'] = rtn
        result['acc_rtn'] = result['rtn'].cumprod()
        acc_rtn = result.iloc[-1, -1]
        return result, acc_rtn