CoinHarbour.xyz Dev Documentation

__init__(self, trading_helper: TradingHelper)

Initializes the algorithm with the trading helper.

on_ticker_recd(self, ticker_data: dict)

Called when a new ticker data is received. The ticker_data object contains fields as described in the Dataset Fields section.

on_algo_start(self)

Called when the algorithm starts.

on_algo_stop(self)

Called when the algorithm stops.

on_order_success(self)

Called when an order is successfully executed.

on_order_error(self, error)

Called when an error occurs during order execution.

open_position(asset: str, long: bool, collateral_usd: float, leverage: float, slippage_percent: float = 0.03)

Opens a new position or adds to an existing position.

Input Parameters:

• asset (string): The asset to trade (e.g., "BTC", "ETH").
• long (boolean): True for long positions, False for short positions.
• collateral_usd (float): The amount of collateral to use for the position in USD.
• leverage (float): The leverage to use for the position. Total Position Value = Collateral * Leverage
• slippage_percent (float, optional): The expected slippage as a percentage. Default is 0.03%.

Output:

Position object or None if the operation fails.

Example:

position = self.trading_helper.open_position("BTC", True, 1000, 2)
print(position)
# Output:
{
    "chain": "arbitrum",
    "index_token_symbol": "BTC",
    "collateral_token_symbol": "USDC",
    "start_token_symbol": "USDC",
    "is_long": true,
    "size_delta_usd": 2000.0,  # position_size_usd = collateral_usd * leverage
    "leverage": 2.0,
    "slippage_percent": 0.03,
    "timestamp": 1679529600,
    "asset_price_at_open": 45000.0,
    "fee": 0.25  # tx_fee (0.15) + network_fee (0.1)
}

close_position(asset: str, long: bool, position_percent_to_close: float = 100, slippage_percent: float = 0.03)

Closes an existing position partially or fully.

Input Parameters:

• asset (string): The asset to trade (e.g., "BTC", "ETH").
• long (boolean): True for long positions, False for short positions.
• position_percent_to_close (float, optional): The percentage of the position to close. Default is 100 (close entire position).
• slippage_percent (float, optional): The expected slippage as a percentage. Default is 0.03%.

Output:

Position object or None if the operation fails.

Example:

position = self.trading_helper.close_position("BTC", True, 50)
print(position)
# Output:
{
    "chain": "arbitrum",
    "index_token_symbol": "BTC",
    "collateral_token_symbol": "USDC",
    "start_token_symbol": "USDC",
    "is_long": true,
    "size_delta_usd": 1000.0,  # Amount being closed (50% of 2000)
    "leverage": 2.0,
    "slippage_percent": 0.03,
    "timestamp": 1679529600,
    "asset_price_at_close": 46000.0,
    "fee": 0.25,  # tx_fee (0.15) + network_fee (0.1)
}

get_usd_balance()

Retrieves the current USD balance.

Output:

float

Example:

balance = self.trading_helper.get_usd_balance()
print(balance)
# Output: 10500.75  # Current USD balance

get_open_positions()

Retrieves all open positions.

Output:

Dictionary of open positions

Example:

positions = self.trading_helper.get_open_positions()
print(positions)
# Output:
{
    "BTC_long": {
        "chain": "arbitrum",
        "index_token_symbol": "BTC",
        "collateral_token_symbol": "USDC",
        "start_token_symbol": "USDC",
        "is_long": true,
        "size_delta_usd": 2000.0,
        "leverage": 2.0,
        "slippage_percent": 0.03,
        "timestamp": 1679529600,
        "asset_price_at_open": 45000.0,
        "fee": 0.25  # tx_fee (0.15) + network_fee (0.1)
    },
    "ETH_short": {
        "chain": "arbitrum",
        "index_token_symbol": "ETH",
        "collateral_token_symbol": "USDC",
        "start_token_symbol": "USDC",
        "is_long": false,
        "size_delta_usd": 1500.0,
        "leverage": 3.0,
        "slippage_percent": 0.03,
        "timestamp": 1679529600,
        "asset_price_at_open": 2800.0,
        "fee": 0.25  # tx_fee (0.15) + network_fee (0.1)
    }
}

get_previous_tickers(asset: str, previous_tickers: int)

Retrieves previous ticker data for a specific asset.

Input Parameters:

• asset (string): The asset to get data for (e.g., "BTC", "ETH").
• previous_tickers (int): The number of previous tickers to retrieve.

Output:

List of dictionaries containing previous ticker data

Example:

tickers = self.trading_helper.get_previous_tickers("BTC", 2)
print(tickers)
# Output:
[
    {
        "timestamp": 1679529600,
        "open": 44950.0,
        "high": 45100.0,
        "low": 44900.0,
        "close": 45000.0,
        "volume": 125.5,
        "quote_volume": 5647500.0,
        "trades": 2450,
        "taker_buy_base_volume": 75.3,
        "taker_buy_quote_volume": 3388500.0
    },
    {
        "timestamp": 1679529660,
        "open": 45000.0,
        "high": 45200.0,
        "low": 44980.0,
        "close": 45150.0,
        "volume": 98.2,
        "quote_volume": 4428090.0,
        "trades": 1850,
        "taker_buy_base_volume": 58.9,
        "taker_buy_quote_volume": 2657385.0
    }
]

Simple Moving Average

class MyAlgorithm:
    def __init__(self, trading_helper):
        self.prices = []
        self.trading_helper = trading_helper

    def on_ticker_recd(self, ticker_data):
        current_price = ticker_data["BTC"]["close"]
        self.prices.append(current_price)

        if len(self.prices) > 5:
            self.prices.pop(0)

        if len(self.prices) == 5:
            sma_5 = sum(self.prices) / 5
            balance = self.trading_helper.get_usd_balance()

            if balance > 0 and current_price > sma_5:
                self.trading_helper.open_position(
                    asset="BTC",
                    long=True,
                    collateral_usd=balance,
                    leverage=1,
                    slippage_percent=0.03,
                )
            else:
                if len(self.trading_helper.get_open_positions()) > 0:
                    self.trading_helper.close_position(
                        asset="BTC",
                        long=True,
                        position_percent_to_close=100,
                        slippage_percent=0.01,
                    )

    def on_algo_start(self):
        pass

    def on_algo_stop(self):
        pass

    def on_order_success(self):
        pass

    def on_order_error(self, error):
        pass  
        

Relative Strength Index

class MyAlgorithm:
    def __init__(self, trading_helper):
        self.prices = []
        self.trading_helper = trading_helper

    def on_ticker_recd(self, ticker_data):
        current_price = ticker_data["BTC"]["close"]
        self.prices.append(current_price)

        if len(self.prices) > 14:
            self.prices.pop(0)

        if len(self.prices) == 14:
            gains = 0
            losses = 0
            for i in range(1, 14):
                gains = gains +(max(0, self.prices[i] - self.prices[i - 1]))
                losses = losses + (max(0, self.prices[i - 1] - self.prices[i]))
            rsi = 100 - (100 / (1 + gains / losses))

            # Open a long position if RSI < 30 (oversold)
            if rsi < 30 :
                self.trading_helper.close_position( 
                    asset="ETH", 
                    long=False, 
                    position_percent_to_close=100, 
                    slippage_percent=0.01 
                )
                balance = self.trading_helper.get_usd_balance()
                self.trading_helper.open_position( 
                    asset="ETH", 
                    long=True, 
                    collateral_usd=balance, 
                    leverage=2, 
                    slippage_percent=0.02 
                )

            # Open a short position if RSI > 70 (overbought)
            elif rsi > 70:
                self.trading_helper.close_position( 
                    asset="ETH", 
                    long=True, 
                    position_percent_to_close=100, 
                    slippage_percent=0.01 
                )
                balance = self.trading_helper.get_usd_balance()
                self.trading_helper.open_position( 
                    asset="ETH", 
                    long=False, 
                    collateral_usd=balance, 
                    leverage=2, 
                    slippage_percent=0.02
                )

    def on_algo_start(self):
        pass

    def on_algo_stop(self):
        pass

    def on_order_success(self):
        pass

    def on_order_error(self, error):
        pass
        

Moving Average Convergence Divergence

class MyAlgorithm:
    def __init__(self, trading_helper):
        self.prices = []
        self.trading_helper = trading_helper
        self.position_open = False

    def on_ticker_recd(self, ticker_data):
        current_price = ticker_data["BTC"]["close"]
        self.prices.append(current_price)

        if len(self.prices) >= 26:
            # Calculate MACD and signal line
            ema12 = sum(self.prices[-12:]) / 12
            ema26 = sum(self.prices[-26:]) / 26
            macd = ema12 - ema26
            signal_line = sum(self.prices[-9:]) / 9
            balance = self.trading_helper.get_usd_balance()
            
            # Determine leverage and collateral dynamically
            macd_diff = abs(macd - signal_line)
            leverage = 1 + min(4, macd_diff / 0.01)  # Max 5x
            collateral_usd = balance * min(1, macd_diff / 0.05)  

            if macd < signal_line and collateral_usd > 0:
                # Open a short position if MACD goes below signal line
                self.trading_helper.open_position(
                    asset="BTC", 
                    long=False, 
                    collateral_usd=collateral_usd, 
                    leverage=leverage, 
                    slippage_percent=0.02
                )
                self.position_open = True

            elif macd > signal_line and self.position_open:
                self.trading_helper.close_position(
                    asset="BTC", 
                    long=False, 
                    position_percent_to_close=100, 
                    slippage_percent=0.01
                )
                self.position_open = False

    def on_algo_start(self):
        pass

    def on_algo_stop(self):
        pass

    def on_order_success(self):
        pass

    def on_order_error(self, error):
        pass
        

Moving Average Crossover

class MyAlgorithm:
    def __init__(self, trading_helper):
        self.short_prices = []
        self.long_prices = []
        self.trading_helper = trading_helper

    def on_ticker_recd(self, ticker_data):
        current_price = ticker_data["ETH"]["close"]
        self.short_prices.append(current_price)
        self.long_prices.append(current_price)

        # Maintain short and long windows (5 and 20 periods)
        if len(self.short_prices) > 5:
            self.short_prices.pop(0)
        if len(self.long_prices) > 20:
            self.long_prices.pop(0)

        # Calculate moving averages
        if len(self.short_prices) == 5 and len(self.long_prices) == 20:
            sma_short = sum(self.short_prices) / 5
            sma_long = sum(self.long_prices) / 20
            balance = self.trading_helper.get_usd_balance()

            # Buy if short-term MA crosses above long-term MA;
            if sma_short > sma_long:
                self.trading_helper.open_position(
                    asset="ETH",
                    long=True,
                    collateral_usd=balance,
                    leverage=1,
                    slippage_percent=0.03,
                )
            elif sma_short < sma_long:
                self.trading_helper.close_position(
                    asset="ETH",
                    long=True,
                    position_percent_to_close=100,
                    slippage_percent=0.01,
                )
    def on_algo_start(self):
        pass

    def on_algo_stop(self):
        pass

    def on_order_success(self):
        pass

    def on_order_error(self, error):
        pass
        

Pairs Trading

class MyAlgorithm:
    def __init__(self, trading_helper):
        self.spread_history = []
        self.trading_helper = trading_helper
        self.position_open = False
        self.pair = ("BTC", "ETH")  # Example pair

    def on_ticker_recd(self, ticker_data):
        btc_price = ticker_data[self.pair[0]]["close"]
        eth_price = ticker_data[self.pair[1]]["close"]

        spread = btc_price / eth_price
        self.spread_history.append(spread)

        if len(self.spread_history) > 30:
            self.spread_history.pop(0)

        if len(self.spread_history) == 30:
            avg_spread = sum(self.spread_history) / len(self.spread_history)
            current_spread = self.spread_history[-1]
            balance = self.trading_helper.get_usd_balance()
            collateral_usd = balance * 0.25
            leverage = 2

            if current_spread > avg_spread * 1.05 and not self.position_open:
                self.trading_helper.open_position(
                    asset=self.pair[0],
                    long=False,
                    collateral_usd=collateral_usd,
                    leverage=leverage,
                    slippage_percent=0.02,
                )
                self.trading_helper.open_position(
                    asset=self.pair[1],
                    long=True,
                    collateral_usd=collateral_usd,
                    leverage=leverage,
                    slippage_percent=0.02,
                )
                self.position_open = True

            elif current_spread < avg_spread * 0.95 and not self.position_open:
                self.trading_helper.open_position(
                    asset=self.pair[0],
                    long=True,
                    collateral_usd=collateral_usd,
                    leverage=leverage,
                    slippage_percent=0.02,
                )
                self.trading_helper.open_position(
                    asset=self.pair[1],
                    long=False,
                    collateral_usd=collateral_usd,
                    leverage=leverage,
                    slippage_percent=0.02,
                )
                self.position_open = True

            elif avg_spread * 0.98 < current_spread < avg_spread * 1.02:
                if self.position_open: 
                    self.trading_helper.close_position(
                        asset=self.pair[0],
                        long=(current_spread < avg_spread),
                        position_percent_to_close=100,
                        slippage_percent=0.01,
                    )
                    self.trading_helper.close_position(
                        asset=self.pair[1],
                        long=(current_spread > avg_spread),
                        position_percent_to_close=100,
                        slippage_percent=0.01,
                    )
                    self.position_open = False
    def on_algo_start(self):
        pass

    def on_algo_stop(self):
        pass

    def on_order_success(self):
        pass

    def on_order_error(self, error):
        pass
        

Bollinger Bands

class MyAlgorithm:
    def __init__(self, trading_helper):
        self.prices = []
        self.trading_helper = trading_helper
        self.open = False
        self.long = None

    def on_ticker_recd(self, ticker_data):
        current_price = ticker_data["BTC"]["close"]
        self.prices.append(current_price)

        if len(self.prices) > 20:
            self.prices.pop(0)

        if len(self.prices) == 20:
            # Calculate Bollinger Bands
            avg_price = sum(self.prices) / 20
            std_dev = (sum((p - avg_price)**2 for p in self.prices)/20)**0.5
            upper_band = avg_price + (2 * std_dev)
            lower_band = avg_price - (2 * std_dev)
            balance = self.trading_helper.get_usd_balance()

            if current_price > upper_band:
                # Open a short position if price breaks above upper band
                if not self.open or self.long:
                    if self.long:
                        # Close any existing long position
                        self.trading_helper.close_position(
                            asset="BTC", 
                            long=True, 
                            position_percent_to_close=100, 
                            slippage_percent=0.01
                        )
                    # Open a short position
                    self.trading_helper.open_position(
                        asset="BTC", 
                        long=False, 
                        collateral_usd=balance / 3, 
                        leverage=2, 
                        slippage_percent=0.02
                    )
                    self.open = True
                    self.long = False

            elif current_price < lower_band:
                # Open a long position if price drops below lower band
                if not self.open or not self.long:
                    if self.open and not self.long:
                        # Close any existing short position
                        self.trading_helper.close_position(
                            asset="BTC", 
                            long=False, 
                            position_percent_to_close=100, 
                            slippage_percent=0.01
                        )
                    # Open a long position
                    self.trading_helper.open_position(
                        asset="BTC", 
                        long=True, 
                        collateral_usd=balance / 3, 
                        leverage=2, 
                        slippage_percent=0.02
                    )
                    self.open = True
                    self.long = True

            elif current_price > avg_price and self.open and not self.long:
                # Close short if price reverts to the mean after upper breakout
                self.trading_helper.close_position(
                    asset="BTC", 
                    long=False, 
                    position_percent_to_close=100, 
                    slippage_percent=0.01
                )
                self.open = False

            elif current_price < avg_price and self.open and self.long:
                # Close long if price reverts to the mean after lower breakout
                self.trading_helper.close_position(
                    asset="BTC", 
                    long=True, 
                    position_percent_to_close=100, 
                    slippage_percent=0.01
                )
                self.open = False
    def on_algo_start(self):
        pass

    def on_algo_stop(self):
        pass

    def on_order_success(self):
        pass

    def on_order_error(self, error):
        pass