数据库模块原理

在量化中,研究策略和实盘交易,都离不开数据。如果要提升回测速度,那么就需要搭建本地数据库。

VNPY集成的数据源

  • 国内期货市场: 米匡推出的RQData数据服务
  • 海外市场: 盈透证券(IB)
  • 数字货币: 币圈交易所(BitMex、币安、火币等等)

1)RQData数据服务

vn.py对米匡的rqdatac库进行了再次封装调用实现。

实现类: RqdataClient
相关代码: vnpy\trader\rqdata.py

demo:从rqdata获取历史行情

操作步骤:

    1) 初始化rqdataClient
    2) 构造请求对象
    3) 从rqdata获取数据
    4) 将数据保存到本地数据库

from vnpy.trader.rqdata import RqdataClient
from vnpy.trader.object import HistoryRequest

#初始化rqdataClient
rqdataClient=RqdataClient()
rqdataClient.init()

#构造请求对象
historyReq=HistoryRequest(
    #symbol='C2105.DCE',
    symbol='C2105',
    exchange=Exchange('DCE'),
    start=datetime(2021,1,1),
    end=datetime(2021,2,1),
    interval=Interval.DAILY
)

#从rqdata获取数据
retData=rqdataClient.query_history(historyReq)

#将数据保存到本地数据库
database_manager.save_bar_data(retData)

2)其它数据源

对应接口交易接口Gateway类的query_history函数,实现历史数据的下载。这里亿盈透证券(IB)接口为例。
实现类: IbGateway(BaseGateway)
相关代码: vnpy\gateway\ib\ib_gateway.py

3)本地数据库

通过对象database_manager实现,该对象通过抽象工厂模式,根据实际情况调用具体的数据库实现类:
相关类: BaseDatabase
相关代码

driver: str = SETTINGS["database.driver"]
module_name: str = f"vnpy.database.{driver}"
try:
    database_manager: BaseDatabase = import_module(module_name).database_manager
except ModuleNotFoundError:
    print(f"找不到数据库驱动{module_name},使用默认的SQLite数据库")
    database_manager: BaseDatabase = import_module("vnpy.database.sqlite").database_manager

代码位置: vnpy\trader\database.py
类的具体实现: vnpy\database

MongoDB

为了提高性能,使用MongoDB,因此,只需要关注MongoDB的实现。
相关类: MongodbDatabase(BaseDatabase)
代码位置: vnpy\database\mongodb\mongodb_database.py

在使用monggoDB时,使用了mongoengine的ORM功能,实现了以下三个对象:

  • DbBarData:对应数据库的db_bar_data集合
  • DbTickData:对应数据库的db_tick_data
  • DbBarOverView:对应数据库的db_bar_overview

demo:获取本地数据库的数据


from datetime import datetime
from vnpy.trader.database import database_manager
from vnpy.trader.object import Interval
from vnpy.trader.constant import Exchange


barDatas=database_manager.load_bar_data(
    'C2105',
    Exchange('DCE'),
    Interval.MINUTE,
    start=datetime(2021,1,1),
    end=datetime(2021,1,10)
)


for bar in barDatas:
    print(f"{bar.datetime} {bar.symbol}:open_price:{bar.open_price}")


barOverViews=database_manager.get_bar_overview()


for barOverView in barOverViews:
    print(f"{barOverView.symbol} start:{barOverView.start} end:{barOverView.end} count:{barOverView.count}")

一句话概括:用好database_manager这个对象,提供了如下方法:

  • def load_bar_data( self, symbol: str, exchange: Exchange,  interval: Interval,  start: datetime, end: datetime) -> List[BarData]
  • def load_tick_data(self,  symbol: str, exchange: Exchange,  start: datetime,  end: datetime ) -> List[TickData]:
  • def get_bar_overview(self) -> List[BarOverview]
  • def delete_bar_data( self,  symbol: str,  exchange: Exchange, interval: Interval ) -> int:
  • def delete_tick_data(self,  symbol: str, exchange: Exchange ) -> int: