Figure31/pattern-finder
0
1import aiohttp2import pandas as pd3import time4import json5import asyncio6from datetime import datetime, timedelta7from typing import Dict, List, Optional, Union, Tuple8import numpy as np9import pytz10 11class BaseDataProvider:12 """13 Base class for data providers14 Adapted from the original BaseOHLCVProvider15 """16 17 async def get_historical_ohlcv(18 self,19 symbol: str,20 interval: str,21 start_time: Optional[str] = None,22 end_time: Optional[str] = None,23 days: int = 36524 ) -> pd.DataFrame:25 """26 Fetch historical OHLCV data27 28 Args:29 symbol: Asset symbol (e.g., "BTC")30 interval: Time interval (e.g., "1m", "5m", "1h", "1d")31 start_time: Optional ISO format start time32 end_time: Optional ISO format end time33 days: Number of days of historical data (default: 365)34 35 Returns:36 pandas.DataFrame with OHLCV data37 """38 raise NotImplementedError("Subclasses must implement this method")39 40 41class BinanceDataProvider(BaseDataProvider):42 """43 Data provider using Binance API for BTCUSDT spot market data44 """45 46 def __init__(self):47 self.base_url = "https://api.binance.us/api/v3"48 self._max_retries = 349 self._request_delay = 0.1 # 100ms between requests50 51 async def get_historical_ohlcv(52 self,53 symbol: str,54 interval: str,55 start_time: Optional[str] = None,56 end_time: Optional[str] = None,57 days: int = 36558 ) -> pd.DataFrame:59 """60 Fetch historical OHLCV data from Binance61 62 Args:63 symbol: Asset symbol (e.g., "BTC", "ETH")64 interval: Time interval (e.g., "1m", "5m", "1h", "1d")65 start_time: Optional ISO format start time66 end_time: Optional ISO format end time67 days: Number of days of historical data (default: 365)68 69 Returns:70 pandas.DataFrame with OHLCV data71 """72 # Format symbol for Binance (add USDT suffix)73 if symbol.upper() == "ETH":74 formatted_symbol = "ETHUSDT"75 else:76 # Default to BTC if not explicitly ETH77 formatted_symbol = "BTCUSDT"78 79 # Determine time range80 if end_time:81 end = datetime.fromisoformat(end_time.replace('Z', '+00:00'))82 else:83 end = datetime.now()84 85 if start_time:86 start = datetime.fromisoformat(start_time.replace('Z', '+00:00'))87 else:88 start = end - timedelta(days=days)89 90 # Convert to milliseconds for Binance API91 start_ms = int(start.timestamp() * 1000)92 end_ms = int(end.timestamp() * 1000)93 94 # Calculate time spans for chunking requests95 time_span_ms = end_ms - start_ms96 97 # Binance API URL for klines (candlestick data)98 url = f"{self.base_url}/klines"99 100 # For long time periods, we need to make multiple requests due to the 1000 candle limit101 chunk_results = await self._fetch_data_in_chunks(url, formatted_symbol, interval, start_ms, end_ms)102 103 if not chunk_results:104 # If API fails, just return empty dataframe with proper error message105 print("Error: Failed to fetch data from Binance API")106 return pd.DataFrame()107 108 # Combine all chunks109 all_klines = []110 for chunk in chunk_results:111 all_klines.extend(chunk)112 113 # Convert to DataFrame114 df = pd.DataFrame(all_klines, columns=[115 "timestamp", "open", "high", "low", "close", "volume",116 "close_time", "quote_volume", "trades_count",117 "taker_buy_volume", "taker_buy_quote_volume", "ignored"118 ])119 120 # Convert types121 for col in ["open", "high", "low", "close", "volume"]:122 df[col] = pd.to_numeric(df[col])123 124 # Ensure proper sorting and reset index125 df = df.sort_values("timestamp").reset_index(drop=True)126 127 # Return only the columns we need128 return df[["timestamp", "open", "high", "low", "close", "volume"]]129 130 async def _fetch_data_in_chunks(131 self, 132 url: str, 133 symbol: str, 134 interval: str, 135 start_ms: int, 136 end_ms: int,137 chunk_size: int = 1000138 ) -> List[List[List]]:139 """140 Fetch data in chunks to handle the 1000 candle limit of Binance API141 142 Args:143 url: API endpoint URL144 symbol: Trading pair symbol145 interval: Candle interval146 start_ms: Start time in milliseconds147 end_ms: End time in milliseconds148 chunk_size: Maximum number of candles per request149 150 Returns:151 List of kline data chunks152 """153 # Calculate the approximate size of each chunk based on the interval154 interval_ms = self._interval_to_milliseconds(interval)155 if interval_ms == 0:156 return []157 158 # Calculate the number of candles needed159 total_candles = (end_ms - start_ms) // interval_ms160 161 # Calculate chunk points (start timestamps for each chunk)162 chunk_points = []163 current = start_ms164 165 while current < end_ms:166 chunk_points.append(current)167 # Each chunk will fetch up to 1000 candles or until end_time168 current += interval_ms * chunk_size169 170 # Add the end time to ensure we get the complete range171 if chunk_points[-1] + (interval_ms * chunk_size) < end_ms:172 chunk_points.append(end_ms - (interval_ms * chunk_size))173 174 # Prepare async tasks to fetch each chunk175 try:176 # Use a longer timeout for cloud deployment177 timeout = aiohttp.ClientTimeout(total=30) # 30 second timeout178 async with aiohttp.ClientSession(timeout=timeout) as session:179 tasks = []180 181 for i, chunk_start in enumerate(chunk_points):182 chunk_end = min(chunk_start + (interval_ms * chunk_size), end_ms)183 184 # Prepare the parameters for this chunk185 params = {186 "symbol": symbol,187 "interval": interval,188 "startTime": chunk_start,189 "endTime": chunk_end,190 "limit": chunk_size191 }192 193 # Create task for this chunk194 task = self._fetch_chunk(session, url, params, f"Chunk {i+1}/{len(chunk_points)}")195 tasks.append(task)196 197 # Run all tasks concurrently with a small delay between each to avoid rate limits198 results = []199 for i, task in enumerate(tasks):200 # Add a small delay between requests to avoid rate limits201 if i > 0:202 await asyncio.sleep(self._request_delay)203 204 try: 205 chunk_result = await task206 if chunk_result:207 results.append(chunk_result)208 except Exception as e:209 print(f"Task {i} failed with error: {str(e)}")210 211 return results212 except Exception as e:213 print(f"Fatal error in fetch_data_in_chunks: {str(e)}")214 # In case of catastrophic failure, return empty results215 return []216 217 async def _fetch_chunk(218 self, 219 session: aiohttp.ClientSession, 220 url: str, 221 params: Dict,222 chunk_id: str223 ) -> Optional[List[List]]:224 """225 Fetch a single chunk of data from Binance API226 227 Args:228 session: aiohttp ClientSession229 url: API endpoint URL230 params: Request parameters231 chunk_id: Identifier for this chunk (for logging)232 233 Returns:234 List of kline data or None if request failed235 """236 # Add retries for reliability237 for attempt in range(self._max_retries):238 try:239 async with session.get(url, params=params) as response:240 if response.status != 200:241 try:242 error_text = await response.text()243 print(f"Error fetching {chunk_id}: {response.status} - {error_text}")244 except:245 print(f"Error fetching {chunk_id}: Status {response.status}, could not read response text")246 247 # If we hit a rate limit, wait and try again248 if response.status == 429:249 retry_after = int(response.headers.get('Retry-After', 1))250 print(f"Rate limited, retrying after {retry_after} seconds")251 await asyncio.sleep(retry_after)252 continue253 elif response.status in [500, 502, 503, 504]:254 # Server error, wait longer255 wait_time = 2 ** attempt # Exponential backoff256 print(f"Server error, retrying after {wait_time} seconds")257 await asyncio.sleep(wait_time)258 continue259 260 return None261 262 try: 263 klines = await response.json()264 except Exception as e:265 print(f"Failed to parse JSON response for {chunk_id}: {str(e)}")266 # Try to get the raw text267 try:268 text = await response.text()269 print(f"Raw response: {text[:200]}...") # First 200 chars270 except:271 print("Could not read response text")272 273 # Wait and retry274 await asyncio.sleep(1)275 continue276 277 if not klines or not isinstance(klines, list):278 print(f"No valid data returned for {chunk_id}")279 return None280 281 print(f"Successfully fetched {len(klines)} candles for {chunk_id}")282 return klines283 284 except aiohttp.ClientConnectorError as e:285 print(f"Connection error for {chunk_id} (attempt {attempt+1}/{self._max_retries}): {str(e)}")286 await asyncio.sleep(2) # Longer wait for connection issues287 except asyncio.TimeoutError:288 print(f"Timeout error for {chunk_id} (attempt {attempt+1}/{self._max_retries})")289 await asyncio.sleep(2) # Longer wait for timeouts290 except Exception as e:291 print(f"Exception during {chunk_id} fetch (attempt {attempt+1}/{self._max_retries}): {str(e)}")292 await asyncio.sleep(1) # Wait before retry293 294 # All retries failed295 print(f"All retries failed for {chunk_id}")296 return None297 298 def _interval_to_milliseconds(self, interval: str) -> int:299 """300 Convert interval string to milliseconds301 302 Args:303 interval: Interval string (e.g. "1m", "1h", "1d")304 305 Returns:306 Interval in milliseconds307 """308 # Parse the interval string309 unit = interval[-1]310 value = int(interval[:-1])311 312 # Convert to milliseconds313 if unit == 'm':314 return value * 60 * 1000315 elif unit == 'h':316 return value * 60 * 60 * 1000317 elif unit == 'd':318 return value * 24 * 60 * 60 * 1000319 elif unit == 'w':320 return value * 7 * 24 * 60 * 60 * 1000321 else:322 return 0 # Invalid interval323 324 # Simulated data function removed as it's not needed for production use325 326 327class CompositeBTCDataProvider(BaseDataProvider):328 """329 Composite implementation for BTC data330 Uses multiple providers with failover capability331 Adapted from the original CompositeCoinDataProvider332 """333 334 def __init__(self, data_providers: List[BaseDataProvider] = None):335 """336 Initialize with a list of data providers, or create a default Binance provider337 """338 if data_providers is None:339 # Default to using Binance provider340 self.data_providers = [BinanceDataProvider()]341 else:342 self.data_providers = data_providers343 344 async def get_historical_ohlcv(345 self,346 symbol: str,347 interval: str,348 start_time: Optional[str] = None,349 end_time: Optional[str] = None,350 days: int = 365351 ) -> pd.DataFrame:352 """353 Fetch historical OHLCV data using multiple providers with failover354 Tries each provider in order until successful355 356 Args:357 symbol: Asset symbol (e.g., "BTC")358 interval: Time interval (e.g., "1m", "5m", "1h", "1d")359 start_time: Optional ISO format start time360 end_time: Optional ISO format end time361 days: Number of days of historical data (default: 365)362 363 Returns:364 pandas.DataFrame with OHLCV data365 """366 for provider in self.data_providers:367 try:368 data = await provider.get_historical_ohlcv(369 symbol=symbol,370 interval=interval,371 start_time=start_time,372 end_time=end_time,373 days=days374 )375 376 if not data.empty:377 return data378 except Exception as e:379 print(f"Provider {provider.__class__.__name__} failed: {str(e)}")380 continue381 382 # If all providers fail, return empty DataFrame383 return pd.DataFrame()