Custom FX Forward Carry Strategy

New user? The best way to learn SigTech is to follow the steps in our user guide.

Restricted data: You will only have access to the data used in this notebook if your organisation has specifically purchased it. To check your current data access, view Data. If you would like to access more data, please contact sales@sigtech.com.

Changes will not be saved: Edits made to SigTech’s example notebooks like this will only persist for the duration of your session. When you restart the research environment, this notebook will have been restored to its original state, and any changes you have made will have been lost. To make permanent changes, copy and paste the content to a new notebook in one of your workspaces.

Introduction

The purpose of this notebook is to show how to create a custom FX carry startegy expressed in the forward market.

Environment

This section will import relevant internal and external libraries, as well as setting up the platform environment.

import sigtech.framework as sig

import datetime as dtm
import pandas as pd

env = sig.init(env_date=dtm.date(2024, 2, 22))
env[sig.config.FORCE_OFFSETTING_OTC_TRADES] = True

Universe

CCYS = [
    'EURUSD',
    'GBPUSD',
    'USDCAD',
    'NZDUSD',
    'AUDUSD',
]

Strategy

class CustomFXStrategy(sig.RollingFXForwardStrategy):
    def __post_init__(self):
        super().__post_init__()
        self.fx_market = sig.FXMarket.instance()

    def _generate_fwd_rates(self, quote_dates, forward_tenor: str='3M'):
        maturities = [self.fx_market.fx_forward_date(
            self.forward_over,
            self.forward_under,
            date,
            forward_tenor
        ) for date in quote_dates]
        spot_dates = quote_dates + dtm.timedelta(days=2)
        return self.fx_market.calculate_forward_rate_batch(
            self.forward_over,
            self.forward_under,
            quote_dates,
            maturities,
            spot_dates=spot_dates
        )

    def _get_fx_prices(self):
        self.spot_rates = sig.obj.get(f'{self.long_currency}{self.currency} CURNCY').history()
        self.fwd_rates = pd.Series(self._generate_fwd_rates(
            self.spot_rates.index,
            self.forward_tenor
        ), index=self.spot_rates.index)

    def _calendar(self):
        return self.fx_market.schedule_stub(self.forward_over, self.forward_under).holidays

    def strategy_initialization(self, dt):
        self._get_fx_prices()
        for d in sig.SchedulePeriodic(
            self.first_entry_date,
            self.end_date,
            self._calendar(),
            frequency='2M',
        ).all_data_dates():
            self.add_method(d, self.close_fxfwd)
            self.add_method(d, self.trade_fxfwd)

    def close_fxfwd(self, dt):
        # Reset position, if any
        for instrument, quantity in self.positions.iterate_instruments(dt):
            if isinstance(instrument, sig.FXForward):
                execution_dt = self.execution_dt_from_datetime(instrument, dt)
                self.add_position_target(
                    dt,
                    instrument_name=instrument.name,
                    units=0.0,
                    transaction_type=self.forward_transaction_type,
                    payment_currency=self.currency,
                    execution_dt=execution_dt,
                    unit_type='WEIGHT'
                )

    def trade_fxfwd(self, dt):
        size_date = self.size_date_from_decision_dt(dt)
        notional = self.valuation_price(size_date) * self.notional_multiplier

        # Take position according to diff in spot vs fwd rate
        dt_str = dt.date().strftime("%Y-%m-%d %H:%M:%S")
        sign = 1 if self.spot_rates[dt_str] > self.fwd_rates[dt_str] else -1
        forward_maturity = self.fx_market.instance().hedge_schedule(
            dt.date(),
            self.calculation_end_date(),
            self.forward_over,
            self.forward_under,
            '3M'
        ).next_data_date(dt.date())

        self.add_trade(
            dt,
            sig.FXForward,
            self.notional_multiplier * notional * sign,
            over=self.forward_over,
            under=self.forward_under,
            maturity=forward_maturity,
        )

def create_fx_fwd_carry_strategies(base_ccy: str, quote_ccy: str):
    strat = CustomFXStrategy(
        long_currency=base_ccy,
        currency=quote_ccy,
        forward_tenor='3M',
        start_date=dtm.date(2018,1,4),
        end_date=dtm.date(2021,1,4),
        custom_roll_offset=2,
        include_trading_costs=False,
        total_return=False,
        ticker=f'{base_ccy}{quote_ccy} FX CARRY'
    )
    strat.build()
    return strat.name

fx_carry_strategies = {
    ccy: create_fx_fwd_carry_strategies(ccy[:3], ccy[3:])
    for ccy in CCYS
}

Portfolio construction

basket = sig.BasketStrategy(
    start_date=dtm.date(2018, 1, 4),
    end_date=dtm.date(2021, 1, 4),
    currency='USD',
    constituent_names=fx_carry_strategies.values(),
    rebalance_frequency='EOM',
    weights=[1/len(CCYS)] * len(CCYS),
    ticker='FX CARRY BASKET'
)

Performance report

basket.plot.performance()