PeriodicIntradayStrategy

PeriodicIntradayStrategy#

class sigtech.framework.strategies.strategy.PeriodicIntradayStrategy

Baseclasses: Strategy

Intra-day strategy which executes on a regular schedule.

Parameters are the same as in the base Strategy.

Additional attributes:

  • strategy_frequency: Frequency used to build schedule of the strategy. Optional, 1 minute by default.

  • execution_delay: Delay added to execution datetime. Optional, 1 minute by default.

  • holidays: Holiday calendar used to build schedule of the strategy. Optional, taken from trading manager if not specified.

  • start_time: Start time of the strategy each day. Optional, Trading manager open time by default.

  • end_time: End time of the strategy each day. Optional, Trading manager close time by default.

  • timezone: Timezone used to build schedule of the strategy.

Example:

import datetime as dtm
import pytz
from typing import Optional

class ExampleStrategy(sig.PeriodicIntradayStrategy):
    futures_group: Optional[str] = 'NG COMDTY FUTURES GROUP'
    trade_timings: dict

    def __post_init__(self):
        super().__post_init__()
        self.fut_grp = sig.obj.get(self.futures_group)
        self.timezone = self.fut_grp.exchange().timezone

    def process_step(self, dt):
        d = dt.date()

        # 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(pytz.timezone(self.timezone).localize(dtm.datetime.combine(d, trade_time)),
                               self.trade, asset=active_contract.name, direction=trade_dir)

    def trade(self, dt, asset, direction):
        self.add_position_target(dt, asset, direction, unit_type='WEIGHT')

tst = ExampleStrategy(currency='USD', start_date=dtm.date(2020, 1, 4), end_date=dtm.date(2020, 1, 10), trade_timings={1: dtm.time(hour=1), 0: dtm.time(hour=11)})
tst.history()
end_time: Optional[time]
execution_delay: Optional[timedelta]
holidays: Optional[str]
start_time: Optional[time]
strategy_frequency: Optional[timedelta]
timezone: Optional[str]
process_step(dt: datetime)

Process a step of the strategy at dt :param dt: datetime :return:

strategy_extension(from_dt: datetime, to_dt: datetime)

Update the strategy contents for an expanded history. It should be replaced if a short run’s process initialization is not equivalent to continuing the strategy.

Parameters:
  • from_dt – Start datetime for history expansion.

  • to_dt – End datetime for history expansion.

strategy_initialization(dt: datetime) None

Initial decision run on the start date of the strategy.

Parameters:

dt – Reference datetime.