Dashboard - How to backtest custom script

[Nicoalz]

Hi, Im new to Hummingbot and Im trying to backtest a custom script I've made.
The goal of the the strategy would be to check prices differences between 2 perp exchanges and if the spread is more than X%, I open a long on exchangeA and a short on exchangeB (or the opposite) and when the price spread is less than Y% I close the trade.
So I would like to backtest it to find good values and also understand how all of this work.
In docs I can't find custom script being backtesting instead of https://github.com/hummingbot/hummingbot/blob/master/scripts/utility/backtest_mm_example.py
But it's not a real script executing strategy.
I also watched videos on the Dashboard and Deploy repos but it seems to be backtesting on pre existing strategies.
So I would like to know if I can backtest my strategy using my script or I should code another one which has the same logic but which is not using the function like buy and sell etc,
At the beginning I thought I could launch a command, like backtest --script ... and it will replace the 'current data' with candles or any test data and also automaticly understand that the buy should be a fake buy etc to backtest.
Currently, I can launch the script with paper trades using Hummingbot, it's working but I'm looking for backtesting it.

Can anyone help me on all of this ?

Details:
Platform: macOS 14.4.1
IDE: PyCharm
Humming bot version = "20240703"

I launch script with
start --script custom_script.py
And the file is in :
scripts/custom_script.py

Here is my code (not a python expert btw) :

from decimal import Decimal
from typing import Dict, Optional

from hummingbot.core.data_type.common import OrderType
from hummingbot.strategy.script_strategy_base import ScriptStrategyBase

class OpportunityInfo:
    def __init__(self, name: str, long_pair: str, short_pair: str, long_exchange: str, short_exchange: str,
                 long_price: Decimal, short_price: Decimal, spread: Decimal, is_spread_above_threshold: bool, is_longA_shortB: bool):
        self.name = name
        self.long_pair = long_pair
        self.short_pair = short_pair
        self.long_exchange = long_exchange
        self.short_exchange = short_exchange
        self.long_price = long_price
        self.short_price = short_price
        self.spread = spread
        self.is_spread_above_threshold = is_spread_above_threshold
        self.is_longA_shortB = is_longA_shortB

class CurrentArbitrage:
    def __init__(self, opportunity: OpportunityInfo, long_order_id: str, short_order_id: str, long_amount: Decimal, short_amount: Decimal):
        self.opportunity = opportunity
        self.long_order_id = long_order_id
        self.short_order_id = short_order_id
        self.long_amount = long_amount
        self.short_amount = short_amount


class PerpPriceSpreadArbitrage(ScriptStrategyBase):
    exchange_A = "binance_paper_trade"
    exchange_B = "kucoin_paper_trade"
    base = "BTC"
    quote_A = "USDT"
    quote_B = "USDT" # in cases where it could be different
    trading_pair_A = f"{base}-{quote_A}"
    trading_pair_B = f"{base}-{quote_B}"
    price_spread_open = Decimal("0.002")  # 0.2%
    price_spread_close = Decimal("0.00005")  # 0.005%
    order_amount_usd = Decimal("100")
    current_arbitrage: Optional[CurrentArbitrage] = None
    markets = {exchange_A: {trading_pair_A},
               exchange_B: {trading_pair_B}}

    def on_tick(self):
        if self.current_arbitrage is not None:
            self.handle_current_arbitrage()
            return # Skip tick if there is an active order

        prices_A = self.get_all_prices(self.exchange_A, self.trading_pair_A)
        prices_B = self.get_all_prices(self.exchange_B, self.trading_pair_B)

        opportunity_longA_shortB = self.create_opportunity("longA_shortB", self.exchange_A, self.exchange_B, prices_A, prices_B)
        opportunity_longB_shortA = self.create_opportunity("longB_shortA", self.exchange_B, self.exchange_A, prices_B, prices_A)

        best_opportunity = self.get_best_opportunity(opportunity_longA_shortB, opportunity_longB_shortA)

        if best_opportunity is None:
            return

        if self.should_execute_opportunity(best_opportunity):
            self.log_opportunity(best_opportunity)
            self.execute_arbitrage(best_opportunity)
        else:
            self.logger().info("No profitable opportunities, best spread: %.6f" % best_opportunity.spread)


    def execute_arbitrage(self, opportunity: OpportunityInfo):
        # Place orders
        short_amount = self.get_trade_amount(self.order_amount_usd, opportunity.short_price)
        long_amount = self.get_trade_amount(self.order_amount_usd, opportunity.long_price)
        long_order_id = self.place_market_order(opportunity.long_exchange, opportunity.long_pair, long_amount, True, opportunity.long_price)
        short_order_id = self.place_market_order(opportunity.short_exchange, opportunity.short_pair, short_amount, False, opportunity.short_price)
        self.current_arbitrage = CurrentArbitrage(opportunity, long_order_id, short_order_id, long_amount, short_amount)

    def place_market_order(self, exchange: str, trading_pair: str, amount: Decimal, is_buy: bool, price:Decimal):
        self.logger().info(f"Placing market order on {exchange} for {trading_pair}")
        if is_buy:
            return self.connectors[exchange].buy(trading_pair, amount, OrderType.MARKET, price)
        else:
            return self.connectors[exchange].sell(trading_pair, amount, OrderType.MARKET, price)

    def on_order_filled(self, order_id: str, amount: Decimal, price: Decimal):
        self.logger().warning(f"Order filled: {order_id}: {amount} @{price}")

    def handle_current_arbitrage(self):
        arb_prices = self.get_arb_prices(self.current_arbitrage.opportunity)
        spread = self.calculate_spread(arb_prices[0], arb_prices[1])
        if self.should_close_arbitrage(spread):
            self.close_arbitrage()
        else:
            self.logger().info(f"Current spread: {spread:.6f}, waiting for it to close")

    def close_arbitrage(self):
        self.logger().info("Closing arbitrage")
        self.place_market_order(self.current_arbitrage.opportunity.long_exchange, self.current_arbitrage.opportunity.long_pair, self.current_arbitrage.long_amount, False, self.current_arbitrage.opportunity.long_price)
        self.place_market_order(self.current_arbitrage.opportunity.short_exchange, self.current_arbitrage.opportunity.short_pair, self.current_arbitrage.short_amount, True, self.current_arbitrage.opportunity.short_price)
        self.current_arbitrage = None

    def get_arb_prices(self, opportunity: OpportunityInfo):
        if opportunity.is_longA_shortB:
            long_price = self.connectors[self.exchange_A].get_price(self.trading_pair_A, True)
            short_price = self.connectors[self.exchange_B].get_price(self.trading_pair_B, False)
        else:
            long_price = self.connectors[self.exchange_B].get_price(self.trading_pair_B, True)
            short_price = self.connectors[self.exchange_A].get_price(self.trading_pair_A, False)
        return long_price, short_price

    def get_all_prices(self, exchange: str, trading_pair: str) -> Dict[str, Decimal]:
        return {
            "long": Decimal(self.connectors[exchange].get_price(trading_pair, True)),
            "short": Decimal(self.connectors[exchange].get_price(trading_pair, False))
        }

    def create_opportunity(self, name: str, long_exchange: str, short_exchange: str,
                           long_prices: Dict[str, Decimal], short_prices: Dict[str, Decimal]) -> OpportunityInfo:
        long_price = long_prices["long"]
        short_price = short_prices["short"]
        spread = self.calculate_spread(long_price, short_price)
        is_spread_above_threshold = spread > self.price_spread_open
        is_longA_shortB = name == "longA_shortB"
        if is_longA_shortB:
            trading_pair_long = self.trading_pair_A
            trading_pair_short = self.trading_pair_B
        else:
            trading_pair_long = self.trading_pair_B
            trading_pair_short = self.trading_pair_A

        return OpportunityInfo(
            name=name,
            long_pair=trading_pair_long,
            short_pair=trading_pair_short,
            is_longA_shortB=is_longA_shortB,
            long_exchange=long_exchange,
            short_exchange=short_exchange,
            long_price=long_price,
            short_price=short_price,
            spread=spread,
            is_spread_above_threshold=is_spread_above_threshold
        )

    def should_execute_opportunity(self, opportunity: OpportunityInfo) -> bool:
        return opportunity.spread > self.price_spread_open

    def should_close_arbitrage(self, spread: Decimal) -> bool:
        return spread <= self.price_spread_close

    def log_opportunity(self, opportunity: OpportunityInfo):
        self.logger().info(f"Best opportunity: {opportunity.name}")
        self.logger().info(f"Long on {opportunity.long_exchange}  @{opportunity.long_price}")
        self.logger().info(f"Short on {opportunity.short_exchange} @{opportunity.short_price}")
        self.logger().info(f"Spread: {opportunity.spread:.6f}")

    @staticmethod
    def calculate_spread(long_price: Decimal, short_price: Decimal) -> Decimal:
        return abs(Decimal("1") - long_price / short_price)

    @staticmethod
    def get_best_opportunity(*opportunities: OpportunityInfo) -> Optional[OpportunityInfo]:
        return max(opportunities, key=lambda x: x.spread) if opportunities else None

    @staticmethod
    def get_trade_amount(amount_usd: Decimal, price_usd: Decimal) -> Decimal:
        return amount_usd / price_usd
       ```
        
        

[Oh-Little-Papaya]

me too. how to backtest custom script

[rapcmia]

Hi @Nicoalz good day
Backtesting using custom script is not yet available on our dashboard however its also been discussed that its a good feature to add on later versions. Currently the dev team is hands full on their current tasks and we can't provide a timeframe yet.

Also im moving this issue to dashboard repo 🙏🏼