如何从Coinmarketcap提取全量加密货币符号并修改现有代码

修改说明 #

1. 替换原有手动输入币种符号的逻辑，新增全币种符号批量抓取功能，一次性获取平台所有加密货币符号
2. 优化代码结构：将通用函数定义移到循环外部，避免重复定义消耗资源
3. 新增请求头伪装浏览器访问，降低被反爬拦截的概率
4. 新增请求间隔逻辑，避免短时间高频请求触发平台限流
5. 优化文件读写逻辑，减少重复打开文件的性能损耗

完整修改后代码 #

import time
import requests
import re
import pandas as pd

# 基础配置
SYMBOL_FILE_PATH = '/Users/DearUser/crypto_symbol_3.txt'
START_DATE = '2021-11-09'
END_DATE = '2021-11-11'
# 请求头伪装浏览器
REQUEST_HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
}

def get_timestamp(datetime_str: str):
    return int(time.mktime(time.strptime(datetime_str, '%Y-%m-%d')))

def get_crypto_quotes(symbol: str, start_date: str, end_date: str):
    start = get_timestamp(start_date)
    end = get_timestamp(end_date)
    url = f'https://web-api.coinmarketcap.com/v1/cryptocurrency/ohlcv/historical?symbol={symbol}&convert=USD&time_start={start}&time_end={end}'
    resp = requests.get(url, headers=REQUEST_HEADERS, timeout=10)
    resp.raise_for_status()
    return resp.json()

def get_all_crypto_symbols():
    """批量获取所有加密货币符号"""
    url = 'https://coinmarketcap.com/coins/views/all/'
    resp = requests.get(url, headers=REQUEST_HEADERS, timeout=15)
    resp.raise_for_status()
    # 正则匹配页面内嵌的初始币种数据
    data_match = re.search(r'window\.__INITIAL_STATE__ = (.*?);', resp.text)
    if not data_match:
        raise Exception('未能获取到币种数据')
    # 提取所有币种符号并去重
    symbol_pattern = re.compile(r'"symbol":"([A-Z0-9]+?)"')
    symbols = list(set(symbol_pattern.findall(data_match.group(1))))
    return symbols

if __name__ == '__main__':
    # 第一步：获取全币种符号列表
    print('正在获取全币种列表...')
    all_symbols = get_all_crypto_symbols()
    print(f'共获取到{len(all_symbols)}个加密货币符号')

    # 读取已存在的符号记录，避免重复写入
    try:
        with open(SYMBOL_FILE_PATH, 'r', encoding='utf-8') as f:
            existed_content = f.read()
    except FileNotFoundError:
        existed_content = ''
        open(SYMBOL_FILE_PATH, 'a').close()

    # 第二步：遍历所有符号执行原有逻辑
    for symbol in all_symbols:
        try:
            print(f'正在处理：{symbol}')
            # 跳过已存在的符号
            if symbol in existed_content:
                print(f'{symbol} 已存在，跳过')
                time.sleep(0.5)
                continue

            data = get_crypto_quotes(symbol, START_DATE, END_DATE)
            data_flat = [quote['quote']['USD'] for quote in data['data']['quotes']]
            df = pd.DataFrame(data_flat)
            market_cap = df.iat[1, 5]

            if market_cap == 0:
                print(f'{symbol} 市值为0，跳过')
                time.sleep(0.5)
                continue

            # 写入符合要求的记录
            with open(SYMBOL_FILE_PATH, 'a', encoding='utf-8') as f:
                f.write(f'"{symbol}": {market_cap},\n')
            existed_content += f'"{symbol}": {market_cap},\n'
            print(f'{symbol} 处理完成')
            # 增加请求间隔，避免触发限流
            time.sleep(1)
        except Exception as e:
            print(f'{symbol} 处理失败，跳过，错误信息：{str(e)}')
            time.sleep(0.5)
            continue

    # 输出最终文件内容
    print('\n最终文件内容：')
    with open(SYMBOL_FILE_PATH, 'r', encoding='utf-8') as f:
        print(f.read())

注意事项 #

• 可根据实际运行情况调整time.sleep()的参数值，降低被平台反爬拦截的概率
• 若需要统计其他日期的市值数据，直接修改START_DATE和END_DATE两个参数即可
• 若出现全币种列表获取失败的情况，可尝试将请求头中的User-Agent替换为你当前浏览器的实际UA值

内容的提问来源于stack exchange，提问作者Raboon