Custom Swap 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 strategy from the base Strategy class

Environment

This section imports the relevant internal and external libraries, and sets up the platform environment.

import datetime as dtm
import seaborn as sns
import pandas as pd
import numpy as np

import sigtech.framework as sig
sns.set(rc={'figure.figsize':(18, 6)})

env = sig.init(env_date=dtm.date(2024, 2, 22))

from sigtech.framework.instruments.ir_otc import InterestRateSwap
from sigtech.framework.schedules.schedule import SchedulePeriodic
from sigtech.framework.infra.cal import imm_date

class SwapCarryStrategy(sig.DailyStrategy):

    ccy_calendar: str = 'TARGET CALENDAR'
    rebalance_frequency: str
    dv01_budget: float # DV01 budget (in USD$) to allocate to each leg before weighting by carry/vol """
    # List of tuples of (ccy, fwd start, maturity)
    all_legs: list[tuple[str, str, str]] = [('EUR', '2Y', '3Y'), ('EUR', '5Y', '5Y')]
    irs_group: sig.InterestRateSwapGroup = sig.obj.get('EUR INTEREST RATE SWAP GROUP')

    ############ UTILITY FUNCTIONS #################################################################


    def _get_rebalance_schedule(self) -> list:
        """ Gets the rebalance schedule as next data date + rest of schedule """
        first_date = self.calendar_schedule().next_data_date(self.start_date)
        all_dates = SchedulePeriodic(
            self.start_date, self.calculation_end_date(),
            self.ccy_calendar,
            frequency=self.rebalance_frequency,
        ).all_data_dates()[1:]
        return [first_date] + all_dates


    def _get_next_dec_imm_date(self, dt) -> dtm.date:
        """ Gets the next December 3rd Wed IMM date given a date """
        # Get 1 Dec date for the given year
        dec_som = dtm.date(dt.year, 12, 1)
        next_dec_imm = imm_date(dec_som, "EUR")

        if next_dec_imm < dt.date():
            dec_som = dtm.date(dt.year + 1, 12, 1)
            return imm_date(dec_som, "EUR")
        else:
            return next_dec_imm


    def _get_1y_carry(self, ccy: str, fwd_start: str, tenor: str, dt: dtm.date) -> float:
        """
        Gets the 1-year carry at a given date defined as follows:
        > i.e 2y3y rate - 1y3y rate
        """
        fwd_start_1 = f"{int(fwd_start[:-1])-1}Y"

        r1 = self.irs_group.data_df(tenor=tenor,
                                    fwd_tenor=fwd_start_1,
                                    trade_date=dt,
                                    rate_only=True).squeeze()
        r2 = self.irs_group.data_df(tenor=tenor,
                                    fwd_tenor=fwd_start,
                                    trade_date=dt,
                                    rate_only=True).squeeze()

        return r1 - r2

    #################################################################################################


    def strategy_initialization(self, dt):
        """ Sets up the strategy, scheduling the future trading actions """
        # Schedule all trading actions
        for d in self._get_rebalance_schedule():
            self.add_method(d, self._trade_processor)


    def _trade_processor(self, dt):
        """ Main trading function """
        # Exit out of existing swaps
        for instr, quantity in self.positions.iterate_instruments(dt):
            if isinstance(instr, sig.InterestRateSwap):
                self.add_position_target(dt, instr.name, 0.0, transaction_type="roll")



        position_size = {}

        # Get all position_size
        for leg in self.all_legs:
            _ccy = leg[0]
            _fwd = leg[1]
            _ten = leg[2]

            # Generate sizing
            historical_rates = self.irs_group.data_df(tenor=_ten,
                                                      fwd_tenor=_fwd,
                                                      start_date=dt.date() - dtm.timedelta(days=365),
                                                      rate_only=True).squeeze()
            # Get leg vol
            _vol = (historical_rates[:dt.date()].rolling(252).std() * np.sqrt(252))[-1]
            # Get carry
            _carry = self._get_1y_carry(_ccy, _fwd, _ten, dt.date())

            # Store size as vol-adjusted carry
            position_size[leg] = _carry / _vol


        # Get start date of all swaps to trade
        start_date = self._get_next_dec_imm_date(dt)

        # Loop again to trade the individual legs
        for leg in self.all_legs:
            _ccy = leg[0]
            _fwd = leg[1]
            _ten = leg[2]

            # Calculate final DV01 scaled by weight
            size = self.dv01_budget * position_size[leg]

            # Trade
            self.add_trade(
                dt=dt,
                instrument_name=InterestRateSwap,
                swap_currency=_ccy,
                tenor=_ten,
                start_date=start_date + pd.Timedelta(f"{int(_fwd[:-1])*365} days"),
                units=size,
                value_post_start=False,
                unit_type='TRADE',
            )

strategy = SwapCarryStrategy(
    currency='EUR',
    start_date=dtm.date(2020, 1, 5),
    initial_cash=1e7,
    rebalance_frequency='1M',
    dv01_budget=1e6,
    include_trading_costs=False,
    total_return=False,
    ccy_calendar = 'TARGET CALENDAR',
    all_legs = [('EUR', '2Y', '3Y'), ('EUR', '5Y', '5Y')]
)

strategy.build()

strategy.history().plot();