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()) |
@prodpeak , Hi, how can i contact you regarding your work?
Share your telegram username and I'll DM you
Hi, how can i contact you regarding your work?
I have been looking for an answer to this problem for a long time
v4
now that's doesn't work:')
@prodpeak hi sir please how can i joind you on telegram
@prodpeak heya, I wanna checkout your code, tele is https://t.me/stevegremory
Hi @prodpeak , were you able to find a solution to this. can i please have a look at the code my telegram is https://t.me/kratkrish
Hi, how can i contact you regarding your work?
@BOTtellegrams
I have been looking for an answer to this problem for a long time
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
@prodpeak , Hi, how can i contact you regarding your work?