Developer Documentation

This documentation is intended for developers who want to extend the infinity-grid trading bot with new exchanges, strategies, or other functionality. It provides guidance on the architecture, interfaces, and best practices for contributing to the project.

🏛️ Architecture Overview

The infinity-grid trading bot follows a modular architecture with clear separation of concerns:

  1. Core Components: State machine, engine, event bus, and other foundational elements.

  2. Interfaces: Well-defined contracts for exchange operations, notifications, etc.

  3. Models: Data structures representing configurations, orders, balances, etc.

  4. Adapters: Implementations of interfaces for specific exchanges or services.

  5. Strategies: Trading algorithms that use the adapters to execute trades.

  6. Services: Supporting functionality like database access and notifications.

  7. Infrastructure: Low-level components like database management.

🩻 Project Structure

Schematic Project Structure
infinity_grid/
├── exceptions.py
├── adapters/              # Implementations of interfaces
│   ├── exchanges/         # Exchange-specific adapters
│   │   └── kraken.py      # Kraken exchange adapter
│   └── notification.py    # Notification service adapters
├── core/                  # Core functionality
│   ├── cli.py             # Command-line interface
│   ├── engine.py          # The main engine driving the algorithm
│   ├── event_bus.py       # Event management
│   └── state_machine.py   # Bot state management
├── infrastructure/        # Low-level components
│   └── database.py        # Database schemas and access
├── interfaces/            # Interface definitions
│   ├── exchange.py        # Exchange interface
│   └── notification.py    # Notification interface
├── models/                # Data models
│   ├── configuration.py   # Configuration models
│   └── exchange.py        # Exchange-related models
├── services/              # Supporting services
│   ├── database.py        # Database service
│   └── notification.py    # Notification service
└── strategies/            # Trading strategies
    ├── grid_base.py       # Base grid strategy
    ├── grid_hodl.py       # HODL grid strategy
    ├── grid_sell.py       # Sell grid strategy
    ├── c_dca.py           # Dollar-cost averaging strategy
    └── swing.py           # Swing trading strategy

🧩 Extension Points

The infinity-grid bot is designed to be extended in several ways:

  1. Adding New Exchanges: Implement the exchange interfaces to support additional trading platforms.

  2. Creating New Strategies: Extend the base strategy classes to implement new trading algorithms.

  3. Custom Notifications: Extend the notification adapters for different notification channels.

  4. Enhanced Monitoring: Add new metrics or monitoring capabilities.

🔌 Adding a New Exchange Adapter

To add support for a new exchange, you need to create an adapter that implements both the REST and WebSocket interfaces:

  1. Create a new file in infinity_grid/adapters/exchanges/ named after the exchange (e.g., binance.py)

  2. Implement the REST interface by creating a class that inherits from infinity_grid.interfaces.exchange.IExchangeRESTService

  3. Implement the WebSocket interface by creating a class that inherits from infinity_grid.interfaces.exchange.IExchangeWebSocketService

  4. Define the exchange domain by creating an instance of infinity_grid.models.exchange.ExchangeDomain with the exchange’s specific naming conventions

  5. Adapt to exchange-specific behavior by ensuring that the implemented interfaces behave as expected. Feel free to orientate on existing adapters.

Sample structure for a new exchange adapter
from infinity_grid.interfaces.exchange import IExchangeRESTService, IExchangeWebSocketService
from infinity_grid.models.exchange import ExchangeDomain

# Define exchange-specific constants
NEW_EXCHANGE_DOMAIN = ExchangeDomain(
    EXCHANGE="NewExchange",
    BUY="buy",
    SELL="sell",
    OPEN="open",
    CLOSED="closed",
    CANCELED="canceled",
    EXPIRED="expired",
    PENDING="pending",
)

class NewExchangeRESTService(IExchangeRESTService):
    """Implementation of the REST interface for NewExchange"""

    def __init__(self, api_public_key, api_secret_key, state_machine, base_currency, quote_currency):
        # Initialize the REST client
        # Set up any exchange-specific configuration

    def check_api_key_permissions(self):
        # Implement permission checking logic

    # Implement all other required methods from IExchangeRESTService

class NewExchangeWebSocketService(IExchangeWebSocketService):
    """Implementation of the WebSocket interface for NewExchange"""

    def __init__(self, api_public_key, api_secret_key, state_machine, event_bus):
        # Initialize the WebSocket client
        # Set up event handlers

    # Implement all required methods from IExchangeWebSocketService

📈 Creating a New Trading Strategy

To create a new trading strategy, you should extend one of the existing base strategies or create a completely new one:

  1. Create a new file in infinity_grid/strategies/ named after your strategy (e.g., my_strategy.py)

  2. Extend an appropriate base class such as infinity_grid.strategies.grid_base.GridStrategyBase for grid-based strategies

  3. Implement the required methods based on your trading logic

  4. Use the state machine to manage the strategy’s state transitions

  5. Handle events using the event bus for asynchronous operations

Sample structure for a completely new trading strategy
from infinity_grid.strategies.grid_base import GridStrategyBase
from infinity_grid.core.state_machine import States
from infinity_grid.exceptions import BotStateError

class MyCustomStrategy(GridStrategyBase):
    """This is a new sample strategy, based on the GridStrategyBase

    NOTE: When deviating from grid-like trading, make sure to mock, disable,
          or overwrite all functions of the GridStrategyBase.
    """

    async def run(self):
        """Main entry point for the strategy"""
        try:
            # Initialize strategy-specific components
            await self._initialize()

            # Start trading
            await self._start_trading()

        except Exception as exc:
            self._state_machine.transition_to(States.ERROR)
            raise BotStateError("Strategy initialization failed") from exc

    async def _initialize(self):
        """Initialize strategy-specific components"""
        # Implement initialization logic

    async def _start_trading(self):
        """Start the trading process"""
        # Implement trading logic

    # Implement other strategy-specific methods

📝 Testing Your Extensions

When developing new adapters or strategies, it’s important to test them thoroughly:

  1. Unit Tests: Create unit tests in the tests/unit/ directory to test your implementations.

  2. Integration Tests: Add integration tests in tests/integration/ for end-to-end testing.

  3. Mocking: Use mocking to simulate exchange responses without making actual API calls.

🤖 State Machine and Error Handling

The infinity-grid uses a state machine that must be used for transitioning to states. It allows the algorithm to determine the current state and perform actions on that, e.g. not placing orders during the initialization or when being in an error state.

When developing on the infinity-grid, the transition to the desired state can be achieved as demonstrated below. Additionally, the BotStateError must be raised in order to exit the current context.

Use of the state machine during error handling
try:
    ...
except Exception as exc:
    self.__state_machine.transition_to(States.ERROR)
    raise BotStateError("Ooops!") from exc
class infinity_grid.core.state_machine.States(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Represents the state of the trading bot

ERROR = 4

The algorithm encountered an error and is handling it.

INITIALIZING = 1

The algorithm is initializing from the beginning.

RUNNING = 2

The algorithm is running.

SHUTDOWN_REQUESTED = 3

A request to shutdown is being processed.

🛰️ Exchange Interfaces

Exchange adapters must implement all functions of the listed interfaces in order to work properly. Please also read the docstrings of required functions carefully to avoid errors and misbehavior.

class infinity_grid.interfaces.exchange.IExchangeRESTService(api_public_key: str, api_secret_key: str, state_machine: StateMachine, base_currency: str, quote_currency: str)

Bases: ABC

Interface for exchange operations.

abstract cancel_order(txid: str) → None

Cancel an order based on transaction/order ID.

Parameters:

txid (str) – The order ID of the order to cancel.

abstract check_api_key_permissions() → None

Check if the API key permissions are set correctly. This function ideally runs various requests to API endpoints to ensure that they are available by the passed API credentials.

abstract check_exchange_status(tries: int = 0) → None

Check if the exchange is online and operational.

Parameters:

tries (int) – Tries to take to check if the exchange is available, defaults to 0

abstract create_order(*, ordertype: str, side: str, volume: float, price: float, userref: int, validate: bool = False, oflags: str | None = None) → CreateOrderResponseSchema

Create a new order.

Parameters:

  • ordertype (str) – The kind of the order, e.g. limit

  • side (str) – The side of the order, e.g. buy or sell

  • volume (float) – The volume of the order

  • price (float) – The price to place the order

  • userref (int) – The user reference number to refer

  • validate (bool, optional) – If the order should just be validated, but not placed, similar to a dry run, defaults to False

  • oflags (str | None, optional) – Additional order flags to use, defaults to None

Return type:

CreateOrderResponseSchema

abstract get_account_balance() → dict[str, float]

Get the account balances.

Note: This function is currently only needed by the Kraken adapter in

order to check the permissions.

abstract get_asset_pair_info() → AssetPairInfoSchema

Get available asset pair info from the exchange.

abstract get_balances() → list[AssetBalanceSchema]

Get the current balances of the account.

abstract get_closed_orders(userref: int, trades: bool) → dict[str, Any]

Get closed orders for a userref with an optional limit.

Note: This function is currently only needed by the Kraken adapter in

order to check the permissions.

abstract get_exchange_domain() → ExchangeDomain

Return the exchange-specific naming conventions.

abstract get_open_orders(userref: int, trades: bool | None = None) → list[OrderInfoSchema]

Retrieve all open orders based on a user reference number.

Parameters:

  • userref (int) – The reference number to filter orders for

  • trades (bool | None, optional) – Include trades in the result, defaults to None

Return type:

list[OrderInfoSchema]

abstract get_order_with_retry(txid: str, tries: int = 0, max_tries: int = 5, exit_on_fail: bool = True) → OrderInfoSchema

Get order information with retry logic.

If exit_on_fail is True, the algorithm must transition to the error state of the state machine and terminate accordingly. For more information please conduct the documentation of the state machine of the infinity-grid.

Parameters:

  • txid (str) – The transaction or order ID

  • tries (int, optional) – The current try to retrieve the order information, defaults to 0

  • max_tries (int, optional) – The maximum number of tries, defaults to 5

  • exit_on_fail (bool, optional) – Exit the algorithm via error state if the order could not be retrieved, defaults to True

Return type:

OrderInfoSchema

abstract get_orders_info(txid: str | None) → OrderInfoSchema | None

Retrieves order information from the exchange based on passed transaction ID.

Parameters:

txid – The transaction or order ID to filter for , defaults to None

Type:

str | None

abstract get_pair_balance() → PairBalanceSchema

Get the balance for a specific currency pair.

abstract get_system_status() → str

Get the current system status of the exchange.

Must return "online" to succeed.

abstract property rest_altname: str

Returns the alternative name for the given base and quote currency.

The return value must match the value of the pair property of the infinity_grid.models.exchange.OrderInfoSchema. This value is often used to compar the pair with pairs of orders and messages being processed.

This method must be implemented with the @cached_property or @property decorator.

Examples for the Kraken Crypto Asset Exchange adapter - BTCUSD for Bitcoin and US Dollar - DOTUSD for Polkadot and US Dollar - AVAXEUR for Avalanche and Euro

abstract property rest_symbol: str

Returns the symbol for the given base and quote currency.

This method must be implemented with the @cached_property or @property decorator.

Examples for the Kraken Crypto Asset Exchange adapter - BTC/USD for Bitcoin and US Dollar - DOT/USD for Polkadot and US Dollar - AVAX/EUR for Avalanche and Euro

abstract truncate(amount: float | Decimal | str, amount_type: str) → str

Truncate amount according to exchange precision. This allows to exactly match the precision that some exchanges might require.

Truncate XBTUSD
1>>> print(truncate(amount=0.123456789, amount_type="volume"))
20.12345678
3
4>>> print(truncate(amount=21123.12849829993, amount_type="price")))
521123.1
Parameters:

  • amount (float | Decimal | str) – The value that needs to be truncated

  • amount_type (str) – What the amount represents. Either "price" or "volume".

Return type:

str

abstract property ws_symbol: str

Returns the websocket symbol for the given base and quote currency.

The return value is used to subscribe and un-subscribe from websocket channels.

This method must be implemented with the @cached_property or @property decorator.

class infinity_grid.interfaces.exchange.IExchangeWebSocketService(api_public_key: str, api_secret_key: str, event_bus: EventBus, state_machine: StateMachine)

Bases: ABC

Interface for exchange websocket operations.

abstract async close() → None

Close the websocket connection.

abstract get_required_subscriptions(rest_api: IExchangeRESTService) → list[dict[str, Any]]

Returns the list of required subscriptions for the trading strategy.

This method should return exchange-specific subscription parameters needed for ticker and execution channels.

Parameters:

rest_api (IExchangeRESTService) – The REST API service instance to access symbol information

Returns:

List of subscription parameter dictionaries

Return type:

list[dict[str, Any]]

abstract async on_message(message: OnMessageSchema) → None

Function called on every received message.

abstract async start() → None

Start the websocket connection.

abstract async subscribe(params: dict[str, Any]) → None

Subscribe to a specific channel and pair.

🔬 Strategies

This module contains the base class for grid-based trading strategies.

All grid-based strategies should inherit from this class and implement the required methods and override exiting ones as needed.

class infinity_grid.strategies.grid_base.GridStrategyBase(config: BotConfigDTO, event_bus: EventBus, state_machine: StateMachine, db: DBConnect)

Base class for grid-like strategies.

This class is intended to be used as base class for subsequent strategies and contains lots of protected and private functions that might need to be disabled when being used in custom strategies that do not belong to GridHODL, GridSell, SWING, or CDCA.

__add_missed_sell_orders() → None

This functions can create sell orders in case there is at least one executed buy order that is missing its sell order.

Missed sell orders came into place when a buy was executed and placing the sell failed. An entry to the missed sell order id table is added right before placing a sell order.

__cancel_all_open_buy_orders() → None

Cancels all open buy orders and removes them from the orderbook.

__check_configuration_changes() → None

Checking if the database content match with the setup parameters of this instance. A change may happen in case the bot configuration is updated.

Checking if the order size or the interval have changed, requiring all open buy orders to be cancelled.

__check_lowest_cancel_of_more_than_n_buy_orders() → None

Cancel the lowest buy order if new higher buy was placed because of an executed sell order.

__check_n_open_buy_orders() → None

Ensures that there are n open buy orders and will place orders until n.

This function uses a timeout mechanism used in the following scenario

  • WHEN there are not enough funds available AND the maximum investment limit is not reached AND there are not enough open buy orders THEN the function triggers the timeout and exits early

    … to prevent to check for the users balances via REST API on each ticker update. If the timeout mechanism would not be employed, the instance might run into rate limit errors.

__check_near_buy_orders() → None

Cancel buy orders that are next to each other. Only the lowest buy order will survive. This is to avoid that the bot buys at the same price multiple times.

Other functions handle the eventual cancellation of a very low buy order to avoid falling out of the price range.

__check_pending_txids() → bool

Skip checking the price range, because first all missing orders must be assigned. Otherwise this could lead to double trades.

Returns False if okay and True if check_price_range must be skipped.

__check_price_range() → None

Checks if the orders prices match the conditions of the bot respecting the current price.

If the price (self.ticker) raises to high, the open buy orders will be canceled and new buy orders below the price respecting the interval will be placed.

__prepare_for_trading() → None

This function gets triggered once during the setup of the algorithm. It prepares the algorithm for live trading by checking the asset pair parameters, syncing the local with the upstream orderbook, place missing sell orders that not get through because of e.g. “missing funds”, and updating the orderbook.

This function must be sync, since it must block until the setup is done. FIXME: The naming of the function is not ideal

__retrieve_asset_information() → None

Retrieve the asset pair information from the exchange.

This includes: - Estimated fee - Cost decimals - Amount per grid plus + fee

__shift_buy_orders_up() → bool

Checks if the buy order prices are not to low. If there are too low, they get canceled and the check_price_range function is triggered again to place new buy orders.

Returns True if the orders get canceled and the check_price_range functions stops.

__sync_order_book() → None

This function only gets triggered once during the setup of the algorithm.

It checks:

  • if the orderbook is up to date, remove filled, closed, and canceled orders.

  • if the local orderbook for changes - comparison with upstream orderbook

  • and will place new orders if filled.

__update_order_book_handle_closed_order(closed_order: OrderInfoSchema) → None

Gets executed when an order of the local orderbook was closed in the upstream orderbook during the update_orderbook function in the init of the algorithm.

This function triggers the Notification message of the executed order and places a new order.

__update_orderbook_get_open_orders() → list[OrderInfoSchema]

Retrieve all open orders from the upstream exchange that are related to the current trading pair.

_assign_all_pending_transactions() → None

Assign all pending transactions to the orderbook.

_assign_order_by_txid(txid: str) → None

Assigns an order by its txid to the orderbook.

  • Option 1: Removes them from the pending txids and appends it to

    the orderbook

  • Option 2: Updates the info of the order in the orderbook

There is no need for checking the order status, since after the order was added to the orderbook, the algorithm will handle any removals in case of closed orders.

_check_extra_sell_order() → None

Checks if an extra sell order can be placed. This only applies for the SWING strategy.

_get_buy_order_price(last_price: float) → float

Returns the order price for the next buy order.

_get_current_buy_prices() → Iterable[float]

Returns a list of the prices of open buy orders.

_get_sell_order_price(last_price: float) → float

Returns the order price. Also assigns a new highest buy price to configuration if there was a new highest buy.

_handle_arbitrage(side: str, order_price: float, txid_to_delete: str | None = None) → None

Handles the arbitrage between buy and sell orders.

The existence of this function is mainly justified due to the sleep statement at the end.

_handle_cancel_order(txid: str) → None

Cancels an order by txid, removes it from the orderbook, and checks if there there was some volume executed which can be sold later.

NOTE: The orderbook is the “gate keeper” of this function. If the order

is not present in the local orderbook, nothing will happen.

For post-only buy orders - if these were cancelled by the exchange, they are still in the local orderbook and will be handled just like regular calls of the handle_cancel_order of the algorithm.

For orders that were cancelled by the algorithm, these will cancelled via API and removed from the orderbook. The incoming “canceled” message by the websocket will be ignored, as the order is already removed from the orderbook.

property _investment: float

Returns the current investment based on open orders.

property _max_investment_reached: bool

Returns True if the maximum investment is reached.

_new_sell_order(order_price: float, txid_to_delete: str | None = None) → None

Places a new sell order.

This method should be implemented by the concrete strategy classes.

get_value_of_orders(orders: Iterable) → float

Returns the overall invested quote that is invested

handle_filled_order_event(txid: str) → None

Gets triggered by a filled order event from the on_message function.

It fetches the filled order info (using some tries).

new_buy_order(order_price: float, txid_to_delete: str | None = None) → None

Places a new buy order.

on_message(message: OnMessageSchema) → None

Handle incoming messages from the WebSocket connection(s).

async run() → None

Main entry point when starting a trading strategy.

  • Connects to the exchange API.

  • Checks exchange status and API key permissions.

  • Subscribes to the required WebSocket channels.

  • Runs the main loop.

send_status_update() → None

Send a message to the Notification channel with the current status.

async stop() → None

Stop the WebSocket connection(s).

class infinity_grid.strategies.c_dca.CDCAStrategy(config: BotConfigDTO, event_bus: EventBus, state_machine: StateMachine, db: DBConnect)
_check_extra_sell_order() → None

Not applicable for cDCA strategy.

_get_sell_order_price(last_price: float) → float

Returns the order price for the next sell order.

_new_sell_order(order_price: float, txid_to_delete: str | None = None) → None

Places a new sell order.

class infinity_grid.strategies.grid_hodl.GridHODLStrategy(config: BotConfigDTO, event_bus: EventBus, state_machine: StateMachine, db: DBConnect)
_check_extra_sell_order() → None

GridHODL does not support extra sell orders, since the base asset is accumulated over time.

_new_sell_order(order_price: float, txid_to_delete: str | None = None) → None

Places a new sell order.

class infinity_grid.strategies.grid_sell.GridSellStrategy(config: BotConfigDTO, event_bus: EventBus, state_machine: StateMachine, db: DBConnect)
_check_extra_sell_order() → None

Not applicable for GridSell strategy.

_get_sell_order_price(last_price: float, extra_sell: bool = False) → float

Returns the sell order price depending. Also assigns a new highest buy price to configuration if there was a new highest buy.

_new_sell_order(order_price: float, txid_to_delete: str | None = None) → None

Places a new sell order.

class infinity_grid.strategies.swing.SwingStrategy(config: BotConfigDTO, event_bus: EventBus, state_machine: StateMachine, db: DBConnect)
_check_extra_sell_order() → None

Checks if an extra sell order can be placed.

_get_extra_sell_order_price(last_price: float) → float

Returns the sell order price depending. Also assigns a new highest buy price to configuration if there was a new highest buy.

_new_sell_order(order_price: float, txid_to_delete: str | None = None) → None

Places a new sell order.

🎛️ Models and Schemas

Exchange models and schemas for the Infinity Grid trading bot.

This module contains Pydantic models that define the structure and validation rules for exchange-related data such as orders, balances, and market updates. All schemas include appropriate validators to ensure data integrity.

class infinity_grid.models.exchange.AssetBalanceSchema(*, asset: Annotated[str, MinLen(min_length=1)], balance: Annotated[float, Ge(ge=0)], hold_trade: Annotated[float, Ge(ge=0)])

Schema for a single asset balance.

asset: str

The asset name

balance: float

The total balance of the asset

hold_trade: float

The balance of an asset that is held in trades

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class infinity_grid.models.exchange.AssetPairInfoSchema(*, base: str, quote: str, aclass_base: str, aclass_quote: str, lot_decimals: int, cost_decimals: int, fees_maker: list[list[float]])

Schema for required asset pair information

aclass_base: str

The asset class of the base (e.g. “currency”, “tokenized_asset”)

aclass_quote: str

The asset class of the quote (e.g. “currency”, “tokenized_asset”)

base: str

The base currency, e.g. “XXBT”

cost_decimals: int

Number of decimals for cost/quote, e.g. 5

fees_maker: list[list[float]]

Fees for maker orders, e.g. [[0, 0.25], [10000, 0.2], …]

lot_decimals: int

Number of decimals for lot/base size, e.g. 8

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

quote: str

The quote currency, e.g. “ZUSD”

class infinity_grid.models.exchange.CreateOrderResponseSchema(*, txid: Annotated[str, MinLen(min_length=1)])

Schema for the response of a create order operation

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

txid: str

The transaction ID of an order or trade

class infinity_grid.models.exchange.ExchangeDomain(*, EXCHANGE: str, BUY: str, SELL: str, OPEN: str, CLOSED: str, CANCELED: str, EXPIRED: str, PENDING: str)

Basic information on how exchange-specific naming conventions.

BUY: str

How the “buy” or “long” side is named

CANCELED: str

How the order state “canceled” is named

CLOSED: str

How the order state “closed” is named

EXCHANGE: str

The name of the exchange

EXPIRED: str

How the order state “expired” is named

OPEN: str

How the order state “open” is named

PENDING: str

How the order state “pending” is named

SELL: str

How the “sell” or “short” side is named

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class infinity_grid.models.exchange.ExecutionsUpdateSchema(*, order_id: Annotated[str, MinLen(min_length=1)], exec_type: str)

Schema for execution update data

exec_type: str

The execution type, e.g. “new”, “filled”, “cancelled”

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

order_id: str

The order ID

class infinity_grid.models.exchange.OnMessageSchema(*, channel: Annotated[str, MinLen(min_length=1)], type: str | None = None, ticker_data: TickerUpdateSchema | None = None, executions: list[ExecutionsUpdateSchema] | None = None)

Schema for WebSocket message data

channel: str

The channel of the message, e.g. “ticker” or “executions”

executions: list[ExecutionsUpdateSchema] | None

Any executions

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ticker_data: TickerUpdateSchema | None

The ticker data

type: str | None

Type of the message, e.g. “update” or “snapshot”

class infinity_grid.models.exchange.OrderInfoSchema(*, pair: Annotated[str, MinLen(min_length=1)], price: Annotated[float, Gt(gt=0)], side: str, status: str, txid: Annotated[str, MinLen(min_length=1)], userref: Annotated[int, Ge(ge=0)], vol_exec: Annotated[float, Ge(ge=0)], vol: Annotated[float, Gt(gt=0)])

Schema for order information

classmethod clean_pair(v: str) → str

Remove any ‘/’ characters from the pair field

Ensuring that the pair is always the “altname”, e.g. “XBT/USD” will be transformed to “XBTUSD”. This is necessary for consistency across different parts of the application that expect the pair without slashes.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pair: str

Asset pair name (altname) without “/”, e.g. BTCUSD

price: float

The order price

side: str

The order side

status: str

The order status e.g. “open”, “closed”, “canceled”

txid: str

The transaction ID

userref: int

The user reference number of the order

validate_volume_relationship() → Self

Validate that executed volume doesn’t exceed total volume

vol: float

The volume of the order

vol_exec: float

The executed volume of the order

class infinity_grid.models.exchange.PairBalanceSchema(*, base_balance: Annotated[float, Ge(ge=0)], quote_balance: Annotated[float, Ge(ge=0)], base_available: Annotated[float, Ge(ge=0)], quote_available: Annotated[float, Ge(ge=0)])

A schema required for providing information about total and available assets.

base_available: float

The available balance of the base currency

base_balance: float

The current balance (total) of the base currency

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

quote_available: float

The available balance of the quote currency

quote_balance: float

The current balance (total) of the quote currency

class infinity_grid.models.exchange.TickerUpdateSchema(*, symbol: Annotated[str, MinLen(min_length=1)], last: Annotated[float, Gt(gt=0)])

Schema for ticker update data

last: float

The last traded price of a trading pair based on this ticker

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

symbol: str

The trading pair symbol of the ticker

🩹 Exceptions

Custom exceptions for the Infinity Grid trading bot.

exception infinity_grid.exceptions.BotStateError

Custom exception for terminating the algorithm due to an error state.

This exception must only be raised in functions within the running loop, that would otherwise continue running or returning something time before the algorithm is terminating due to an error state.

message

The error message to be displayed.

Type:

str

Example:

def func():
    try:
        do_something()
    except Exception as exc:
        message = "Exception while processing message."
        LOG.error(msg=message, exc_info=exc)
        self.state_machine.transition_to(States.ERROR)
        raise GridBotErrorState(message) from exc
exception infinity_grid.exceptions.MetricsServerError

Custom exception for metrics server related errors.

This exception is raised when the metrics server encounters errors during startup, shutdown, or request handling.

message

The error message describing what went wrong.

Type:

str

Example:

try:
    await metrics_server.start()
except Exception as exc:
    raise MetricsServerError("Failed to start metrics server") from exc
exception infinity_grid.exceptions.UnknownOrderError

Exception used to be raised when an order that was requested is unknown.