full_returns_calculation.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "9e73f819",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import multiprocess\n",
    "\n",
    "symbols_list_file_name = \"binance_BTC_from_2019_05_01_to_2022_04_30.json\"\n",
    "with open(symbols_list_file_name, \"r\") as f:\n",
    "    symbols_list = json.load(f)\n",
    "\n",
    "def calculate_returns(symbol):\n",
    "    import os\n",
    "    import pandas as pd\n",
    "    import numpy as np\n",
    "    import pprint as pp\n",
    "    import datetime as dt\n",
    "    import pathlib\n",
    "    import re\n",
    "    import json\n",
    "    import math\n",
    "    exchange = \"binance\"\n",
    "    # data_folder_name = \"full_datasets_by_asset_\" + exchange\n",
    "    data_folder_name = \"full_datasets_against_btc_\" + exchange\n",
    "    extension = \".gz\"\n",
    "    regex_match_to_strip = \"\\.csv.gz$\"\n",
    "    # returns_folder_name = \"full_returns\"\n",
    "    returns_folder_name = \"full_returns_against_btc\"\n",
    "    pattern = \"(.*?)_(.*)\" # date_ticker regex pattern\n",
    "    first_day_first_minute = dt.datetime(2019, 5, 1)\n",
    "    last_day_plus_one_first_minute = dt.datetime(2022, 5, 1)\n",
    "\n",
    "    date_ticker_regex = re.compile(pattern)\n",
    "    initial_time = dt.datetime.utcfromtimestamp(0)\n",
    "    theoretical_first_minute = math.floor(\n",
    "        (first_day_first_minute - initial_time) / dt.timedelta(minutes=1)\n",
    "    )\n",
    "    theoretical_last_minute_plus_one = math.floor(\n",
    "        (last_day_plus_one_first_minute - initial_time) / dt.timedelta(minutes=1)\n",
    "    )\n",
    "\n",
    "\n",
    "    def time_is_in_day(a, timestamp):\n",
    "        # is timestamp (in ms) within 1 day in the future of a (datetime object)?\n",
    "        b = dt.datetime.utcfromtimestamp(timestamp/1000000)\n",
    "        return a <= b < a + dt.timedelta(days=1)\n",
    "\n",
    "    def data_integrity_test(df, date_ticker_regex):\n",
    "        date_ticker_regex_result = date_ticker_regex.match(data_name)\n",
    "        file_name_date = date_ticker_regex_result.group(1)\n",
    "        file_name_ticker = date_ticker_regex_result.group(2)\n",
    "        file_name_datetime_object = dt.datetime.strptime(\n",
    "            file_name_date, \"%Y-%m-%d\"\n",
    "        )\n",
    "\n",
    "        first_time = df.iloc[0]['timestamp']\n",
    "        last_time = df.iloc[-1]['timestamp']\n",
    "\n",
    "        if not pd.Index(df['timestamp']).is_monotonic:\n",
    "            print(\"Timestamp is not monotonic for \" + data_name)\n",
    "\n",
    "        if not pd.Index(df['id']).is_monotonic:\n",
    "            print(\"ID is not monotonic for \" + data_name)\n",
    "\n",
    "        if not (time_is_in_day(file_name_datetime_object, first_time) and \n",
    "                time_is_in_day(file_name_datetime_object, last_time)):\n",
    "            print(\"Time is out of range for \" + data_name)\n",
    "\n",
    "        if df.iloc[0]['symbol'] != file_name_ticker:\n",
    "            print(\"Ticker is wrong for \" + data_name)\n",
    "\n",
    "    pathlib.Path(\n",
    "        os.path.join(os.getcwd(), returns_folder_name)\n",
    "    ).mkdir(parents=True, exist_ok=True)\n",
    "\n",
    "    def keep_latest_trades(df):\n",
    "        df_latests = df.groupby(['minute', 'side']).timestamp.transform(max)\n",
    "        df.drop(df[df.timestamp != df_latests].index, inplace=True)\n",
    "        df.drop(columns=['timestamp'], inplace=True)\n",
    "\n",
    "    def compute_best_sells(df):\n",
    "        sell_df = df[df.side == 'sell'].drop(columns=['side'])\n",
    "        sell_latests_priciest = sell_df.groupby(['minute']).price.transform(max)\n",
    "        sell_df = sell_df[sell_df.price == sell_latests_priciest]\n",
    "        return sell_df.groupby(['minute']).tail(1)\n",
    "\n",
    "    def compute_best_buys(df):\n",
    "        buy_df = df[df.side == 'buy'].drop(columns=['side'])\n",
    "        buy_latests_cheapest = buy_df.groupby(['minute']).price.transform(min)\n",
    "        buy_df = buy_df[buy_df.price == buy_latests_cheapest]\n",
    "        return buy_df.groupby(['minute']).tail(1)\n",
    "\n",
    "    def add_missing_minutes(df):\n",
    "        return (\n",
    "            df\n",
    "            .set_index('minute')\n",
    "            .reindex(\n",
    "                range(theoretical_first_minute, theoretical_last_minute_plus_one), \n",
    "                fill_value=np.NaN\n",
    "            ).reset_index()\n",
    "               )\n",
    "\n",
    "    def fill_in_missing_trades(df):\n",
    "        df.fillna(method='ffill', inplace=True) # best price from prev. min(s)\n",
    "        df.fillna(method='bfill', inplace=True) # only affects start of dataset\n",
    "\n",
    "    def compute_mid_prices(big_sell_df, big_buy_df):\n",
    "        big_sell_df = big_sell_df.rename(columns={'price': 'sell'})\n",
    "        big_buy_df = big_buy_df.drop(columns=['minute'])\n",
    "        big_buy_df = big_buy_df.rename(columns={'price': 'buy'})\n",
    "        big_df = pd.concat([big_sell_df, big_buy_df], axis=1)\n",
    "        big_df['mid'] = (big_df['sell'] + big_df['buy']) / 2\n",
    "        return big_df.drop(columns=['sell', 'buy'])\n",
    "\n",
    "    def compute_log_returns(big_df):\n",
    "        big_df['log_return'] = np.log(big_df['mid'] / big_df['mid'].shift(1))\n",
    "        big_df.drop(columns=['mid'], inplace=True)\n",
    "        big_df.fillna(value=0, inplace=True)\n",
    "\n",
    "    dir_name = os.path.join(os.getcwd(), data_folder_name, symbol)\n",
    "    sell_df_list = []\n",
    "    buy_df_list = []\n",
    "    for root, _, file_names in os.walk(dir_name):\n",
    "        for file_name in file_names:\n",
    "            if file_name.endswith(extension):\n",
    "                file_path = os.path.join(root, file_name)\n",
    "                data_name = re.sub(regex_match_to_strip, '', file_name)\n",
    "#                 print(f\"Walked to {file_path}\")\n",
    "                try:\n",
    "                    df = pd.read_csv(file_path, compression='gzip')\n",
    "                    df.drop(columns=['timestamp'], inplace=True)\n",
    "                    df.rename(\n",
    "                        columns={'local_timestamp': 'timestamp'}, inplace=True\n",
    "                    )\n",
    "                    data_integrity_test(df, date_ticker_regex)\n",
    "                    \n",
    "                    df = df[['timestamp', 'side', 'price']]\n",
    "                    df.loc[:,'minute'] = df.timestamp.apply(\n",
    "                        lambda x: math.floor(\n",
    "                            dt.timedelta(microseconds=x) / dt.timedelta(minutes=1)\n",
    "                        )\n",
    "                    )\n",
    "                    \n",
    "                    keep_latest_trades(df)\n",
    "                    \n",
    "                    sell_df = compute_best_sells(df)\n",
    "                    sell_df_list.append(sell_df)\n",
    "                    \n",
    "                    buy_df = compute_best_buys(df)\n",
    "                    buy_df_list.append(buy_df)\n",
    "                except:\n",
    "                    print(\"No data\")\n",
    "    # write stuff for that symbol\n",
    "    big_sell_df = pd.concat(sell_df_list, ignore_index=True)\n",
    "    big_buy_df = pd.concat(buy_df_list, ignore_index=True)\n",
    "    \n",
    "    big_sell_df = add_missing_minutes(big_sell_df)\n",
    "    big_buy_df = add_missing_minutes(big_buy_df)\n",
    "    fill_in_missing_trades(big_sell_df)\n",
    "    fill_in_missing_trades(big_buy_df)\n",
    "    \n",
    "    big_df = compute_mid_prices(big_sell_df, big_buy_df)\n",
    "    compute_log_returns(big_df)\n",
    "    \n",
    "    # structure: returns_folder_name/symbol/this df as csv.gz\n",
    "    save_file_name = symbol + '.csv.gz'\n",
    "    save_path = os.path.join(os.getcwd(), returns_folder_name, save_file_name)\n",
    "    big_df.to_csv(save_path, compression='gzip', index=False)\n",
    "    print(f\"Saved returns for {symbol}\")\n",
    "    \n",
    "num_workers = multiprocess.cpu_count()\n",
    "with multiprocess.Pool(num_workers) as p:\n",
    "    p.map(calculate_returns, symbols_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f04817ce",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.7.9"
  },
  "vscode": {
   "interpreter": {
    "hash": "88dc6c63d85bacc0fe6a08fac069e7b25e0df598b9d0da137b9635f1d8d13f4d"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}