Documented Exchange.py

src/Exchange.py

@@ -11,9 +11,21 @@ logger = logging.getLogger(__name__)
 
 
 class Exchange:
+    """
+    A class to interact with a cryptocurrency exchange using the CCXT library.
+    """
+
     _market_cache = {}
 
-    def __init__(self, name: str, api_keys: Dict[str, str], exchange_id: str):
+    def __init__(self, name: str, api_keys: Dict[str, str] | None, exchange_id: str):
+        """
+        Initializes the Exchange object.
+
+        Parameters:
+            name (str): The name of this exchange instance.
+            api_keys (Dict[str, str]): Dictionary containing 'key' and 'secret' for API authentication.
+            exchange_id (str): The ID of the exchange as recognized by CCXT. Example('binance')
+        """
         self.name = name
         self.api_key = api_keys['key'] if api_keys else None
         self.api_key_secret = api_keys['secret'] if api_keys else None
@@ -26,6 +38,12 @@ class Exchange:
         self.symbols_n_precision = {}
 
     def _connect_exchange(self) -> ccxt.Exchange:
+        """
+        Connects to the exchange using the CCXT library.
+
+        Returns:
+            ccxt.Exchange: An instance of the CCXT exchange class.
+        """
         exchange_class = getattr(ccxt, self.exchange_id)
         if not exchange_class:
             logger.error(f"Exchange {self.exchange_id} is not supported by CCXT.")
@@ -47,10 +65,29 @@ class Exchange:
 
     @staticmethod
     def datetime_to_unix_millis(dt: datetime) -> int:
+        """
+        Converts a datetime object to Unix milliseconds.
+        Parameters:
+            dt (datetime): The datetime object to convert.
+        Returns:
+            int: The Unix timestamp in milliseconds.
+        """
         return int(dt.timestamp() * 1000)
 
     def _fetch_historical_klines(self, symbol: str, interval: str,
                                  start_dt: datetime, end_dt: datetime = None) -> pd.DataFrame:
+        """
+        Fetches historical OHLCV data for a given symbol and interval.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+            interval (str): The interval for the OHLCV data (e.g., '1d').
+            start_dt (datetime): The start date for fetching data.
+            end_dt (datetime, optional): The end date for fetching data. Defaults to the current UTC time.
+
+        Returns:
+            pd.DataFrame: A DataFrame containing the OHLCV data.
+        """
         if end_dt is None:
             end_dt = datetime.utcnow()
 
@@ -92,6 +129,15 @@ class Exchange:
             return pd.DataFrame(columns=['open_time', 'open', 'high', 'low', 'close', 'volume'])
 
     def _fetch_price(self, symbol: str) -> float:
+        """
+        Fetches the current price for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            float: The current price.
+        """
         try:
             ticker = self.client.fetch_ticker(symbol)
             return float(ticker['last'])
@@ -100,6 +146,15 @@ class Exchange:
             return 0.0
 
     def _fetch_min_qty(self, symbol: str) -> float:
+        """
+        Fetches the minimum quantity for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            float: The minimum quantity.
+        """
         try:
             market_data = self.exchange_info[symbol]
             return float(market_data['limits']['amount']['min'])
@@ -108,6 +163,15 @@ class Exchange:
             return 0.0
 
     def _fetch_min_notional_qty(self, symbol: str) -> float:
+        """
+        Fetches the minimum notional quantity for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            float: The minimum notional quantity.
+        """
         try:
             market_data = self.exchange_info[symbol]
             return float(market_data['limits']['cost']['min'])
@@ -116,6 +180,16 @@ class Exchange:
             return 0.0
 
     def _fetch_order(self, symbol: str, order_id: str) -> object:
+        """
+        Fetches an order by its ID for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+            order_id (str): The ID of the order.
+
+        Returns:
+            object: The order details.
+        """
         try:
             return self.client.fetch_order(order_id, symbol)
         except ccxt.BaseError as e:
@@ -123,6 +197,12 @@ class Exchange:
             return None
 
     def _set_symbols(self) -> List[str]:
+        """
+        Sets the list of available symbols on the exchange.
+
+        Returns:
+            List[str]: A list of trading symbols.
+        """
         try:
             markets = self.client.fetch_markets()
             symbols = [market['symbol'] for market in markets if market['active']]
@@ -132,6 +212,12 @@ class Exchange:
             return []
 
     def _set_balances(self) -> List[Dict[str, Union[str, float]]]:
+        """
+        Sets the balances of the account.
+
+        Returns:
+            List[Dict[str, Union[str, float]]]: A list of balances with asset, balance, and PnL.
+        """
         if self.api_key and self.api_key_secret:
             try:
                 account_info = self.client.fetch_balance()
@@ -148,6 +234,12 @@ class Exchange:
             return [{'asset': 'N/A', 'balance': 0, 'pnl': 0}]
 
     def _set_exchange_info(self) -> dict:
+        """
+        Sets the exchange information and caches it.
+
+        Returns:
+            dict: The exchange information.
+        """
         if self.exchange_id in Exchange._market_cache:
             return Exchange._market_cache[self.exchange_id]
 
@@ -160,21 +252,60 @@ class Exchange:
             return {}
 
     def get_client(self) -> object:
+        """
+        Returns the CCXT client instance.
+
+        Returns:
+            object: The CCXT client.
+        """
         return self.client
 
     def get_avail_intervals(self) -> Tuple[str, ...]:
+        """
+        Returns the available time intervals for OHLCV data.
+
+        Returns:
+            Tuple[str, ...]: A tuple of available intervals.
+        """
         return self.intervals
 
     def get_exchange_info(self) -> dict:
+        """
+        Returns the exchange information.
+
+        Returns:
+            dict: The exchange information.
+        """
         return self.exchange_info
 
     def get_symbols(self) -> List[str]:
+        """
+        Returns the list of available symbols.
+
+        Returns:
+            List[str]: A list of trading symbols.
+        """
         return self.symbols
 
     def get_balances(self) -> List[Dict[str, Union[str, float]]]:
+        """
+        Returns the balances of the account.
+
+        Returns:
+            List[Dict[str, Union[str, float]]]: A list of balances with asset, balance, and PnL.
+        """
         return self.balances
 
     def get_symbol_precision_rule(self, symbol: str) -> int:
+        """
+        Returns the precision rule for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            int: The precision rule.
+        """
         r_value = self.symbols_n_precision.get(symbol)
         if r_value is None:
             self._set_precision_rule(symbol)
@@ -183,36 +314,125 @@ class Exchange:
 
     def get_historical_klines(self, symbol: str, interval: str,
                               start_dt: datetime, end_dt: datetime = None) -> pd.DataFrame:
+        """
+        Returns historical OHLCV data for a given symbol and interval.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+            interval (str): The interval for the OHLCV data (e.g., '1d').
+            start_dt (datetime): The start date for fetching data.
+            end_dt (datetime, optional): The end date for fetching data. Defaults to the current UTC time.
+
+        Returns:
+            pd.DataFrame: A DataFrame containing the OHLCV data.
+        """
         return self._fetch_historical_klines(symbol=symbol, interval=interval,
                                              start_dt=start_dt, end_dt=end_dt)
 
     def get_price(self, symbol: str) -> float:
+        """
+        Returns the current price for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            float: The current price.
+        """
         return self._fetch_price(symbol)
 
     def get_min_qty(self, symbol: str) -> float:
+        """
+        Returns the minimum quantity for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            float: The minimum quantity.
+        """
         return self._fetch_min_qty(symbol)
 
     def get_min_notional_qty(self, symbol: str) -> float:
+        """
+        Returns the minimum notional quantity for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+
+        Returns:
+            float: The minimum notional quantity.
+        """
         return self._fetch_min_notional_qty(symbol)
 
     def get_order(self, symbol: str, order_id: str) -> object:
+        """
+        Returns an order by its ID for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+            order_id (str): The ID of the order.
+
+        Returns:
+            object: The order details.
+        """
         return self._fetch_order(symbol, order_id)
 
     def place_order(self, symbol: str, side: str, type: str, timeInForce: str,
                     quantity: float, price: float = None) -> Tuple[str, object]:
+        """
+        Places an order on the exchange.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+            side (str): The side of the order ('buy' or 'sell').
+            type (str): The type of the order ('limit' or 'market').
+            timeInForce (str): The time-in-force policy ('GTC', 'IOC', etc.).
+            quantity (float): The quantity of the order.
+            price (float, optional): The price of the order for limit orders.
+
+        Returns:
+            Tuple[str, object]: A tuple containing the result ('Success' or 'Failure') and the order details or None.
+        """
         result, msg = self._place_order(symbol=symbol, side=side, type=type,
                                         timeInForce=timeInForce, quantity=quantity, price=price)
         return result, msg
 
     def _set_avail_intervals(self) -> Tuple[str, ...]:
+        """
+        Sets the available intervals for OHLCV data.
+
+        Returns:
+            Tuple[str, ...]: A tuple of available intervals.
+        """
         return tuple(self.client.timeframes.keys())
 
     def _set_precision_rule(self, symbol: str) -> None:
+        """
+        Sets the precision rule for a given symbol.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+        """
         market_data = self.exchange_info[symbol]
         precision = market_data['precision']['amount']
         self.symbols_n_precision[symbol] = precision
 
     def _place_order(self, symbol: str, side: str, type: str, timeInForce: str, quantity: float, price: float = None) -> Tuple[str, object]:
+        """
+        Places an order on the exchange.
+
+        Parameters:
+            symbol (str): The trading symbol (e.g., 'BTC/USDT').
+            side (str): The side of the order ('buy' or 'sell').
+            type (str): The type of the order ('limit' or 'market').
+            timeInForce (str): The time-in-force policy ('GTC', 'IOC', etc.).
+            quantity (float): The quantity of the order.
+            price (float, optional): The price of the order for limit orders.
+
+        Returns:
+            Tuple[str, object]: A tuple containing the result ('Success' or 'Failure') and the order details or None.
+        """
         def format_arg(value: float) -> float:
             precision = self.symbols_n_precision.get(symbol, 8)
             return float(f"{value:.{precision}f}")
@@ -242,6 +462,12 @@ class Exchange:
             return 'Failure', None
 
     def get_active_trades(self) -> List[Dict[str, Union[str, float]]]:
+        """
+        Returns a list of active trades.
+
+        Returns:
+            List[Dict[str, Union[str, float]]]: A list of active trades with symbol, side, quantity, and price.
+        """
         if self.api_key and self.api_key_secret:
             try:
                 positions = self.client.fetch_positions()
@@ -262,6 +488,12 @@ class Exchange:
             return []
 
     def get_open_orders(self) -> List[Dict[str, Union[str, float]]]:
+        """
+        Returns a list of open orders.
+
+        Returns:
+            List[Dict[str, Union[str, float]]]: A list of open orders with symbol, side, quantity, and price.
+        """
         if self.api_key and self.api_key_secret:
             try:
                 open_orders = self.client.fetch_open_orders()
@@ -280,3 +512,41 @@ class Exchange:
                 return []
         else:
             return []
+
+
+# Usage Examples
+
+# Example 1: Initializing the Exchange class
+api_keys = {
+    'key': 'your_api_key',
+    'secret': 'your_api_secret'
+}
+exchange = Exchange(name='Binance', api_keys=api_keys, exchange_id='binance')
+
+# Example 2: Fetching historical data
+start_date = datetime(2022, 1, 1)
+end_date = datetime(2022, 6, 1)
+historical_data = exchange.get_historical_klines(symbol='BTC/USDT', interval='1d',
+                                                 start_dt=start_date, end_dt=end_date)
+print(historical_data)
+
+# Example 3: Fetching the current price of a symbol
+current_price = exchange.get_price(symbol='BTC/USDT')
+print(f"Current price of BTC/USDT: {current_price}")
+
+# Example 4: Placing a limit buy order
+order_result, order_details = exchange.place_order(symbol='BTC/USDT', side='buy', type='limit',
+                                                   timeInForce='GTC', quantity=0.001, price=30000)
+print(order_result, order_details)
+
+# Example 5: Getting account balances
+balances = exchange.get_balances()
+print(balances)
+
+# Example 6: Fetching open orders
+open_orders = exchange.get_open_orders()
+print(open_orders)
+
+# Example 7: Fetching active trades
+active_trades = exchange.get_active_trades()
+print(active_trades)