sashaboulouds · October 6, 2025 16:16 · prodpeak · Nov 28, 2024 · prodpeak · Nov 28, 2024
diff --git a/dexscreener_trending_solana_pairs_websocket_scraper.py b/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())
No results found