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:
