Intraday strategies#
This page shows how to create intraday strategies on the SigTech platform—iteratively going through different intraday strategies, where each iteration will add more complexity as well as explain the functionality being used.
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.instruments.futures import Future
from sigtech.framework.infra.objects.dtypes import (
AnyType, IntType, StringType, DictType)
from collections import defaultdict
import datetime as dtm
import pandas as pd
import pytz
import seaborn as sns
import numpy as np
sns.set(rc={'figure.figsize': (18, 6)})
if not sig.config.is_initialised():
env = sig.init()
env[sig.config.CUSTOM_DATA_SOURCE_CONFIG] = [
('[A-Z]{6} CURNCY', 'REFINITIV'),
]
Intraday strategies#
Overview#
Introduction
Empty strategy
Basic strategy
Intermediate strategy: FX
Intermediate strategy: Futures
This section covers the concept of creating intraday strategies on the SigTech platform. The SigTech backtesting engine operates as a timeline, where processes, such as trades and margin adjustments, are queued. Once all processes have been queued, the timeline gets run-through over the time period.
On each of the trading processes queued on the timeline, there are three key points in time:
Decision time: the point where the order is generated, and queued on the timeline to be executed at the execution time.
Sizing time: in relation to the decision point, this is a previous point in time used for sizing the trade that is to take place. for example, calculating the order size by using the price data at the sizing time.
Execution time: this is where the queued order gets executed and assumed filled.
These time points are essential to all trading actions taking place on the timeline.
In the following sub-sections, a number of intraday strategies are created. The idea is that the first strategy will be a strategy in its most basic form, while the subsequent strategies will become increasingly more complex.
Empty strategy#
Strategy intro
The first strategy to be created is an empty strategy—a strategy that does nothing. Although there’s no logic in terms of trading behaviour, there’s still a few steps that need to be taken for it to operate properly on the timeline.
Strategy behaviour
The strategy does nothing apart from holding the initial cash specified when instantiating the strategy from the start_date
to the end_date
.
Key functionality
When creating a strategy, a strategy class needs to be defined. In this case it is named
EmptyStrategy
.The created strategy class inherits from the
Strategy
class—all strategies created on the SigTech platform need to inherit from theStrategy
class to make use of the functionality in the library framework.The method
strategy_initialization
is inherited from theStrategy
class, and needs to be overwritten to avoid throwing an exception—eventhough apass
statement is enough, as shown below.
class EmptyStrategy(sig.Strategy):
def strategy_initialization(self, dt):
pass
empty_strategy = EmptyStrategy(
currency='USD',
start_date=dtm.date(2020, 10, 7),
initial_cash=10000,
total_return=False,
ticker='EMPTY'
)
Python:
empty_strategy.history().head()
Output:
2020-10-07 10000.0
2020-10-08 10000.0
2020-10-09 10000.0
2020-10-12 10000.0
2020-10-13 10000.0
Name: (LastPrice, EOD, EMPTY STRATEGY), dtype: float64
Basic strategy#
Strategy intro
In this next iteration of an intraday strategy we introduce the concept of processes. Processes are different actions that are scheduled on the timeline, which in turn are enacted as the timeline is replayed.
Strategy behaviour
The following strategy does the following actions every day:
Buys front future contract at 01:00
Sells front future contract at 11:00
Sweeps PnL at 18:59:59
Key functionality
See previous strategies for already introduced functionality.
add_method
: allows for adding methods or functions to the timeline which manipulate the strategy in some way. In this case, the methodtrade
to the timeline. The method takes a reference time as a parameter—the point in time the process will relate to. Additional keyword arguments can be provided to theadd_method
method.size_dt_from_decision_dt
: specifies when the position sizing is done relative to the decision date-time. This method usually simply subtracts adatetime.timedelta
from thedecision_dt
argument.execution_dt_from_datetime
: similar to the above, this method is expected to modify the passeddecision_dt
to specify when the order is filled. Optionally, one can use theinstrument
argument to make theexecution_dt
instrument dependent.add_position_target
: no matter what the current position holdings are at a given time, theadd_position_target
method will change the position of an asset based on a weight fraction of the overall portfolio, if the argumentunit_type
are used with the valueWEIGHT
.add_margin_cleanup
: this method applies the variation margin based on the latest price movement, and cash gets added/removed to/from the cash account from/to the spot margin account.
class BasicStrategy(sig.Strategy):
futures_group = StringType(default='NG COMDTY FUTURES GROUP')
trade_timings = DictType(AnyType())
def __init__(self):
super().__init__()
self.fut_grp = sig.obj.get(self.futures_group)
self.calendar = sig.calendar.build_calendar(
self.fut_grp.exchange().holidays)
self.timezone = pytz.timezone(self.fut_grp.exchange().timezone)
def strategy_initialization(self, dt):
for _d in self.calendar.range(self.start_date, self.end_date):
d = _d.astype(dtm.datetime)
# Gets active contract on date
active_contract = self.fut_grp.active_contract(d, True)
# Loops through the entry and exit times
for trade_dir, trade_time in self.trade_timings.items():
# Adds the process trade to the timeline
self.add_method(
self.timezone.localize(dtm.datetime.combine(d, trade_time)),
self.trade,
asset=active_contract.name,
direction=trade_dir
)
# Runs a clean up margin process at end of day
self.add_margin_cleanup(d, Future)
def trade(self, dt, asset, direction):
self.add_position_target(dt, asset, direction, unit_type='WEIGHT')
def size_dt_from_decision_dt(self, decision_dt):
return decision_dt - dtm.timedelta(minutes=1)
def execution_dt_from_datetime(self, instrument, dt):
return dt + dtm.timedelta(minutes=1)
basic_strategy = BasicStrategy(
currency='USD',
start_date=dtm.date(2020, 1, 4),
end_date=dtm.date(2020, 1, 10),
total_return=False,
ticker='BASIC',
trade_timings={1: dtm.time(hour=1), 0: dtm.time(hour=11)},
initial_cash=1e8
)
Input:
basic_strategy.history()
Output:
2020-01-06 9.979126e+07
2020-01-07 9.958345e+07
2020-01-08 9.937866e+07
2020-01-09 9.917229e+07
2020-01-10 9.917229e+07
Name: (LastPrice, EOD, BASIC STRATEGY), dtype: float64
View example actions on the timeline over a certain period of time, using the interactive_portfolio_table
:
fut_tz = sig.obj.get('NG COMDTY FUTURES GROUP').exchange().timezone
basic_strategy.plot.portfolio_table(
'ACTION_PTS',
start_dt=dtm.datetime(2020, 1, 4, tzinfo=pytz.timezone(fut_tz)),
end_dt=dtm.datetime(2020, 1, 10, tzinfo=pytz.timezone(fut_tz)),
)
Intermediate strategy: FX#
Strategy intro
In this iteration of intraday strategies, the concept of signals gets included where the signal is momentum-based and applied to the spot FX market.
Strategy behaviour
This strategy calculates a simple momentum signal on each minute on each business day. The calculated momentum signal will be used as a weight representation of the portfolio, i.e. the higher the value of the momentum signal, the larger the position in the relevant asset. The rebalance process creates a trade based on the difference between the target and the current weight in relevant asset.
Key functionality
Tip: see previous strategies for already introduced functionality.
add_fx_spot_trade
: this method allows for performing FX spot trades.
class IntermediateFXStrategy(sig.Strategy):
long_span = IntType(default=200)
short_span = IntType(default=50)
asset = StringType(default='EURUSD CURNCY')
def __init__(self):
super().__init__()
self.signals = defaultdict(float)
self.ts = defaultdict(float)
def calculate_signal(self, instrument):
self.ts[instrument.name] = sig.obj.get(self.asset).intraday_history(period=dtm.timedelta(minutes=1))
short_ma = self.ts[instrument.name].ewm(span=self.short_span).mean()
long_ma = self.ts[instrument.name].ewm(span=self.long_span).mean()
ma_signal = (short_ma - long_ma) * (2 / (self.long_span - self.short_span))
return ma_signal * 100
def strategy_initialization(self, dt):
self.signals[self.asset] = self.calculate_signal(sig.obj.get(self.asset))
for d in sig.obj.get(self.asset).calendar_schedule().data_dates(self.start_date, self.end_date):
for t in pd.date_range("00:00", "23:59", freq="1min"):
dt = pytz.UTC.localize(dtm.datetime.combine(d, t.time()))
self.add_method(dt, self.trade, asset=self.asset)
def trade(self, dt, asset):
target_weight = self.signals[asset].asof(dt)
size_price = self.ts[asset][dt - dtm.timedelta(minutes=1)]
current_pos = self.positions.get_cash_value(dt, f'{asset[:3]} CASH')
current_pos_strat_ccy = current_pos * size_price
nav = self.positions.valuation(dt, self.currency)
current_weight = current_pos_strat_ccy / nav
adjust_weight = target_weight - current_weight
price_adjust = size_price if adjust_weight > 0 else 1
trade_notional = adjust_weight * nav / price_adjust
self.add_fx_spot_trade(dt, over=asset[3:6], under=asset[:3],
notional=trade_notional,
execution_dt=dt + dtm.timedelta(minutes=1))
def size_dt_from_decision_dt(self, decision_dt):
return decision_dt - dtm.timedelta(minutes=1)
def execution_dt_from_datetime(self, instrument, dt):
return dt + dtm.timedelta(minutes=1)
Python:
intermediate_strat_fx = IntermediateFXStrategy(
currency='USD',
start_date=dtm.date(2020,11,24),
end_date=dtm.date(2020,11,30),
total_return=False,
initial_cash=1e8
)
intermediate_strat_fx.history()
Output:
2020-11-24 9.999976e+07
2020-11-25 9.999925e+07
2020-11-26 9.999918e+07
2020-11-27 9.999919e+07
2020-11-30 9.999905e+07
Name: (LastPrice, EOD, USD C7B2184A IFXS STRATEGY), dtype: float64
intermediate_strat_fx.plot.portfolio_table(
'ACTION_PTS',
start_dt = dtm.datetime(2020, 11, 24, 23, 30, 0, tzinfo=pytz.UTC),
end_dt = dtm.datetime(2020, 11, 25, 0, 0, 0, tzinfo=pytz.UTC),
)
Intermediate strategy: Futures#
Strategy intro
This strategy operates the same as the previously defined strategy, but with Futures instead of FX Spot.
Strategy behaviour
This strategy calculates a simple momentum signal on each minute, on each business date. The calculated momentum signal is used as a weight representation of the portfolio—the higher the value of the momentum signal, the larger the position in the relevant asset. The rebalance process creates a trade based on the difference between the target and the current weight in the relevant asset.
class IntermediateFuturesStrategy(sig.Strategy):
long_span = IntType(default=200)
short_span = IntType(default=50)
futures_group = StringType(default='CL COMDTY FUTURES GROUP')
def __init__(self):
super().__init__()
self.signals = defaultdict(float)
self.fut_group = sig.obj.get(self.futures_group)
self.calendar = sig.calendar.build_calendar(
self.fut_group.exchange().holidays)
self.open_time = self.fut_group.exchange().exchange_open
self.close_time = self.fut_group.exchange().exchange_close
self.timezone = pytz.timezone(self.fut_group.exchange().timezone)
def calculate_signal(self, instrument):
ts = instrument.intraday_history(
period=dtm.timedelta(minutes=1))
short_ma = ts.ewm(span=self.short_span).mean()
long_ma = ts.ewm(span=self.long_span).mean()
ma_signal = (short_ma - long_ma) * (2 / (self.long_span - self.short_span))
return ma_signal * 1000
def strategy_initialization(self, dt):
for _d in self.calendar.range(self.start_date, self.end_date):
d = _d.astype(dtm.datetime)
active_contract = self.fut_group.active_contract(d, True)
if active_contract.name not in self.signals:
self.signals[active_contract.name] = self.calculate_signal(
active_contract)
t = self.timezone.localize(dtm.datetime.combine(d, self.open_time))
close_dt = self.timezone.localize(
dtm.datetime.combine(d, self.close_time))
while t < close_dt:
weight = self.signals[active_contract.name].asof(dt)
if np.isnan(weight):
weight = 0
self.add_method(t, self.trade,
asset=active_contract.name, weight=weight)
t += dtm.timedelta(minutes=1)
self.add_method(t, self.trade, asset=active_contract.name, weight=0)
self.add_margin_cleanup(d, Future)
def trade(self, dt, asset, weight):
self.add_position_target(dt, asset, weight, unit_type='WEIGHT')
def size_dt_from_decision_dt(self, decision_dt):
return decision_dt - dtm.timedelta(minutes=1)
def execution_dt_from_datetime(self, instrument, dt):
return dt + dtm.timedelta(minutes=1)
intermediate_strat_futures = IntermediateFuturesStrategy(
currency='USD',
start_date=dtm.date(2019,1,5),
end_date=dtm.date(2019,1,15),
)
Input:
intermediate_strat_futures.history()
Output:
2019-01-07 996.125492
2019-01-08 992.448788
2019-01-09 988.778176
2019-01-10 985.346968
2019-01-11 981.868839
2019-01-14 978.526895
2019-01-15 978.592130
Name: (LastPrice, EOD, USD 1E02DEF8 IFS STRATEGY), dtype: float64
intermediate_strat_futures.plot.portfolio_table(
'ACTION_PTS',
start_dt = dtm.datetime(2019, 1, 7, 21, 10, 0, tzinfo=pytz.UTC),
end_dt = dtm.datetime(2019, 1, 8, 0, 0, 0, tzinfo=pytz.UTC),
)