strategy

Trend-Following Futures System

A trend-following strategy for futures based on technical analysis.

You can clone and edit this example there (tab Examples).


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
2022-06-20T00
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"))
100% (33631792 of 33631792) |############| Elapsed Time: 0:00:00 Time:  0:00:00

We will use WMA and ROCP from qnt.xr_talib to measure trend.

In [3]:
help(xrtl.WMA)
Help on function WMA in module qnt.xr_talib:

WMA(data: xarray.core.dataarray.DataArray, timeperiod: int = 30) -> xarray.core.dataarray.DataArray
    Weighted Moving Average (Overlap Studies)
    Parameters:
        timeperiod: 30
    Input:
        data: time series
    Output:
        double series

In [4]:
help(xrtl.ROCP)
Help on function ROCP in module qnt.xr_talib:

ROCP(data: xarray.core.dataarray.DataArray, timeperiod: int = 14) -> xarray.core.dataarray.DataArray
     Rate of change Percentage: (real-prevPrice)/prevPrice (Momentum Indicators)
    Input:
        data: time series
    Parameters:
        timeperiod: 14
    Output:
        double series

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()