get_historical_candle_stick_data.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The autoreload extension is already loaded. To reload it, use:\n",
      "  %reload_ext autoreload\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "import ccxt\n",
    "import datetime\n",
    "import time\n",
    "from tqdm import tqdm\n",
    "from termcolor import colored\n",
    "import os\n",
    "\n",
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Retrieve candle stick data \n",
    "- requesting data from phemex upto a certain point with chunking to prevent rate liming"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initialize the exchange - Using Phemex\n",
    "# exchange = ccxt.phemex({\"enableRateLimit\": True})\n",
    "exchange = ccxt.binance()  # need a VPN if your in the US\n",
    "\n",
    "# Define the symbols\n",
    "symbol = \"BTCUSDT\"  # This is the correct symbol for Phemex USDT-margined perpetual\n",
    "timeframe = \"5m\"  # You can change this to any of the supported timeframes\n",
    "\n",
    "# Calculate start and end times\n",
    "end_time = datetime.datetime.now()\n",
    "start_time = end_time - datetime.timedelta(days=365 * 5)  # 1 year worth of data\n",
    "\n",
    "\"\"\"\n",
    "'timeframes': {\n",
    "    '1s': '1s',  # spot only for now\n",
    "    '1m': '1m',\n",
    "    '3m': '3m',\n",
    "    '5m': '5m',\n",
    "    '15m': '15m',\n",
    "    '30m': '30m',\n",
    "    '1h': '1h',\n",
    "    '2h': '2h',\n",
    "    '4h': '4h',\n",
    "    '6h': '6h',\n",
    "    '8h': '8h',\n",
    "    '12h': '12h',\n",
    "    '1d': '1d',\n",
    "    '3d': '3d',\n",
    "    '1w': '1w',\n",
    "    '1M': '1M',\n",
    "},\n",
    "\"\"\"\n",
    "\n",
    "\n",
    "# Function to fetch data in chunks\n",
    "def fetch_ohlcv_data(start_time, end_time, timeframe):\n",
    "    all_candles = []\n",
    "    current_time = start_time\n",
    "\n",
    "    timeframe_durations = {\n",
    "        \"1s\": datetime.timedelta(seconds=1),\n",
    "        \"1m\": datetime.timedelta(minutes=1),\n",
    "        \"3m\": datetime.timedelta(minutes=3),\n",
    "        \"5m\": datetime.timedelta(minutes=5),\n",
    "        \"15m\": datetime.timedelta(minutes=15),\n",
    "        \"30m\": datetime.timedelta(minutes=30),\n",
    "        \"1h\": datetime.timedelta(hours=1),\n",
    "        \"2h\": datetime.timedelta(hours=2),\n",
    "        \"4h\": datetime.timedelta(hours=4),\n",
    "        \"6h\": datetime.timedelta(hours=6),\n",
    "        \"8h\": datetime.timedelta(hours=8),\n",
    "        \"12h\": datetime.timedelta(hours=12),\n",
    "        \"1d\": datetime.timedelta(days=1),\n",
    "        \"3d\": datetime.timedelta(days=3),\n",
    "        \"1w\": datetime.timedelta(weeks=1),\n",
    "        \"1M\": datetime.timedelta(days=30),  # Approximation for 1 month\n",
    "    }\n",
    "\n",
    "    if timeframe not in timeframe_durations:\n",
    "        raise ValueError(f\"Unsupported timeframe: {timeframe}\")\n",
    "\n",
    "    duration = timeframe_durations[timeframe]\n",
    "\n",
    "    # Calculate total number of iterations for progress bar\n",
    "    total_iterations = (end_time - start_time) // duration\n",
    "\n",
    "    # Start the timer\n",
    "    start_time_perf = time.time()\n",
    "\n",
    "    # Initialize progress bar\n",
    "    pbar = tqdm(\n",
    "        total=total_iterations, desc=colored(\"Fetching data\", \"cyan\"), ncols=100\n",
    "    )\n",
    "\n",
    "    while current_time < end_time:\n",
    "        try:\n",
    "            pbar.set_postfix_str(f\"Current time: {current_time}\")\n",
    "            candles = exchange.fetch_ohlcv(\n",
    "                symbol,\n",
    "                timeframe,\n",
    "                since=int(current_time.timestamp() * 1000),\n",
    "                limit=1000,\n",
    "            )\n",
    "\n",
    "            if not candles:\n",
    "                break\n",
    "\n",
    "            all_candles.extend(candles)\n",
    "\n",
    "            # Update current_time to the last candle's time + 1 timeframe\n",
    "            last_candle_time = datetime.datetime.fromtimestamp(candles[-1][0] / 1000)\n",
    "            current_time = last_candle_time + duration\n",
    "\n",
    "            # Update progress bar\n",
    "            pbar.update(len(candles))\n",
    "\n",
    "            # Print colored log message\n",
    "            tqdm.write(\n",
    "                colored(\n",
    "                    f\"Fetched {len(candles)} candles. Next fetch from {current_time}\",\n",
    "                    \"green\",\n",
    "                )\n",
    "            )\n",
    "\n",
    "            # Sleep to respect rate limits\n",
    "            time.sleep(exchange.rateLimit / 1000)\n",
    "\n",
    "        except ccxt.NetworkError as e:\n",
    "            tqdm.write(\n",
    "                colored(\n",
    "                    f\"Network error occurred: {str(e)}. Retrying in 10 seconds...\",\n",
    "                    \"yellow\",\n",
    "                )\n",
    "            )\n",
    "            time.sleep(10)\n",
    "        except ccxt.ExchangeError as e:\n",
    "            tqdm.write(colored(f\"Exchange error occurred: {str(e)}. Stopping.\", \"red\"))\n",
    "            break\n",
    "\n",
    "    # Close progress bar\n",
    "    pbar.close()\n",
    "\n",
    "    # End the timer\n",
    "    end_time_perf = time.time()\n",
    "\n",
    "    # Calculate the execution time\n",
    "    execution_time = end_time_perf - start_time_perf\n",
    "\n",
    "    print(\n",
    "        colored(f\"Data fetching completed in {execution_time:.2f} seconds\", \"magenta\")\n",
    "    )\n",
    "\n",
    "    return all_candles, execution_time\n",
    "\n",
    "\n",
    "# Fetch the data and measure performance\n",
    "candles, fetch_time = fetch_ohlcv_data(start_time, end_time, timeframe)\n",
    "\n",
    "# Create DataFrame\n",
    "df = pd.DataFrame(\n",
    "    candles, columns=[\"Timestamp\", \"Open\", \"High\", \"Low\", \"Close\", \"Volume\"]\n",
    ")\n",
    "\n",
    "# Convert Timestamp to datetime and set as index\n",
    "df[\"Timestamp\"] = pd.to_datetime(df[\"Timestamp\"], unit=\"ms\")\n",
    "df.set_index(\"Timestamp\", inplace=True)\n",
    "\n",
    "# Convert columns to appropriate types\n",
    "df = df.astype(\n",
    "    {\n",
    "        \"Open\": \"float64\",\n",
    "        \"High\": \"float64\",\n",
    "        \"Low\": \"float64\",\n",
    "        \"Close\": \"float64\",\n",
    "        \"Volume\": \"float64\",\n",
    "    }\n",
    ")\n",
    "\n",
    "print(df.dtypes)  # Print data types\n",
    "print(df.head())  # Print the first few rows of the dataframe\n",
    "\n",
    "\n",
    "# Function to calculate expected number of candles\n",
    "def calculate_expected_candles(start_time, end_time, timeframe):\n",
    "    timeframe_minutes = {\n",
    "        \"1s\": 1 / 60,\n",
    "        \"1m\": 1,\n",
    "        \"3m\": 3,\n",
    "        \"5m\": 5,\n",
    "        \"15m\": 15,\n",
    "        \"30m\": 30,\n",
    "        \"1h\": 60,\n",
    "        \"2h\": 120,\n",
    "        \"4h\": 240,\n",
    "        \"6h\": 360,\n",
    "        \"8h\": 480,\n",
    "        \"12h\": 720,\n",
    "        \"1d\": 1440,\n",
    "        \"3d\": 4320,\n",
    "        \"1w\": 10080,\n",
    "        \"1M\": 43200,  # Approximation for 1 month (30 days)\n",
    "    }\n",
    "    if timeframe not in timeframe_minutes:\n",
    "        raise ValueError(f\"Unsupported timeframe: {timeframe}\")\n",
    "    duration = end_time - start_time\n",
    "    total_minutes = duration.total_seconds() / 60\n",
    "    expected_candles = total_minutes / timeframe_minutes[timeframe]\n",
    "    return int(expected_candles)\n",
    "\n",
    "\n",
    "# Calculate and print expected number of candles\n",
    "expected_candles = calculate_expected_candles(start_time, end_time, timeframe)\n",
    "print(colored(f\"Expected number of candles: {expected_candles}\", \"cyan\"))\n",
    "print(colored(f\"Actual number of candles: {len(df)}\", \"cyan\"))\n",
    "print(colored(f\"Total execution time: {fetch_time:.2f} seconds\", \"magenta\"))\n",
    "\n",
    "df  # Return the dataframe for display in Jupyter"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generate a dynamic output file name & save data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_exchange_data(df, exchange, symbol, timeframe, file_format=\"parquet\"):\n",
    "    # Extract the exchange name\n",
    "    exchange_name = exchange.id.lower()  # This will give us 'binance' or 'phemex'\n",
    "\n",
    "    # Check if 'Timestamp' is the index\n",
    "    if df.index.name == \"Timestamp\":\n",
    "        start_date = df.index.min().strftime(\"%Y%m%d\")\n",
    "        end_date = df.index.max().strftime(\"%Y%m%d\")\n",
    "    else:\n",
    "        start_date = df[\"Timestamp\"].min().strftime(\"%Y%m%d\")\n",
    "        end_date = df[\"Timestamp\"].max().strftime(\"%Y%m%d\")\n",
    "\n",
    "    # Validate and set the file format\n",
    "    if file_format.lower() not in [\"csv\", \"parquet\"]:\n",
    "        raise ValueError(\"Invalid file format. Choose 'csv' or 'parquet'.\")\n",
    "\n",
    "    file_format = file_format.lower()\n",
    "\n",
    "    # Create the directory structure\n",
    "    base_dir = \"./saved_candlestick_data\"\n",
    "    if exchange_name == \"binance\":\n",
    "        directory = os.path.join(base_dir, exchange_name, file_format)\n",
    "    else:  # for phemex and potentially other exchanges\n",
    "        directory = os.path.join(\n",
    "            base_dir, exchange_name, \"parquet\"\n",
    "        )  # Always use 'parquet' for non-Binance exchanges\n",
    "\n",
    "    os.makedirs(directory, exist_ok=True)\n",
    "\n",
    "    # Create the dynamic filename\n",
    "    if exchange_name == \"binance\":\n",
    "        filename = f\"{exchange_name}_{symbol}_{timeframe}_{start_date}_{end_date}.{file_format}\"\n",
    "    else:  # for phemex and potentially other exchanges\n",
    "        filename = f\"{symbol}_{timeframe}_{start_date}_{end_date}.parquet\"  # Always use .parquet for non-Binance exchanges\n",
    "\n",
    "    output_filename = os.path.join(directory, filename)\n",
    "\n",
    "    # Save the file in the specified format\n",
    "    if file_format == \"csv\" and exchange_name == \"binance\":\n",
    "        df.to_csv(output_filename)\n",
    "    else:  # parquet\n",
    "        df.to_parquet(output_filename)\n",
    "\n",
    "    print(f\"Data saved as {file_format.upper()} file: {output_filename}\")\n",
    "\n",
    "    return output_filename\n",
    "\n",
    "\n",
    "# Save the data in both CSV and Parquet formats\n",
    "csv_file = save_exchange_data(df, exchange, symbol, timeframe, file_format=\"csv\")\n",
    "parquet_file = save_exchange_data(\n",
    "    df, exchange, symbol, timeframe, file_format=\"parquet\"\n",
    ")\n",
    "\n",
    "print(f\"Data saved as CSV: {csv_file}\")\n",
    "print(f\"Data saved as Parquet: {parquet_file}\")\n",
    "\n",
    "csv_size = os.path.getsize(csv_file)\n",
    "parquet_size = os.path.getsize(parquet_file)\n",
    "\n",
    "print(f\"CSV file size: {csv_size / 1024 / 1024:.2f} MB\")\n",
    "print(f\"Parquet file size: {parquet_size / 1024 / 1024:.2f} MB\")\n",
    "print(f\"Parquet file is {csv_size / parquet_size:.2f}x smaller than CSV\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!eza -la --tree saved_candlestick_data/binance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}