import ccxt
import pandas as pd
import time
import os
from datetime import datetime
import numpy as np
from dotenv import load_dotenv
# Load environment variables from .env file (for API keys)
load_dotenv()
class BinanceTrader:
"""
A trading bot for Binance using the CCXT library.
Handles market data retrieval, order placement, and execution monitoring.
"""
def __init__(self, api_key=None, api_secret=None, testnet=True):
"""
Initialize the Binance trader.
Args:
api_key: Binance API key
api_secret: Binance API secret
testnet: Whether to use testnet (default: True)
"""
# Use environment variables if not provided
self.api_key = api_key or os.getenv('BINANCE_API_KEY')
self.api_secret = api_secret or os.getenv('BINANCE_API_SECRET')
if not self.api_key or not self.api_secret:
raise ValueError("API key and secret must be provided or set as environment variables")
# Initialize exchange
exchange_id = 'binance'
exchange_class = getattr(ccxt, exchange_id)
# Exchange configuration
exchange_config = {
'apiKey': self.api_key,
'secret': self.api_secret,
'enableRateLimit': True, # Important to avoid getting banned
}
# Use testnet for testing if enabled
if testnet:
exchange_config.update({
'options': {
'defaultType': 'future', # Use futures market on testnet
},
'urls': {
'api': {
'public': 'https://testnet.binancefuture.com/fapi/v1',
'private': 'https://testnet.binancefuture.com/fapi/v1',
},
}
})
self.exchange = exchange_class(exchange_config)
self.symbols = {} # Cache for symbol info
print(f"{'Testnet' if testnet else 'Production'} Binance trader initialized")
def get_markets(self):
"""Get all available markets"""
markets = self.exchange.load_markets()
return markets
def fetch_ticker(self, symbol):
"""
Fetch current ticker information for a symbol.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
Returns:
Ticker information
"""
ticker = self.exchange.fetch_ticker(symbol)
return ticker
def fetch_ohlcv(self, symbol, timeframe='1h', limit=100):
"""
Fetch OHLCV data for a symbol.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
timeframe: Timeframe (e.g., '1m', '5m', '1h', '1d')
limit: Number of candles to fetch
Returns:
DataFrame with OHLCV data
"""
ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
def fetch_balance(self):
"""
Fetch account balance.
Returns:
Balance information
"""
balance = self.exchange.fetch_balance()
return balance
def place_market_order(self, symbol, side, amount):
"""
Place a market order.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
side: 'buy' or 'sell'
amount: Amount to buy or sell
Returns:
Order information
"""
try:
order = self.exchange.create_order(
symbol=symbol,
type='market',
side=side,
amount=amount
)
print(f"Placed {side} market order for {amount} {symbol}")
return order
except Exception as e:
print(f"Error placing market order: {e}")
return None
def place_limit_order(self, symbol, side, amount, price):
"""
Place a limit order.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
side: 'buy' or 'sell'
amount: Amount to buy or sell
price: Limit price
Returns:
Order information
"""
try:
order = self.exchange.create_order(
symbol=symbol,
type='limit',
side=side,
amount=amount,
price=price
)
print(f"Placed {side} limit order for {amount} {symbol} at {price}")
return order
except Exception as e:
print(f"Error placing limit order: {e}")
return None
def fetch_order(self, order_id, symbol):
"""
Fetch information about an order.
Args:
order_id: Order ID
symbol: Trading pair symbol (e.g., 'BTC/USDT')
Returns:
Order information
"""
try:
order = self.exchange.fetch_order(order_id, symbol)
return order
except Exception as e:
print(f"Error fetching order: {e}")
return None
def fetch_open_orders(self, symbol=None):
"""
Fetch open orders.
Args:
symbol: Trading pair symbol (optional)
Returns:
List of open orders
"""
try:
orders = self.exchange.fetch_open_orders(symbol)
return orders
except Exception as e:
print(f"Error fetching open orders: {e}")
return []
def cancel_order(self, order_id, symbol):
"""
Cancel an order.
Args:
order_id: Order ID
symbol: Trading pair symbol (e.g., 'BTC/USDT')
Returns:
Cancellation result
"""
try:
result = self.exchange.cancel_order(order_id, symbol)
print(f"Cancelled order {order_id}")
return result
except Exception as e:
print(f"Error cancelling order: {e}")
return None
def calculate_sma(self, symbol, timeframe='1h', period=20):
"""
Calculate Simple Moving Average.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
timeframe: Timeframe for candles
period: SMA period
Returns:
SMA value
"""
df = self.fetch_ohlcv(symbol, timeframe, limit=period+10)
df['sma'] = df['close'].rolling(window=period).mean()
return df['sma'].iloc[-1]
def calculate_ema(self, symbol, timeframe='1h', period=20):
"""
Calculate Exponential Moving Average.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
timeframe: Timeframe for candles
period: EMA period
Returns:
EMA value
"""
df = self.fetch_ohlcv(symbol, timeframe, limit=period*3) # Get more data for better EMA
df['ema'] = df['close'].ewm(span=period, adjust=False).mean()
return df['ema'].iloc[-1]
def simple_strategy(self, symbol, timeframe='1h'):
"""
A simple trading strategy based on EMA crossover.
Args:
symbol: Trading pair symbol (e.g., 'BTC/USDT')
timeframe: Timeframe for analysis
Returns:
Signal: 'buy', 'sell', or 'hold'
"""
df = self.fetch_ohlcv(symbol, timeframe, limit=50)
# Calculate EMAs
df['ema_short'] = df['close'].ewm(span=9, adjust=False).mean()
df['ema_long'] = df['close'].ewm(span=21, adjust=False).mean()
# Generate signals
current_price = df['close'].iloc[-1]
ema_short = df['ema_short'].iloc[-1]
ema_long = df['ema_long'].iloc[-1]
ema_short_prev = df['ema_short'].iloc[-2]
ema_long_prev = df['ema_long'].iloc[-2]
# Check for crossover
if ema_short > ema_long and ema_short_prev <= ema_long_prev:
return 'buy', current_price
elif ema_short < ema_long and ema_short_prev >= ema_long_prev:
return 'sell', current_price
else:
return 'hold', current_price
# Example usage
if __name__ == "__main__":
try:
# Initialize trader (using environment variables for API keys)
trader = BinanceTrader(testnet=True)
# Get all available markets
markets = trader.get_markets()
print(f"Available markets: {len(markets)}")
# Symbol to trade
symbol = 'BTC/USDT'
# Fetch and display ticker information
ticker = trader.fetch_ticker(symbol)
print(f"\n{symbol} ticker:")
print(f"Last price: {ticker['last']}")
print(f"Bid: {ticker['bid']}, Ask: {ticker['ask']}")
print(f"24h change: {ticker['change']}, % change: {ticker['percentage']}%")
print(f"24h volume: {ticker['quoteVolume']} USDT")
# Fetch and display OHLCV data
ohlcv = trader.fetch_ohlcv(symbol, timeframe='1h', limit=5)
print(f"\nRecent {symbol} OHLCV data (1h):")
print(ohlcv)
# Fetch and display balance
balance = trader.fetch_balance()
print("\nAccount balance (available):")
for currency in ['USDT', 'BTC', 'ETH']:
if currency in balance:
print(f"{currency}: {balance[currency].get('free', 0)}")
# Run simple strategy
signal, price = trader.simple_strategy(symbol)
print(f"\nStrategy signal for {symbol}: {signal.upper()} at {price}")
# Place a small limit order based on strategy (for demonstration)
if signal in ['buy', 'sell'] and input(f"Place {signal} order? (y/n): ").lower() == 'y':
amount = 0.001 # Small BTC amount
# Calculate price with small offset from current price
offset = 0.01 # 1% offset
if signal == 'buy':
price = price * (1 - offset) # Lower price for buy
else:
price = price * (1 + offset) # Higher price for sell
# Place limit order
order = trader.place_limit_order(
symbol=symbol,
side=signal,
amount=amount,
price=price
)
if order:
print(f"\nOrder placed:")
print(f"ID: {order['id']}")
print(f"Status: {order['status']}")
# Wait a bit and fetch the order
time.sleep(2)
updated_order = trader.fetch_order(order['id'], symbol)
print(f"\nUpdated order status: {updated_order['status']}")
# Cancel the order if it's still open
if updated_order['status'] in ['open', 'created']:
confirm = input("Cancel order? (y/n): ")
if confirm.lower() == 'y':
trader.cancel_order(order['id'], symbol)
# Fetch and display open orders
open_orders = trader.fetch_open_orders(symbol)
print(f"\nOpen orders for {symbol}: {len(open_orders)}")
for order in open_orders:
print(f"ID: {order['id']}, Type: {order['type']}, Side: {order['side']}, Amount: {order['amount']}, Price: {order.get('price')}")
except Exception as e:
print(f"Error running example: {e}")