In [ ]:
# Import libraries
import xarray as xr
import xarray.ufuncs as xruf
import numpy as np
import pandas as pd
import qnt.output as qnout
import qnt.ta as qnta
import qnt.data as qndata
import qnt.stats as qnstats
import qnt.graph as qngraph
import datetime as dt
import plotly.graph_objs as go
import xarray.ufuncs as xruf
import time
Load futures data¶
Quantnet provides data for 75 global derivatives. The underlying assets are currencies, cross-rates, indices, bonds, energy and metals from the world's futures exchanges.
Suppose we want to download the data for the last 20 years. One can use the following function:
In [ ]:
fut_data = qndata.futures.load_data(tail = 20*365)
# The complete list
fut_data.asset
In [ ]:
# we can see historical data on a chart
trend_fig = [
go.Scatter(
x = fut_data.sel(asset = 'F_DX').sel(field = 'close').to_pandas().index,
y = fut_data.sel(asset = 'F_DX').sel(field = 'close'),
line = dict(width=1,color='black'))]
# draw chart
fig = go.Figure(data = trend_fig)
fig.update_yaxes(fixedrange=False) # unlock vertical scrolling
fig.show()
Weights allocation
This function calculates positions using wma and roc as trend indicators.
In [ ]:
def calc_positions(futures, ma_periods, roc_periods, sideways_threshold):
""" Calculates positions for given data(futures) and parameters """
close = futures.sel(field='close')
# calculate MA
ma = qnta.lwma(close, ma_periods)
# calcuate ROC
roc = qnta.roc(ma, roc_periods)
# positive trend direction
positive_trend = roc > sideways_threshold
# negtive trend direction
negative_trend = roc < -sideways_threshold
# sideways
sideways_trend = abs(roc) <= sideways_threshold
# 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)
# 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)
return position
Select asset and adjust parameters:
In [ ]:
asset = 'F_DX' ###
sdat = fut_data.sel(asset=asset).dropna('time','any')
sout = calc_positions(sdat, 40, 6, 1)
sout = xr.concat([sout], pd.Index([asset], name='asset'))
ssta = qnstats.calc_stat(fut_data, sout)
display(ssta.to_pandas().tail())
performance = ssta.to_pandas()["equity"]
qngraph.make_plot_filled(performance.index, performance, name="PnL (Equity)", type="log")
This function calculate positions for multiple instruments with different parameters.
In [ ]:
def calc_output_all(data, params):
positions = data.sel(field='close').copy(True)
positions[:] = np.nan
for futures_name in params.keys():
p = params[futures_name]
futures_data = data.sel(asset=futures_name).dropna('time','any')
p = calc_positions(futures_data, p['ma_periods'], p['roc_periods'], p['sideways_threshold'])
positions.loc[{'asset':futures_name, 'time':p.time}] = p
return positions
In [ ]:
# say we select futures and their parameters for technical algorithm
params = {
'F_NY': {
'ma_periods': 200,
'roc_periods': 5,
'sideways_threshold': 2,
},
'F_GX': {
'ma_periods': 200,
'roc_periods': 20,
'sideways_threshold': 2
},
'F_DX': {
'ma_periods': 40,
'roc_periods': 6,
'sideways_threshold': 1
},
}
futures_list = list(params.keys())
# form the output
output = calc_output_all(fut_data.sel(asset = futures_list), params)
# check the output
qnout.check(output, fut_data)
# write the result
qnout.write(output)
# show statistics
stat = qnstats.calc_stat(fut_data, output.sel(time=slice('2006-01-01', None)))
display(stat.to_pandas().tail())
# show plot with profit and losses:
performance = stat.to_pandas()["equity"]
qngraph.make_plot_filled(performance.index, performance, name="PnL (Equity)", type="log")
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
import qnt.ta as qnta
def load_data(period):
return qndata.futures_load_data(tail=period)
def calc_positions(futures, ma_periods, roc_periods, sideways_threshold):
""" Calculates positions for given data(futures) and parameters """
close = futures.sel(field='close')
# calculate MA
ma = qnta.lwma(close, ma_periods)
# calcuate ROC
roc = qnta.roc(ma, roc_periods)
# positive trend direction
positive_trend = roc > sideways_threshold
# negtive trend direction
negative_trend = roc < -sideways_threshold
# sideways
sideways_trend = abs(roc) <= sideways_threshold
# 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)
# 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)
return position
def calc_output_all(data, params):
positions = data.sel(field='close').copy(True)
positions[:] = np.nan
for futures_name in params.keys():
p = params[futures_name]
futures_data = data.sel(asset=futures_name).dropna('time','any')
p = calc_positions(futures_data, p['ma_periods'], p['roc_periods'], p['sideways_threshold'])
positions.loc[{'asset':futures_name, 'time':p.time}] = p
return positions
# say we select futures and their parameters for technical algorithm
params = {
'F_NY': {
'ma_periods': 200,
'roc_periods': 5,
'sideways_threshold': 2,
},
'F_GX': {
'ma_periods': 200,
'roc_periods': 20,
'sideways_threshold': 2
},
'F_DX': {
'ma_periods': 40,
'roc_periods': 6,
'sideways_threshold': 1
},
}
futures_list = list(params.keys())
def strategy(data):
output = calc_output_all(data.sel(asset = futures_list), params)
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
)