This template a simple implementation of a trend-following strategy.
We will start to prototype with a single-pass implementation and finish with a multi-pass implementation.
In [ ]:
# Import basic libraries for manipulating data.
# Please refer to xarray.pydata.org for xarray documentation.
# xarray works optimally with N-dimensional datasets in Python
# and is well suited for financial datasets with labels "time",
# "field" and "asset". xarray data structures can also be easily
# converted to pandas dataframes.
import xarray as xr
import xarray.ufuncs as xruf
import numpy as np
import pandas as pd
# Import quantnet libraries.
import qnt.data as qndata # data loading and manipulation
import qnt.stats as qnstats # key statistics
import qnt.graph as qngraph # graphical tools
import qnt.xr_talib as xrtl # technical analysis indicators (talib)
import qnt.output as qnout
# display function for fancy displaying:
from IPython.display import display
# lib for charts
import plotly.graph_objs as go
In [ ]:
# Load all available data since given date.
# It is possible to set a max_date in the call in order to
# develop the system on a limited in-sample period and later
# test the system on unseen data after max_date.
# A submission will be accepted only if no max_date is set,
# as submissions will be evaluated on live data on a daily basis.
data = qndata.futures.load_data(min_date='2000-01-01', dims=("time", "field", "asset"))
We will use WMA and ROCP from qnt.xr_talib to measure trend.
In [ ]:
help(xrtl.WMA)
In [ ]:
help(xrtl.ROCP)
Let's implement strategy based on WMA using one asset:
In [ ]:
stock_name = 'F_GC'
# select only 1 stock
stock = data.sel(asset=stock_name).dropna('time', 'all')
pd_time = stock.time.to_pandas()
close = stock.sel(field='close')
# chart with prices
price_fig = [
go.Candlestick(
x=stock.time.to_pandas(),
open=stock.sel(field='open').values,
high=stock.sel(field='high').values,
low=stock.sel(field='low').values,
close=stock.sel(field='close').values,
name=stock_name
)
]
# calculate MA
ma = xrtl.WMA(close, timeperiod=20) # you can use also SMA, EMA, etc.
# calcuate ROC
roc = xrtl.ROCP(ma, timeperiod=1)
# We suppose, when abs(roc) < sideways_threshold, the trend is sideways.
sideways_threshold = 0.01
# positive trend direction
positive_trend = roc > sideways_threshold
# negtive trend direction
negative_trend = roc < -sideways_threshold
# sideways
sideways_trend = abs(roc) <= sideways_threshold
# This is a street magic. We will elliminate sideway
# We suppose that a sideways trend after a positive trend is also positive
side_positive_trend = positive_trend.where(sideways_trend == False).ffill('time').fillna(False)
# and a sideways trend after a negative trend is also negative
side_negative_trend = negative_trend.where(sideways_trend == False).ffill('time').fillna(False)
# charts with trend indicator
trend_fig = [
go.Scatter(
x = pd_time,
y = ma,
name='ma',
line = dict(width=1,color='orange')
),
go.Scatter(
x = pd_time,
y = ma.where(side_positive_trend),
name='side-positive-trend',
line = dict(width=1,color='green')
),
go.Scatter(
x = pd_time,
y = ma.where(side_negative_trend),
name='side-negative-trend',
line = dict(width=1,color='red')
),
go.Scatter(
x = pd_time,
y = ma.where(positive_trend),
name='positive-trend',
line = dict(width=3,color='green')
),
go.Scatter(
x = pd_time,
y = ma.where(negative_trend),
name='negative-trend',
line = dict(width=3,color='red')
)
]
# define signals
buy_signal = positive_trend
buy_stop_signal = side_negative_trend
sell_signal = negative_trend
sell_stop_signal = side_positive_trend
# calc positions
position = close.copy(True)
position[:] = np.nan
position = xr.where(buy_signal, 1, position)
position = xr.where(sell_signal, -1, position)
position = xr.where(xruf.logical_and(buy_stop_signal, position.ffill('time') > 0), 0, position)
position = xr.where(xruf.logical_and(sell_stop_signal, position.ffill('time') < 0), 0, position)
position = position.ffill('time').fillna(0)
# calc real orders
real_buy = xruf.logical_and(position > 0, position.shift(time=1) <= 0)
real_sell = xruf.logical_and(position < 0, position.shift(time=1) >= 0)
real_stop = xruf.logical_and(position == 0, position.shift(time=1) != 0)
# plot orders chart
signals_fig=[
go.Scatter(
x=close.loc[real_buy].time.to_pandas(),
y=close.loc[real_buy],
mode="markers",
hovertext='buy',
name="buy",
marker_size=9,
marker_color='green'
),
go.Scatter(
x=close.loc[real_sell].time.to_pandas(),
y=close.loc[real_sell],
mode="markers",
hovertext='sell',
name="sell",
marker_size=9,
marker_color='red'
),
go.Scatter(
x=close.loc[real_stop].time.to_pandas(),
y=close.loc[real_stop],
mode="markers",
hovertext='stop',
name="stop",
marker_size=9,
marker_color='gray'
),
]
# draw chart
fig = go.Figure(data = price_fig + trend_fig + signals_fig)
fig.update_yaxes(fixedrange=False) # unlock vertical scrolling
fig.show()
# calc stats
position_with_asset = xr.concat([position], pd.Index([stock_name], name='asset'))
stats = qnstats.calc_stat(data, position_with_asset)
display(stats.to_pandas().tail())
performance = stats.loc[:,"equity"]
# draw performance chart
fig = go.Figure(data = [
go.Scatter(
x=performance.time.to_pandas(),
y=performance,
hovertext='performance',
)
])
fig.update_yaxes(fixedrange=False) # unlock vertical scrolling
fig.show()
Now, implement the strategy on multiple assets. At first, we use single-pass approach, because calculation is much faster.
In [ ]:
close = data.sel(field='close')
# trend
ma = xrtl.WMA(close, timeperiod=200)
roc = xrtl.ROCP(ma, timeperiod=20)
sideways_threshold = 0.02
positive_trend = roc > sideways_threshold
negative_trend = roc < -sideways_threshold
sideways_trend = abs(roc) <= sideways_threshold
side_positive_trend = positive_trend.where(sideways_trend == False).ffill('time').fillna(False)
side_negative_trend = negative_trend.where(sideways_trend == False).ffill('time').fillna(False)
# signals
buy_signal = positive_trend
buy_stop_signal = side_negative_trend
sell_signal = negative_trend
sell_stop_signal = side_positive_trend
# calc positions
position = close.copy(True)
position[:] = np.nan
# align signals
buy_signal = xr.align(buy_signal, position, join='right')[0]
buy_stop_signal = xr.align(buy_stop_signal, position, join='right')[0]
sell_signal = xr.align(sell_signal, position, join='right')[0]
sell_stop_signal = xr.align(sell_stop_signal, position, join='right')[0]
# apply signals to position
position = xr.where(buy_signal, 1, position)
position = xr.where(sell_signal, -1, position)
fp = position.ffill('time')
position = xr.where(xruf.logical_and(buy_stop_signal, fp > 0), 0, position)
position = xr.where(xruf.logical_and(sell_stop_signal, fp < 0), 0, position)
position = position.ffill('time').fillna(0)
# position normalization
output = position/abs(position).sum('asset')
# check your output
qnout.check(output, data)
# write writing output to file:
qnout.write(output)
#calc and print stats
stats = qnstats.calc_stat(data, output.sel(time=slice('2006-01-01')))
# display(output.to_pandas().tail())
display(stats.to_pandas().tail())
Notice: despite this result, it may be a good strategy. But it does not work well for all futures. It is a good idea to select suitable futures for this strategy. Another point is that the effective parameters may differ for every asset. So, it makes sense to select the right parameters for specific futures. See the template "Trend-Following with Custom Arguments" to get more information about this idea.
Multi-pass implementation
Now, let's use multi-pass approach to verify the strategy. It is much slower but it is the best way to properly test it and to avoid looking-forward.
In [ ]:
%%javascript
IPython.OutputArea.prototype._should_scroll = function(lines) { return false; }
// disable widget scrolling
In [ ]:
# In your final submission you can remove/deactivate all the other cells to reduce the checking time.
# The checking system will run this book multiple times for every trading day within the in-sample period.
# Every pass the available data will be isolated till the current day.
# qnt.backtester is optimized to work with the checking system.
# The checking system will override test_period=1 to make your strategy to produce weights for 1 day per pass.
import xarray as xr
import numpy as np
import qnt.ta as qnta
import qnt.backtester as qnbt
import qnt.data as qndata
import qnt.xr_talib as xrtl
import xarray.ufuncs as xruf
def load_data(period):
return qndata.futures_load_data(tail=period)
def strategy(data):
close = data.sel(field='close')
# trend
ma = xrtl.WMA(close, timeperiod=200)
roc = xrtl.ROCP(ma, timeperiod=20)
sideways_threshold = 0.02
positive_trend = roc > sideways_threshold
negative_trend = roc < -sideways_threshold
sideways_trend = abs(roc) <= sideways_threshold
side_positive_trend = positive_trend.where(sideways_trend == False).ffill('time').fillna(False)
side_negative_trend = negative_trend.where(sideways_trend == False).ffill('time').fillna(False)
# signals
buy_signal = positive_trend
buy_stop_signal = side_negative_trend
sell_signal = negative_trend
sell_stop_signal = side_positive_trend
# calc positions
position = close.copy(True)
position[:] = np.nan
# align signals
buy_signal = xr.align(buy_signal, position, join='right')[0]
buy_stop_signal = xr.align(buy_stop_signal, position, join='right')[0]
sell_signal = xr.align(sell_signal, position, join='right')[0]
sell_stop_signal = xr.align(sell_stop_signal, position, join='right')[0]
# apply signals to position
position = xr.where(buy_signal, 1, position)
position = xr.where(sell_signal, -1, position)
fp = position.ffill('time')
position = xr.where(xruf.logical_and(buy_stop_signal, fp > 0), 0, position)
position = xr.where(xruf.logical_and(sell_stop_signal, fp < 0), 0, position)
position = position.ffill('time').fillna(0)
# position normalization
output = position/abs(position).sum('asset')
return output.isel(time=-1)
weights = qnbt.backtest(
competition_type="futures",
load_data=load_data,
lookback_period=5 * 365,
start_date='2006-01-01',
strategy=strategy
)