In [1]:
# Import Zipline, the open source backester, and a few other libraries that we will use
import zipline
from zipline import TradingAlgorithm
from zipline.api import schedule_function, order_target_percent, order_percent, record, symbol, history, add_history,set_commission,set_slippage
from zipline.api import date_rules, time_rules, commission, slippage, order_target, get_datetime
import pytz
from datetime import datetime
import numpy as np
import pandas as pd
from matplotlib import pyplot
from scipy.optimize import minimize
import time
import math
In [2]:
stocks = symbols(['XLF', 'XLE', 'XLU', 'XLK', 'XLB', 'XLP', 'XLY', 'XLI', 'XLV'])
data = get_pricing(
stocks,
start_date='2003-01-01',
end_date = '2015-06-23',
frequency='daily'
)
data.price.plot()
Out[2]:
In [6]:
# Define the algorithm - this should look familiar from the Quantopian IDE
# For more information on writing algorithms for Quantopian
# and these functions, see https://www.quantopian.com/help
def initialize(context):
schedule_function(trade,
date_rule=date_rules.week_end(),
time_rule=time_rules.market_close(minutes=1),
half_days=False)
context.lookback = 252*2 # 252*n = n years
context.stocks = symbols(['XLF', 'XLE', 'XLU', 'XLK', 'XLB', 'XLP', 'XLY', 'XLI', 'XLV'])
set_commission(commission.PerShare(cost=0.00))
set_slippage(slippage.VolumeShareSlippage(volume_limit=1, price_impact=0))
context.i = 0
add_history(context.lookback, '1d', 'price')
def handle_data(context, data):
context.i += 1
if context.i < context.lookback:
return
def trade(context,data):
if context.i < context.lookback:
return
w = estimate(context,data)
# Close all current positions
for stock in context.portfolio.positions:
if stock not in w.index:
order_target(stock, 0)
#Order
for i,n in enumerate(w):
if w.index[i] in data:
if w.index[i] in context.portfolio.positions: #If position already exists
order_target_percent(w.index[i], n)
else:
order_percent(w.index[i], n)
def estimate(context,data):
# Structure a portfolio such that it maximizes dividend yield while lowering volatility
# Covar estimation
price_history = history(context.lookback, frequency="1d", field='price')
ret = (price_history/price_history.shift(5)-1).dropna(axis=0)[5::5]
ret.columns = context.stocks
n = len(ret.columns)
covar = ret.cov()
w = min_var_weights(covar, context.stocks)
# LOG EVERYTHING
#esigma = np.sqrt(np.dot(w.T,np.dot(covar,w)))*math.sqrt(52)*100
#log.info('Expected Volatility: '+str(esigma)+'%')
#w = pd.Series(np.round(1000*w)/1000)
#log.info('Number of Stocks: '+str(sum(abs(w)>0)))
#log.info('Current Leverage: '+str(context.account.leverage))
#w.index = ret.columns
#log.info(w)
#print(get_datetime())
#print(w)
return w #Cash Cushion of X%
def min_var_weights(cov, stocks):
'''
Returns a dictionary of sid:weight pairs.
'''
cov = pd.DataFrame(2*cov)
x = np.array([0.]*(len(cov)+1))
#x = np.ones(len(cov) + 1)
x[-1] = 1.0
p = lagrangize(cov)
weights = np.linalg.solve(p, x)[:-1]
weights = pd.Series(weights, index=stocks)
return weights
def lagrangize(df):
'''
Utility funcion to format a DataFrame
in order to solve a Lagrangian sysem.
'''
df = df
df['lambda'] = np.ones(len(df))
z = np.ones(len(df) + 1)
x = np.ones(len(df) + 1)
z[-1] = 0.0
x[-1] = 1.0
m = [i for i in df.as_matrix()]
m.append(z)
return pd.DataFrame(np.array(m))
In [7]:
# Analyze is a post-hoc analysis method available on Zipline.
# It accepts the context object and 'perf' which is the output
# of a Zipline backtest. This API is currently experimental,
# and will likely change before release.
def analyze(context, perf):
perf.portfolio_value.plot()
In [8]:
# NOTE: This cell will take a few minutes to run.
# Create algorithm object passing in initialize and
# handle_data functions
algo_obj = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data
)
# HACK: Analyze isn't supported by the parameter-based API, so
# tack it directly onto the object.
algo_obj._analyze = analyze
# Run algorithm
perf_manual = algo_obj.run(data.price)
In [ ]:
perf_manual = perf_manual[perf_manual.index >= datetime(2005,1,1)]
In [ ]:
perf_manual.returns.plot()
In [ ]:
def calculate_max_drawdown(rets):
compounded_returns = []
cur_return = 0.0
for r in rets:
try:
cur_return += math.log(1.0 + r)
# this is a guard for a single day returning -100%, if returns are
# greater than -1.0 it will throw an error because you cannot take
# the log of a negative number
except ValueError:
log.debug("{cur} return, zeroing the returns".format(
cur=cur_return))
cur_return = 0.0
compounded_returns.append(cur_return)
cur_max = None
max_drawdown = None
for cur in compounded_returns:
if cur_max is None or cur > cur_max:
cur_max = cur
drawdown = (cur - cur_max)
if max_drawdown is None or drawdown < max_drawdown:
max_drawdown = drawdown
if max_drawdown is None:
return 0.0
return 1.0 - math.exp(max_drawdown)
In [ ]:
calculate_max_drawdown(perf_manual.returns)
In [ ]:
In [ ]: