[Add]CTA策略实盘交易引擎加入对RQData的数据支持

examples/VnTrader/CTA_setting.json

@@ -2,18 +2,18 @@
     {
         "name": "double ema",
         "className": "DoubleMaStrategy",
-        "vtSymbol": "rb1805"
+        "vtSymbol": "rb1905"
     },
 
     {
         "name": "atr rsi",
         "className": "AtrRsiStrategy",
-        "vtSymbol": "IC1802"
+        "vtSymbol": "IC1901"
     },
 
     {
         "name": "king keltner",
         "className": "KkStrategy",
-        "vtSymbol": "IH1802"
+        "vtSymbol": "IH1901"
     }
 ]

examples/VnTrader/VT_setting.json

@@ -16,5 +16,8 @@
 
 	"tdPenalty": ["IF", "IH", "IC"],
 
-	"maxDecimal": 4
+	"maxDecimal": 4,
+
+	"rqUsername": "",
+	"rqPassword": ""
 }

vnpy/trader/VT_setting.json

@@ -16,5 +16,8 @@
 
 	"tdPenalty": ["IF", "IH", "IC"],
 
-	"maxDecimal": 4
+	"maxDecimal": 4,
+
+	"rqUsername": "",
+	"rqPassword": ""
 }

vnpy/trader/app/ctaStrategy/ctaEngine.py

@@ -20,6 +20,7 @@
 from vnpy.trader.vtGateway import VtSubscribeReq, VtOrderReq, VtCancelOrderReq, VtLogData
 from vnpy.trader.vtFunction import todayDate, getJsonPath
 from vnpy.trader.app import AppEngine
+from vnpy.trader.vtGlobal import globalSetting
 
 from .ctaBase import *
 from .strategy import STRATEGY_CLASS
@@ -74,6 +75,15 @@ def __init__(self, mainEngine, eventEngine):
         # 引擎类型为实盘
         self.engineType = ENGINETYPE_TRADING
 
+        # RQData数据服务
+        self.rq = None
+
+        # RQData能获取的合约代码列表
+        self.rqSymbolSet = set()
+
+        # 初始化RQData服务
+        self.initRqData()
+
         # 注册日式事件类型
         self.mainEngine.registerLogEvent(EVENT_CTA_LOG)
 
@@ -343,6 +353,12 @@ def insertData(self, dbName, collectionName, data):
     #----------------------------------------------------------------------
     def loadBar(self, dbName, collectionName, days):
         """从数据库中读取Bar数据，startDate是datetime对象"""
+        # 优先尝试从RQData获取数据
+        if dbName == MINUTE_DB_NAME and collectionName.upper() in self.rqSymbolSet:
+            l = self.loadRqBar(collectionName, days)
+            return l
+
+        # 如果没有则从数据库中读取数据
         startDate = self.today - timedelta(days)
 
         d = {'datetime':{'$gte':startDate}}
@@ -663,4 +679,63 @@ def getPriceTick(self, strategy):
             return contract.priceTick
         return 0
 
-
+    #----------------------------------------------------------------------
+    def initRqData(self):
+        """初始化RQData客户端"""
+        # 检查是否填写了RQData配置
+        username = globalSetting.get('rqUsername')
+        password = globalSetting.get('rqPassword')
+        if not username or not password:
+            print globalSetting
+            return
+
+        # 加载RQData
+        try:
+            import rqdatac as rq
+        except ImportError:
+            print 'import fail'
+            return
+
+        # 登录RQData
+        self.rq = rq
+        self.rq.init(username, password)
+
+        # 获取本日可交易合约代码
+        try:
+            df = self.rq.all_instruments(type='Future', date=datetime.now())
+            for ix, row in df.iterrows():
+                self.rqSymbolSet.add(row['order_book_id'])
+        except RuntimeError:
+            print 'download fail'
+            pass
+
+    #----------------------------------------------------------------------
+    def loadRqBar(self, symbol, days):
+        """从RQData加载K线数据"""
+        endDate = datetime.now()
+        startDate = endDate - timedelta(days)
+
+        df = self.rq.get_price(symbol.upper(), 
+                               frequency='1m', 
+                               fields=['open', 'high', 'low', 'close', 'volume'],
+                               start_date=startDate,
+                               end_date=endDate)
+
+        l = []
+
+        for ix, row in df.iterrows():
+            bar = VtBarData()
+            bar.symbol = symbol
+            bar.vtSymbol = symbol
+            bar.open = row['open']
+            bar.high = row['high']
+            bar.low = row['low']
+            bar.close = row['close']
+            bar.volume = row['volume']
+            bar.datetime = row.name
+            bar.date = bar.datetime.strftime("%Y%m%d")
+            bar.time = bar.datetime.strftime("%H:%M:%S")
+
+            l.append(bar)
+
+        return l

vnpy/trader/vtEngine.py

@@ -113,6 +113,8 @@ def connect(self, gatewayName):
 
         if gateway:
             gateway.connect()
+
+        self.dbConnect()
 
     #----------------------------------------------------------------------
     def subscribe(self, subscribeReq, gatewayName):
@@ -196,7 +198,7 @@ def dbConnect(self):
             # 读取MongoDB的设置
             try:
                 # 设置MongoDB操作的超时时间为0.5秒
-                self.dbClient = MongoClient(globalSetting['mongoHost'], globalSetting['mongoPort'], connectTimeoutMS=500)
+                self.dbClient = MongoClient(globalSetting['mongoHost'], globalSetting['mongoPort'], serverSelectionTimeoutMS=500)
 
                 # 调用server_info查询服务器状态，防止服务器异常并未连接成功
                 self.dbClient.server_info()
@@ -208,6 +210,7 @@ def dbConnect(self):
                     self.eventEngine.register(EVENT_LOG, self.dbLogging)
 
             except ConnectionFailure:
+                self.dbClient = None
                 self.writeLog(text.DATABASE_CONNECTING_FAILED)
 
     #----------------------------------------------------------------------