4. 使用requests获取历史Tick数据：以Dukascopy为例

做量化交易的朋友都知道，Tick数据是外汇市场的「原子」——每一笔成交记录。但获取它，说实话，挺折腾的。市面上很多数据商卖得死贵，还藏着掖着。我个人习惯是直接从Dukascopy扒数据，这家瑞士银行公开了历史Tick数据，而且免费。

今天我就带你手撸一个HTTP请求，把Dukascopy的Tick数据拽下来，再解析成我们能用的格式。

4.1 Dukascopy的数据接口长什么样？

Dukascopy的API其实是个「伪装的静态文件服务器」。它把Tick数据按时间切片，存成了CSV文件。你只要拼对URL，就能下载。

URL的构造规则是这样的：

https://datafeed.dukascopy.com/datafeed/EURUSD/2024/01/02/00h_ticks.bi5

拆开来看：

• EURUSD — 交易品种，注意是大写
• 2024/01/02 — 年/月/日
• 00h_ticks.bi5 — 小时级别的Tick文件，00代表UTC 0点

嗯，这里要注意：Dukascopy用的是UTC时间。你在中国，想拿北京时间下午2点的数据，得算成UTC 6点。我刚开始搞的时候，没注意时区，拿到的数据全是凌晨的，白忙活半天。

4.2 构建HTTP请求，把数据拽下来

Python的requests库，大家应该都用过。但Dukascopy有个坑——它需要你带上特定的请求头，否则直接返回403。

我一般这么写：

import requests

def fetch_dukascopy_tick(symbol, year, month, day, hour):
    """
    获取Dukascopy某小时的Tick数据
    """
    url = f"https://datafeed.dukascopy.com/datafeed/{symbol}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.bi5"
    
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Accept': '*/*',
        'Accept-Encoding': 'gzip, deflate',
        'Connection': 'keep-alive'
    }
    
    try:
        resp = requests.get(url, headers=headers, timeout=30)
        resp.raise_for_status()
        return resp.content
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 测试：拿2024年1月2日UTC 0点的EURUSD Tick数据
raw_data = fetch_dukascopy_tick('EURUSD', 2024, 1, 2, 0)
print(f"拿到数据了，大小: {len(raw_data)} 字节")

为什么会这样？Dukascopy的服务器会检查User-Agent，没有的话直接拒绝。我建议你直接用上面这个头，我试过很多次，稳得很。

4.3 解析.bi5文件——Dukascopy的「加密」格式

下载下来的文件后缀是.bi5，不是CSV。这是Dukascopy自己搞的二进制压缩格式。说白了，就是先把Tick数据写成CSV，然后用LZMA压缩，再改个后缀名。

解析步骤：

1. 用lzma模块解压
2. 按行读取CSV文本
3. 解析每一行的字段

代码实现：

import lzma
import csv
import io

def parse_bi5_to_ticks(raw_bytes):
    """
    将.bi5原始字节解析为Tick数据列表
    """
    if raw_bytes is None or len(raw_bytes) == 0:
        return []
    
    try:
        # 第一步：LZMA解压
        decompressed = lzma.decompress(raw_bytes)
        text = decompressed.decode('utf-8')
        
        # 第二步：按行解析CSV
        reader = csv.reader(io.StringIO(text))
        ticks = []
        
        for row in reader:
            if len(row) != 5:
                continue  # 跳过格式异常的行
            
            tick = {
                'timestamp': int(row[0]),  # 毫秒时间戳
                'bid': float(row[1]) / 100000,  # 原始数据是整数，要除以10万
                'ask': float(row[2]) / 100000,
                'bid_volume': float(row[3]),
                'ask_volume': float(row[4])
            }
            ticks.append(tick)
        
        return ticks
    
    except Exception as e:
        print(f"解析失败: {e}")
        return []

# 接上面的例子
ticks = parse_bi5_to_ticks(raw_data)
print(f"解析出 {len(ticks)} 条Tick记录")
if ticks:
    print(f"第一条: {ticks[0]}")

4.4 完整流程：从下载到解析，一步到位

把上面两个函数串起来，再加个时间循环，就能拿到一整天的Tick数据了。

from datetime import datetime, timedelta

def get_full_day_ticks(symbol, year, month, day):
    """
    获取某一天所有小时的Tick数据
    """
    all_ticks = []
    
    for hour in range(24):
        raw = fetch_dukascopy_tick(symbol, year, month, day, hour)
        ticks = parse_bi5_to_ticks(raw)
        all_ticks.extend(ticks)
        
        if len(ticks) > 0:
            print(f"小时 {hour:02d}: {len(ticks)} 条")
    
    # 按时间戳排序
    all_ticks.sort(key=lambda x: x['timestamp'])
    
    return all_ticks

# 拿2024年1月2日全天的EURUSD Tick数据
day_ticks = get_full_day_ticks('EURUSD', 2024, 1, 2)
print(f"全天共 {len(day_ticks)} 条Tick数据")

你想想看，有了这个基础函数，你就能批量拉取任意时间段的数据了。我自己的历史数据库，就是用这个脚本跑了三天三夜建起来的。

4.5 知识体系总览

下面这张图，帮你理清整个流程的逻辑关系：

4.6 进阶：直接解析JSON格式

Dukascopy其实也提供了JSON格式的接口，但用得不多。我个人觉得CSV格式更稳定，毕竟这么多年没变过。不过如果你非要JSON，可以试试这个：

# Dukascopy的JSON接口（不常用，但存在）
json_url = f"https://datafeed.dukascopy.com/datafeed/{symbol}/{year}/{month:02d}/{day:02d}/{hour:02d}h_ticks.json"

但说实话，我不推荐用JSON。原因有两个：一是文件体积比.bi5大好几倍，二是解析速度慢。做量化交易，数据量大了之后，每一毫秒都很宝贵。

好了，这一章的核心内容就这些。你只要掌握了URL构造、HTTP请求、LZMA解压、CSV解析这四步，Dukascopy的Tick数据就任你取用了。下一章我们聊聊怎么把这些数据存进数据库，方便后续回测和分析。

---