slightly modified dynamic spread script

dynamic_spread_and_band_script.py

from decimal import Decimal
from datetime import datetime

import time
from hummingbot.script.script_base import ScriptBase
from hummingbot.core.event.events import BuyOrderCompletedEvent, SellOrderCompletedEvent
from os.path import realpath, join

s_decimal_1 = Decimal("1")
LOGS_PATH = realpath(join(__file__, "../../logs/"))
SCRIPT_LOG_FILE = f"{LOGS_PATH}/logs_script.log"


def log_to_file(file_name, message):
    with open(file_name, "a+") as f:
        f.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S") + " - " + message + "\n")

class InventoryCost:

    def __init__(self, script_base):
        self.script_base = script_base
        self.base_asset, self.quote_asset = self.script_base.pmm_market_info.trading_pair.split("-")
        self.base_balance = Decimal("0")
        self.quote_balance = Decimal("0")
        self.update_balances()
        self.update_inv_values()

        # initialize start cost At start assume current value as cost
        self.current_inv_cost = self.base_inv_value
        self.current_avg_cost = self.script_base.mid_price


    def update_balances(self):
        # Check what is the current balance of each asset
        market_info = self.script_base.pmm_market_info.exchange
        self.base_balance = self.script_base.all_total_balances[market_info].get(self.base_asset, self.base_balance)
        self.quote_balance = self.script_base.all_total_balances[market_info].get(self.quote_asset, self.quote_balance)

    def update_inv_values(self):
        # calculate the current value and it's proportion
        self.base_inv_value = Decimal(self.base_balance * self.script_base.mid_price)
        self.total_inv_value = Decimal(self.base_inv_value + self.quote_balance)

    def inventory_buy(self, cost, amount):
        self.current_inv_cost += Decimal(cost)
        self.base_balance += Decimal(amount)
        self.current_avg_cost = Decimal(self.current_inv_cost / self.base_balance)

    def inventory_sell(self, cost, amount):
        cost = Decimal(cost)
        if self.current_inv_cost < cost:
            self.current_inv_cost = 0
        else:
            self.current_inv_cost -= cost
        self.base_balance -= Decimal(amount)
        if self.base_balance == 0:
            self.current_avg_cost = Decimal("0")
        else:
            self.current_avg_cost = Decimal(self.current_inv_cost / self.base_balance)

    @property
    def current_cost(self):
        return self.current_avg_cost


class SpreadsAdjustedOnVolatility(ScriptBase):
    """
    Demonstrates how to adjust bid and ask spreads based on price volatility.
    The volatility, in this example, is simply a price change compared to the previous cycle regardless of its
    direction, e.g. if price changes -3% (or 3%), the volatility is 3%.
    To update our pure market making spreads, we're gonna smooth out the volatility by averaging it over a short period
    (short_period), and we need a benchmark to compare its value against. In this example the benchmark is a median
    long period price volatility (you can also use a fixed number, e.g. 3% - if you expect this to be the norm for your
    market).
    For example, if our bid_spread and ask_spread are at 0.8%, and the median long term volatility is 1.5%.
    Recently the volatility jumps to 2.6% (on short term average), we're gonna adjust both our bid and ask spreads to
    1.9%  (the original spread - 0.8% plus the volatility delta - 1.1%). Then after a short while the volatility drops
    back to 1.5%, our spreads are now adjusted back to 0.8%.
    """

    # Let's set interval and sample sizes as below.
    # These numbers are for testing purposes only (in reality, they should be larger numbers)
    # interval is a interim which to pick historical mid price samples from, if you set it to 5, the first sample is
    # the last (current) mid price, the second sample is a past mid price 5 seconds before the last, and so on.
    interval = 5
    # short_period is how many interval to pick the samples for the average short term volatility calculation,
    # for short_period of 3, this is 3 samples (5 seconds interval), of the last 15 seconds
    short_period = 3
    # long_period is how many interval to pick the samples for the median long term volatility calculation,
    # for long_period of 10, this is 10 samples (5 seconds interval), of the last 50 seconds
    long_period = 10
    last_stats_logged = 0
    volatility_granularity = Decimal("0.0002")

    # Let's set the upper bound of the band to 0.05% away from the inventory cost price
    band_upper_bound_pct = Decimal("0.0005")
    # turn off buy band this will use max_sell_price
    use_max_price = True
    # Let's set the lower bound of the band to 0.02% away from the inventory cost price
    band_lower_bound_pct = Decimal("0.0002")
    # Let's sample mid prices once every 5 seconds
    avg_interval = 5
    # Let's average the last 3 samples
    avg_length = 6

    # how to handle price that drops below cost, valid values (stop/spread)
    sell_protect_method = "spread"
    bid_protect_method = "spread"

    def __init__(self):
        super().__init__()
        self.original_bid_spread = None
        self.original_ask_spread = None
        self.modified_bid_spread = None
        self.modified_ask_spread = None
        self.avg_inventory_cost = None
        self.avg_short_volatility = None
        self.median_long_volatility = None
        self.band_upper_bound = None
        self.band_lower_bound = None
        self.inventory_cost = None
        self.price_source = None
        self.sell_override_spread = None
        self.bid_override_spread = None
        self.max_sell_price = None

    def volatility_msg(self, include_mid_price=False):
        if self.avg_short_volatility is None or self.median_long_volatility is None:
            return f"short_volatility: N/A  long_volatility: N/A"
        uband_msg = ""
        lband_msg = ""
        if self.band_upper_bound:
            uband_msg = f"upper_bound_price: {self.band_upper_bound:.8g} "
        if self.max_sell_price:
            uband_msg = f"{uband_msg}max_sell_price: {self.max_sell_price:.8g}"
        if self.band_lower_bound:
            lband_msg = f"lower_bound_price: {self.band_lower_bound:.8g} "
        if self.inventory_cost:
            lband_msg = f"{lband_msg}inventory cost: {self.inventory_cost.current_cost:.8g}"

        msgs = [
            f"short_volatility: {self.avg_short_volatility:.2%}  " \
               f"long_volatility: {self.median_long_volatility:.2%}",
            f"avg_mid_price: {self.avg_mid_price(self.avg_length, self.avg_interval):<15}",
            f"original_bid_spread: {self.original_bid_spread:.2%}" \
               f" original_ask_spread: {self.original_ask_spread:.2%}",
            f"mod_bid_spread: {self.modified_bid_spread:.2%}" \
               f" mod_ask_spread: {self.modified_ask_spread:.2%}",
            f"{uband_msg}",
            f"{lband_msg}",
            f"buy_levels: {self.pmm_parameters.buy_levels}  sell_levels: {self.pmm_parameters.sell_levels}"
        ]
        return "\n".join([msg for msg in msgs if msg])

    def _initialize_spreads(self):
        # First, let's keep the original spreads.
        if self.original_bid_spread is None:
            self.original_bid_spread = self.modified_bid_spread = self.pmm_parameters.bid_spread
        if self.original_ask_spread is None:
            self.original_ask_spread = self.modified_ask_spread = self.pmm_parameters.ask_spread

        # check for user changes to spread
        if self.modified_bid_spread != self.pmm_parameters.bid_spread:
            self.original_bid_spread = self.pmm_parameters.bid_spread

        if self.modified_ask_spread != self.pmm_parameters.ask_spread:
            self.original_ask_spread = self.pmm_parameters.ask_spread

    def _calculate_volatility(self):
        # Average volatility (price change) over a short period of time, this is to detect recent sudden changes.
        self.avg_short_volatility = self.avg_price_volatility(self.interval, self.short_period)
        # Median volatility over a long period of time, this is to find the market norm volatility.
        # We use median (instead of average) to find the middle volatility value - this is to avoid recent
        # spike affecting the average value.
        self.median_long_volatility = self.median_price_volatility(self.interval, self.long_period)

    def _initialize_inventory_cost(self):
        # init the cost_price delegate if not set up
        if self.inventory_cost is None:
            self.inventory_cost = InventoryCost(self)

    def _adjust_spreads(self, spread_adjustment, check_price):
        # Show the user on what's going, you can remove this statement to stop the notification.
        #self.notify(f"avg_short_volatility: {self.avg_short_volatility} median_long_volatility: {self.median_long_volatility} "
        #             f"spread_adjustment: {spread_adjustment}")
        new_bid_spread = self.original_bid_spread + spread_adjustment
        # Let's not set the spreads below the originals, this is to avoid having spreads to be too close
        # to the mid price.
        old_bid_spread = self.pmm_parameters.bid_spread
        self._adjust_bid_spread(new_bid_spread, check_price)

        old_ask_spread = self.pmm_parameters.ask_spread
        new_ask_spread = self.original_ask_spread + spread_adjustment
        self._adjust_ask_spread(new_ask_spread, check_price)

        if old_bid_spread != new_bid_spread or old_ask_spread != new_ask_spread:
            #self.log(self.volatility_msg(True))
            log_to_file(SCRIPT_LOG_FILE, self.volatility_msg(True))
            log_to_file(SCRIPT_LOG_FILE, f"spreads adjustment: Old Value: {old_bid_spread:.2%} "
                                         f"New Value: {new_bid_spread:.2%}")


    def _adjust_ask_spread(self, new_spread, check_price):
        if self.sell_override_spread and self.sell_override_spread > new_spread:
            new_spread = self.sell_override_spread
            if new_spread != self.pmm_parameters.ask_spread:
                self.notify(f"avg_mid_price ({check_price:.8g}) is below lower_band ({self.band_lower_bound:.8g}), adjusting sell spread to {new_spread:.2%}")
        else:
            self.sell_override_spread = None

        new_ask_spread = max(self.original_ask_spread, new_spread)
        if new_ask_spread != self.pmm_parameters.ask_spread:
            self.modified_ask_spread = self.pmm_parameters.ask_spread = new_ask_spread
            return True
        else:
            return False

    def _adjust_bid_spread(self, new_spread, check_price):
        if self.bid_override_spread and self.bid_override_spread > new_spread:
            new_spread = self.bid_override_spread
            if new_spread != self.pmm_parameters.bid_spread:
                self.notify(f"avg_mid_price ({check_price:.8g}) is above upper_band ({self.band_upper_bound:.8g}), adjusting bid spread to {new_spread:.2%}")
        else:
            self.bid_override_spread = None

        new_bid_spread = max(self.original_bid_spread, new_spread)
        if new_bid_spread != self.pmm_parameters.bid_spread:
            self.modified_bid_spread = self.pmm_parameters.bid_spread = new_bid_spread
            return True
        else:
            return False

    def _calculate_sell(self, check_price):
        if check_price <= self.band_lower_bound:
            if self.sell_protect_method == "spread":
                price_diff = self.band_lower_bound - check_price
                spread_pct = price_diff / self.band_lower_bound
                self.sell_override_spread = self.round_by_step(spread_pct, Decimal("0.0001"))
            else:
                self._stop_sells(check_price, "lower_band", self.band_lower_bound)

        else:
            self._resume_sells()

    def _calculate_buy(self, check_price):
        if check_price >= self.band_upper_bound:
            if self.bid_protect_method == "spread":
                price_diff = check_price - self.band_upper_bound
                spread_pct = price_diff / self.band_upper_bound
                self.bid_override_spread = self.round_by_step(spread_pct, Decimal("0.0001"))
            else:
                self._stop_buys(check_price, "upper_band", self.band_upper_bound)
        else:
            self._resume_buys()

    def _stop_buys(self, check_price, limit_param_name, limit_param_value):
        if self.pmm_parameters.buy_levels != 0:
            self.pmm_parameters.buy_levels = 0
            self.notify(f"Stopping buys, avg_mid_price ({check_price:.8g}) is above {limit_param_name} ({limit_param_value:.8g})")

    def _stop_sells(self, check_price, limit_param_name, limit_param_value):
        if self.pmm_parameters.sell_levels != 0:
            self.pmm_parameters.sell_levels = 0
            self.notify(f"Stopping sells, avg_mid_price ({check_price:.8g}) is below {limit_param_name} ({limit_param_value:.8g})")

    def _resume_buys(self):
        if self.bid_protect_method == "spread" and self.bid_override_spread:
            self.bid_override_spread = None
            self.notify("resuming buys")
        else:
            if self.pmm_parameters.buy_levels != self.pmm_parameters.order_levels:
                self.pmm_parameters.buy_levels = self.pmm_parameters.order_levels
                self.notify("resuming buys")

    def _resume_sells(self):
        if self.sell_protect_method == "spread" and self.sell_override_spread:
            self.sell_override_spread = None
            self.notify("resuming sells")
        else:
            if self.pmm_parameters.sell_levels != self.pmm_parameters.order_levels:
                self.pmm_parameters.sell_levels = self.pmm_parameters.order_levels
                self.notify("resuming sells")


    def on_tick(self):
        self._initialize_spreads()
        self._initialize_inventory_cost()

        self._calculate_volatility()

        # If the bot just got started, we'll not have these numbers yet as there is not enough mid_price sample size.
        # We'll start to have these numbers after interval * long_term_period (150 seconds in this example).
        if self.avg_short_volatility is None or self.median_long_volatility is None:
            return

        # Let's log some stats once every 5 minutes
        if time.time() - self.last_stats_logged > 60 * 5:
            log_to_file(SCRIPT_LOG_FILE, self.volatility_msg(True))
            self.last_stats_logged = time.time()

        # This volatility delta will be used to adjust spreads.
        delta = self.avg_short_volatility - self.median_long_volatility
        # Let's round the delta into granular volitility increment to ignore noise and to avoid adjusting the spreads too often.
        spread_adjustment = self.round_by_step(delta, self.volatility_granularity)

        avg_mid_price = self.avg_mid_price(self.avg_length, self.avg_interval)
        line_price = self.inventory_cost.current_cost
        self.price_source = "COST"
        if line_price is None:
            line_price = avg_mid_price
            self.price_source = "AVG"
        if avg_mid_price is None:
            avg_mid_price = self.mid_price
            line_price = avg_mid_price
            self.price_source = "MID"

        if self.use_max_price and self.max_sell_price:
            self.band_upper_bound = self.max_sell_price * (s_decimal_1 + self.band_upper_bound_pct)
        else:
            self.band_upper_bound =  avg_mid_price * (s_decimal_1 + self.band_upper_bound_pct)

        self.band_lower_bound = line_price * (s_decimal_1 - self.band_lower_bound_pct)
        # When mid_price reaches the upper bound, we expect the price to bounce back as such we don't want be a buyer
        # (as we can probably buy back at a cheaper price later).
        # If you anticipate the opposite, i.e. the price breaks out on a run away move, you can protect your inventory
        # by stop selling (setting the sell_levels t 0).
        self._calculate_buy(avg_mid_price)

        # When mid_price reaches the lower bound, we don't want to be a seller.
        self._calculate_sell(avg_mid_price)

        self._adjust_spreads(spread_adjustment, avg_mid_price)

    def on_buy_order_completed(self, event: BuyOrderCompletedEvent):
        self.notify(f"buy order completed => ( price: {event.quote_asset_amount}, amount: {event.base_asset_amount} )")
        self.inventory_cost.inventory_buy(event.quote_asset_amount, event.base_asset_amount)
        return

    def on_sell_order_completed(self, event: SellOrderCompletedEvent):
        self.notify(f"sell order completed => ( price: {event.quote_asset_amount}, amount: {event.base_asset_amount} )")
        sell_price = Decimal(event.quote_asset_amount / event.base_asset_amount)
        if self.max_sell_price is None or sell_price > self.max_sell_price:
            self.max_sell_price = sell_price
        self.inventory_cost.inventory_sell(event.quote_asset_amount, event.base_asset_amount)
        return


    def on_status(self) -> str:
        return self.volatility_msg()

spreads_adjusted_on_volatility_script.py

from decimal import Decimal
from datetime import datetime
import time
from hummingbot.script.script_base import ScriptBase
from os.path import realpath, join

s_decimal_1 = Decimal("1")
LOGS_PATH = realpath(join(__file__, "../../logs/"))
SCRIPT_LOG_FILE = f"{LOGS_PATH}/logs_script.log"


def log_to_file(file_name, message):
    with open(file_name, "a+") as f:
        f.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S") + " - " + message + "\n")


class SpreadsAdjustedOnVolatility(ScriptBase):
    """
    Demonstrates how to adjust bid and ask spreads based on price volatility.
    The volatility, in this example, is simply a price change compared to the previous cycle regardless of its
    direction, e.g. if price changes -3% (or 3%), the volatility is 3%.
    To update our pure market making spreads, we're gonna smooth out the volatility by averaging it over a short period
    (short_period), and we need a benchmark to compare its value against. In this example the benchmark is a median
    long period price volatility (you can also use a fixed number, e.g. 3% - if you expect this to be the norm for your
    market).
    For example, if our bid_spread and ask_spread are at 0.8%, and the median long term volatility is 1.5%.
    Recently the volatility jumps to 2.6% (on short term average), we're gonna adjust both our bid and ask spreads to
    1.9%  (the original spread - 0.8% plus the volatility delta - 1.1%). Then after a short while the volatility drops
    back to 1.5%, our spreads are now adjusted back to 0.8%.
    """

    # Let's set interval and sample sizes as below.
    # These numbers are for testing purposes only (in reality, they should be larger numbers)
    # interval is a interim which to pick historical mid price samples from, if you set it to 5, the first sample is
    # the last (current) mid price, the second sample is a past mid price 5 seconds before the last, and so on.
    interval = 5
    # short_period is how many interval to pick the samples for the average short term volatility calculation,
    # for short_period of 3, this is 3 samples (5 seconds interval), of the last 15 seconds
    short_period = 3
    # long_period is how many interval to pick the samples for the median long term volatility calculation,
    # for long_period of 10, this is 10 samples (5 seconds interval), of the last 50 seconds
    long_period = 10
    last_stats_logged = 0

    def __init__(self):
        super().__init__()
        self.original_bid_spread = None
        self.original_ask_spread = None
        self.modified_bid_spread = None
        self.modified_ask_spread = None
        self.avg_short_volatility = None
        self.median_long_volatility = None

    def volatility_msg(self, include_mid_price=False):
        if self.avg_short_volatility is None or self.median_long_volatility is None:
            return "short_volatility: N/A  long_volatility: N/A"
        mid_price_msg = f"  mid_price: {self.mid_price:<15}" if include_mid_price else ""
        return f"short_volatility: {self.avg_short_volatility:.2%}  " \
               f"long_volatility: {self.median_long_volatility:.2%}{mid_price_msg}" \
               f"  original_bid_spread: {self.original_bid_spread:.2%}" \
               f" original_ask_spread: {self.original_ask_spread:.2%}" \
               f"  mod_bid_spread: {self.modified_bid_spread:.2%}" \
               f" mod_ask_spread: {self.modified_ask_spread:.2%}"


    def on_tick(self):
        # First, let's keep the original spreads.
        if self.original_bid_spread is None:
            self.original_bid_spread = self.modified_bid_spread = self.pmm_parameters.bid_spread
        if self.original_ask_spread is None:
            self.original_ask_spread = self.modified_ask_spread = self.pmm_parameters.ask_spread

        # check for user changes to spread
        if self.modified_bid_spread != self.pmm_parameters.bid_spread:
            self.original_bid_spread = self.pmm_parameters.bid_spread

        if self.modified_ask_spread != self.pmm_parameters.ask_spread:
            self.original_ask_spread = self.pmm_parameters.ask_spread


        # Average volatility (price change) over a short period of time, this is to detect recent sudden changes.
        self.avg_short_volatility = self.avg_price_volatility(self.interval, self.short_period)
        # Median volatility over a long period of time, this is to find the market norm volatility.
        # We use median (instead of average) to find the middle volatility value - this is to avoid recent
        # spike affecting the average value.
        self.median_long_volatility = self.median_price_volatility(self.interval, self.long_period)

        # If the bot just got started, we'll not have these numbers yet as there is not enough mid_price sample size.
        # We'll start to have these numbers after interval * long_term_period (150 seconds in this example).
        if self.avg_short_volatility is None or self.median_long_volatility is None:
            return

        # Let's log some stats once every 5 minutes
        if time.time() - self.last_stats_logged > 60 * 5:
            log_to_file(SCRIPT_LOG_FILE, self.volatility_msg(True))
            self.last_stats_logged = time.time()

        # This volatility delta will be used to adjust spreads.
        delta = self.avg_short_volatility - self.median_long_volatility
        # Let's round the delta into 0.02% increment to ignore noise and to avoid adjusting the spreads too often.
        spread_adjustment = self.round_by_step(delta, Decimal("0.0002"))
        # Show the user on what's going, you can remove this statement to stop the notification.
        #self.notify(f"avg_short_volatility: {self.avg_short_volatility} median_long_volatility: {self.median_long_volatility} "
        #             f"spread_adjustment: {spread_adjustment}")
        new_bid_spread = self.original_bid_spread + spread_adjustment
        # Let's not set the spreads below the originals, this is to avoid having spreads to be too close
        # to the mid price.
        new_bid_spread = max(self.original_bid_spread, new_bid_spread)
        old_bid_spread = self.pmm_parameters.bid_spread
        if new_bid_spread != self.pmm_parameters.bid_spread:
            self.modified_bid_spread = self.pmm_parameters.bid_spread = new_bid_spread

        new_ask_spread = self.original_ask_spread + spread_adjustment
        new_ask_spread = max(self.original_ask_spread, new_ask_spread)
        old_ask_spread = self.pmm_parameters.ask_spread
        if new_ask_spread != self.pmm_parameters.ask_spread:
            self.modified_ask_spread = self.pmm_parameters.ask_spread = new_ask_spread

        if old_bid_spread != new_bid_spread or old_ask_spread != new_ask_spread:
            log_to_file(SCRIPT_LOG_FILE, self.volatility_msg(True))
            log_to_file(SCRIPT_LOG_FILE, f"spreads adjustment: Old Value: {old_bid_spread:.2%} "
                                         f"New Value: {new_bid_spread:.2%}")

    def on_status(self) -> str:
        return self.volatility_msg()