USER
# fetch.py
import asyncio
import json
import os
import sys
import websockets
import ccxt.async_support as ccxt
import pandas as pd
import numpy as np
import traceback
from datetime import datetime
from asgiref.sync import sync_to_async
import django
from loguru import logger
# Initialize Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'binancebot.settings')
django.setup()
from signals.models import Kline
# Remove default Loguru handlers and configure new ones
logger.remove()
logger.add("fetch_data.log", rotation="20 MB", retention="7 days", level="DEBUG",
format="{time} | {level} | {message}") # File logging
logger.add(sys.stdout, level="INFO") # Console logging
class DataFetcher:
def __init__(self, symbols, timeframes=['5m', '15m', '1h']):
self.symbols = symbols
self.timeframes = [tf.lower() for tf in timeframes]
self.exchange = ccxt.binanceusdm({'enableRateLimit': True})
self.data_lock = asyncio.Lock()
self.symbol_map = {}
self.streams = []
self.stream_urls = []
self.max_streams_per_connection = 100 # Adjust based on testing
logger.debug(f"Initialized DataFetcher with symbols: {self.symbols}, timeframes: {self.timeframes}")
async def initialize_exchange(self):
await self.exchange.load_markets()
for symbol in self.symbols:
market = self.exchange.market(symbol)
ws_symbol = market['id'].lower() # e.g., 'btcusdt'
self.symbol_map[ws_symbol] = symbol
for timeframe in self.timeframes:
self.streams.append(f"{ws_symbol}@kline_{timeframe}")
# Create WebSocket URLs with limited streams per connection
for i in range(0, len(self.streams), self.max_streams_per_connection):
stream_subset = self.streams[i:i + self.max_streams_per_connection]
url = f"wss://fstream.binance.com/stream?streams={'/'.join(stream_subset)}"
self.stream_urls.append(url)
logger.debug(f"WebSocket URLs: {self.stream_urls}")
async def fetch_historical_data(self):
semaphore = asyncio.Semaphore(3) # Limit concurrency
tasks = []
for symbol in self.symbols:
for timeframe in self.timeframes:
tasks.append(self.fetch_symbol_historical_data(symbol, timeframe, semaphore))
await asyncio.gather(*tasks)
async def fetch_symbol_historical_data(self, symbol, timeframe, semaphore):
async with semaphore:
try:
since = None # Fetch all available data
all_bars = []
limit = 1000 # Binance allows up to 1000 bars per request
while True:
bars = await self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, since=since, limit=limit)
if not bars:
break
all_bars += bars
since = bars[-1][0] + 1 # Prevent fetching the last bar again
if len(bars) < limit:
break
if all_bars:
df = pd.DataFrame(all_bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
df['timestamp'] = df['timestamp'].dt.tz_convert('Africa/Nairobi')
logger.info(f"Fetched {len(df)} bars for {symbol} {timeframe}")
# Save to database
await self.save_klines(symbol, timeframe, df)
except Exception as e:
logger.error(f"Error fetching historical data for {symbol} {timeframe}: {e}\n{traceback.format_exc()}")
@sync_to_async
def save_klines(self, symbol, timeframe, df):
klines_to_create = []
for _, row in df.iterrows():
klines_to_create.append(Kline(
symbol=symbol,
timeframe=timeframe,
timestamp=row['timestamp'],
open=row['open'],
high=row['high'],
low=row['low'],
close=row['close'],
volume=row['volume']
))
Kline.objects.bulk_create(klines_to_create, ignore_conflicts=True)
logger.info(f"Saved {len(klines_to_create)} klines for {symbol} {timeframe} to the database.")
async def handle_websocket(self, url):
retry_count = 0
max_retries = 5
backoff = 5
while retry_count < max_retries:
try:
async with websockets.connect(
url,
ping_interval=20,
ping_timeout=20,
close_timeout=10
) as websocket:
logger.info(f"Connected to WebSocket: {url}")
retry_count = 0 # Reset on successful connection
backoff = 5
async for message in websocket:
data = json.loads(message)
if 'data' not in data:
continue
kline_data = data['data']['k']
if not kline_data['x']:
continue # Only process closed klines
ws_symbol = kline_data['s'].lower()
symbol = self.symbol_map.get(ws_symbol)
timeframe = kline_data['i']
if timeframe not in self.timeframes:
continue
timestamp = pd.to_datetime(kline_data['t'], unit='ms', utc=True).tz_convert('Africa/Nairobi')
kline = Kline(
symbol=symbol,
timeframe=timeframe,
timestamp=timestamp,
open=float(kline_data['o']),
high=float(kline_data['h']),
low=float(kline_data['l']),
close=float(kline_data['c']),
volume=float(kline_data['v'])
)
# Save or update the kline in the database
await self.save_kline(kline)
except Exception as e:
logger.error(f"WebSocket connection error: {e}\n{traceback.format_exc()}")
retry_count += 1
sleep_time = backoff * retry_count
logger.info(f"Retrying WebSocket connection in {sleep_time} seconds (Attempt {retry_count}/{max_retries})")
await asyncio.sleep(sleep_time)
@sync_to_async
def save_kline(self, kline):
Kline.objects.update_or_create(
symbol=kline.symbol,
timeframe=kline.timeframe,
timestamp=kline.timestamp,
defaults={
'open': kline.open,
'high': kline.high,
'low': kline.low,
'close': kline.close,
'volume': kline.volume,
}
)
logger.debug(f"Saved kline for {kline.symbol} {kline.timeframe} at {kline.timestamp}")
async def start_websockets(self):
tasks = []
for url in self.stream_urls:
tasks.append(asyncio.create_task(self.handle_websocket(url)))
await asyncio.gather(*tasks)
async def run(self):
try:
logger.info("Initializing exchange...")
await self.initialize_exchange()
logger.info("Fetching historical data...")
await self.fetch_historical_data()
logger.info("Starting WebSocket connections...")
await self.start_websockets()
finally:
await self.exchange.close()
logger.info("Exchange connection closed.")
async def main():
exchange = ccxt.binanceusdm({'enableRateLimit': True})
await exchange.load_markets()
markets = exchange.markets
# Filter for USDT perpetual futures pairs
usdt_pairs = []
for symbol, market in markets.items():
if market.get('active') and market.get('contract') and market.get('type') == 'swap' and market.get('linear') and market.get('quote') == 'USDT':
usdt_pairs.append(symbol)
usdt_pairs = list(set(usdt_pairs))
logger.info(f"Total Futures USDT Pairs: {len(usdt_pairs)}")
if not usdt_pairs:
logger.warning("No USDT pairs found after filtering. Please check the filtering criteria.")
return
# Optionally, sort symbols by volatility or other criteria here
# For simplicity, we'll proceed with the filtered list
fetcher = DataFetcher(symbols=usdt_pairs, timeframes=['5m', '15m', '1h'])
await fetcher.run()
if __name__ == "__main__":
if sys.platform.startswith('win'):
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(main())
Between fetch.py code above, and fetch_data.py below which one is more efficient in fetching crypto prices from binance
# fetch_data.py
import asyncio
import json
import os
import sys
import websockets
import ccxt.async_support as ccxt
import pandas as pd
import traceback
from datetime import datetime
from asgiref.sync import sync_to_async
import django
from loguru import logger
# Initialize Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'binancebot.settings')
django.setup()
from signals.models import Kline
# Remove default Loguru handlers and configure new ones
logger.remove()
logger.add("fetch_data.log", rotation="20 MB", retention="7 days", level="DEBUG",
format="{time} | {level} | {message}") # File logging
logger.add(sys.stdout, level="INFO") # Console logging
class DataFetcher:
def __init__(self, symbols, timeframes=['1m', '5m', '15m', '1h']):
self.symbols = symbols
self.timeframes = [tf.lower() for tf in timeframes]
self.exchange = ccxt.binanceusdm({'enableRateLimit': True})
self.data_lock = asyncio.Lock()
self.symbol_map = {}
self.streams = []
self.stream_urls = []
self.max_streams_per_connection = 100 # Adjust based on testing
logger.debug(f"Initialized DataFetcher with symbols: {self.symbols}, timeframes: {self.timeframes}")
async def initialize_exchange(self):
await self.exchange.load_markets()
for symbol in self.symbols:
market = self.exchange.market(symbol)
ws_symbol = market['id'].lower() # e.g., 'btcusdt'
self.symbol_map[ws_symbol] = symbol
for timeframe in self.timeframes:
self.streams.append(f"{ws_symbol}@kline_{timeframe}")
# Create WebSocket URLs with limited streams per connection
for i in range(0, len(self.streams), self.max_streams_per_connection):
stream_subset = self.streams[i:i + self.max_streams_per_connection]
url = f"wss://fstream.binance.com/stream?streams={'/'.join(stream_subset)}"
self.stream_urls.append(url)
logger.debug(f"WebSocket URLs: {self.stream_urls}")
async def fetch_historical_data(self):
semaphore = asyncio.Semaphore(5) # Adjusted semaphore for optimal concurrency
tasks = []
for symbol in self.symbols:
for timeframe in self.timeframes:
tasks.append(self.fetch_symbol_historical_data(symbol, timeframe, semaphore))
await asyncio.gather(*tasks)
async def fetch_symbol_historical_data(self, symbol, timeframe, semaphore):
async with semaphore:
try:
since = None # Fetch all available data
all_bars = []
limit = 1000 # Binance allows up to 1000 bars per request
while True:
bars = await self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, since=since, limit=limit)
if not bars:
break
all_bars += bars
since = bars[-1][0] + 1 # Prevent fetching the last bar again
if len(bars) < limit:
break
if all_bars:
df = pd.DataFrame(all_bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
df['timestamp'] = df['timestamp'].dt.tz_convert('Africa/Nairobi')
logger.info(f"Fetched {len(df)} bars for {symbol} {timeframe}")
# Save to database
await self.save_klines(symbol, timeframe, df)
except ccxt.NetworkError as e:
logger.error(f"Network error while fetching data for {symbol} {timeframe}: {str(e)}")
except ccxt.ExchangeError as e:
logger.error(f"Exchange error while fetching data for {symbol} {timeframe}: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error fetching data for {symbol} {timeframe}: {str(e)}\n{traceback.format_exc()}")
@sync_to_async
def save_klines(self, symbol, timeframe, df):
# Validate and clean data before saving
df.dropna(inplace=True)
if df.empty:
logger.warning(f"No valid data to save for {symbol} {timeframe}")
return
klines_to_create = []
for _, row in df.iterrows():
klines_to_create.append(Kline(
symbol=symbol,
timeframe=timeframe,
timestamp=row['timestamp'],
open=row['open'],
high=row['high'],
low=row['low'],
close=row['close'],
volume=row['volume']
))
Kline.objects.bulk_create(klines_to_create, ignore_conflicts=True)
logger.info(f"Saved {len(klines_to_create)} klines for {symbol} {timeframe} to the database.")
async def handle_websocket(self, url):
retry_count = 0
max_retries = 5
backoff = 1
while retry_count < max_retries:
try:
async with websockets.connect(
url,
ping_interval=20,
ping_timeout=20,
close_timeout=10
) as websocket:
logger.info(f"Connected to WebSocket: {url}")
retry_count = 0 # Reset on successful connection
async for message in websocket:
data = json.loads(message)
if 'data' not in data:
continue
kline_data = data['data']['k']
if not kline_data['x']:
continue # Only process closed klines
ws_symbol = kline_data['s'].lower()
symbol = self.symbol_map.get(ws_symbol)
timeframe = kline_data['i']
if timeframe not in self.timeframes:
continue
timestamp = pd.to_datetime(kline_data['t'], unit='ms', utc=True).tz_convert('Africa/Nairobi')
kline = Kline(
symbol=symbol,
timeframe=timeframe,
timestamp=timestamp,
open=float(kline_data['o']),
high=float(kline_data['h']),
low=float(kline_data['l']),
close=float(kline_data['c']),
volume=float(kline_data['v'])
)
# Save or update the kline in the database
await self.save_kline(kline)
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"WebSocket connection closed: {str(e)}")
except websockets.exceptions.InvalidStatusCode as e:
logger.error(f"Invalid status code when connecting to WebSocket: {str(e)}")
break
except Exception as e:
logger.error(f"Unexpected WebSocket error: {str(e)}\n{traceback.format_exc()}")
retry_count += 1
sleep_time = backoff * retry_count
logger.info(f"Retrying WebSocket connection in {sleep_time} seconds (Attempt {retry_count}/{max_retries})")
await asyncio.sleep(sleep_time)
@sync_to_async
def save_kline(self, kline):
# Data validation and cleansing
if any(val is None for val in [kline.open, kline.high, kline.low, kline.close, kline.volume]):
logger.warning(f"Invalid data encountered for {kline.symbol} {kline.timeframe} at {kline.timestamp}")
return
Kline.objects.update_or_create(
symbol=kline.symbol,
timeframe=kline.timeframe,
timestamp=kline.timestamp,
defaults={
'open': kline.open,
'high': kline.high,
'low': kline.low,
'close': kline.close,
'volume': kline.volume,
}
)
logger.debug(f"Saved kline for {kline.symbol} {kline.timeframe} at {kline.timestamp}")
async def start_websockets(self):
tasks = []
for url in self.stream_urls:
tasks.append(asyncio.create_task(self.handle_websocket(url)))
await asyncio.gather(*tasks)
async def run(self):
try:
logger.info("Initializing exchange...")
await self.initialize_exchange()
logger.info("Fetching historical data...")
await self.fetch_historical_data()
logger.info("Starting WebSocket connections...")
await self.start_websockets()
finally:
await self.exchange.close()
logger.info("Exchange connection closed.")
async def main():
exchange = ccxt.binanceusdm({'enableRateLimit': True})
await exchange.load_markets()
markets = exchange.markets
# Filter for USDT perpetual futures pairs
usdt_pairs = []
for symbol, market in markets.items():
if market.get('active') and market.get('contract') and market.get('type') == 'swap' and market.get('linear') and market.get('quote') == 'USDT':
usdt_pairs.append(symbol)
usdt_pairs = list(set(usdt_pairs))
logger.info(f"Total Futures USDT Pairs: {len(usdt_pairs)}")
if not usdt_pairs:
logger.warning("No USDT pairs found after filtering. Please check the filtering criteria.")
return
fetcher = DataFetcher(symbols=usdt_pairs, timeframes=['1m', '5m', '15m', '1h'])
await fetcher.run()
if __name__ == "__main__":
if sys.platform.startswith('win'):
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(main())