""" Bybit Integration Examples Based on official pybit library documentation and examples """ import os from pybit.unified_trading import HTTP import logging # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def create_bybit_session(testnet=True): """Create a Bybit HTTP session. Args: testnet (bool): Use testnet if True, live if False Returns: HTTP: Bybit session object """ api_key = os.getenv('BYBIT_API_KEY') api_secret = os.getenv('BYBIT_API_SECRET') if not api_key or not api_secret: raise ValueError("BYBIT_API_KEY and BYBIT_API_SECRET must be set in environment") session = HTTP( testnet=testnet, api_key=api_key, api_secret=api_secret, ) logger.info(f"Created Bybit session (testnet: {testnet})") return session def get_account_info(session): """Get account information and balances.""" try: # Get account info account_info = session.get_wallet_balance(accountType="UNIFIED") logger.info(f"Account info: {account_info}") return account_info except Exception as e: logger.error(f"Error getting account info: {e}") return None def get_ticker_info(session, symbol="BTCUSDT"): """Get ticker information for a symbol. Args: session: Bybit HTTP session symbol: Trading symbol (default: BTCUSDT) """ try: ticker = session.get_tickers(category="linear", symbol=symbol) logger.info(f"Ticker for {symbol}: {ticker}") return ticker except Exception as e: logger.error(f"Error getting ticker for {symbol}: {e}") return None def get_orderbook(session, symbol="BTCUSDT", limit=25): """Get orderbook for a symbol. Args: session: Bybit HTTP session symbol: Trading symbol limit: Number of price levels to return """ try: orderbook = session.get_orderbook( category="linear", symbol=symbol, limit=limit ) logger.info(f"Orderbook for {symbol}: {orderbook}") return orderbook except Exception as e: logger.error(f"Error getting orderbook for {symbol}: {e}") return None def place_limit_order(session, symbol="BTCUSDT", side="Buy", qty="0.001", price="50000"): """Place a limit order. Args: session: Bybit HTTP session symbol: Trading symbol side: "Buy" or "Sell" qty: Order quantity as string price: Order price as string """ try: order = session.place_order( category="linear", symbol=symbol, side=side, orderType="Limit", qty=qty, price=price, timeInForce="GTC" # Good Till Cancelled ) logger.info(f"Placed order: {order}") return order except Exception as e: logger.error(f"Error placing order: {e}") return None def place_market_order(session, symbol="BTCUSDT", side="Buy", qty="0.001"): """Place a market order. Args: session: Bybit HTTP session symbol: Trading symbol side: "Buy" or "Sell" qty: Order quantity as string """ try: order = session.place_order( category="linear", symbol=symbol, side=side, orderType="Market", qty=qty ) logger.info(f"Placed market order: {order}") return order except Exception as e: logger.error(f"Error placing market order: {e}") return None def get_open_orders(session, symbol=None): """Get open orders. Args: session: Bybit HTTP session symbol: Trading symbol (optional, gets all if None) """ try: params = {"category": "linear", "openOnly": True} if symbol: params["symbol"] = symbol orders = session.get_open_orders(**params) logger.info(f"Open orders: {orders}") return orders except Exception as e: logger.error(f"Error getting open orders: {e}") return None def cancel_order(session, symbol, order_id): """Cancel an order. Args: session: Bybit HTTP session symbol: Trading symbol order_id: Order ID to cancel """ try: result = session.cancel_order( category="linear", symbol=symbol, orderId=order_id ) logger.info(f"Cancelled order {order_id}: {result}") return result except Exception as e: logger.error(f"Error cancelling order {order_id}: {e}") return None def get_position(session, symbol="BTCUSDT"): """Get position information. Args: session: Bybit HTTP session symbol: Trading symbol """ try: positions = session.get_positions( category="linear", symbol=symbol ) logger.info(f"Position for {symbol}: {positions}") return positions except Exception as e: logger.error(f"Error getting position for {symbol}: {e}") return None def get_trade_history(session, symbol="BTCUSDT", limit=50): """Get trade history. Args: session: Bybit HTTP session symbol: Trading symbol limit: Number of trades to return """ try: trades = session.get_executions( category="linear", symbol=symbol, limit=limit ) logger.info(f"Trade history for {symbol}: {trades}") return trades except Exception as e: logger.error(f"Error getting trade history for {symbol}: {e}") return None # Example usage if __name__ == "__main__": # Create session (testnet by default) session = create_bybit_session(testnet=True) # Get account info account_info = get_account_info(session) # Get ticker ticker = get_ticker_info(session, "BTCUSDT") # Get orderbook orderbook = get_orderbook(session, "BTCUSDT") # Get open orders open_orders = get_open_orders(session) # Get position position = get_position(session, "BTCUSDT") # Note: Uncomment below to actually place orders (use with caution) # order = place_limit_order(session, "BTCUSDT", "Buy", "0.001", "30000") # market_order = place_market_order(session, "BTCUSDT", "Buy", "0.001")