Simulations & scenarios#

This page shows how simulation adapters can be added for a simple momentum strategy.

SigTech platform’s flexible data adapter system allows data to be replaced with artificial data. This is useful for scenario analysis, running simulations and checking strategies for possible overfitting.

Learn more: Example notebooks

Environment#

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

Learn more: Environment setup

import sigtech.framework as sig
from sigtech.framework.infra.data_adapter.simulations.simulation_adapter import SimAdapter
from sigtech.framework.infra.objects.dtypes import StringType, IntType, DictType
from sigtech.framework.schedules import SchedulePeriodic
from sigtech.framework.schedules.strategy import StrategySchedule

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

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

if not sig.config.is_initialised():
    env = sig.config.init()

Dates to be used when running analysis and scenarios:

past_date = dtm.datetime(2019, 1, 4, 11, 0)
future_date = dtm.datetime(2021, 10, 1, 11, 0)

The original data & tradable instrument #

Construct a simple momentum signal of the Eurostoxx index and create a tradable version of the index to trade using the signal. The index is plotted in the following example.

Use obj.get, a generic object retrieving API that can be used on any platform object, such as individual instruments or strategies. This goes to a database, an object cache or builds the object on the fly.

index_ts = sig.obj.get('SX5E INDEX').history()
instr_sx5e = sig.TradableIndex(underlying='SX5E INDEX', currency='EUR')
instr_sx5e.history().loc[:past_date].plot();

A signal is defined in the example below by providing a new signal function, which retrieves the index data and evaluates the average return for the last 21 periods. The negative sign of this average is then used for a mean-reversion signal.

Defining a simple mean-reversion strategy #

def signal_fn():
    mean_reversion_signal = - \
        np.sign(sig.obj.get('SX5E INDEX').history(
        ).pct_change().rolling(window=21).mean())
    return mean_reversion_signal.dropna().to_frame('SX5E')

signal_obj = sig.Signal(function=lambda: signal_fn())

In the following example the building block SignalStrategy is used to turn a time series of signals into a tradable strategy which can be backtested.

Learn more: Signal Strategy

start_date = dtm.date(2010, 1, 4)
example_strategy_1 = sig.SignalStrategy(
    start_date=start_date,
    end_date=past_date,
    currency='EUR',
    signal_name=signal_obj.name,
    rebalance_frequency='EOM',
    allocation_function=sig.signal_library.allocation.identity,
    instrument_mapping={'SX5E': instr_sx5e.name},
    convert_long_short_weight=False,
)

Python:

example_strategy_1.history().plot()

Output:

Modifying the data using a simulation adapter #

As the historic performance has been obtained, the performance that would be achieved if the data were modified is important. This can be achieved by introducing a simulation adapter, which can be dynamically added or removed from the data service. It obtained the raw data from the underlying data adapters and feed a modified version to the user. This allows various modifications to be performed.

A few examples include:

Moving a window of original data to a new time to be replayed. The replay can either be in absolute, difference or return space.
Perturbing the original data.
Substituting the original data with randomly generated data.

In the following example we obtain a simulation adapter, push it on to the data service and add a modifier. In this case the modifier acts on the Eurostoxx index data by shifted the original level by a normally distributed random quantity.

sim_adapter = SimAdapter()
sim_adapter.push_to_data_service()

sim_adapter.add_normal_timeseries_modifier(
    'SX5E INDEX',
    original_weight=1.0,
    change_type='PRICE',
    mean_override=0.0,
    std_override=10.0
)

The caches in the platform need clearing to allow the new data to flow through:

sig.config.configured_env().object.clear_object_data()

Now the modifier has been added we can retrieve the data and see a perturbation compared to the original level.

(sig.obj.get('SX5E INDEX').history() - index_ts).loc[:past_date].plot();

The strategy can be run multiple times using different perturbations, which are defined by a seed on the simulation adapter. The returns are all plotted below for 128 runs and the total return each time is recorded.

returns = []
for s in range(128):
    sim_adapter.clear_simulation()
    sig.config.configured_env().object.clear_object_data()
    sim_adapter.seed = s
    total_return = example_strategy_1.history().dropna()
    returns.append((total_return.iloc[-1] / total_return.iloc[0]) - 1)
    np.log(total_return).plot(color='r', alpha=0.01)

The distribution of returns:

pd.Series(returns).hist(bins=10);

As the current simulation adapter is no longer needed, remove it from the data service and clear the caches:

sim_adapter.remove_from_data_service()
sig.config.configured_env().object.clear_object_data()
sig.config.configured_env().fx_cross.clear_cache()

Future scenarios #

In the following example an FX hedged version of the previous strategy is run in to the future.

The FX Forward Hedging Strategy #

The previous strategy is based in EUR. We can wrap it in by using the FXForwardHedgingStrategy to hedge the FX risk for a USD version.

hedged_instr_nky = sig.FXForwardHedgingStrategy(
    start_date=start_date,
    end_date=past_date,
    strategy_name=example_strategy_1.name,
    currency='USD',
    hedging_tenor='1M',
    exposure_rebalance_threshold=0.02,
    hedge_rebalance_threshold=0.02,
    use_holdings_for_hedge_ccy = True,
)

The performance of the hedged version is shown against the unhedged version:

hedged_instr_nky.history().plot()
example_strategy_1.history().plot();

Adding a simulation adapter to replay history #

Create a simulation adapter and push it to the data service. In this example, add modifiers to move data from the past in to the future. As forward is now being included in the strategy, move the curves:

future_sim_adapter = SimAdapter()
future_sim_adapter.push_to_data_service()

replay_start = dtm.date(2010, 1, 1)

future_sim_adapter.add_move_timeseries_modifier('SX5E INDEX',
                                                change_type='RETURN',
                                                original_start=replay_start,
                                                fill_start=past_date.date())
future_sim_adapter.add_move_timeseries_modifier('EURUSD CURNCY',
                                                change_type='RETURN',
                                                original_start=replay_start,
                                                fill_start=past_date.date())
future_sim_adapter.add_move_curve_modifier('USD.D CURVE',
                                           original_start=replay_start,
                                           fill_start=past_date.date())
future_sim_adapter.add_move_curve_modifier('USD.FX CURVE',
                                           original_start=replay_start,
                                           fill_start=past_date.date())
future_sim_adapter.add_move_curve_modifier('EUR.FX CURVE',
                                           original_start=replay_start,
                                           fill_start=past_date.date())

sig.config.configured_env().object.clear_object_data()

The retrieved data now shows the returns being replayed in to the future:

ax = sig.obj.get('SX5E INDEX').history().plot()
ax.axvline(past_date, color='k', linestyle='--')
ax.axvline(replay_start, color='k', linestyle='--', alpha=0.5);

ax = sig.obj.get('EURUSD CURNCY').history().plot()
ax.axvline(past_date, color='k', linestyle='--')
ax.axvline(replay_start, color='k', linestyle='--', alpha=0.5);

The strategies can be redefined to run up to this future date. The performance is then calculated:

start_date = dtm.date(2010, 1, 4)
future_example_strategy = sig.SignalStrategy(
    start_date=start_date,
    end_date=future_date,
    currency='EUR',
    signal_name=signal_obj.name,
    rebalance_frequency='EOM',
    allocation_function=sig.signal_library.allocation.identity,
    instrument_mapping={'SX5E': instr_sx5e.name},
    convert_long_short_weight=False,
)

hedged_instr = sig.FXForwardHedgingStrategy(
    start_date=start_date,
    end_date=future_date,
    strategy_name=future_example_strategy.name,
    currency='USD',
    hedging_tenor='1M',
    exposure_rebalance_threshold=0.02,
    hedge_rebalance_threshold=0.02,
)

ax = hedged_instr.history().plot(color='r')
future_example_strategy.history().plot(ax=ax, color='b')
ax.axvline(past_date, color='k', linestyle='--');

FX Barrier Hedging Strategy #

We can also hedge using FX Barrier options. The following example defines a strategy to do this:

class FXBarrierHedgingStrategy(sig.DailyStrategy):
    # Input Checking
    strategy_name = StringType(required=True)
    maturity = StringType(required=True)
    rebalance_frequency = StringType(required=True)
    group_name = StringType(required=True)
    bdc = StringType(default=sig.calendar.BDC_PRECEDING)

    def __init__(self):
        super(FXBarrierHedgingStrategy, self).__init__()
        self.option_group = sig.obj.get(self.group_name)
        self._strategy = sig.obj.get(self.strategy_name)
        self.over = self._strategy.currency
        self.under = self.currency

    def schedule_information(self):
        """ set the holiday calendar for rebalancing - Here we use the the same holidays as the global trading manager """
        return StrategySchedule(holidays=sig.TradingManager.instance().trading_holidays)

    def get_rebalance_dates(self):
        """
        helper function to get all the rebalance dates based on the rebalance schedule and market holidays
        """
        first_date = self.calendar_schedule().next_data_date(self.start_date)
        all_dates = sig.SchedulePeriodic(self.start_date, self.calculation_end_date(),
                                     self.history_schedule().approximate_holidays(),
                                     frequency=self.rebalance_frequency, bdc=self.bdc).all_data_dates()[1:]
        return [first_date] + all_dates

    def strategy_initialization(self, dt):
        """ This is the first function to start the strategy building process. """
        self.add_method(self.first_entry_date, self.push_strategy)

        rebalance_dates = self.get_rebalance_dates()

        for d in rebalance_dates:
            self.add_method(d, self.enter_barrier_option_positions)

    def push_strategy(self, dt):

        self.add_position_target(dt, self.strategy_name, 1.0, unit_type='WEIGHT')
        self.add_method(self.valuation_dt(dt.date()),
                           self.clean_up_foreign_cash)

    def clean_up_foreign_cash(self, dt):
        for cash_instrument, quantity in self.positions.iterate_cash_positions(dt):
            if cash_instrument.currency != self.currency:
                self.add_fx_spot_trade(dt, self.currency,
                                       cash_instrument.currency, -quantity)

    def enter_barrier_option_positions(self, dt):
        """ This process determines the barrier option parameters. """

        # first close out existing fx forwards
        for instrument, quantity in self.positions.iterate_instruments(dt):
            if instrument.is_option():
                self.add_position_target(dt, instrument.name, 0.0, unit_type='WEIGHT')

        size_date = self.size_date_from_decision_dt(dt)
        notional_ccy = self.valuation_price(size_date, ccy=self.over)

        maturity_date = self.option_group.convert_maturity_tenor_to_date(
            size_date, self.maturity, target_maturity_weekday=3)

        # Use the spot price
        atm = sig.obj.get(self.option_group.underlying).history().asof(pd.to_datetime(size_date))

        barrier_option = self.option_group.get_option(
            'Put', atm, size_date, maturity_date, barrier_type='KI', barrier=atm * 0.99)

        self.add_position_target(dt, barrier_option.name, notional_ccy)

ki_hedged_instr = FXBarrierHedgingStrategy(
    currency='USD',
    strategy_name=future_example_strategy.name,
    start_date=start_date,
    end_date=past_date,
    group_name='EURUSD OTC OPTION GROUP',
    maturity='3M',
    rebalance_frequency='3M',
)

ax = ki_hedged_instr.history().plot(color='r')
example_strategy_1.history().plot(ax=ax, color='b')
ax.axvline(past_date, color='k', linestyle='--');

Input:

ki_hedged_instr.timeline_holdings.display(
    full=True,
    start_dt=ki_hedged_instr.valuation_dt(start_date),
    end_dt=ki_hedged_instr.valuation_dt(dtm.date(2010, 5, 1))
)