Some Python Trality examples

BOLLINGER_BANDS.py

'''
In this bot the possibility to use bollinger bands is presented.
If you are new and trying to get started
we recommend checking our https://app.trality.com/masterclass
For more information on each part of a trading bot,
please visit our documentation pages: https://docs.trality.com
'''

'''
Initializing state object and setting number_offset_trades to 0 before the first bot execution.
'''
def initialize(state):
    state.number_offset_trades = 0;


@schedule(interval="1h", symbol="BTCUSDT")
def handler(state, data):

    '''
    1) Compute indicators from data
    '''

    bbands = data.bbands(20, 2)

    # on erroneous data return early (indicators are of NoneType)
    if bbands is None:
        return

    bbands_lower = bbands["bbands_lower"].last
    bbands_upper = bbands["bbands_upper"].last

    current_price = data.close_last

    '''
    2) Fetch portfolio
        > check liquidity (in quoted currency)
        > resolve buy value
    '''

    portfolio = query_portfolio()
    balance_quoted = portfolio.excess_liquidity_quoted
    # we invest only 80% of available liquidity
    buy_value = float(balance_quoted) * 0.80


    '''
    3) Fetch position for symbol
        > has open position
        > check exposure (in base currency)
    '''

    position = query_open_position_by_symbol(data.symbol,include_dust=False)
    has_position = position is not None

    '''
    4) Resolve buy or sell signals
        > create orders using the order api
        > print position information
    '''

    if current_price < bbands_lower and not has_position:
        print("-------")
        print("Buy Signal: creating market order for {}".format(data.symbol))
        print("Buy value: ", buy_value, " at current market price: ", data.close_last)

        order_market_value(symbol=data.symbol, value=buy_value)

    elif current_price > bbands_upper and has_position:
        print("-------")
        log_msg = "Sell Signal: closing {} position with exposure {} at current market price {}"
        print(log_msg.format(data.symbol,float(position.exposure),data.close_last))

        close_position(data.symbol)

    '''
    5) Check strategy profitability
        > print information profitability on every offsetting trade
    '''

    if state.number_offset_trades < portfolio.number_of_offsetting_trades:

        pnl = query_portfolio_pnl()
        print("-------")
        print("Accumulated Pnl of Strategy: {}".format(pnl))

        offset_trades = portfolio.number_of_offsetting_trades
        number_winners = portfolio.number_of_winning_trades
        print("Number of winning trades {}/{}.".format(number_winners,offset_trades))
        print("Best trade Return : {:.2%}".format(portfolio.best_trade_return))
        print("Worst trade Return : {:.2%}".format(portfolio.worst_trade_return))
        print("Average Profit per Winning Trade : {:.2f}".format(portfolio.average_profit_per_winning_trade))
        print("Average Loss per Losing Trade : {:.2f}".format(portfolio.average_loss_per_losing_trade))
        # reset number offset trades
        state.number_offset_trades = portfolio.number_of_offsetting_trades

EMA_CROSS_AND_RSI.py

'''
In this bot an example of using the exponential moving average in combination with RSI is presented.
If you are new and trying to get started we recommend checking our https://app.trality.com/masterclass
For more information on each part of a trading bot, please visit our documentation pages: https://docs.trality.com
'''

'''
Initializing state object and setting number_offset_trades to 0 before the first bot execution.
'''
def initialize(state):
    state.number_offset_trades = 0;


@schedule(interval="1h", symbol="BTCUSDT")
def handler(state, data):
    '''
    1) Compute indicators from data
    '''

    ema_long_ind = data.ema(40)
    ema_short_ind = data.ema(20)
    rsi_ind = data.rsi(14)

    # on erroneous data return early (indicators are of NoneType)
    if rsi_ind is None or ema_long_ind is None or ema_short_ind is None:
        return

    ema_long = ema_long_ind.last # Getting latest value for ema_long_ind from data object
    ema_short = ema_short_ind.last
    rsi = rsi_ind.last

    current_price = data.close_last

    '''
    2) Fetch portfolio
        > check liquidity (in quoted currency)
        > resolve buy value
    '''

    portfolio = query_portfolio()
    balance_quoted = portfolio.excess_liquidity_quoted
    # we invest only 80% of available liquidity
    buy_value = float(balance_quoted) * 0.80


    '''
    3) Fetch position for symbol
        > has open position
        > check exposure (in base currency)
    '''

    position = query_open_position_by_symbol(data.symbol,include_dust=False)
    has_position = position is not None

    '''
    4) Resolve buy or sell signals
        > create orders using the order api
        > print position information

    '''
    if ema_short > ema_long and rsi < 30 and not has_position:
        print("-------")
        print("Buy Signal: creating market order for {}".format(data.symbol))
        print("Buy value: ", buy_value, " at current market price: ", data.close_last)

        order_market_value(symbol=data.symbol, value=buy_value)  # creating market order

    elif ema_short < ema_long and rsi > 70 and has_position:
        print("-------")
        logmsg = "Sell Signal: closing {} position with exposure {} at current market price {}"
        print(logmsg.format(data.symbol,float(position.exposure),data.close_last))

        close_position(data.symbol)  # closing position

    '''
    5) Check strategy profitability
        > print information profitability on every offsetting trade
    '''

    if state.number_offset_trades < portfolio.number_of_offsetting_trades:

        pnl = query_portfolio_pnl()
        print("-------")
        print("Accumulated Pnl of Strategy: {}".format(pnl))

        offset_trades = portfolio.number_of_offsetting_trades
        number_winners = portfolio.number_of_winning_trades
        print("Number of winning trades {}/{}.".format(number_winners,offset_trades))
        print("Best trade Return : {:.2%}".format(portfolio.best_trade_return))
        print("Worst trade Return : {:.2%}".format(portfolio.worst_trade_return))
        print("Average Profit per Winning Trade : {:.2f}".format(portfolio.average_profit_per_winning_trade))
        print("Average Loss per Losing Trade : {:.2f}".format(portfolio.average_loss_per_losing_trade))
        # reset number offset trades
        state.number_offset_trades = portfolio.number_of_offsetting_trades

MACD_CROSS.py

'''
In this bot an examle using MACD Crossover is presented.
If you are new and trying to get started we recommend checking our https://app.trality.com/masterclass
For more information on each part of a trading bot, please visit our documentation pages: https://docs.trality.com
'''

'''
Initializing state object and setting number_offset_trades to 0 before the first bot execution.
'''
def initialize(state):
    state.number_offset_trades = 0;


@schedule(interval="1h", symbol="BTCUSDT")
def handler(state, data):

    '''
    1) Compute indicators from data
    '''

    macd_ind = data.macd(12,26,9)

    # on erroneous data return early (indicators are of NoneType)
    if macd_ind is None:
        return

    signal = macd_ind['macd_signal'].last
    macd = macd_ind['macd'].last

    current_price = data.close_last

    '''
    2) Fetch portfolio
        > check liquidity (in quoted currency)
        > resolve buy value
    '''

    portfolio = query_portfolio()
    balance_quoted = portfolio.excess_liquidity_quoted
    # we invest only 80% of available liquidity
    buy_value = float(balance_quoted) * 0.80

    '''
    3) Fetch position for symbol
        > has open position
        > check exposure (in base currency)
    '''

    position = query_open_position_by_symbol(data.symbol,include_dust=False)
    has_position = position is not None

    '''
    4) Resolve buy or sell signals
        > create orders using the order api
        > print position information

    '''
    if macd > signal and not has_position:
        print("-------")
        print("Buy Signal: creating market order for {}".format(data.symbol))
        print("Buy value: ", buy_value, " at current market price: ", data.close_last)
        order_market_value(symbol=data.symbol, value=buy_value) # creating market order

    elif macd < signal and has_position:
        print("-------")
        logmsg = "Sell Signal: closing {} position with exposure {} at current market price {}"
        print(logmsg.format(data.symbol,float(position.exposure),data.close_last))
        close_position(data.symbol) # closing position


    '''
    5) Check strategy profitability
        > print information profitability on every offsetting trade
    '''

    if state.number_offset_trades < portfolio.number_of_offsetting_trades:

        pnl = query_portfolio_pnl()
        print("-------")
        print("Accumulated Pnl of Strategy: {}".format(pnl))

        offset_trades = portfolio.number_of_offsetting_trades
        number_winners = portfolio.number_of_winning_trades
        print("Number of winning trades {}/{}.".format(number_winners,offset_trades))
        print("Best trade Return : {:.2%}".format(portfolio.best_trade_return))
        print("Worst trade Return : {:.2%}".format(portfolio.worst_trade_return))
        print("Average Profit per Winning Trade : {:.2f}".format(portfolio.average_profit_per_winning_trade))
        print("Average Loss per Losing Trade : {:.2f}".format(portfolio.average_loss_per_losing_trade))
        # reset number offset trades
        state.number_offset_trades = portfolio.number_of_offsetting_trades

MARGIN.py

'''
The code example is a long short QQE strategy using the new Margin Trading functionality on Trality.
This strategy can only be deployed on Binance accounts with Margin Trading functionality.

Learn more about Margin Trading in our documentation where you can find more example strategies: https://docs.trality.com/trality-code-editor/margin
If you are new and trying to get started we recommend checking our https://app.trality.com/masterclass
'''

TRADE_SIZE = 500
MIN_TRADE_SIZE = 50

@enable_margin_trading()
def initialize(state):
    state.buy_order = None
    state.sell_order = None

@schedule(interval="1d", symbol="BTCUSDT", window_size=200)
def handler(state, data):
    portfolio = query_portfolio()
    pos = query_open_position_by_symbol(data.symbol)

    # use qqe indicator as the trend
    trend = -1 if data.qqe(21, 5, 4.2)["trend"][-1]  < 0 else +1

    exposure = 0.0 if pos is None else float(pos.exposure)

    # plot data
    plot_line("leverage", query_position_weight(data.symbol), data.symbol)
    plot_line("exposure", exposure, data.symbol)
    plot_line("value", portfolio.portfolio_value, data.symbol)

    if trend > 0:
        target_pos = TRADE_SIZE / data.close[-1]
        trade_size = target_pos - exposure

        # check trade is size is large enough to be accepted by the exchange
        if trade_size * data.close[-1] > +MIN_TRADE_SIZE :
            print(f"buying {trade_size:.5f} {data.base}")
            state.buy_order = margin_order_market_amount(symbol = "BTCUSDT", amount = trade_size,
                                                    side_effect = OrderMarginSideEffect.AutoDetermine)

    if trend < 0:
        target_pos = -TRADE_SIZE / data.close[-1]
        trade_size = target_pos - exposure

        # check trade is size is large enough to be accepted by the exchange
        if trade_size * data.close[-1] < -MIN_TRADE_SIZE :
            print(f"selling {trade_size:.5f} {data.base}")
            state.sell_order = margin_order_market_amount(symbol = "BTCUSDT", amount = trade_size,
                                                    side_effect = OrderMarginSideEffect.AutoDetermine)

OPTIMIZER.py

'''
The code example is an ema cross over bot with 2 parameters ema_period_short and ema_period_long to be optimized.
If the following code is optimized with 3 segments then a total of 9 versions of the strategy will be back tested.
The final backtest will have the results of the backtest with the best objective function.

Make sure to activate the Optimizer inside the Advanced Settings of the backtester on the right hand-side.

Learn more about the Optimizer in our documentation: https://docs.trality.com/trality-code-editor/api-documentation/optimizer
If you are new and trying to get started we recommend checking our https://app.trality.com/masterclass
'''

@parameter(name="ema_period_short", type="float", default=20, min=15, max=25)
@parameter(name="ema_period_long", type="float", default=40, min=35, max=50)
def initialize(state, params):
    print(f"running long: {params.ema_period_long}, short: {params.ema_period_short}")

@schedule(interval="1h", symbol="BTCUSDT", window_size=200)
def handler(state, data, params):

    ema_short = data.ema(params.ema_period_short).last
    ema_long = data.ema(params.ema_period_long).last

    position = query_open_position_by_symbol(data.symbol,include_dust=False)
    has_position = position is not None

    if ema_short > ema_long and not has_position:
        order_market_target(symbol=data.symbol, target_percent=0.8)

    elif ema_short < ema_long and has_position:
        close_position(data.symbol)

SKELETON.py

'''
Welcome to the Base-Skeleton of the code-based Trality Bot Creator.
If you are new and trying to get started
we recommend checking our https://app.trality.com/masterclass
For more information on each part of a trading bot,
please visit our documentation pages: https://docs.trality.com
'''

'''
Each bot can have one optional initializer function. These take
a single argument containing a python dictionary that you can write
and read to - values stored in this state object will always persist
throughout the lifetime of a bot.
'''
def initialize(state):
    state.run = 0

'''
A bot can specify one or multiple handler functions for different data resolutions,
and one or more symbols that the handler function should operate on. In order for a
function to be recognized as a handler function it must be decorated with a
schedule decorator. Note the selection of a time interval and symbol as arguments
of the decorator.
'''
@schedule(interval="1h", symbol="BTCUSDT")
def handler(state, data):

    '''
    Within the handler, you can then call the variables you wish to use and define the logic with which the bot should operate.
    Each handlers receives two arguments: state and data. While the state object is the same dictionary you will receive during initialization,
    data is filled with the data you request in the decorator.
    '''

    '''
    You can access a variety of different data and indicators from the data object, e.g. the last closing price
    of the candle within the timeframe you requested.
    '''
    last_closing_price = data.close_last

    '''
    Lastly you can also use the log function to write to the console, make orders, query your balance, etc.
    Feel free to experiment with your own ideas!
    '''
    if state.run % 2 == 0:
        print("-- run {}".format(state.run))
        print("closing price {}".format(last_closing_price))
    state.run += 1

TWO_INTERVALS.py

'''
In this bot the possibility to use different intervals of the same symbols is presented.
If you are new and trying to get started we recommend checking our https://app.trality.com/masterclass
For more information on each part of a trading bot, please visit our documentation pages: https://docs.trality.com
'''

'''
In this example we will be using the state object which can be used to pass information between handlers.
Here, we initialize a variable indicating the current trend on the state object directly.
'''
def initialize(state):
    state.trend = None
    state.number_offset_trades = 0

'''
handler_long is used in this context to identify trends in a higher interval. This information is then
passed along using the state object.
'''
@schedule(interval= "1h", symbol="BTCUSDT")
def handler_long(state, data):
    ema_long_ind = data.ema(40)
    ema_short_ind = data.ema(15)

    # on erroneous data return early (indicators are of NoneType)
    if ema_long_ind is None or ema_short_ind is None:
        return

    ema_long = ema_long_ind.last
    ema_short = ema_short_ind.last


    if ema_short > ema_long:
        state.trend = "upward"
    else:
        state.trend = "downward"

'''
handler_short operates on a smaller interval than handler_long and uses the trends from the higher
interval to create orders.
'''
@schedule(interval = "15m", symbol = "BTCUSDT")
def handler_short(state,data):
    rsi_ind = data.rsi(14)

    # again check on erroneous data return early (indicator is of NoneType)
    if rsi_ind is None:
        return

    rsi = rsi_ind.last

    # this code block is querying the portfolio to see if it has any open positions,
    # querying the amount of quoted balance we have free
    # and the amount of the traded asset that we have free.
    portfolio = query_portfolio()
    has_position = has_open_position(data.symbol, include_dust=False)
    balance_quoted = float(query_balance_free(data.quoted))
    balance_base = float(query_balance_free(data.base))
    buy_amount = balance_quoted / data.close_last

    # if conditions are met place buy order
    if state.trend == "upward" and rsi < 40 and balance_base<buy_amount and not has_position:
        print("Buying {} {}@{}".format(data.symbol, buy_amount, data.close_last))
        order_market_amount(symbol=data.symbol, amount=buy_amount)

    # if conditions are met close position
    if state.trend == "downward" and rsi > 60 and has_position:
        print("Selling {} {}@{}".format(data.symbol, balance_base, data.close_last))
        close_position(data.symbol)

    '''
    5) Check strategy profitability
        > print information profitability on every offsetting trade
    '''

    if state.number_offset_trades < portfolio.number_of_offsetting_trades:

        pnl = query_portfolio_pnl()
        print("-------")
        print("Accumulated Pnl of Strategy: {}".format(pnl))

        offset_trades = portfolio.number_of_offsetting_trades
        number_winners = portfolio.number_of_winning_trades
        print("Number of winning trades {}/{}.".format(number_winners,offset_trades))
        print("Best trade Return : {:.2%}".format(portfolio.best_trade_return))
        print("Worst trade Return : {:.2%}".format(portfolio.worst_trade_return))
        print("Average Profit per Winning Trade : {:.2f}".format(portfolio.average_profit_per_winning_trade))
        print("Average Loss per Losing Trade : {:.2f}".format(portfolio.average_loss_per_losing_trade))
        # reset number offset trades
        state.number_offset_trades = portfolio.number_of_offsetting_trades

TWO_PARAMETERS.py

'''
In this bot template the possibility to use different symbols is represented.
If you are new and trying to get started we recommend checking our https://app.trality.com/masterclass
For more information on each part of a trading bot, please visit our documentation pages: https://docs.trality.com
The selection of the given symbols in this example is arbitrary. This could have been made following some data analyses or
i.e. analyzing the correlations between these two symbols.
The logic is simple: a general signal is computed for each asset used in this bot,
and orders are made for a single asset in case the signal agrees for all symbols,
In this example, we use a combination of EMA crossover and RSI, and invest roughly half our portfolio
into a single asset.
'''

SYMBOLS = ["BTCUSDT", "ETHUSDT"]

SIGNAL_BUY = 1
SIGNAL_SELL = 2
SIGNAL_IGNORE = 3

def initialize(state):
   state.signals = {}  # This is where signals will be stored for each symbol
   state.signal_parameters = [20, 40, 14]

# Compute a signal for a specific symbol pair
def compute_signal(data, short_n, long_n, rsi_n):
    # Computing indicators from data
    ema_short_ind = data.ema(short_n)
    ema_long_ind = data.ema(long_n)
    rsi_ind = data.rsi(rsi_n)

    # On erroneous data return early (indicators are of NoneType)
    if ema_short_ind is None or ema_long_ind is None or rsi_ind is None:
        return

    ema_short = ema_short_ind.last  # Getting latest value for ema_short from data object
    ema_long = ema_long_ind.last
    rsi = rsi_ind.last

    if ema_short > ema_long and rsi < 40:
        signal = SIGNAL_BUY
    elif ema_short < ema_long and rsi > 60:
        signal = SIGNAL_SELL
    else:
        signal = SIGNAL_IGNORE
    return signal

# Store signal in state
def resolve_ema_signal(state, data):
    state.signals[data.symbol] = compute_signal(data, *state.signal_parameters)

# Check if all signals agree
def resolve_action(state):
    signals = list(state.signals.values())
    if all(x == SIGNAL_BUY for x in signals):
        return SIGNAL_BUY
    elif all(x == SIGNAL_SELL for x in signals):
        return SIGNAL_SELL
    else:
        return SIGNAL_IGNORE

@schedule(interval= "1h", symbol=SYMBOLS)
def handler(state, dataMap):

    # Resolve all signals
    for symbol, data in dataMap.items():
        resolve_ema_signal(state, data)

    # Resolve action
    action = resolve_action(state)

    # Skip early
    if action == SIGNAL_IGNORE:
        return

    # Let's say we're buying/selling ETH if BTC agrees
    data = dataMap[SYMBOLS[1]]

    print("Resolved {} signal for {}".format("BUY" if action is SIGNAL_BUY else "SELL", data.symbol))

    # This code block is querying the portfolio to see if it has any open positions,
    # querying the amount of quoted balance we have free
    # and the amount of the traded asset that we have free.
    has_position = has_open_position(data.symbol, include_dust=False)
    balance_base = float(query_balance_free(data.base))
    balance_quoted = float(query_balance_free(data.quoted))
    buy_amount = balance_quoted / data.close_last * 0.5

    # Depending on given signals sets buy or sell order
    if action == SIGNAL_BUY and balance_base<buy_amount and not has_position:
        print("-> Buying {}".format(data.symbol))
        order_market_amount(symbol=data.symbol,amount=buy_amount)
    elif action == SIGNAL_SELL and has_position:
        print("-> Selling {}".format(data.symbol))
        close_position(data.symbol)

Also an OCO example would be nice

from enum import Enum

class Mode(Enum):
    Buy = 1
    Sell = 2

@enable_margin_trading()
def initialize(state):
    state.order = None
    state.tp_order = None
    state.sl_order = None

    state.mode = Mode.Buy

def has_position(data):
    free = query_balance_free(data.base)
    locked = query_balance_locked(data.base)
    borrowed = query_balance_borrowed(data.base)
    interest = query_balance_interest(data.base)

    # calculate risk exposure
    exposure = free+locked-borrowed-interest

    price = Decimal(f"{data.close[-1]}")
    limit = symbol_limits(data.symbol)
    has_exposure = abs(exposure) * price > Decimal(limit.costMin)
    has_liabilities = (borrowed+interest) * price > Decimal(limit.costMin)

    return has_exposure or has_liabilities

@schedule(interval="1h", symbol="ETHBUSD")
def handler(state, data):

    has_pos = has_position(data)
    if not has_pos:

        free = query_balance_free(data.base)
        borrowed = query_balance_borrowed(data.base)
        interest = query_balance_interest(data.base)
        if (borrowed + interest) > 0 and free > (borrowed + interest):
            # clean up any residual borrowed amount by repaying loans
            print(f"repaying {data.base} borrowed {borrowed} interest: {interest}")
            margin_repay(data.base, borrowed + interest)
        elif state.mode == Mode.Buy:
            state.mode = Mode.Sell
            # buy margin order with borrowing
            value = float(query_portfolio_value()) * 1.1
            state.order = margin_order_market_value(symbol = data.symbol, value = value, side_effect = OrderMarginSideEffect.AutoDetermine)

            with OrderScope.one_cancels_others():
                # adjust for fees since purchase amount will be smaller
                order_qty = subtract_order_fees(state.order.quantity)
                tp_price = data.close[-1] * 1.01
                sl_price = data.close[-1] / 1.01

                # exit orders should repay.
                print("long stops", state.order.quantity, order_qty)
                state.tp_order = margin_order_iftouched_market_amount(symbol = data.symbol, amount = -order_qty, stop_price=tp_price,
                                                    side_effect = OrderMarginSideEffect.Repay)
                state.sl_order = margin_order_iftouched_market_amount(symbol = data.symbol, amount = -order_qty, stop_price=sl_price,
                                                    side_effect = OrderMarginSideEffect.Repay)

        elif state.mode == Mode.Sell:
            state.mode = Mode.Buy
            # sell short with borrowing
            value = float(query_portfolio_value()) * -0.5
            state.order = margin_order_market_value(symbol = data.symbol, value = value, side_effect = OrderMarginSideEffect.AutoDetermine)

            with OrderScope.one_cancels_others():
                # fees of sale are paid in quoted so no adjustment for fees is needed
                order_qty = state.order.quantity
                tp_price = data.close[-1] / 1.01   # take profit is below price
                sl_price = data.close[-1] * 1.01   # stop loss is above current price

                # exit orders should repay.
                # exit orders should be buys to close
                # note: this will leave dust when the position is closed due to accumulated interest
                print("short stops", state.order.quantity, order_qty)
                state.tp_order = margin_order_iftouched_market_amount(symbol = data.symbol, amount = order_qty, stop_price=tp_price,
                                                    side_effect = OrderMarginSideEffect.Repay)
                state.sl_order = margin_order_iftouched_market_amount(symbol = data.symbol, amount = order_qty, stop_price=sl_price,
                                                    side_effect = OrderMarginSideEffect.Repay)

Most examples should be modified to use f-strings - My 2cts