Data Strategies

Primer on CLS data strategies.

Introduction

CLS operates the largest multi-currency cash settlement system. CLS Market Data is a comprehensive suite of FX alternative data products designed to provide quality insight and analytics both on a timely and historical for financial efficiency, visibility and control.

These data sets can be accessed and analysed in the SigTech platform.

This data can be used for creating new strategies.

In this notebook we give examples of defining strategies that trade FX spot based on signals created using the CLS data.

A notebook containing all the code used in this page can be accessed in the research environment: Example notebooks.

Environment

This section will import relevant internal and external libraries, as well as setting up the platform environment. For further information in regards to setting up the environment, see Environment setup.

from sigtech.framework.analytics.cls_data import ClsData, CLS_DATASETS, CLS_FREQ, CLS_INSTRUMENTS, CLS_PUBLICATION_FREQ, \
    resample_reduced, remove_weekends, CLS_CCY_CROSSES
from sigtech.framework.infra.objects.dtypes import AnyType
from sigtech.timeline.utils.date_utils import dt_to_utc
import sigtech.framework as sig

from collections import defaultdict
import pytz
import datetime as dtm
import numpy as np
import pandas as pd
import seaborn as sns

sns.set(rc={'figure.figsize': (18, 6)})

if not sig.config.is_initialised():
    env = sig.init(env_date=dtm.date(2019, 1, 4), data_date=dtm.date(2021, 6, 21))
    env[sig.config.CUSTOM_DATA_SOURCE_CONFIG] = [
        ('[A-Z]{6} CURNCY', 'CLS'),
    ]

A CLS data class can then be created. This instance can then provide access to the various data sets and parameters.

We initialise the class with a list of default currencies to retrieve data for.

This set of currency crosses will be used when running the strategies.

EXCLUDED_CCYS = {'USDILS', 'EURHUF', 'EURDKK', 'USDHUF', 'USDDKK', 'USDKRW', 'USDSGD', 'USDMXN'}
ccys_included = list(set(CLS_CCY_CROSSES) - EXCLUDED_CCYS)
cls_data = ClsData(ccys_included[:3])

CLS Strategy Examples

Intraday mean-reversion strategies

In the strategy constructed below the CLS data is used to calculate a 24 hour VWAP, using different measures of volume. The strategy then trades on the assumption that the price will revert to that level.

The signal below is constructed by:

• Calculating a 24 Hour VWAP based on the volume data. This can also be calculated based on the counter-party specific activity.

• Looking at the deviation from the last price, normalising and then using this as a weight.

• The strategy trades spot with a 1 hour delay to the signal calculation.

The excess returns are plotted below, together with some performance metrics. A strong performance can be observed, however these strategies suffer with a large turnover. Here we exclude transaction costs to give a cleaner measure of the signals predictive power. A benefit can be observed for incorporating the counter-party information, with the use of Fund trading activity giving the greatest improvement.

def vol_weight(data, weights, window):
    sum_prod = weights.multiply(data, axis=0).dropna()
    sum_vol = weights.reindex(sum_prod.index).rolling(window=window).sum()
    sum_prod = sum_prod.rolling(window=window).sum()
    return sum_prod.divide(sum_vol, axis=0).dropna()

class VWAPStrategy(sig.Strategy):
    # Specify Strategy parameters
    ccy_crosses = AnyType(required=True)
    cpty_group = AnyType(default=None)
    
    def __init__(self):
        super().__init__()
        self.signals = defaultdict(float)
        for cross in self.ccy_crosses:
            self.signals[cross] = self.calculate_signal(cross)
        df = pd.concat(self.signals, axis=1).ffill().reindex(self.ccy_crosses, axis=1).fillna(0)
        self.signals = remove_weekends(df)
    
    def calculate_signal(self, cross):
        base, quote = cross[:3], cross[3:]
        prices = cls_data.data(cls_data.SPOT_PRICE_5MIN_DAILY, cross_list=[cross])[cross]['twap'].ffill().dropna()
        _ , hourly_sizes = cls_data.spot_hourly_flow_df([cross])
        
        # An offset of 1 is applied for windows with no flow
        hourly_sizes = hourly_sizes[(base, quote, self.cpty_group)] + 1
        hourly_prices = resample_reduced(prices, rule='1H', method='last')
        
        rolling_hourly_vwap = vol_weight(hourly_prices, hourly_sizes, window=24)
        signal = rolling_hourly_vwap - hourly_prices
        signal = signal.divide(signal.rolling(window=24*10).std()).dropna()
        signal.index = [pytz.utc.localize(dt) for dt in signal.index]
        return signal
                             
    def strategy_initialization(self, dt):
        # setup strategy hourly rebalancing.
        for rdt in self.signals.loc[self.calculation_start_dt():self.calculation_end_dt()].index:
            if rdt > dt:
                self.add_method(rdt, self.rebalance, use_trading_manager_calendar=False)
                
    def rebalance(self, dt):
        value = self.positions.valuation(dt, self.currency) / len(self.ccy_crosses)
        trades = defaultdict(float)
        for cross in self.ccy_crosses:
            base, target = cross[:3], cross[3:]
            signal = self.signals[cross].asof(dt - dtm.timedelta(hours=1))
            direction = 0.5 * min(max(signal, -2), 2)
            trades[base] += direction * value
            trades[target] -= direction * value
        
        for ccy in trades:
            if ccy == self.currency:
                continue
            base_fx = self.strategy_timeline.instrument_store.fx_data.fx(self.currency, ccy, dt_to_utc(dt), 0)
            current_base = self.positions.get_cash_value(dt, '{} CASH'.format(ccy))
            to_trade = (trades[ccy] / base_fx) - current_base
            
            self.add_fx_spot_trade(dt, self.currency, ccy, to_trade, execution_dt=dt, use_trading_manager_calendar=False)
        
    def size_dt_from_decision_dt(self, decision_dt):
        return decision_dt - dtm.timedelta(hours=1)
        
    def execution_dt_from_datetime(self, instrument, _dt):
        return dt

vwap_strategy_nb = VWAPStrategy(currency='USD', start_date=dtm.date(2018, 1, 4), end_date=dtm.date(2019, 1, 4),
                                include_trading_costs=False, ccy_crosses=cls_data.ccy_crosses,
                                cpty_group='Non-Bank Financial-Bank')
vwap_nb_ts = vwap_strategy_nb.history()

vwap_strategy_fb = VWAPStrategy(currency='USD', start_date=dtm.date(2018, 1, 4), end_date=dtm.date(2019, 1, 4),
                                include_trading_costs=False, ccy_crosses=cls_data.ccy_crosses,
                                cpty_group='Fund-Bank')
vwap_fb_ts = vwap_strategy_fb.history()

vwap_strategy_cb = VWAPStrategy(currency='USD', start_date=dtm.date(2018, 1, 4), end_date=dtm.date(2019, 1, 4),
                                include_trading_costs=False, ccy_crosses=cls_data.ccy_crosses,
                                cpty_group='Corporate-Bank')
vwap_cb_ts = vwap_strategy_cb.history()

vwap_strategy_bs = VWAPStrategy(currency='USD', start_date=dtm.date(2018, 1, 4), end_date=dtm.date(2019, 1, 4),
                                include_trading_costs=False, ccy_crosses=cls_data.ccy_crosses,
                                cpty_group='BuySide-SellSide')
vwap_bs_ts = vwap_strategy_bs.history()

The total return histories are plotted below.

Python

df = pd.concat({'VWAP Non-Bank': vwap_nb_ts, 'VWAP Fund': vwap_fb_ts,
                'VWAP Corporate': vwap_cb_ts, 'VWAP Buyside/Sellside': vwap_bs_ts}, axis=1)

df.plot(title='VWAP Reversion Signal Strategies');

Output

Python

sig.PerformanceReport(df, cash=sig.CashIndex.from_currency('USD')).report()

Output

Delayed flow direction strategy

Here we define a strategy which trades in the same direction as the flow was 24 hours prior. This makes use of the autocorrelation observed in the flows.

This strategy:

• Combines the the cross flows to create currency flows.

• Combines the direction of the flows for the counter-party groupings, with a negative multiplier for the corporate flow.

• Looks at this combined direction 1 day prior to the next hour and trades in the same direction.

class DelayedFlowStrategy(sig.DailyStrategy):
    ccy_crosses = AnyType(required=True)
    
    def __init__(self):
        super().__init__()
        vol_flow, flow_size_base = cls_data.spot_hourly_flow_df(cross_list=self.ccy_crosses)
        self.flow = vol_flow.sum(axis=1, level=(0,2))
        
        self.signals = defaultdict(float)
        for cross in self.ccy_crosses:
            self.signals[cross] = self.calculate_signal(cross)
        df = pd.concat(self.signals, axis=1).ffill().reindex(self.ccy_crosses, axis=1).fillna(0) # .rolling(window=2).mean()
        self.signals = remove_weekends(df)
        
    def calculate_signal(self, cross):
        base, quote = cross[:3], cross[3:]
        flow_direction = np.sign(self.flow[base] - self.flow[quote])
        signal = flow_direction[['BuySide-SellSide', 'Fund-Bank', 'Non-Bank Financial-Bank']].sum(axis=1)
        signal -= flow_direction['Corporate-Bank']
        signal /= 4
        signal.index = [dt.replace(tzinfo=pytz.utc) for dt in signal.index]
        signal = signal.rolling(window=2).mean()
        
        return remove_weekends(signal)

    def strategy_initialization(self, dt):
        
        # setup strategy hourly rebalancing.
        for rdt in self.signals.loc[self.calculation_start_dt():self.calculation_end_dt()].index:
            if rdt > dt:
                self.add_method(rdt, self.rebalance, use_trading_manager_calendar=False)
                
    def rebalance(self, dt):
        value = self.positions.valuation(dt, self.currency) / len(self.ccy_crosses)
        trades = defaultdict(float)
        for cross in self.ccy_crosses:
            base, target = cross[:3], cross[3:]
            signal = self.signals[cross].shift(22).asof(dt) 
            trades[base] += signal * value
            trades[target] -= signal * value
        
        for ccy in trades:
            if ccy == self.currency:
                continue
            base_fx = self.strategy_timeline.instrument_store.fx_data.fx(self.currency, ccy, dt_to_utc(dt), 0)
            current_base = self.positions.get_cash_value(dt, '{} CASH'.format(ccy))
            to_trade = (trades[ccy] / base_fx) - current_base
            
            self.add_fx_spot_trade(dt, self.currency, ccy, to_trade, execution_dt=dt, use_trading_manager_calendar=False)

An instance of this strategy is created and the total return is plotted below. The transaction costs are excluded to give a representation of the signal performance without transaction cost assumptions.

Python

delayed_flow_strategy = DelayedFlowStrategy(currency='USD', start_date=dtm.date(2018, 1, 4),
                                            end_date=dtm.date(2019, 1, 4),
                                            include_trading_costs=False, ccy_crosses=cls_data.ccy_crosses)
delayed_flow_ts = delayed_flow_strategy.history()
delayed_flow_ts.plot(figsize=(16, 6), title='One day delayed flow strategy');

Output

Python

sig.PerformanceReport(delayed_flow_ts, cash=sig.CashIndex.from_currency('USD')).report()

Output

Forward data based strategies

The following strategy, uses a observed tendency for spot to move in the opposite direction to the price move:

• It evaluates the spot return multiplied by volume during the life of all previous forwards.

• Groups these returns by delivery date, evaluates their negative direction and then takes a 3 month average to construct the signal.

class ForwardDeliveryBiasStrategy(sig.DailyStrategy):
    ccy_crosses = AnyType(required=True)
    
    def __init__(self):
        super().__init__()
        self.signals = defaultdict(float)
        for cross in self.ccy_crosses:
            self.signals[cross] += self.calculate_signal(cross)
        self.signals = pd.concat(self.signals, axis=1).ffill()
    
    def calculate_signal(self, cross):
        base, quote = cross[:3], cross[3:]
        fwds = cls_data.data(cls_data.FWD_VOLUME_HOURLY_DAILY)
        prices = cls_data.spot_prices_df()[(base, quote)]
        
        # Filter forward data
        fwd = fwds[cross].dropna()
        fwd = fwd[fwd['volume'] > 0]
        
        # Attach utc delivery datetime
        fwd['delivery_dt'] = [pytz.timezone('Europe/London').localize(dtm.datetime.combine(d, dtm.time(22,0))).astimezone(pytz.utc).replace(tzinfo=None) 
                         for d in fwd.index.get_level_values(1)]
        
        # Calculate spot returns over each forward
        fwd['trade_prices'] = prices.reindex(fwd.index.get_level_values(0), method='ffill').values
        fwd['delivery_prices'] = prices.reindex(fwd['delivery_dt'], method='ffill').values
        fwd['fwd_return'] = (fwd['delivery_prices'] / fwd['trade_prices']) - 1
        fwd['scaled_fwd_return'] = (fwd['fwd_return'] * fwd['volume'])
        fwd = fwd.dropna()

        signal = -np.sign(fwd[['scaled_fwd_return', 'delivery_dt']].groupby('delivery_dt').mean().iloc[:,0])
        
        # Resample to 11pm each day - introduces one hour lag for processing and trading
        daily_prices = resample_reduced(prices, rule='24H', method='last', base=23)
        signal = signal.reindex(daily_prices.index, method='ffill').ffill().fillna(0)
        
        # Take 3 month rolling average
        signal = signal.rolling(window=63).mean().dropna()
        
        # Add timezone and remove weekends.
        signal.index = [pytz.utc.localize(dt) for dt in signal.index]
        signal = remove_weekends(signal)
        
        return signal

    def strategy_initialization(self, dt):
        
        for rdt in self.signals.loc[self.calculation_start_dt():self.calculation_end_dt()].index:
            if rdt > dt:
                self.add_method(rdt, self.rebalance, use_trading_manager_calendar=False)
            
    def rebalance(self, dt):
        value = self.positions.valuation(dt, self.currency) / len(self.ccy_crosses)
        trades = defaultdict(float)
        for cross in self.ccy_crosses:
            base, target = cross[:3], cross[3:]
            
            # lag introduced by signal reindexing to 23:00
            signal = self.signals[cross].asof(dt)
                
            trades[base] += signal * value
            trades[target] -= signal * value
        
        for ccy in trades:
            if ccy == self.currency:
                continue
            base_fx = self.strategy_timeline.instrument_store.fx_data.fx(self.currency, ccy, dt_to_utc(dt), 0)
            current_base = self.positions.get_cash_value(dt, '{} CASH'.format(ccy))
            to_trade = (trades[ccy] / base_fx) - current_base
            
            self.add_fx_spot_trade(dt, self.currency, ccy, to_trade, execution_dt=dt, use_trading_manager_calendar=False)

Now the strategy has been defined we can initiate instances that use different counterparty groupings and run the strategies. We exclude transaction costs here to show the signal performance excluding cost assumptions.

The total return history is plotted below.

Python

fwd_del_bias_strategy = ForwardDeliveryBiasStrategy(currency='USD', start_date=dtm.date(2018, 1, 4),
                                                    end_date=dtm.date(2019, 1, 4),
                                                    include_trading_costs=False, ccy_crosses=cls_data.ccy_crosses)
fwd_del_bias_ts = fwd_del_bias_strategy.history()
fwd_del_bias_ts.plot(figsize=(16, 6), title='Forward return reversion strategy');

Output

Python

sig.PerformanceReport(fwd_del_bias_ts, cash=sig.CashIndex.from_currency('USD')).report()

Output