Collect every 10 seconds all h6 trending Solana pairs on DexScreener — and save to a .csv file 🦅

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())