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 [1]:
# 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 [2]:
# 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 [3]:
help(xrtl.WMA)
In [4]:
help(xrtl.ROCP)
Let's implement strategy based on WMA using one asset:
In [5]:
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()