단순이동평균(SMA) 및 백 테스팅 소개

파이썬으로 나만의 투자 전략 세우기 - 3주차

1. 데이터 이해하기
• 실습으로 대체

2. 간단한 매수 후 보유 전략
1. 수익률 산출
1. 로그 수익률 사용
2. 처음 대비 현재 가치율(이후 creturns)
1. 로그 수익률을 sum()한 다음 exp() 적용
2. Pn/P0 과 비슷함
3. 처음 대비 누적 수익 가치율
1. cumsum() 후 exp() 적용
2. price 차트와 동일하게 형성됨

3. 성능 지표
1. 특정 기간동안 risk
1. 로그수익률의 표준 편차 * root특정기간
2. 누적 최대치 계산하기(이후 cummax)
1. creturns에 cummax() 사용
3. drawdown 값 계산
1. cummax - creturns 로 계산

4. SMA 교차전략 - 개요
1. if SMA_S>SMA_L? Long Position: Short Position

5. SMA 교차전략 정의하기
1. SMA_S, SMA_L 정의
1. rolling().mean() 사용
2. POSITION 계산
1. if SMA_S>SMA_L? 1: -1

6. 백터화된 백 테스팅 전략
1. 전략 수익률 계산(이하 strategy)
1. 전일 포지션 * 오늘 수익률
2. 전략 누적 수익률 계산(cstrategy)
1. strategy를 cumsum() 하고 exp()로 계산

7. 최적의 SMA 전략 찾기
1. run_strategy() 정의

def run_strategy(SMA):
  data = df.copy()
  data["returns"] = np.log(data.price.div(data.price.shift(1)))
  data["SMA_S"] = data.price.rolling(int(SMA[0])).mean()
  data["SMA_L"] = data.price.rolling(int(SMA[1])).mean()
  data.dropna(inplace = True)
  
  data["position"] = np.where(data["SMA_S"] > data["SMA_L"], 1, -1)
  data["strategy"] = data.position.shift(1) * data["returns"]
  data.dropna(inplace = True)

  return data[["returns", "strategy"]].sum().apply(np.exp)

2. 최소값을 찾도록 리턴값에 음수 처리

def run_strategy(SMA):
  data = df.copy()
  data["returns"] = np.log(data.price.div(data.price.shift(1)))
  data["SMA_S"] = data.price.rolling(int(SMA[0])).mean()
  data["SMA_L"] = data.price.rolling(int(SMA[1])).mean()
  data.dropna(inplace = True)
  
  data["position"] = np.where(data["SMA_S"] > data["SMA_L"], 1, -1)
  data["strategy"] = data.position.shift(1) * data["returns"]
  data.dropna(inplace = True)

  return -data[["returns", "strategy"]].sum().apply(np.exp)[-1]

3. scipy의 brute()를 이용한 최적값 찾기

import scipy.optimize as optimize
from scipy.optimize import brute

brute(run_strategy, ((20, 40, 1), (150, 200, 1)), finish=None)

 

단순이동평균(SMA) 및 백 테스팅 소개 2/2 (황영진)

파이썬으로 나만의 투자 전략 세우기 - 3주차

1. SMA와 백테스팅

4. SMA란?

특정한 기간 동안의 주식 종가를 단순 평균으로 계산하는 방법

5. 백테스팅이란?

트레이딩 전략에 과거의 데이터를 적용하여 수익성을 평가하는 것

2. OOP를 사용한 SMA 백 테스트 클래스 작성하기

첫 번째, 클래스 필드

6. symbol은 어떤 column의 Price를 선택할 것인지 지정
테스트 csv



7. SMA_S: 단기 SMA window size
8. SMA_L: 장기 SMA window size
9. start: 시작 Date
10. end: 종료 Date
11. tc: 거래비용 값
12. data: get_data() 함수를 통해 저장되는 데이터 프레임 값
13. results: 추가적인 속성을 위해 작성 -> 나중에 strategy에 대한 데이터를 담는다.

def __init__(self, symbol, SMA_S, SMA_L, start, end, tc):
    self.symbol = symbol
    self.SMA_S = SMA_S
    self.SMA_L = SMA_L
    self.start = start
    self.end = end
    self.tc = tc
    self.results = None 
    self.get_data() # 객체 초기화하는 과정에서 data 값 넣어주기

두 번째, 클래스의 함수 6개

14. get_data(self)

csv 파일을 읽어 데이터프레임으로 이동평균선 값 (Short, Long)을 보여주는 메소드로 객체를 초기화 할때도 사용된다.

코드

def get_data(self):
        raw = pd.read_csv("forex_pairs.csv", parse_dates = ["Date"], index_col = "Date")
        raw = raw[self.symbol].to_frame().dropna()
        raw = raw.loc[self.start:self.end].copy()
				# 심볼의 이름을 price로 바꾸고 inplace=True를 통해 원본값을 수정
        raw.rename(columns={self.symbol: "price"}, inplace=True)
        raw["returns"] = np.log(raw / raw.shift(1)) # 일일 수익률
        raw["SMA_S"] = raw["price"].rolling(self.SMA_S).mean() # add short sma
        raw["SMA_L"] = raw["price"].rolling(self.SMA_L).mean()
        self.data = raw# 새로운 속성 data 추가 후 raw 값으로 할당
        return raw # 해당함수를 독립적으로 사용할 수 있으므로 값 return

15. set_parameters(self, SMA_S = None, SMA_L = None)

데이터 프레임의 설정된 SMA_S (= Short), SMA_L (= Long) 의 값을 변경하여 새롭게 적용된 Window size를 데이터 프레임에 반영하는 메소드

코드

def set_parameters(self, SMA_S = None, SMA_L = None):
    if SMA_S is not None:
        self.SMA_S = SMA_S
        self.data["SMA_S"] = self.data["price"].rolling(self.SMA_S).mean()
    if SMA_L is not None:
        self.SMA_L = SMA_L
        self.data["SMA_L"] = self.data["price"].rolling(self.SMA_L).mean()

16. test_strategy(self) ⭐

SMA_S 와 SMA_L 이 교차 할때 Long or Short 포지션이 나오게 되고, 다음 4가지의 정보를 데이터프레임에 추가하는 메소드

코드

def test_strategy(self):
    data = self.data.copy().dropna()
    # SMA_S 가 SMA_L 보다 크면 다음날 Long Position (= 1), 
    # 작으면 다음날 Short Position(= -1)
    data["position"] = np.where(data["SMA_S"] > data["SMA_L"], 1, -1)
    # position은 다음날의 Position 이므로 shift(1) 해서 일일 수익률 returns에 곱한다.
    data["strategy"] = data["position"].shift(1) * data["returns"]
    data.dropna(inplace=True)
    # 명확한 수익 %를 구하기 위해서 e(자연로그)에 해당 값을 거듭제곱 해준다 -> log 제거
    data["creturns"] = data["returns"].cumsum().apply(np.exp)
    data["cstrategy"] = data["strategy"].cumsum().apply(np.exp)
    self.results = data
    
    # 마지막 cstrategy 값은 전략적으로 배팅했을때 최종값이며 절대적인 성과를 의미한다.
    perf = data["cstrategy"].iloc[-1] # absolute performance
    # outperf는 그냥 보유하고 있을때 수익 대비해서 전략적으로 배팅했을때의 수익을 비교하고
    # 그 값의 차이를 저장
    outperf = perf - data["creturns"].iloc[-1] # outperformance 
    return round(perf, 6), round(outperf, 6)

17. plot_results(self)

creturns 값과 cstrategy 값을 그래프로 그려주는 메소드

코드

def plot_results(self):
  if self.results is None:
      print("No results to plot yet. Run a strategy.")
  else:
      title = "{} | SMA_S = {} | SMA_L = {}".format(self.symbol, self.SMA_S, self.SMA_L)
      self.results[["creturns", "cstrategy"]].plot(title=title, figsize=(12, 8))
            

18. update_and_run(self, SMA)

short window size와 long window size를 받아서 "절대적인 성과”를 반환하는 메소드

• 절대적인 성과란?
• 그냥 보유하고 있을때 수익(creturns) 대비해서 전략적으로 배팅했을때의 수익(cstrategy)을 비교하여 그 차이값을 구한 값을 의미
• 절대적인 성과값은 추후 brute force 알고리즘에 사용되는 값이다.
따라서 brute force 알고리즘에 의해 최소화 되도록, 음의 절대 성과를 리턴해야하므로 (-)를 붙여 리턴한다.
코드

def update_and_run(self, SMA):
	  self.set_parameters(int(SMA[0]), int(SMA[1]))
	  return -self.test_strategy()[0]

19. optimize_parameters(self, SMA_S_range, SMA_L_range)

short window의 Range와 long window의 Range를 받아서 최적의 short window size와 long window size를 구하는 메소드

코드

 def optimize_parameters(self, SMA_S_range, SMA_L_range):
	  opt = brute(self.update_and_run, (SMA_S_range, SMA_L_range), finish=None)
	  # opt는 최적의 매개변수(Short size, Long size)를 의미
	  # self.update_and_run(opt)는 음의 절대적인 성과를 의미
	  return opt, -self.update_and_run(opt)

Using
tester.optimize_parameters((10, 50, 1), (100, 252, 1))
# output
(array([ 46., 137.]), 2.526694)
1. 최적의 short size는 46
2. 최적의 long size는 137
3. 절대적인 성과의 값은 2.526694으로 1달러 투자시 약 2.52 달러가 될 수 있다.

---

전체 코드 ⭐⭐

class SMABacktester():
    def __init__(self, symbol, SMA_S, SMA_L, start, end):
        self.symbol = symbol
        self.SMA_S = SMA_S
        self.SMA_L = SMA_L
        self.start = start
        self.end = end
        self.results = None
        self.get_data()
        
    def get_data(self):
        raw = pd.read_csv("forex_pairs.csv", parse_dates = ["Date"], index_col = "Date")
        raw = raw[self.symbol].to_frame().dropna()
        raw = raw.loc[self.start:self.end].copy()
        raw.rename(columns={self.symbol: "price"}, inplace=True)
        raw["returns"] = np.log(raw / raw.shift(1))
        raw["SMA_S"] = raw["price"].rolling(self.SMA_S).mean() # add short sma
        raw["SMA_L"] = raw["price"].rolling(self.SMA_L).mean()
        self.data = raw
        return raw
    
    def set_parameters(self, SMA_S = None, SMA_L = None):
        if SMA_S is not None:
            self.SMA_S = SMA_S
            self.data["SMA_S"] = self.data["price"].rolling(self.SMA_S).mean()
        if SMA_L is not None:
            self.SMA_L = SMA_L
            self.data["SMA_L"] = self.data["price"].rolling(self.SMA_L).mean()
            
    def test_strategy(self):
        data = self.data.copy().dropna()
        data["position"] = np.where(data["SMA_S"] > data["SMA_L"], 1, -1)
        data["strategy"] = data["position"].shift(1) * data["returns"]
        data.dropna(inplace=True)
        data["creturns"] = data["returns"].cumsum().apply(np.exp)
        data["cstrategy"] = data["strategy"].cumsum().apply(np.exp)
        self.results = data
        
        perf = data["cstrategy"].iloc[-1] # absolute performance
        outperf = perf - data["creturns"].iloc[-1] # outperformance 
        return round(perf, 6), round(outperf, 6)
    
    def plot_results(self):
        if self.results is None:
            print("No results to plot yet. Run a strategy.")
        else:
            title = "{} | SMA_S = {} | SMA_L = {}".format(self.symbol, self.SMA_S, self.SMA_L)
            self.results[["creturns", "cstrategy"]].plot(title=title, figsize=(12, 8))
            
    def update_and_run(self, SMA):
        self.set_parameters(int(SMA[0]), int(SMA[1]))
        return -self.test_strategy()[0]
    
    def optimize_parameters(self, SMA_S_range, SMA_L_range):
        opt = brute(self.update_and_run, (SMA_S_range, SMA_L_range), finish=None)
        # opt는 최적의 매개변수(Short size, Long size)를 의미
        # self.update_and_run(opt)는 음의 절대적인 성과를 의미
        return opt, -self.update_and_run(opt)

세 번째, Class Description

20. class에 대한 설명
21. shift+tab (on Mac) 을 통한 함수 description

---

Code - 코드참조

class SMABacktester():
    ''' Class for the vectorized backtesting of SMA-based trading strategies.

    Attributes
    ==========
    symbol: str
        ticker symbol with which to work with
    SMA_S: int
        time window in days for shorter SMA
    SMA_L: int
        time window in days for longer SMA
    start: str
        start date for data retrieval
    end: str
        end date for data retrieval
        
        
    Methods
    =======
    get_data:
        retrieves and prepares the data
        
    set_parameters:
        sets one or two new SMA parameters
        
    test_strategy:
        runs the backtest for the SMA-based strategy
        
    plot_results:
        plots the performance of the strategy compared to buy and hold
        
    update_and_run:
        updates SMA parameters and returns the negative absolute performance (for minimization algorithm)
        
    optimize_parameters:
        implements a brute force optimization for the two SMA parameters
    '''
    
    def __init__(self, symbol, SMA_S, SMA_L, start, end):
        self.symbol = symbol
        self.SMA_S = SMA_S
        self.SMA_L = SMA_L
        self.start = start
        self.end = end
        self.results = None
        self.get_data()
    
    def __repr__(self):
        rep = "SMABacktester(symbol = {}, SMA_S = {}, SMA_L = {}, start = {}, end = {})"
        return rep.format(self.symbol, self.SMA_S, self.SMA_L, self.start, self.end)  
    
    def get_data(self):
        ''' Retrieves and prepares the data.
        '''
        raw = pd.read_csv("forex_pairs.csv", parse_dates = ["Date"], index_col = "Date")
        raw = raw[self.symbol].to_frame().dropna()
        raw = raw.loc[self.start:self.end].copy()
        raw.rename(columns={self.symbol: "price"}, inplace=True)
        raw["returns"] = np.log(raw / raw.shift(1))
        raw["SMA_S"] = raw["price"].rolling(self.SMA_S).mean() # add short sma
        raw["SMA_L"] = raw["price"].rolling(self.SMA_L).mean()
        self.data = raw
        return raw
    
    def set_parameters(self, SMA_S = None, SMA_L = None):
        ''' Updates SMA parameters and resp. time series.
        '''
        if SMA_S is not None:
            self.SMA_S = SMA_S
            self.data["SMA_S"] = self.data["price"].rolling(self.SMA_S).mean()
        if SMA_L is not None:
            self.SMA_L = SMA_L
            self.data["SMA_L"] = self.data["price"].rolling(self.SMA_L).mean()
            
    def test_strategy(self):
        ''' Backtests the trading strategy.
        '''
        data = self.data.copy().dropna()
        data["position"] = np.where(data["SMA_S"] > data["SMA_L"], 1, -1)
        data["strategy"] = data["position"].shift(1) * data["returns"]
        data.dropna(inplace=True)
        data["creturns"] = data["returns"].cumsum().apply(np.exp)
        data["cstrategy"] = data["strategy"].cumsum().apply(np.exp)
        self.results = data
        
        perf = data["cstrategy"].iloc[-1] # absolute performance
        outperf = perf - data["creturns"].iloc[-1] # outperformance 
        return round(perf, 6), round(outperf, 6)
    
    def plot_results(self):
        ''' Plots the cumulative performance of the trading strategy compared to buy and hold.
        '''
        if self.results is None:
            print("No results to plot yet. Run a strategy.")
        else:
            title = "{} | SMA_S = {} | SMA_L = {}".format(self.symbol, self.SMA_S, self.SMA_L)
            self.results[["creturns", "cstrategy"]].plot(title=title, figsize=(12, 8))
            
    def update_and_run(self, SMA):
        ''' Updates SMA parameters and returns the negative absolute performance (for minimization algorithm).

        Parameters
        ==========
        SMA: tuple
            SMA parameter tuple
        '''
        self.set_parameters(int(SMA[0]), int(SMA[1]))
        return -self.test_strategy()[0]
    
    def optimize_parameters(self, SMA_S_range, SMA_L_range):
        ''' Finds global maximum given the SMA parameter ranges.

        Parameters
        ==========
        SMA_S_range, SMA_L_range: tuple
            tuples of the form (start, end, step size)
        '''
        opt = brute(self.update_and_run, (SMA_S_range, SMA_L_range), finish=None)
        return opt, -self.update_and_run(opt)

추가된 __repr__(self) 알아보기



네 번째, 거래비용 값 적용하기

⚠️ 거래비용 값 적용하기 전에 거래비용에 대해 알아보자!!!
⭐거래비용 Trades and Trading Costs (1)

---

• SMABacktester 클래스에 Trading cost 값을 의미하는 tc 필드 추가

전략값을 뽑아내는 test_strategy() 함수에서 전략대로 포지션 취했을때 수익을 나타내는 strategy 값의 거래비용인 tc 값을 빼고 그 값을 기준으로 cstrategy 계산

# determine when a trade takes place
data["trades"] = data.position.diff().fillna(0).abs() # 포지션변경 여부를 파악하기위해 사용
data.strategy = data.strategy - data.trades * self.tc
data["cstrategy"] = data["strategy"].cumsum().apply(np.exp)



💡
‼️ SMA 크로스오버 전략은 거래비용의 Cost가 필요한 전략이 아니다. → 거래가 많이 일어나지 않기 때문 → 많은 거래가 일어나는 전략을 사용하게 되는 경우 거래비용 적용시 Optimization의 결과가 달라질 수 있다.

3. 최종 SMABacktester 클래스 코드

class SMABacktester(): # with ptc 
    ''' Class for the vectorized backtesting of SMA-based trading strategies.

    Attributes
    ==========
    symbol: str
        ticker symbol with which to work with
    SMA_S: int
        time window in days for shorter SMA
    SMA_L: int
        time window in days for longer SMA
    start: str
        start date for data retrieval
    end: str
        end date for data retrieval
    tc: float
        proportional transaction costs per trade
        
        
    Methods
    =======
    get_data:
        retrieves and prepares the data
        
    set_parameters:
        sets one or two new SMA parameters
        
    test_strategy:
        runs the backtest for the SMA-based strategy
        
    plot_results:
        plots the performance of the strategy compared to buy and hold
        
    update_and_run:
        updates SMA parameters and returns the negative absolute performance (for minimization algorithm)
        
    optimize_parameters:
        implements a brute force optimization for the two SMA parameters
    '''
    
    def __init__(self, symbol, SMA_S, SMA_L, start, end, tc):
        self.symbol = symbol
        self.SMA_S = SMA_S
        self.SMA_L = SMA_L
        self.start = start
        self.end = end
        self.tc = tc
        self.results = None 
        self.get_data()
        
    def __repr__(self):
        return "SMABacktester(symbol = {}, SMA_S = {}, SMA_L = {}, start = {}, end = {})".format(self.symbol, self.SMA_S, self.SMA_L, self.start, self.end)
        
    def get_data(self):
        ''' Retrieves and prepares the data.
        '''
        raw = pd.read_csv("forex_pairs.csv", parse_dates = ["Date"], index_col = "Date")
        raw = raw[self.symbol].to_frame().dropna()
        raw = raw.loc[self.start:self.end]
        raw.rename(columns={self.symbol: "price"}, inplace=True)
        raw["returns"] = np.log(raw / raw.shift(1))
        raw["SMA_S"] = raw["price"].rolling(self.SMA_S).mean()
        raw["SMA_L"] = raw["price"].rolling(self.SMA_L).mean()
        self.data = raw
        
    def set_parameters(self, SMA_S = None, SMA_L = None):
        ''' Updates SMA parameters and resp. time series.
        '''
        if SMA_S is not None:
            self.SMA_S = SMA_S
            self.data["SMA_S"] = self.data["price"].rolling(self.SMA_S).mean()
        if SMA_L is not None:
            self.SMA_L = SMA_L
            self.data["SMA_L"] = self.data["price"].rolling(self.SMA_L).mean()
            
    def test_strategy(self):
        ''' Backtests the trading strategy.
        '''
        data = self.data.copy().dropna()
        data["position"] = np.where(data["SMA_S"] > data["SMA_L"], 1, -1)
        data["strategy"] = data["position"].shift(1) * data["returns"]
        data.dropna(inplace=True)
        
        # determine when a trade takes place
        data["trades"] = data.position.diff().fillna(0).abs()
        
        # subtract transaction costs from return when trade takes place
        data.strategy = data.strategy - data.trades * self.tc
        
        data["creturns"] = data["returns"].cumsum().apply(np.exp)
        data["cstrategy"] = data["strategy"].cumsum().apply(np.exp)
        self.results = data
        
        perf = data["cstrategy"].iloc[-1] # absolute performance of the strategy
        outperf = perf - data["creturns"].iloc[-1] # out-/underperformance of strategy
        return round(perf, 6), round(outperf, 6)
    
    def plot_results(self):
        ''' Plots the cumulative performance of the trading strategy
        compared to buy and hold.
        '''
        if self.results is None:
            print("No results to plot yet. Run a strategy.")
        else:
            title = "{} | SMA_S = {} | SMA_L = {} | TC = {}".format(self.symbol, self.SMA_S, self.SMA_L, self.tc)
            self.results[["creturns", "cstrategy"]].plot(title=title, figsize=(12, 8))
        
    def update_and_run(self, SMA):
        ''' Updates SMA parameters and returns the negative absolute performance (for minimization algorithm).

        Parameters
        ==========
        SMA: tuple
            SMA parameter tuple
        '''
        self.set_parameters(int(SMA[0]), int(SMA[1]))
        return -self.test_strategy()[0]
    
    def optimize_parameters(self, SMA1_range, SMA2_range):
        ''' Finds global maximum given the SMA parameter ranges.

        Parameters
        ==========
        SMA1_range, SMA2_range: tuple
            tuples of the form (start, end, step size)
        '''
        opt = brute(self.update_and_run, (SMA1_range, SMA2_range), finish=None)
        return opt, -self.update_and_run(opt)

사용 하기 코드

ptc = 0.00007
tester = SMABacktester("EURUSD=X", 50, 200, "2004-01-01", "2020-06-30", ptc)
tester.test_strategy()
tester.optimize_parameters((25, 50, 1), (100, 200, 1))

 

변동성 돌파 전략 (박철종)

터틀 트레이딩 4개명

22. 승산이 있는 트레이딩에 임하라
23. 리스크를 관리하라
24. 일관성 있는 태도를 유지하라
25. 단순성 초점을 맞춰라
 

변동성 돌파 전략



• 추세를 이용한 투자 방식에 속한다.
• 추세란 한번 방향이 정해져서 움직이면 쉽게 그 방향을 바꾸지 않고 지속되는 현상이다.
 



 



• 전날 Range = 전날 가격의 모든 움직임을 나타내는 값
• 전날 Range의 일정 수치를 벗어난다 = 의미 있는 모멘텀이 생겼다.
• 따라서 그 시점에 사도 모멘텀이 유지가 되어 우위가 있다.
 



• 변동성 돌파 전략 매수 시점 계산법
 



• 현금 비중을 조절하여 MDD를 조절하여 리스크를 줄인다.
 
트레이딩 3주차 발표.pptx
1093.8KB
 
 
 

---

 
 
본 스터디는 Udemy의 <【한글자막】 알고리즘 거래와 투자의 기술적 분석 with Python> 강의를 활용해 진행됐습니다. 강의에 대한 자세한 정보는 아래에서 확인하실 수 있습니다.
 
www.udemy.com
https://www.udemy.com/course/best-python/?utm_medium=udemyads&utm_source=wj-etc&utm_campaign=study_pming&utm_content=best-python&utm_term=220512
 
프밍 스터디는 Udemy Korea와 함께 합니다.
 
 

---