3. K线数据获取
做量化交易，第一步就是搞定数据。没有数据，你什么策略都跑不起来。说实话，币安的API文档写得不算差，但新手第一次上手还是容易踩坑。今天我就带你走一遍——怎么从币安拿到1小时和4小时的K线数据，然后存成能直接用的格式。
3.1 准备工作：与基础库
你得先有个币安账户，然后创建一个。创建的时候别勾太多权限——只开“读取”就够了，不然万一泄露了，后果很严重。
我个人习惯用 python-binance 这个库，它把很多底层细节封装好了，写起来省心。当然你也可以直接用 requests 手撸HTTP请求，但我觉得没必要重复造轮子。
安装命令：
pip install python-binance导入库，设置密钥：
from binance.client import Client
= "你的"
api_secret = "你的API_SECRET"
client = Client(, api_secret)
小技巧：密钥别写死在代码里。用环境变量或者配置文件，我一般放 .env 文件里，然后用 python-dotenv 加载。
3.2 K线接口的核心参数
币安K线接口的路径是 /api/v3/klines，或者用 client.get_klines()。关键参数就几个：
| 参数名 | 说明 | 举例 |
|---|---|---|
symbol |
交易对 | BTCUSDT |
interval |
K线周期 | 1h、4h |
startTime |
起始时间戳（毫秒） | 1700000000000 |
endTime |
结束时间戳（毫秒） | 1700080000000 |
limit |
最多返回条数，默认500，最大1000 | 1000 |
注意： startTime 和 endTime 都是可选的，但如果你不指定，默认返回最近500根K线。这容易让你拿到的数据不是你想要的区间。我建议永远显式指定时间范围。
3.3 获取1小时K线数据
举个例子，我要取2025年1月1日到1月30日，BTCUSDT的1小时K线。先算一下时间戳（毫秒）。
from datetime import datetime
import pandas as pd
start = int(datetime(2025, 1, 1).timestamp() * 1000)
end = int(datetime(2025, 1, 30).timestamp() * 1000)
klines = client.get_klines(
symbol='BTCUSDT',
interval=Client.KLINE_INTERVAL_1HOUR,
startTime=start,
endTime=end,
limit=1000
)
# 转换成DataFrame
columns = [
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_asset_volume', 'number_of_trades',
'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'
]
df_1h = pd.DataFrame(klines, columns=columns)
# 时间戳转成可读时间
df_1h['open_time'] = pd.to_datetime(df_1h['open_time'], unit='ms')
df_1h['close_time'] = pd.to_datetime(df_1h['close_time'], unit='ms')
print(df_1h.head())
嗯，这里有个坑：返回的K线数据，价格和成交量都是字符串，不是数字。你想想看，直接拿来算指标肯定会报错。所以记得转一下类型：
numeric_cols = ['open', 'high', 'low', 'close', 'volume']
df_1h[numeric_cols] = df_1h[numeric_cols].apply(pd.to_numeric, errors='coerce')
3.4 获取4小时K线，以及批量拉取
4小时K线同理，把interval换成 Client.KLINE_INTERVAL_4HOUR 就行了。不过要注意：1小时的数据量会是4小时的4倍，所以你需要更小心地处理请求次数限制。
币安对IP的请求频率有限制——每分钟最多1200次权重请求。get_klines 每次的权重是1，所以理论上你可以一直刷。但如果你程序里写了个死循环不断请求，很快就会触发风控。我曾经因为没加 sleep，直接被封了10分钟，那段时间策略全停了，亏了不少。
所以我建议：如果时间跨度超过1000根K线，就分片请求：
def fetch_kline_batch(symbol, interval, start_time, end_time):
all_klines = []
current_start = start_time
while current_start < end_time:
resp = client.get_klines(
symbol=symbol,
interval=interval,
startTime=current_start,
endTime=end_time,
limit=1000
)
if not resp:
break
all_klines.extend(resp)
# 下一次起始时间 = 最后一条的close_time + 1毫秒
current_start = resp[-1][6] + 1
import time
time.sleep(0.1) # 礼貌性等待
return all_klines
3.5 数据存储：本地CSV还是数据库？
这得看你数据量。单次分析用CSV完全够了，但要是每天跑增量更新，我建议用SQLite或者Parquet。我自己做回测时常用CSV，因为方便用Excel看一眼数据质量。
df_1h.to_csv('BTCUSDT_1h_202501.csv', index=False)
记得存的时候把时间列设成索引，后面做时间序列分析会省很多麻烦。
3.6 数据获取的整体流程
下面这张图概括了咱们刚刚讲的整个过程：
3.7 常见问题与避坑
- 数据不全？ 检查startTime和endTime是不是严格按照毫秒时间戳。我遇到过直接用秒戳传进去的，结果返回空数组。白白浪费了半天排查。
- 频率限制被ban？ 别并发请求太多。我个人的做法：每请求完一批就sleep 100ms，稳得很。
- K线周期搞混？ 1H和4H的K线时间戳不同。4H的K线是每4小时一根，开收时间都是整点（0,4,8...）。你用的时候注意对齐。
- 缺失值怎么处理？ 币安接口不会返回缺失的K线。如果你需要补全（比如回测），最好用前一根填充或插值。
我曾经踩过的坑： 有一次没加 endTime 参数，默认只返回最近500根。结果我拿着2023年的数据跑回测，实际用的却是2025年的数据，策略效果完全不对，白跑了三个小时。所以一定要显式指定时间范围，并在保存文件名里带上日期。
好了，到这里你应该能轻松从币安拉取1小时和4小时的K线数据了。数据获取是整个量化流程的地基——地基稳了，后面才能盖高楼。