[파이썬 주식] 변동성돌파전략 - 11. pb추가 백테스트

가장 어려운 것은 앞으로 이런 전략이 통할 종목을 찾는 것 같다. 

 

## 1. 라이브러리 임포트

from Investar import Analyzer
mk = Analyzer.MarketDB()
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import time
import seaborn as sns
import warnings
warnings.filterwarnings('ignore')
# matplotlib 한글 폰트 출력코드

import matplotlib
from matplotlib import font_manager, rc
import platform

try : 
    if platform.system() == 'Windows':
    # 윈도우인 경우
        font_name = font_manager.FontProperties(fname="c:/Windows/Fonts/malgun.ttf").get_name()
        rc('font', family=font_name)
    else:    
    # Mac 인 경우
        rc('font', family='AppleGothic')
except : 
    pass
matplotlib.rcParams['axes.unicode_minus'] = False   

## 2. 종목 불러오기

## 종목 불러오기
import pandas as pd
"""KRX로부터 상장기업 목록 파일을 읽어와서 데이터프레임으로 반환"""
url = 'http://kind.krx.co.kr/corpgeneral/corpList.do?method='\
    'download&searchType=13'
krx = pd.read_html(url, header=0)[0]
krx = krx[['종목코드', '회사명']]
krx = krx.rename(columns={'종목코드': 'code', '회사명': 'company'})
krx.code = krx.code.map('{:06d}'.format)
krx = krx.set_index('code')
krx

codes = list(krx.index)
companys = [krx.loc[code,'company'] for code in codes]

 

## 3. 백테스트 종목 선정

백테스트 조건 

k=0.5

목표가 > 3MA

pb<0.6

def 돌려보기(codes, start_date, end_date):
    def load_data(code, k, start='2020-01-01'):
        
        df=mk.get_daily_price(code, start, end_date)
        df['변동폭'] = df['high'] - df['low']
        df['목표가'] = df['open'] + df['변동폭'].shift(1)*k
        df['어제종가'] = df['close'].shift(1)
        df['내일시가'] = df['open'].shift(-1)
        df['어제거래량'] = df['volume'].shift(1)
        df['그제거래량'] = df['volume'].shift(2)
        df['시가-어제종가'] = df['open']-df['어제종가']
        df['MA3_yes'] = df['close'].rolling(window=3).mean().shift(1)
        df['MA20'] = df['close'].rolling(window=20).mean() 
        df['stddev'] = df['close'].rolling(window=20).std() 
        df['upper'] = df['MA20'] + (df['stddev'] * 2)
        df['lower'] = df['MA20'] - (df['stddev'] * 2)
        df['PB'] = (df['close'] - df['lower']) / (df['upper'] - df['lower'])
        df['PB_yes'] = df['PB'].shift(1)
        df['bandwidth'] = (df['upper'] - df['lower']) / df['MA20'] * 100 # ①
        df['TP'] = (df['high'] + df['low'] + df['close']) / 3

        return df

    start = time.time()
    a=[]
    b=[]
    c=[]
    d=[]
    e=[]
    f=[]
    g=[]
    h=[]
    i=[]
    j=[]
    k=[]
    l=[]
    m=[]
    n=[]
    name=[]
    #name2=[]
    iteration=0
    for code in codes:
        iteration=iteration+1    
        if iteration%500==0: #알림용
            print(iteration)
        try:
    
            df = load_data(code, 0.5, start=start_date)
            기간 = df.shape[0] # 기간수
            cond = ( df['high'] > df['목표가'] ) & ( df['목표가'] > df['MA3_yes'] ) &(df['PB_yes']<0.6)# 구매조건
            df=df[cond]

            df['수익률'] = df['내일시가']/df['목표가']*0.9975 - 0.006 #0.9975 수수료, 0.002 슬리피지
            df['승패'] = np.where(df['수익률']>1, 1, 0)
            df=df.iloc[:-2]

            조건만족횟수 = df.shape[0] # 조건만족 수
            조건만족비율 = 조건만족횟수/기간
            조건승률 = df['승패'].value_counts()[1] / len(df['승패'])
            #최근승률 = df[-1:].승패.value_counts()[1]/ len(df[-10:].승패)
            보유수익률 = (df['close'][-1]/df['close'][0]*0.9975-0.006-1)
            돌파수익률 = (df.수익률.cumprod()[-1]-1)
            최대수익률 = (df.loc[df.수익률.idxmax()].수익률-1)
            평균수익률 = df.수익률.mean()-1
            중앙수익률 = df.수익률.median()-1
            수익률표준편차 = df.std()['수익률']
            최대손실률 = (df.loc[df.수익률.idxmin()].수익률-1)
            기간수익률 = df.수익률.cumprod().iloc[-1]
            돌파비율 = 돌파수익률-보유수익률
            N = (df.index[-1] - df.index[0]).days / 252
            M = df.dropna().shape[0]
            CAGR = (기간수익률 ** (1/N))-1
            기하평균수익률 = (기간수익률 **(1/M))-1
            name.append(code)
            #name2.append(krx.loc[code])
            a.append(조건만족횟수)
            b.append(조건만족비율)
            c.append(조건승률)
            #d.append(최근승률)
            e.append(보유수익률)
            f.append(돌파수익률)
            g.append(최대수익률)
            h.append(평균수익률)
            i.append(중앙수익률)
            j.append(최대손실률)
            k.append(CAGR)
            l.append(기하평균수익률)
            m.append(수익률표준편차)
            n.append(돌파비율)
        except :
            pass
        df=pd.DataFrame({"종목이름":name,"조건만족횟수":a,"조건만족비율":b,"조건승률":c,"보유수익률":e,"돌파수익률":f,"평균수익률":h,"수익률표준편차":m,"중앙수익률":i,"최대수익률":g,"최대손실률":j,"돌파비율":n,"CAGR":k,"기하수익률":l})
    print("완료 소요시간 :", time.time() - start)  # 현재시각 - 시작시간 = 실행 시간
    return df
    
codes = companys
start_date='2020-01-01'
end_date = '2021-02-11'
df = 돌려보기(codes, start_date, end_date)
df

조건

cond = (df['조건승률']>0.55 ) &(df['기하수익률']>0) & (df['최대손실률']>-0.04)
print(df[cond].shape[0])
print("평균승률 : {:.2f}".format(df[cond].조건승률.mean()))
print("평균수익률 : {:.2f}".format(df[cond].평균수익률.mean()))
print("중앙수익률 : {:.2f}".format(df[cond].중앙수익률.mean()))
print("돌파수익률 : {:.2f}".format(df[cond].돌파수익률.mean()))
df[cond].sort_values(by='기하수익률',ascending=False)

종목뽑기

codes = list(df.loc[cond,"종목이름"])

['LG전자', '유투바이오', '티와이홀딩스', '바이브컴퍼니', '아진산업', '이노와이어리스', '무진메디', '인카금융서비스', '서울반도체', '에이스테크', '메디쎄이', 'KEC', '이아이디', '소룩스', '솔브레인', '이오플로우', '클래시스', '현우산업', '루켄테크놀러지스', '한올바이오파마', '소마젠', '엔에프씨', '한화투자증권', '파트론', '씨엔티드림', '나무가', '원포유', '코셋']

 

## 4. 실적 좋았던 종목의 누적수익률 그려보기

def backtest(code, k, start):
    df=pd.DataFrame()
    df = df=mk.get_daily_price(code, start)
    df['변동폭'] = df['high']-df['low']
    df['목표가'] = df['open'] + df['변동폭'].shift(1)*k
    df['MA3_yes'] = df.close.rolling(window=3).mean().shift(1)
    df['내일시가'] = df.open.shift(-1)
    df['MA20'] = df['close'].rolling(window=20).mean() 
    df['stddev'] = df['close'].rolling(window=20).std() 
    df['upper'] = df['MA20'] + (df['stddev'] * 2)
    df['lower'] = df['MA20'] - (df['stddev'] * 2)
    df['PB'] = (df['close'] - df['lower']) / (df['upper'] - df['lower'])
    df['PB_yes'] = df['PB'].shift(1)
    cond = ( df['high'] > df['목표가'] ) & ( df['목표가'] > df['MA3_yes'] ) & (df['PB_yes']<0.6)
    df.loc[cond,'수익률'] = df.loc[cond,'내일시가']/df.loc[cond,'목표가']*0.9975 - 0.006 #0.9975 수수료, 0.002 슬리피지
    return df['수익률']
    
returns = pd.DataFrame()
for code in codes:
    df2 = backtest(code,k=0.5,start='2020-01-01')
    returns[code] = df2
    time.sleep(0.01)
returns.set_index(returns.reset_index()['date'].apply(lambda x:pd.to_datetime(x)),inplace=True)
returns.set_index(returns.index.strftime("%Y-%m-%d"),inplace=True)

plt.figure(figsize=(12,8))
returns.mean(axis=1).cumprod().plot()

▶ 조건을 만족하는 종목들 간 하루 평균을 수익률로 하여 누적수익률을 구하면, 20배까지 나온다..

중간에 보이듯이 점프뛰는 구간이 있는데 한종목이 20% 넘었을 경우이다. 

 

## 5. 일자별 매수 종목 수에 따른 수익률 비교

실제로는 자금의 20% 비율로 5종목을 대상으로 매매하고 있으므로,

100% 1종목 / 50%씩 2종목 / 20%씩 5종목 으로 구분하여 비교해보겠다.

import random
returns_=pd.DataFrame()
returns_['min']= returns.min(axis=1)
returns_['mean']= returns.mean(axis=1)
returns_['median']= returns.median(axis=1)
returns_['max']= returns.max(axis=1)
returns_['count'] = returns.count(axis=1)
returns_['승패'] = returns_['mean'].map(lambda x:1 if x>1 else 0 )
returns_['count_rev'] = returns_['count'].map(lambda x : x/5 if 0<x<5 else 1) #5종목 이내일 때는 1/5
returns_['mean_rev'] = returns_['mean']**(returns_['count_rev'])
for date in list(returns_.index):
    codes_date = returns.loc[date].dropna()
    if len(codes_date) >=5:
        codes_rand = random.sample(range(0,len(codes_date)),5)
        returns_.loc[date, 'mean_rand5'] = codes_date[codes_rand].mean()
    else:
        returns_.loc[date, 'mean_rand5'] = returns_.loc[date,'mean']**(returns_.loc[date,'count_rev'])

returns_['count_rev2'] = returns_['count'].map(lambda x : x/2 if 0<x<2 else 1) #5종목 이내일 때는 1/5
returns_['mean_rev2'] = returns_['mean']**(returns_['count_rev2'])
for date in list(returns_.index):
    codes_date = returns.loc[date].dropna()
    if len(codes_date) >=2:
        codes_rand = random.sample(range(0,len(codes_date)),2)
        returns_.loc[date, 'mean_rand2'] = codes_date[codes_rand].mean()
    else:
        returns_.loc[date, 'mean_rand2'] = returns_.loc[date,'mean']**(returns_.loc[date,'count_rev2'])
for date in list(returns_.index):
    codes_date = returns.loc[date].dropna()
    if len(codes_date) >=1:
        codes_rand = random.sample(range(0,len(codes_date)),1)
        #returns_.loc[date, 'mean_rand1'] = codes_date[codes_rand].mean()
        returns_.loc[date, 'mean_rand1'] = codes_date[random.sample(list(codes_date.index),1)[0]]
    else:
        returns_.loc[date, 'mean_rand1'] = returns_.loc[date,'mean']



#print("5종목 평균수익률 : {:.2f}".format(returns_['mean_rev'].dropna().cumprod().iloc[-1]))
#print("2종목 평균수익률 : {:.2f}".format(returns_['mean_rev2'].dropna().cumprod().iloc[-1]))
print("1종목 임의평균수익률 : {:.2f}".format(returns_['mean_rand1'].dropna().cumprod().iloc[-1]))
print("2종목 임의평균수익률 : {:.2f}".format(returns_['mean_rand2'].dropna().cumprod().iloc[-1]))
print("5종목 임의평균수익률 : {:.2f}".format(returns_['mean_rand5'].dropna().cumprod().iloc[-1]))

#print("1종목 중앙수익률 : {:.2f}".format(returns_['median'].dropna().cumprod().iloc[-1]))
returns_[['mean_rand1','mean_rand2','mean_rand5']].cumprod().loc[:'2020-10-31'].plot()

## 6. 종목 임의 선정(난수)으로 일자별 매수 종목 수에 따른 수익률 비교(100회)

def 세어보기 (i):
    np.random.seed(i)
    returns_=pd.DataFrame()
    returns_['min']= returns.min(axis=1)
    returns_['mean']= returns.mean(axis=1)
    returns_['median']= returns.median(axis=1)
    returns_['max']= returns.max(axis=1)
    returns_['count'] = returns.count(axis=1)
    returns_['승패'] = returns_['mean'].map(lambda x:1 if x>1 else 0 )
    returns_['count_rev'] = returns_['count'].map(lambda x : x/5 if 0<x<5 else 1) #5종목 이내일 때는 1/5
    returns_['mean_rev'] = returns_['mean']**(returns_['count_rev'])
    for date in list(returns_.index):
        codes_date = returns.loc[date].dropna()
        if len(codes_date) >=5:
            codes_rand = random.sample(range(0,len(codes_date)),5)
            returns_.loc[date, 'mean_rand5'] = codes_date[codes_rand].mean()
        else:
            returns_.loc[date, 'mean_rand5'] = returns_.loc[date,'mean']**(returns_.loc[date,'count_rev'])

    returns_['count_rev2'] = returns_['count'].map(lambda x : x/2 if 0<x<2 else 1) #5종목 이내일 때는 1/5
    returns_['mean_rev2'] = returns_['mean']**(returns_['count_rev2'])
    for date in list(returns_.index):
        codes_date = returns.loc[date].dropna()
        if len(codes_date) >=2:
            codes_rand = random.sample(range(0,len(codes_date)),2)
            returns_.loc[date, 'mean_rand2'] = codes_date[codes_rand].mean()
        else:
            returns_.loc[date, 'mean_rand2'] = returns_.loc[date,'mean']**(returns_.loc[date,'count_rev2'])
    for date in list(returns_.index):
        codes_date = returns.loc[date].dropna()
        if len(codes_date) >=1:
            codes_rand = random.sample(range(0,len(codes_date)),1)
            returns_.loc[date, 'mean_rand1'] = codes_date[random.sample(list(codes_date.index),1)[0]]
        else:
            returns_.loc[date, 'mean_rand1'] = returns_.loc[date,'mean']

    return returns_['mean_rand1'].dropna().cumprod().iloc[-1], returns_['mean_rand2'].dropna().cumprod().iloc[-1], returns_['mean_rand5'].dropna().cumprod().iloc[-1]

mean_rand1 = []
mean_rand2 = []
mean_rand5 = []

start = time.time()
for i in range(100):
    a,b,c = 세어보기(i)
    mean_rand1.append(a)
    mean_rand2.append(b)
    mean_rand5.append(c)
    if (i+1)%20==0:
        print(i+1)
print("완료 소요시간 :", time.time() - start)  # 현재시각 - 시작시간 = 실행 시간
s = pd.DataFrame({"mean_rand1":mean_rand1, "mean_rand2":mean_rand2, "mean_rand5":mean_rand5})
s   

 

난수로 100회 시뮬레이션 해본 결과, 

sns.distplot(s['mean_rand1'],bins=20)

▶ 1종목 100%인 경우, 누적수익률 평균 약 19배

 

sns.distplot(s['mean_rand2'],bins=20)

▶ 2종목 50%인 경우, 누적수익률 평균 약 12배

sns.distplot(s['mean_rand5'],bins=20)

▶ 5종목 20%인 경우, 누적수익률 최종 약 5배