binance_ws_api.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "id": "dLFeWoavs9dS",
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'/root/miniconda3/BINANCE'"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import os\n",
    "os.getcwd()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[{'symb': 'XEMUSDT',\n",
      "  'pct': 700,\n",
      "  'trades': 8,\n",
      "  'avg': 1,\n",
      "  'url': 'https://binance.com/en/futures/XEMUSDT'},\n",
      " {'symb': 'LQTYUSDT',\n",
      "  'pct': 205,\n",
      "  'trades': 58,\n",
      "  'avg': 19,\n",
      "  'url': 'https://binance.com/en/futures/LQTYUSDT'},\n",
      " {'symb': 'DYDXUSDT',\n",
      "  'pct': 178,\n",
      "  'trades': 802,\n",
      "  'avg': 288,\n",
      "  'url': 'https://binance.com/en/futures/DYDXUSDT'}]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import websocket\n",
    "import threading\n",
    "import json\n",
    "from collections import deque\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from pprint import pprint\n",
    "from time import sleep\n",
    "from IPython.display import clear_output\n",
    "from datetime import datetime\n",
    "from tokens import token_list\n",
    "\n",
    "symbols = [s.lower() for s in token_list]\n",
    "\n",
    "# only for testing\n",
    "# symbols = ['ethusdt']\n",
    "\n",
    "sockets = {}\n",
    "\n",
    "# ========= [MAIN CONFIG] =============\n",
    "# default settings\n",
    "INTERVAL = '1m'\n",
    "SHOW_FUTURES = True\n",
    "COLLECT_EVERY_MINUTE = True\n",
    "MAX_KLINE_COUNT = 10\n",
    "COEF_TRADE = 2 # \n",
    "COEF_RANGE = 3 #\n",
    "PCT_THRESHOLD = 80 # %\n",
    "HIGH_THRESHOLD = 50 # %\n",
    "LOW_THRESHOLD = 50 # %\n",
    "# =====================================\n",
    "\n",
    "BINANCE_URL = 'wss://stream.binance.com:9443/ws/{}@kline_'+INTERVAL\n",
    "URL = 'https://binance.com/en/futures/{}' if SHOW_FUTURES else 'https://binance.com/en/trade/{}'\n",
    "\n",
    "\n",
    "def pct_change(old_value, new_value):\n",
    "    change = new_value - old_value\n",
    "    return (change / old_value) * 100\n",
    "\n",
    "def on_message(ws, message, kline_buffer):\n",
    "    data = json.loads(message)\n",
    "    \n",
    "    # check if the message is a kline event\n",
    "    if 'k' in data:\n",
    "        symbol = data['s']\n",
    "        kline_data = data['k']\n",
    "        trade_count = int(kline_data['n'])\n",
    "        is_closed = bool(kline_data['x'])  \n",
    "        open_price = float(kline_data['o'])\n",
    "        close_price = float(kline_data['c'])\n",
    "        high_price = float(kline_data['h'])\n",
    "        low_price = float(kline_data['l'])\n",
    "        event_time = datetime.utcfromtimestamp(data['E'] / 1000).strftime('%H:%M:%S')\n",
    "\n",
    "        if COLLECT_EVERY_MINUTE:\n",
    "            if is_closed:\n",
    "                kline_buffer['trades'].append(trade_count)\n",
    "                kline_buffer['ohlc'].append({'open':open_price,\n",
    "                                             'high':high_price,\n",
    "                                             'low':low_price,\n",
    "                                             'close':close_price})     \n",
    "            else: return\n",
    "        else:\n",
    "            kline_buffer['trades'].append(trade_count)\n",
    "            kline_buffer['ohlc'].append({'open':open_price,\n",
    "                                         'high':high_price,\n",
    "                                         'low':low_price,\n",
    "                                         'close':close_price})            \n",
    "\n",
    "        if len(kline_buffer['trades']) >= MAX_KLINE_COUNT:\n",
    "            average = int(COEF_TRADE * float(np.mean(np.array(kline_buffer['trades']))))\n",
    "            if average and trade_count: # check for non zero\n",
    "                pct = round(pct_change(average,trade_count)) # calc percent change\n",
    "                if pct >= PCT_THRESHOLD: # main logic\n",
    "\n",
    "                    # convert buffer to dataframe\n",
    "                    df = pd.DataFrame(kline_buffer['ohlc'])\n",
    "\n",
    "                    # get ohlc data from last bar\n",
    "                    last_row = df.iloc[-1]\n",
    "                    open = last_row['open']\n",
    "                    high = last_row['high']\n",
    "                    low = last_row['low']\n",
    "                    close = last_row['close']\n",
    "\n",
    "                    # get average 'range' * coef\n",
    "                    df['all_range'] = df['high'] - df['low']\n",
    "                    mean_range = df[df['all_range'] != 0]['all_range'].mean() * COEF_RANGE\n",
    "                    \n",
    "                    # get 'range' from last bar\n",
    "                    range = (high - low) \n",
    "                    \n",
    "                    # filter bars only with long 'range'\n",
    "                    if range > mean_range: \n",
    "                        high_shadow = round((abs(high - max(close, open)) / range) * 100)\n",
    "                        low_shadow = round((abs(low -  min(close, open)) / range) * 100)\n",
    "                        \n",
    "                        # filter by high&low shadows size\n",
    "                        if high_shadow >= HIGH_THRESHOLD or low_shadow >= LOW_THRESHOLD:\n",
    "                            # drop last ohlc row\n",
    "                            df = df[:-1]\n",
    "                            # check if var is less or great than all values in the series\n",
    "                            if high_shadow > low_shadow:\n",
    "                                is_needed_bar = (high > df['high']).all()\n",
    "                                signal = 'SELL'\n",
    "                            elif low_shadow > high_shadow:\n",
    "                                is_needed_bar = (low < df['low']).all()\n",
    "                                signal = 'BUY'\n",
    "                            else: is_needed_bar = False\n",
    "                                \n",
    "                            if is_needed_bar:\n",
    "                                # print(f\"{event_time} ┇ pct_change: +{pct}% ┇ high_shadow: {high_shadow}% ┇ low_shadow: {low_shadow}% ┇ {URL.format(symbol)}\")\n",
    "                                print(f\"{event_time} ┇ signal: {signal} ┇ {URL.format(symbol)}\")\n",
    "\n",
    "\n",
    "\n",
    "def on_error(ws, error):\n",
    "    print(f\"Error: {error}\")\n",
    "\n",
    "def on_close(ws, close_status_code, close_msg):\n",
    "    print(f\"Closed connection: {close_status_code}\")\n",
    "\n",
    "# thread WebSocket connection for a given symbol\n",
    "def socket_thread(symbol):\n",
    "    kline_buffer = {'ohlc': deque(maxlen=MAX_KLINE_COUNT), 'trades': deque(maxlen=MAX_KLINE_COUNT)}\n",
    "    \n",
    "    def on_message_wrapper(ws, message):\n",
    "        nonlocal kline_buffer\n",
    "        on_message(ws, message, kline_buffer)\n",
    "        \n",
    "    websocket.WebSocketApp(BINANCE_URL.format(symbol), on_message=on_message_wrapper, on_error=on_error, on_close=on_close).run_forever()\n",
    "\n",
    "\n",
    "# start WebSocket connections for each symbol\n",
    "for symbol in symbols:\n",
    "    sockets[symbol] = threading.Thread(target=socket_thread, args=(symbol,))\n",
    "    sockets[symbol].start()\n",
    "\n",
    "# keep main thread alive\n",
    "try:\n",
    "    print(f'Размер token_list: {len(token_list)}')\n",
    "    print('Запускаю воркеры ... Ждем результаты ........\\n')\n",
    "    while True: pass\n",
    "except KeyboardInterrupt:\n",
    "    print(\"Stopping WebSocket connections...waiting...\")\n",
    "    for symbol in symbols: sockets[symbol].join()\n",
    "    print(\"WebSocket connections stopped.\")"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "provenance": []
  },
  "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.8.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}