Pairs trading with states iterations

spancham

Hi @support
I am trying to merge my pairs strategy with the Quantiacs states iterations example.
It runs but I am still not certain if my code is correct however.
Can you have a look at my code below please and see if the implementation is correct?
What corrections would you make?
Don't be concerned that I am posting it here, these are not all the parameters or the futures contracts that I would use.

import xarray as xr
import numpy as np
import qnt.data as qndata
import qnt.backtester as qnbt

def load_data(period):
    data = qndata.futures_load_data(tail=period)
    return data

def strategy(data, state):
    
    zthreshold =  1.25
    lookback_trading_days = 60
    markets = ['F_GC', 'F_SI']
    
    close = data.sel(field='close', asset=markets).transpose('time', 'asset')
    last_close = close.ffill('time').isel(time=-1)
    
    # state may be null, so define a default value
    if state is None:
        state = {
            "zscore": last_close,
            "zscore_prev": last_close,
            "output": xr.zeros_like(last_close)
        }

    zscore_prev = state['zscore']
    zscore_prev_prev = state['zscore_prev']
    output_prev = state['output']
    
    # align the arrays to prevent problems in case the asset list changes
    zscore_prev, zscore_prev_prev, last_close = xr.align(zscore_prev, zscore_prev_prev, last_close, join='right') 
    
    s1 = close[-lookback_trading_days:, 0]
    s2 = close[-lookback_trading_days:, 1]
    
    # Compute mean of the spread up to now
    mvavg = np.mean(np.log(s1/s2))
    
    # Compute stdev of the spread up to now
    stdev = np.std(np.log(s1/s2))
    
    # Compute current spread
    current_spread = np.log(s1[-1] / s2[-1])
    
    # Compute z-score
    zscore = (current_spread - mvavg) / stdev if stdev > 0 else 0

    if zscore >= zthreshold:
        output = [-0.5, 0.5]
    elif zscore <= -zthreshold:
        output = [0.5, -0.5]
    else:
        output = output_prev
        
    next_state = {
        "zscore": zscore,
        "zscore_prev": zscore_prev,
        "output": output
    }
    
    return xr.DataArray(output, dims=['asset'], coords=dict(asset=markets)), next_state


weights, state = qnbt.backtest(
    competition_type="futures",  # Futures contest
    lookback_period=100,  # lookback in calendar days
    test_period=16*365,
    strategy=strategy,
    analyze=True,
    build_plots=True,
    collect_all_states=False # if it is False, then the function returns the last state, otherwise - all states
)

Thank you!

support

Hello.

This code looks ok.

But if you don't need zscore_prev for your computation, you don't need to pass it to the state, so this code may be simplified:


import xarray as xr
import numpy as np
import qnt.data as qndata
import qnt.backtester as qnbt

def load_data(period):
    data = qndata.futures_load_data(tail=period)
    return data

def strategy(data, state):
    
    zthreshold =  1.25
    lookback_trading_days = 60
    markets = ['F_GC', 'F_SI']
    
    close = data.sel(field='close', asset=markets).transpose('time', 'asset')
    last_close = close.ffill('time').isel(time=-1)
    
    # state may be null, so define a default value
    prev_output = state
    if prev_output is None:
        prev_output = xr.zeros_like(last_close)
   
    
    s1 = close[-lookback_trading_days:, 0]
    s2 = close[-lookback_trading_days:, 1]
    
    # Compute mean of the spread up to now
    mvavg = np.mean(np.log(s1/s2))
    
    # Compute stdev of the spread up to now
    stdev = np.std(np.log(s1/s2))
    
    # Compute current spread
    current_spread = np.log(s1[-1] / s2[-1])
    
    # Compute z-score
    zscore = (current_spread - mvavg) / stdev if stdev > 0 else 0

    if zscore >= zthreshold:
        output = [-0.5, 0.5]
    elif zscore <= -zthreshold:
        output = [0.5, -0.5]
    else:
        output = prev_output
    
    return xr.DataArray(output, dims=['asset'], coords=dict(asset=markets)), output


weights, state = qnbt.backtest(
    competition_type="futures",  # Futures contest
    lookback_period=100,  # lookback in calendar days
    test_period=16*365,
    strategy=strategy,
    analyze=True,
    build_plots=True,
    collect_all_states=False # if it is False, then the function returns the last state, otherwise - all states
)

Regards.

support

If you set collect_all_states=True, you can analyze all states after evaluation. It may be useful for tracing when print does not suit well.

Just a notice.

Sheikh

@support
Cool, thanks very much!