下载数据

在开始策略回测之前，必须保证数据库内有充足的历史数据。故vnpy提供了历史数据一键下载的功能。

RQData

RQData提供国内股票、ETF、期货以及期权的历史数据。其下载数据功能主要是基于RQData的get_price()函数实现的。

get_price(
    order_book_ids, start_date='2013-01-04', end_date='2014-01-04',
    frequency='1d', fields=None, adjust_type='pre', skip_suspended =False,
    market='cn'
)

在使用前要保证RQData初始化完毕，然后填写以下4个字段信息：

• 本地代码：格式为合约品种+交易所，如IF88.CFFEX、rb88.SHFE；然后在底层通过RqdataClient的to_rq_symbol()函数转换成符合RQData格式，对应RQData中get_price()函数的order_book_ids字段。

• K线周期：可以填1m、1h、d、w，对应get_price()函数的frequency字段。

• 开始日期：格式为yy/mm/dd，如2017/4/21，对应get_price()函数的start_date字段。（点击窗口右侧箭头按钮可改变日期大小）

• 结束日期：格式为yy/mm/dd，如2019/4/22，对应get_price()函数的end_date字段。（点击窗口右侧箭头按钮可改变日期大小）

填写完字段信息后，点击下方“下载数据”按钮启动下载程序，下载成功如图所示。

IB

盈透证券提供外盘股票、期货、期权的历史数据。下载前必须连接好IB接口，因为其下载数据功能主要是基于IbGateway类query_history()函数实现的。

    def query_history(self, req: HistoryRequest):
        """"""
        self.history_req = req
 
        self.reqid += 1
 
        ib_contract = Contract()
        ib_contract.conId = str(req.symbol)
        ib_contract.exchange = EXCHANGE_VT2IB[req.exchange]
 
        if req.end:
            end = req.end
            end_str = end.strftime("%Y%m%d %H:%M:%S")
        else:
            end = datetime.now()
            end_str = ""
 
        delta = end - req.start
        days = min(delta.days, 180)     # IB only provides 6-month data
        duration = f"{days} D"
        bar_size = INTERVAL_VT2IB[req.interval]
 
        if req.exchange == Exchange.IDEALPRO:
            bar_type = "MIDPOINT"
        else:
            bar_type = "TRADES"
 
        self.client.reqHistoricalData(
            self.reqid,
            ib_contract,
            end_str,
            duration,
            bar_size,
            bar_type,
            1,
            1,
            False,
            []
        )
 
        self.history_condition.acquire()    # Wait for async data return
        self.history_condition.wait()
        self.history_condition.release()
 
        history = self.history_buf
        self.history_buf = []       # Create new buffer list
        self.history_req = None
 
        return history

BITMEX

BITMEX交易所提供数字货币历史数据。由于仿真环境与实盘环境行情差异比较大，故需要用实盘账号登录BIMEX接口来下载真实行情数据，其下载数据功能主要是基于BitmexGateway类query_history()函数实现的。

    def query_history(self, req: HistoryRequest):
        """"""
        if not self.check_rate_limit():
            return
 
        history = []
        count = 750
        start_time = req.start.isoformat()
 
        while True:
            # Create query params
            params = {
                "binSize": INTERVAL_VT2BITMEX[req.interval],
                "symbol": req.symbol,
                "count": count,
                "startTime": start_time
            }
 
            # Add end time if specified
            if req.end:
                params["endTime"] = req.end.isoformat()
 
            # Get response from server
            resp = self.request(
                "GET",
                "/trade/bucketed",
                params=params
            )
 
            # Break if request failed with other status code
            if resp.status_code // 100 != 2:
                msg = f"获取历史数据失败，状态码：{resp.status_code}，信息：{resp.text}"
                self.gateway.write_log(msg)
                break
            else:
                data = resp.json()
                if not data:
                    msg = f"获取历史数据为空，开始时间：{start_time}，数量：{count}"
                    break
 
                for d in data:
                    dt = datetime.strptime(
                        d["timestamp"], "%Y-%m-%dT%H:%M:%S.%fZ")
                    bar = BarData(
                        symbol=req.symbol,
                        exchange=req.exchange,
                        datetime=dt,
                        interval=req.interval,
                        volume=d["volume"],
                        open_price=d["open"],
                        high_price=d["high"],
                        low_price=d["low"],
                        close_price=d["close"],
                        gateway_name=self.gateway_name
                    )
                    history.append(bar)
 
                begin = data[0]["timestamp"]
                end = data[-1]["timestamp"]
                msg = f"获取历史数据成功，{req.symbol} - {req.interval.value}，{begin} - {end}"
                self.gateway.write_log(msg)
 
                # Break if total data count less than 750 (latest date collected)
                if len(data) < 750:
                    break
 
                # Update start time
                start_time = bar.datetime + TIMEDELTA_MAP[req.interval]
 
        return history