sashaboulouds/dexscreener_trending_solana_pairs_websocket_scraper.py

## dexscreener_trending_solana_pairs_websocket_scraper.py
# =============================================================================
# Title: DexScreener Crypto Live Prices Scraper
# Description: This script scrape the first 200 h6 trending Solana pairs from DexScreener — every 10 seconds
# Author: Sasha Bouloudnine
# Date: 2024-03-13
#
# Usage:
# - Install websocket using `pip install websockets`.
# - Launch the script.
#
# =============================================================================


import asyncio
import websockets
from datetime import datetime
import os
import base64
import json
import csv
import time

def generate_sec_websocket_key():
    random_bytes = os.urandom(16)
    key = base64.b64encode(random_bytes).decode('utf-8')
    return key

TYPES = ['pairs', 'latestBlock']

DATA = []
FIELDNAMES = [
  "chain_id",
  "dex_id",
  "pair_address",
  "token_address",
  "token_name",
  "token_symbol",
  "token_m5_buys",
  "token_m5_sells",
  "token_h1_buys",
  "token_h1_sells",
  "token_h1_to_m5_buys",
  "token_liquidity",
  "token_market_cap",
  "token_created_at",
  "token_created_since",
  "token_eti",
  "token_header",
  "token_website",
  "token_twitter",
  "token_links",
  "token_img_key",
  "token_price_usd",
  "token_price_change_h24",
  "token_price_change_h6",
  "token_price_change_h1",
  "token_price_change_m5"
]

async def dexscreener_scraper():
    headers = {
      "Host": "io.dexscreener.com",
      "Connection": "Upgrade",
      "Pragma": "no-cache",
      "Cache-Control": "no-cache",
      "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
      "Upgrade": "websocket",
      "Origin": "https://dexscreener.com",
      "Sec-WebSocket-Version": 13,
      "Accept-Encoding": "gzip, deflate, br, zstd",
      "Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
      "Sec-WebSocket-Key": generate_sec_websocket_key()
    }
    uri = "wss://io.dexscreener.com/dex/screener/pairs/h24/1?rankBy[key]=trendingScoreH6&rankBy[order]=desc"
    async with websockets.connect(uri, extra_headers=headers) as websocket:
      while True:
        message_raw = await websocket.recv()
        message = json.loads(message_raw)
        _type = message["type"]
        assert _type in TYPES
        if _type == 'pairs':

          pairs = message["pairs"]
          assert pairs
          for pair in pairs:

            chain_id = pair["chainId"]
            dex_id = pair["dexId"]
            pair_address = pair["pairAddress"]

            assert pair_address

            token_address = pair["baseToken"]["address"]
            token_name = pair["baseToken"]["name"]
            token_symbol = pair["baseToken"]["symbol"]

            token_txns = pair["txns"]

            token_m5_buys = token_txns["m5"]["buys"]
            token_m5_sells = token_txns["m5"]["sells"]

            token_h1_buys = token_txns["h1"]["buys"]
            token_h1_sells = token_txns["h1"]["sells"]

            token_h1_to_m5_buys = round(token_m5_buys*12/token_h1_buys, 2) if token_m5_buys else None

            token_liquidity = pair["liquidity"]["usd"]
            token_market_cap = pair["marketCap"]

            token_created_at_raw = pair["pairCreatedAt"]
            token_created_at = token_created_at_raw / 1000
            token_created_at = datetime.utcfromtimestamp(token_created_at)

            now_utc = datetime.utcnow()
            token_created_since = round((now_utc - token_created_at).total_seconds() / 60, 2)

            token_eti = pair.get("eti", False)
            token_header = pair.get("profile", {}).get("header", False)
            token_website = pair.get("profile", {}).get("website", False)
            token_twitter = pair.get("profile", {}).get("twitter", False)
            token_links = pair.get("profile", {}).get("linkCount", False)
            token_img_key = pair.get("profile", {}).get("imgKey", False)

            token_price_usd = pair["priceUsd"]
            token_price_change_h24 = pair["priceChange"]["h24"]
            token_price_change_h6 = pair["priceChange"]["h6"]
            token_price_change_h1 = pair["priceChange"]["h1"]
            token_price_change_m5 = pair["priceChange"]["m5"]

            VALUES = [
              chain_id,
              dex_id,
              pair_address,
              token_address,
              token_name,
              token_symbol,
              token_m5_buys,
              token_m5_sells,
              token_h1_buys,
              token_h1_sells,
              token_h1_to_m5_buys,
              token_liquidity,
              token_market_cap,
              token_created_at,
              token_created_since,
              token_eti,
              token_header,
              token_website,
              token_twitter,
              token_links,
              token_img_key,
              token_price_usd,
              token_price_change_h24,
              token_price_change_h6,
              token_price_change_h1,
              token_price_change_m5
            ]

            print(token_name, token_price_usd)

            row = dict(zip(FIELDNAMES, VALUES))
            DATA.append(row)

          file_created_at = int(time.time())
          filename = 'dexscreener_%s.csv' % file_created_at
          with open(filename, 'w') as f:
            writer = csv.DictWriter(f, fieldnames=FIELDNAMES, delimiter='\t')
            writer.writeheader()
            for row in DATA:
              writer.writerow(row)
          print('done %s' % filename)
          print('pause 10s :°')
          time.sleep(10)

if __name__ == "__main__":
  asyncio.run(dexscreener_scraper())
	# =============================================================================
	# Title: DexScreener Crypto Live Prices Scraper
	# Description: This script scrape the first 200 h6 trending Solana pairs from DexScreener — every 10 seconds
	# Author: Sasha Bouloudnine
	# Date: 2024-03-13
	#
	# Usage:
	# - Install websocket using `pip install websockets`.
	# - Launch the script.
	#
	# =============================================================================


	import asyncio
	import websockets
	from datetime import datetime
	import os
	import base64
	import json
	import csv
	import time

	def generate_sec_websocket_key():
	random_bytes = os.urandom(16)
	key = base64.b64encode(random_bytes).decode('utf-8')
	return key

	TYPES = ['pairs', 'latestBlock']

	DATA = []
	FIELDNAMES = [
	"chain_id",
	"dex_id",
	"pair_address",
	"token_address",
	"token_name",
	"token_symbol",
	"token_m5_buys",
	"token_m5_sells",
	"token_h1_buys",
	"token_h1_sells",
	"token_h1_to_m5_buys",
	"token_liquidity",
	"token_market_cap",
	"token_created_at",
	"token_created_since",
	"token_eti",
	"token_header",
	"token_website",
	"token_twitter",
	"token_links",
	"token_img_key",
	"token_price_usd",
	"token_price_change_h24",
	"token_price_change_h6",
	"token_price_change_h1",
	"token_price_change_m5"
	]

	async def dexscreener_scraper():
	headers = {
	"Host": "io.dexscreener.com",
	"Connection": "Upgrade",
	"Pragma": "no-cache",
	"Cache-Control": "no-cache",
	"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
	"Upgrade": "websocket",
	"Origin": "https://dexscreener.com",
	"Sec-WebSocket-Version": 13,
	"Accept-Encoding": "gzip, deflate, br, zstd",
	"Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
	"Sec-WebSocket-Key": generate_sec_websocket_key()
	}
	uri = "wss://io.dexscreener.com/dex/screener/pairs/h24/1?rankBy[key]=trendingScoreH6&rankBy[order]=desc"
	async with websockets.connect(uri, extra_headers=headers) as websocket:
	while True:
	message_raw = await websocket.recv()
	message = json.loads(message_raw)
	_type = message["type"]
	assert _type in TYPES
	if _type == 'pairs':

	pairs = message["pairs"]
	assert pairs
	for pair in pairs:

	chain_id = pair["chainId"]
	dex_id = pair["dexId"]
	pair_address = pair["pairAddress"]

	assert pair_address

	token_address = pair["baseToken"]["address"]
	token_name = pair["baseToken"]["name"]
	token_symbol = pair["baseToken"]["symbol"]

	token_txns = pair["txns"]

	token_m5_buys = token_txns["m5"]["buys"]
	token_m5_sells = token_txns["m5"]["sells"]

	token_h1_buys = token_txns["h1"]["buys"]
	token_h1_sells = token_txns["h1"]["sells"]

	token_h1_to_m5_buys = round(token_m5_buys*12/token_h1_buys, 2) if token_m5_buys else None

	token_liquidity = pair["liquidity"]["usd"]
	token_market_cap = pair["marketCap"]

	token_created_at_raw = pair["pairCreatedAt"]
	token_created_at = token_created_at_raw / 1000
	token_created_at = datetime.utcfromtimestamp(token_created_at)

	now_utc = datetime.utcnow()
	token_created_since = round((now_utc - token_created_at).total_seconds() / 60, 2)

	token_eti = pair.get("eti", False)
	token_header = pair.get("profile", {}).get("header", False)
	token_website = pair.get("profile", {}).get("website", False)
	token_twitter = pair.get("profile", {}).get("twitter", False)
	token_links = pair.get("profile", {}).get("linkCount", False)
	token_img_key = pair.get("profile", {}).get("imgKey", False)

	token_price_usd = pair["priceUsd"]
	token_price_change_h24 = pair["priceChange"]["h24"]
	token_price_change_h6 = pair["priceChange"]["h6"]
	token_price_change_h1 = pair["priceChange"]["h1"]
	token_price_change_m5 = pair["priceChange"]["m5"]

	VALUES = [
	chain_id,
	dex_id,
	pair_address,
	token_address,
	token_name,
	token_symbol,
	token_m5_buys,
	token_m5_sells,
	token_h1_buys,
	token_h1_sells,
	token_h1_to_m5_buys,
	token_liquidity,
	token_market_cap,
	token_created_at,
	token_created_since,
	token_eti,
	token_header,
	token_website,
	token_twitter,
	token_links,
	token_img_key,
	token_price_usd,
	token_price_change_h24,
	token_price_change_h6,
	token_price_change_h1,
	token_price_change_m5
	]

	print(token_name, token_price_usd)

	row = dict(zip(FIELDNAMES, VALUES))
	DATA.append(row)

	file_created_at = int(time.time())
	filename = 'dexscreener_%s.csv' % file_created_at
	with open(filename, 'w') as f:
	writer = csv.DictWriter(f, fieldnames=FIELDNAMES, delimiter='\t')
	writer.writeheader()
	for row in DATA:
	writer.writerow(row)
	print('done %s' % filename)
	print('pause 10s :°')
	time.sleep(10)

	if __name__ == "__main__":
	asyncio.run(dexscreener_scraper())