gogo2/docs/exchanges/bybit/examples.py

"""
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")