Last active
October 6, 2025 16:16
-
-
Save sashaboulouds/d6cf5e034e2a505c2337a26e76cf2a83 to your computer and use it in GitHub Desktop.
Collect every 10 seconds all h6 trending Solana pairs on DexScreener — and save to a .csv file 🦅
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # ============================================================================= | |
| # 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()) |
Blender-git
commented
May 22, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment