如何使用 Python 提取加密货币价格数据

·

加密货币全天 24 小时成交,价格波动在分秒之间,交易员、投资者和量化分析师若想占得先机,就需要“实时数据”和“历史数据”两条腿走路。Python 凭借其丰富的金融数据接口数据处理库可视化工具,成为获取并分析加密货币价格的利器。本文将手把手演示如何用 Python 提取加密货币价格数据、清洗与存储,以及快速做出可视化和技术指标,让任何人都能从零构建自己的研究或交易脚本。


为什么要用 Python 提取加密货币价格?


环境搭建:3 分钟搞定依赖

Step 1 安装库

pip install requests pandas matplotlib python-dotenv

Step 2 选择数据源并获取 API Key

数据源免费额度/频率特点
CoinGecko50–100 req/min免注册即可用,币种齐全
Binance REST1200 req/min深度高,支持期货、现货、逐笔
CoinMarketCap333 req/day综合排名、资金面、情绪一目了然

以 Binance 为例:

  1. 登陆后创建 API,只勾选 读取行情
  2. API_KEY=你的key 写入 .env,脚本中用 os.getenv("API_KEY") 读取。

👉 想把海量数据一键拉取,却还不会写复杂签名?点这里看调试秘籍


实时数据:一步拿到最新成交价

通过 CoinGecko 免费接口抓取 BTC 对美元价格:

import requests

url = "https://api.coingecko.com/api/v3/simple/price"
params = {"ids": "bitcoin", "vs_currencies": "usd"}
price = requests.get(url, params=params, timeout=5).json()
print(f"BTC/USDT 最新价: ${price['bitcoin']['usd']}")

输出示例(实时变化):

BTC/USDT 最新价: $86327

扩展技巧:替换 ids 参数即可一次查询 250+ 主流币,如 "ids": "ethereum,solana,dogecoin",实现“网页行情看板”式监控。


历史数据:获取完整 OHLCV 供回测

Binance 的 K 线接口支持最多 1000 根,默认 1 小时频率,可按需调整为 1m、5m、1d、1w。

import pandas as pd
import requests, datetime as dt

symbol  = "BTCUSDT"
interval = "1d"        # 日线
limit    = 500         # 最多 1000 根

url = f"https://api.binance.com/api/v3/klines"
params = dict(symbol=symbol, interval=interval, limit=limit)
raw = requests.get(url, params=params).json()

cols = ['timestamp','open','high','low','close','volume',
        'close_time','quote_vol','trades','taker_base','taker_quote','ignore']
df = pd.DataFrame(raw, columns=cols)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
for col in ['open','high','low','close','volume']:
    df[col] = df[col].astype(float)

数据前 5 行示意:

timestampopenhighlowclosevolume
2024-05-0160670.161555.059668.060550.4104567
……

数据清洗与存储

  1. 时间本地化df.set_index('timestamp', inplace=True),直接支持按时间切片。
  2. 去除重复/异常

    df = df.loc[~df.index.duplicated(keep='first')]
    df = df[df['volume'] > 0]      # 筛掉零成交量异常行情
  3. 本地持久化

    df.to_csv("btc_daily_ohlcv.csv")

👉 还在为分钟级数据的存取头疼?看看极客如何3秒实现极速落盘


分析场景:均线 + 布林带 5 行代码完成

把数据搬进策略笔记本:

import pandas as pd, matplotlib.pyplot as plt

df = pd.read_csv("btc_daily_ohlcv.csv", index_col='timestamp', parse_dates=True)

# 50/200 日均线
df['ma50'] = df['close'].rolling(50).mean()
df['ma200'] = df['close'].rolling(200).mean()

# 布林带 (20,2)
df['basis'] = df['close'].rolling(20).mean()
df['upper'] = df['basis'] + 2*df['close'].rolling(20).std()
df['lower'] = df['basis'] - 2*df['close'].rolling(20).std()

plt.figure(figsize=(14,6))
plt.plot(df.index, df['close'], label='Close')
plt.plot(df.index, df['ma50'], label='MA50')
plt.plot(df.index, df['ma200'], label='MA200')
plt.fill_between(df.index, df['upper'], df['lower'], color='gray', alpha=0.2)
plt.title("BTC/USDT 日线趋势")
plt.legend()
plt.show()

FAQ:被问爆的 5 个高频问题

Q1:免费接口会不会突然失效?
A:官方限制通常以 IP 维度 429 报错。解决方法是:

Q2:需要买 VPN 吗?
A:不需要。遵守各交易所 ToS,用公开 REST 即可。切勿抓取需登录或限制区域数据。

Q3:分钟级数据太占硬盘怎么办?
A:存成 Parquet + Blosc 压缩(pip install pyarrow),相比 CSV 节省 70% 体积;或用云盘冷存,按目录分区日期。

Q4:如何把 WebSocket 数据实时落地?
A:

import asyncio, json, aiofiles, websockets

async def save_stream():
    url = "wss://stream.binance.com:9443/ws/btcusdt@ticker"
    async with websockets.connect(url) as ws, aiofiles.open("tick.log", "a") as f:
        while True:
            tick = await ws.recv()
            await f.write(tick + "\n")

asyncio.run(save_stream())

每秒写入一条 ticker,再进行离线批处理。

Q5:怎么防止程序崩溃导致数据断档?
A:做三重保险

  1. 在循环中捕获异常并记录日志;
  2. 每 100 条持久化一次;
  3. 使用 scheduleAPScheduler 做守护进程+重启策略,确保 7×24 小时无缝。

小结 & 下一步探索

掌握“获取—清洗—分析—可视化”四步后,你已经可以在任何币种上快速复刻类似流程:

现在脚本已跑通,接下来你可以:

  1. 用 Airflow 定时收集 100+ 币种日线;
  2. 用 Streamlit 做交互式行情网页;
  3. 接入券商 API 实现全自动量化交易。

祝你在加密世界一路向上,量化长红!