动量策略

# -*- coding: utf-8 -*-
"""
Momentum strategies are almost the opposite of mean-reversion strategies. 
A typical momentum strategy will buy stocks that have been showing 
an upward trend in hopes that the trend will continue. 
The momentum strategy defined in Clenow’s books 
trades based upon the following rules:

Trade once a week. In his book, Clenow trades every Wednesday, 
but as he notes, which day is completely arbitrary.

Rank stocks in the S&P 500 based on momentum. 
Momentum is calculated by multiplying 
the annualized exponential regression slope 
of the past 90 days by the R^2R 
2 coefficient of the regression calculation.

Position size is calculated using the 20-day Average True Range of each stock, 
multiplied by 10 basis points of the portfolio value.

Only open new positions if the S&P 500 is above its 200-day moving average.

Every week, look to sell stocks that are not in the top 20% momentum ranking, 
or have fallen below their 100 day moving average. 
Buy stocks in the top 20% momentum rankings with remaining cash.

Every other week, rebalance existing positions 
with updated Average True Range values.


quant/survivorship-free at master · teddykoker/quant · GitHub  
https://github.com/teddykoker/quant/tree/master/survivorship-free

"""

from datetime import datetime 
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
#%matplotlib inline
plt.rcParams["figure.figsize"] = (10, 6) # (w, h)
plt.ioff()

import backtrader as bt

from scipy.stats import linregress


#%%

def momentum(closes):
    returns = np.log(closes)
    x = np.arange(len(returns))
    slope, _, rvalue, _, _ = linregress(x, returns)
    return ((1 + slope) ** 252) * (rvalue ** 2)  # annualize slope and multiply by R^2

#file = 'https://github.com/teddykoker/quant/tree/master/survivorship-free/data/tickers.csv'
#tickers = pd.read_csv('survivorship-free/tickers.csv', header=None)[1].tolist()
#tickers = pd.read_csv(file, header=None)[1].tolist()
#stocks = (
#    (pd.concat(
#        [pd.read_csv(f"survivorship-free/{ticker}.csv", index_col='date', parse_dates=True)[
#            'close'
#        ].rename(ticker)
#        for ticker in tickers],
#        axis=1,
#        sort=True)
#    )
#)
#stocks = stocks.loc[:,~stocks.columns.duplicated()]
#



class Momentum(bt.Indicator):
    lines = ('trend',)
    params = (('period', 90),)
    
    def __init__(self):
        self.addminperiod(self.params.period)
    
    def next(self):
        returns = np.log(self.data.get(size=self.p.period))
        x = np.arange(len(returns))
        slope, _, rvalue, _, _ = linregress(x, returns)
        annualized = (1 + slope) ** 252
        self.lines.trend[0] = annualized * (rvalue ** 2)
        
        
class Strategy(bt.Strategy):
    def __init__(self):
        self.i = 0
        self.inds = {}
        self.spy = self.datas[0]
        self.stocks = self.datas[1:]
        
        self.spy_sma200 = bt.indicators.SimpleMovingAverage(self.spy.close,
                                                            period=200)
        for d in self.stocks:
            self.inds[d] = {}
            self.inds[d]["momentum"] = Momentum(d.close, 
                                                period=90)
            self.inds[d]["sma100"] = bt.indicators.SimpleMovingAverage(d.close, 
                                                                       period=100)
            self.inds[d]["atr20"] = bt.indicators.ATR(d, 
                                                      period=20)

    def prenext(self):
        # call next() even when data is not available for all tickers
        self.next()
    
    def next(self):
        if self.i % 5 == 0:
            self.rebalance_portfolio()
        if self.i % 10 == 0:
            self.rebalance_positions()
        self.i += 1
    
    def rebalance_portfolio(self):
        # only look at data that we can have indicators for 
        self.rankings = list(filter(lambda d: len(d) > 100, self.stocks))
        self.rankings.sort(key=lambda d: self.inds[d]["momentum"][0])
        num_stocks = len(self.rankings)
        
        # sell stocks based on criteria
        for i, d in enumerate(self.rankings):
            if self.getposition(self.data).size:
                if i > num_stocks * 0.2 or d < self.inds[d]["sma100"]:
                    self.close(d)
        
        if self.spy < self.spy_sma200:
            return
        
        # buy stocks with remaining cash
        for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]):
            cash = self.broker.get_cash()
            value = self.broker.get_value()
            if cash <= 0:
                break
            if not self.getposition(self.data).size:
                size = value * 0.001 / self.inds[d]["atr20"]
                self.buy(d, size=size)
                
        
    def rebalance_positions(self):
        num_stocks = len(self.rankings)
        
        if self.spy < self.spy_sma200:
            return

        # rebalance all stocks
        for i, d in enumerate(self.rankings[:int(num_stocks * 0.2)]):
            cash = self.broker.get_cash()
            value = self.broker.get_value()
            if cash <= 0:
                break
            size = value * 0.001 / self.inds[d]["atr20"]
            self.order_target_size(d, size)

cerebro = bt.Cerebro(stdstats=False)
cerebro.broker.set_coc(True)

spy = bt.feeds.YahooFinanceData(dataname='SPY',
                                 fromdate=datetime(2012,2,28),
                                 todate=datetime(2018,2,28),
                                 plot=False)
cerebro.adddata(spy)  # add S&P 500 Index

for ticker in tickers:
    df = pd.read_csv(f"survivorship-free/{ticker}.csv",
                     parse_dates=True,
                     index_col=0)
    if len(df) > 100: # data must be long enough to compute 100 day SMA
        cerebro.adddata(bt.feeds.PandasData(dataname=df, plot=False))

cerebro.addobserver(bt.observers.Value)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, riskfreerate=0.0)
cerebro.addanalyzer(bt.analyzers.Returns)
cerebro.addanalyzer(bt.analyzers.DrawDown)
cerebro.addstrategy(Strategy)
results = cerebro.run()
cerebro.plot(iplot=False)[0][0]
print(f"Sharpe: {results[0].analyzers.sharperatio.get_analysis()['sharperatio']:.3f}")
print(f"Norm. Annual Return: {results[0].analyzers.returns.get_analysis()['rnorm100']:.2f}%")
print(f"Max Drawdown: {results[0].analyzers.drawdown.get_analysis()['max']['drawdown']:.2f}%")