tool_get_market_data_parallel.py

import time
import requests
import pandas as pd
from datetime import datetime
import os
from concurrent.futures import ThreadPoolExecutor, as_completed

# API Setup
BASE_URL = "https://api.binance.com/api/v3"
KLINE_ENDPOINT = f"{BASE_URL}/klines"

def ensure_dir_exists(path):
    os.makedirs(path, exist_ok=True)

def fetch_historical_data(symbol, interval, start_date=None, end_date=None):
    if not start_date:
        start_date = datetime(2019, 1, 1)
    if not end_date:
        end_date = datetime.now()

    start_timestamp = int(start_date.timestamp() * 1000)
    end_timestamp = int(end_date.timestamp() * 1000)

    all_klines = []
    current_timestamp = start_timestamp

    while current_timestamp < end_timestamp:
        url = f"{KLINE_ENDPOINT}?symbol={symbol}&interval={interval}&limit=1000&startTime={current_timestamp}"

        try:
            response = requests.get(url)
            response.raise_for_status()
            klines = response.json()

            if not klines:
                break

            all_klines.extend(klines)
            current_timestamp = klines[-1][0] + 1
            print(f"[{symbol}-{interval}] ✅ Up to {datetime.fromtimestamp(current_timestamp / 1000)}")

            time.sleep(0.3)

        except requests.exceptions.RequestException as e:
            raise RuntimeError(f"[{symbol}-{interval}] ❌ Fetch error: {e}")

    if not all_klines:
        return pd.DataFrame()

    df = pd.DataFrame(all_klines, columns=[
        'timestamp', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_asset_volume', 'number_of_trades',
        'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'
    ])

    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)

    numeric_columns = ['open', 'high', 'low', 'close', 'volume', 'quote_asset_volume',
                       'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume']
    df[numeric_columns] = df[numeric_columns].apply(pd.to_numeric, axis=1)
    df = df[~df.index.duplicated(keep='first')]
    return df

def save_to_csv(df, filename):
    df = df.iloc[:-1]
    df.to_csv(filename)
    print(f"💾 Saved to {filename}")

def fetch_and_save(symbol, interval, max_retries=10, backoff_base=2):
    attempt = 0
    while attempt < max_retries:
        try:
            df = fetch_historical_data(symbol, interval)
            if not df.empty:
                filepath = f"datasets/raw/full_scope/{interval}/{symbol}_{interval}_historical_data.csv"
                ensure_dir_exists(os.path.dirname(filepath))
                save_to_csv(df, filepath)
                return
            else:
                print(f"[{symbol}-{interval}] ⚠️ Empty dataset received. Retrying...")

        except Exception as e:
            wait = min(backoff_base ** attempt, 300)  # Cap delay at 5 minutes
            print(f"[{symbol}-{interval}] ❌ Error: {e} — Retrying in {wait}s...")
            time.sleep(wait)
            attempt += 1

    print(f"[{symbol}-{interval}] ❌ Final failure after {max_retries} attempts.")

def main():
    symbols = [
        "ZRXUSDT", "ZILUSDT", "ZECUSDT", "XTZUSDT", "XRPUSDT", "XLMUSDT",
        "WINUSDT", "WANUSDT", "VITEUSDT", "VETUSDT", "USDCUSDT", "TUSDUSDT",
        "TRXUSDT", "TROYUSDT", "THETAUSDT", "TFUELUSDT", "STXUSDT", "RVNUSDT",
        "RLCUSDT", "ONTUSDT", "ONGUSDT", "ONEUSDT", "NULSUSDT", "NKNUSDT",
        "NEOUSDT", "MTLUSDT", "LTCUSDT", "LINKUSDT", "KAVAUSDT", "IOTXUSDT",
        "IOTAUSDT", "IOSTUSDT", "ICXUSDT", "HOTUSDT", "HBARUSDT", "FUNUSDT",
        "FTTUSDT", "FTMUSDT", "FETUSDT", "ETHUSDT", "ETCUSDT", "EOSUSDT",
        "ENJUSDT", "DUSKUSDT"
    ]

    # Total Intervals: ['1m', '3m', '5m', '15m', '30m', '1h', '2h', '4h', '6h', '8h', '12h', '1d', '3d', '1w']
    intervals = ['1m']
    max_workers = 1

    tasks = [(symbol, interval) for interval in intervals for symbol in symbols]

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(fetch_and_save, symbol, interval) for symbol, interval in tasks]
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                print(f"🚨 Unexpected thread failure: {e}")

if __name__ == "__main__":
    main()