Built-in strategies

# indie:lang_version = 5
from math import isnan, nan
from indie import strategy, param, MutSeriesF, color, plot, MainStrategyContext, Optional
from indie.algorithms import Adx, Highest, Lowest
from indie.math import cross
from indie.strategies import Order, order_side, order_status as o_st


def is_not_active(order: Optional[Order]) -> bool:
    return order is None or order.value().status in [o_st.FILLED, o_st.CANCELED, o_st.REJECTED, o_st.PARTIALLY_FILLED_CLOSED]


@strategy('ADX Breakout', overlay_main_pane=True)
@param.int('adx_smooth_period', default=14, title='ADX Smoothing Period', min=1)
@param.int('adx_period', default=14, title='ADX Period', min=1)
@param.float('adx_lower_level', default=18.0, title='ADX Lower Level', min=0.0, max=100.0, step=0.1)
@param.float('profit_target_multiple', default=1.0, title='Profit Target Box Width Multiple', min=0.0, max=10.0, step=0.1)
@param.float('stop_loss_multiple', default=0.5, title='Stop Loss Box Width Multiple', min=0.0, max=10.0, step=0.1)
@param.int('box_look_back', default=20, title='BreakoutBox Lookback Period', min=1)
@param.str('orders_side', default='Both', title='Orders side', options=['Both', 'Long', 'Short'])
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
@plot.line('box_upper_level', title='Box Upper Level', color=color.BLUE)
@plot.line('box_lower_level', title='Box Lower Level', color=color.BLUE)
@plot.background(title='Is ADX Low', color=color.PURPLE(0.1))
@plot.line('stop_loss', title='Stop loss', color=color.RED, line_width=2)
@plot.line('profit_target', title='Profit target', color=color.GREEN, line_width=2)
class Main(MainStrategyContext):
    def __init__(self):
        self._order = Optional[Order]()
        self._upper_exit_order = Optional[Order]()
        self._lower_exit_order = Optional[Order]()

    def calc(self, adx_smooth_period, adx_period, adx_lower_level, profit_target_multiple, stop_loss_multiple, box_look_back, orders_side, order_size, order_size_unit):
        # When the ADX drops below threshold limit, then we consider the pair in consolidation.
        # Set Box around highs and lows of the last box_look_back candles with upper and lower boundaries.
        # When price breaks outside of box, a trade is taken.
        # User can set a profit target and stop loss values:
        # e.g. 0.5 is 50% of the size of the box, 1 is full size, 2 is 200% of the size of the box.

        _, sig, _ = Adx.new(adx_period, adx_smooth_period)
        is_adx_low = sig[0] < adx_lower_level

        box_upper_level = MutSeriesF.new()
        box_lower_level = MutSeriesF.new()
        pos_size = self.trading.position.size
        if pos_size == 0:
            highest = Highest.new(self.high, box_look_back)
            box_upper_level[0] = highest[1]
            lowest = Lowest.new(self.low, box_look_back)
            box_lower_level[0] = lowest[1]
        box_width = box_upper_level[0] - box_lower_level[0]

        profit_target = nan
        stop_loss = nan

        if pos_size == 0 and self._order is None:
            if self._lower_exit_order is not None:
                self._lower_exit_order.value().cancel()
                self._lower_exit_order = None
            if self._upper_exit_order is not None:
                self._upper_exit_order.value().cancel()
                self._upper_exit_order = None
            is_buy_valid = cross(self.close, box_upper_level) and is_adx_low
            is_sell_valid = cross(self.close, box_lower_level) and is_adx_low
            entry_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
            entry_long = is_buy_valid and (orders_side in ['Both', 'Long'])
            if entry_long:
                self._order = self.trading.place_order(order_side.BUY, size=entry_size).submit()
            entry_short = is_sell_valid and (orders_side in ['Both', 'Short'])
            if entry_short:
                self._order = self.trading.place_order(order_side.SELL, size=entry_size).submit()
        elif pos_size != 0:
            self._order = None
            pos_price = self.trading.position.price
            if pos_size > 0:
                profit_target = pos_price + box_width * profit_target_multiple
                stop_loss = pos_price - box_width * stop_loss_multiple
            elif pos_size < 0:
                profit_target = pos_price - box_width * profit_target_multiple
                stop_loss = pos_price + box_width * stop_loss_multiple
            upper_exit_price = max(profit_target, stop_loss)
            lower_exit_price = min(profit_target, stop_loss)
            exit_side = order_side.SELL if pos_size > 0 else order_side.BUY
            if self._lower_exit_order is None or is_not_active(self._lower_exit_order):
                self._lower_exit_order = self.trading.place_order(exit_side, abs(pos_size)).stop(lower_exit_price).submit()
            else:
                self._lower_exit_order.value().amend().stop(lower_exit_price).submit()
            if self._upper_exit_order is None or is_not_active(self._upper_exit_order):
                self._upper_exit_order = self.trading.place_order(exit_side, abs(pos_size)).limit(upper_exit_price).submit()
            else:
                self._upper_exit_order.value().amend().limit(upper_exit_price).submit()

        return (
            box_upper_level[0], box_lower_level[0],
            plot.Background() if is_adx_low else plot.Background(color=color.TRANSPARENT),
            stop_loss, profit_target,
        )

# indie:lang_version = 5
from indie import strategy, param, Var
from indie.strategies import order_side


@strategy('Consecutive Up/Down Strategy', overlay_main_pane=True)
@param.int('consecutive_bars_up', default=3, title='Consecutive bars up', min=1)
@param.int('consecutive_bars_down', default=3, title='Consecutive bars down', min=1)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
def Main(self, consecutive_bars_up, consecutive_bars_down, order_size, order_size_unit):
    price = self.close

    ups = Var[int].new(init=0)
    downs = Var[int].new(init=0)
    if price[0] > price[1]:
        ups.set(ups.get() + 1)
    else:
        ups.set(0)
    if price[0] < price[1]:
        downs.set(downs.get() + 1)
    else:
        downs.set(0)

    pos_size = self.trading.position.size
    desired_pos_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
    if ups.get() >= consecutive_bars_up and pos_size <= 0:
        # reverse previous position if exists or open a new one
        self.trading.place_order(order_side.BUY, size=desired_pos_size + abs(pos_size)).submit()
    elif downs.get() >= consecutive_bars_down and pos_size >= 0:
        self.trading.place_order(order_side.SELL, size=desired_pos_size + abs(pos_size)).submit()

# indie:lang_version = 5
from indie import strategy, param
from indie.strategies import order_side


@strategy('InSide Bar Strategy', overlay_main_pane=True)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
def Main(self, order_size, order_size_unit):
    desired_pos_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
    if self.high[0] < self.high[1] and self.low[0] > self.low[1]:
        pos_size = self.trading.position.size
        if self.close[0] > self.open[0] and pos_size <= 0:
            # reverse previous position if exists or open a new one
            self.trading.place_order(order_side.BUY, size=desired_pos_size + abs(pos_size)).submit()
        elif self.close[0] < self.open[0] and pos_size >= 0:
            self.trading.place_order(order_side.SELL, size=desired_pos_size + abs(pos_size)).submit()

# indie:lang_version = 5
from math import isnan
from indie import strategy, param, MutSeriesF
from indie.algorithms import Macd
from indie.math import cross_over, cross_under
from indie.strategies import order_side


@strategy('MACD Strategy', overlay_main_pane=True)  # Moving Average Convergence Divergence Strategy
@param.int('fast_length', default=12, title='Fast length', min=1)
@param.int('slow_length', default=26, title='Slow length', min=1)
@param.int('macd_length', default=9, title='MACD length', min=1)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
def Main(self, fast_length, slow_length, macd_length, order_size, order_size_unit):
    macd, signal, _ = Macd.new(self.close, fast_length, slow_length, macd_length)
    delta = MutSeriesF.new(macd[0] - signal[0])

    desired_pos_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
    if not isnan(delta[0]):
        pos_size = self.trading.position.size
        if cross_over(delta, 0) and pos_size <= 0:
            # reverse previous position if exists or open a new one
            self.trading.place_order(order_side.BUY, size=desired_pos_size + abs(pos_size)).submit()
        elif cross_under(delta, 0) and pos_size >= 0:
            self.trading.place_order(order_side.SELL, size=desired_pos_size + abs(pos_size)).submit()

# indie:lang_version = 5
from indie import strategy, param, MainStrategyContext, Optional
from indie.algorithms import Change
from indie.strategies import order_side as o_sd, order_status as o_st, Order


def is_not_active(order: Optional[Order]) -> bool:
    return order is None or order.value().status in [o_st.FILLED, o_st.CANCELED, o_st.REJECTED, o_st.PARTIALLY_FILLED_CLOSED]


def is_updatable(order: Optional[Order]) -> bool:
    return order is not None and order.value().status in [o_st.CREATED, o_st.PLACED, o_st.PENDING_PLACED, o_st.PARTIALLY_FILLED]


@strategy('Momentum Strategy', overlay_main_pane=True)
@param.int('length', default=12, title='Momentum length', min=1)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
class Main(MainStrategyContext):
    def __init__(self):
        self.long_order: Optional[Order] = None
        self.short_order: Optional[Order] = None

    def calc(self, length, order_size, order_size_unit):
        mom0 = Change.new(self.close, length)
        mom1 = Change.new(mom0, 1)

        entry_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
        if mom0[0] > 0.0 and mom1[0] > 0.0:
            self.entry(o_sd.BUY, entry_size, self.high[0] + self.info.tick_size)
        else:
            if is_updatable(self.long_order):
                # cancel pending long order if momentum reverses
                self.trading.cancel_order(self.long_order.value().id)
        if mom0[0] < 0.0 and mom1[0] < 0.0:
            self.entry(o_sd.SELL, entry_size, self.low[0] - self.info.tick_size)
        else:
            if is_updatable(self.short_order):
                # cancel pending short order if momentum reverses
                self.trading.cancel_order(self.short_order.value().id)

    def entry(self, side: o_sd, size: float, stop_price: float) -> None:
        order = self.long_order if side == o_sd.BUY else self.short_order

        pos_size = self.trading.position.size
        # reverse previous position if exists or open a new one
        order_size = size + abs(pos_size)
        if pos_size == 0 or pos_size > 0 and side == o_sd.SELL or pos_size < 0 and side == o_sd.BUY:
            # create new order or update existing one
            if is_not_active(order):
                order = self.trading.place_order(side=side, size=order_size).stop(stop_price).submit()
            elif is_updatable(order):
                self.trading.amend_order(order.value().id).size(order_size).stop(stop_price).submit()
        elif is_updatable(order):
            # if desired position is already achieved, cancel existing order
            self.trading.cancel_order(order.value().id)
            order = None

        if side == o_sd.BUY:
            self.long_order = order
        else:
            self.short_order = order

# indie:lang_version = 5
from indie import strategy, param, Var
from indie.algorithms import Sma
from indie.strategies import order_side


@strategy('MovingAvg Cross', overlay_main_pane=True)
@param.int('length', default=9, min=1, title='MA Length')
@param.int('confirm_bars', default=1, min=1, title='Confirm bars')
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
def Main(self, length, confirm_bars, order_size, order_size_unit):
    price = self.close
    ma = Sma.new(price, length)
    pos_size = self.trading.position.size
    desired_pos_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])

    buy_cond = price[0] > ma[0]
    buy_count = Var[int].new(init=0)
    if buy_cond:
        buy_count.set(buy_count.get() + 1)
    else:
        buy_count.set(0)
    if buy_count.get() == confirm_bars and pos_size <= 0:
        # reverse previous position if exists or open a new one
        self.trading.place_order(order_side.BUY, size=desired_pos_size + abs(pos_size)).submit()

    sell_cond = price[0] < ma[0]
    sell_count = Var[int].new(init=0)
    if sell_cond:
        sell_count.set(sell_count.get() + 1)
    else:
        sell_count.set(0)
    if sell_count.get() == confirm_bars and pos_size >= 0:
        self.trading.place_order(order_side.SELL, size=desired_pos_size + abs(pos_size)).submit()

# indie:lang_version = 5
from indie import strategy, param
from indie.strategies import order_side


@strategy('OutSide Bar Strategy', overlay_main_pane=True)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
def Main(self, order_size, order_size_unit):
    desired_pos_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
    if self.high[0] > self.high[1] and self.low[0] < self.low[1]:
        pos_size = self.trading.position.size
        if self.close[0] > self.open[0] and pos_size <= 0:
            # reverse previous position if exists or open a new one
            self.trading.place_order(order_side.BUY, size=desired_pos_size + abs(pos_size)).submit()
        elif self.close[0] < self.open[0] and pos_size >= 0:
            self.trading.place_order(order_side.SELL, size=desired_pos_size + abs(pos_size)).submit()

# indie:lang_version = 5
from indie import strategy, param, MainStrategyContext, Optional
from indie.algorithms import Highest, Lowest
from indie.strategies import order_side as o_sd, order_status as o_st, Order


def is_not_active(order: Optional[Order]) -> bool:
    return order is None or order.value().status in [o_st.FILLED, o_st.CANCELED, o_st.REJECTED, o_st.PARTIALLY_FILLED_CLOSED]


def is_updatable(order: Optional[Order]) -> bool:
    return order is not None and order.value().status in [o_st.CREATED, o_st.PLACED, o_st.PENDING_PLACED, o_st.PARTIALLY_FILLED]


@strategy('Price Channel Strategy', overlay_main_pane=True)
@param.int('length', default=20, title='Lookback window length', min=1)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
class Main(MainStrategyContext):
    def __init__(self):
        self.long_order: Optional[Order] = None
        self.short_order: Optional[Order] = None

    def calc(self, length, order_size, order_size_unit):
        highest = Highest.new(self.high, length)
        lowest = Lowest.new(self.low, length)

        if self.bar_index > length:
            entry_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
            self.entry(o_sd.BUY, entry_size, highest[0])
            self.entry(o_sd.SELL, entry_size, lowest[0])

    def entry(self, side: o_sd, size: float, stop_price: float) -> None:
        order = self.long_order if side == o_sd.BUY else self.short_order

        pos_size = self.trading.position.size
        # reverse previous position if exists or open a new one
        order_size = size + abs(pos_size)
        if pos_size == 0 or pos_size > 0 and side == o_sd.SELL or pos_size < 0 and side == o_sd.BUY:
            # create new order or update existing one
            if is_not_active(order):
                order = self.trading.place_order(side=side, size=order_size).stop(stop_price).submit()
            elif is_updatable(order):
                self.trading.amend_order(order.value().id).size(order_size).stop(stop_price).submit()
        elif is_updatable(order):
            # if desired position is already achieved, cancel existing order
            self.trading.cancel_order(order.value().id)
            order = None

        if side == o_sd.BUY:
            self.long_order = order
        else:
            self.short_order = order

# indie:lang_version = 5
from math import isnan
from indie import strategy, param
from indie.algorithms import Rsi
from indie.math import cross_over, cross_under
from indie.strategies import order_side


@strategy('RSI Strategy', overlay_main_pane=True)  # Relative Strength Index Strategy
@param.int('rsi_length', default=14, title='RSI Length', min=1)
@param.float('oversold', default=30.0, title='Oversold', min=0.0, max=100.0, step=0.1)
@param.float('overbought', default=70.0, title='Overbought', min=0.0, max=100.0, step=0.1)
@param.float('order_size', default=10.0, title='Order size', min=0.1, max=100.0)
@param.str('order_size_unit', default='% of cash', title='Order size unit', options=['% of cash', 'quantity'])
def Main(self, rsi_length, oversold, overbought, order_size, order_size_unit):
    rsi = Rsi.new(self.close, rsi_length)
    pos_size = self.trading.position.size
    desired_pos_size = order_size if order_size_unit == 'quantity' else max(0.0, self.trading.cash * order_size / 100 / self.close[0])
    if not isnan(rsi[0]):
        if cross_over(rsi, oversold) and pos_size <= 0:
            # reverse previous position if exists or open a new one
            self.trading.place_order(order_side.BUY, size=desired_pos_size + abs(pos_size)).submit()
        elif cross_under(rsi, overbought) and pos_size >= 0:
            self.trading.place_order(order_side.SELL, size=desired_pos_size + abs(pos_size)).submit()