[Add]新增fxcmGateway

examples/VnTrader/CTA_setting.json

@@ -2,7 +2,7 @@
     {
         "name": "double ema",
         "className": "DoubleMaStrategy",
-        "vtSymbol": "IF1802"
+        "vtSymbol": "rb1805"
     },
 
     {

vnpy/api/fxcm/test.py

@@ -1,5 +1,7 @@
 # encoding: UTF-8
 
+from time import sleep
+
 from vnfxcm import FxcmApi
 
 url = 'https://api-demo.fxcm.com:443'
@@ -13,11 +15,12 @@ print 'api created'
 api.connect(url, port, token, proxy)
 print api.bearer
 
-#api.getInstruments()
+sleep(20)
+api.getInstruments()
 
-api.subscribe('EUR/USD')
+#api.subscribe('EUR/USD')
-api.subscribe('USD/JPY')
+#api.subscribe('USD/JPY')
-api.subscribe('GBP/USD')
+#api.subscribe('GBP/USD')
 #api.getModel('Summary')
 #api.subscribeModel('Summary')
 

vnpy/api/fxcm/vnfxcm.py

@@ -38,7 +38,8 @@ class FxcmApi(object):
         self.queue = Queue()
         self.reqid = 0
         self.active = False
-        self.reqThread = Thread(target=self.run)
+        self.reqThread = None
+        self.sioThread = None
         
     #----------------------------------------------------------------------
     def connect(self, url, port, token, proxy=''):
@@ -48,24 +49,23 @@ class FxcmApi(object):
         self.token = token
         self.proxy = proxy
         
-        self.initSocketIO()
-        self.generateBearer()
-        self.generateHeaders()
-        
         self.active = True
+        
+        self.reqThread = Thread(target=self.runReq)
         self.reqThread.start()
         
-        self.sioThread = Thread(target=self.sio.wait)
+        self.sioThread = Thread(target=self.runSio)
         self.sioThread.start()
         
     #----------------------------------------------------------------------
     def stop(self):
         """停止"""
-        self.active = False
+        if self.active:
-        self.reqThread.join()
+            self.active = False
-        
+            self.reqThread.join()
-        self.sio._close()
+            
-        self.sioThread.join()
+            self.sio._close()
+            self.sioThread.join()
         
     #----------------------------------------------------------------------
     def initSocketIO(self):
@@ -100,14 +100,22 @@ class FxcmApi(object):
         }
         
     #----------------------------------------------------------------------
-    def run(self):
+    def runReq(self):
-        """连续运行"""
+        """处理主动请求"""
         while self.active:
             try:
                 d = self.queue.get(timeout=1)
                 self.processReq(d)
             except Empty:
                 pass
+    
+    #----------------------------------------------------------------------
+    def runSio(self):
+        """处理回调数据"""
+        self.initSocketIO()
+        self.generateBearer()
+        self.generateHeaders()        
+        self.sio.wait()
         
     #----------------------------------------------------------------------
     def sendReq(self, method, uri, params, callback):
@@ -155,21 +163,6 @@ class FxcmApi(object):
         else:
             self.onError(u'HTTP请求失败，错误代码%s' %resp.status_code)
         
-    #----------------------------------------------------------------------
-    def onConnect(self):
-        """连接回调"""
-        print 'onConnect'
-        
-    #----------------------------------------------------------------------
-    def onDisconnect(self):
-        """断开回调"""
-        print 'onClose'
-        
-    #----------------------------------------------------------------------
-    def onError(self, error, reqid):
-        """错误回调"""
-        print 'onError', error
-            
     #----------------------------------------------------------------------
     def getInstruments(self):
         """查询合约代码"""
@@ -292,6 +285,21 @@ class FxcmApi(object):
         reqid = self.sendReq(self.METHOD_POST, uri, params, self.onDeleteOrder)
         return reqid    
     
+    #----------------------------------------------------------------------
+    def onConnect(self):
+        """连接回调"""
+        print 'onConnect'
+        
+    #----------------------------------------------------------------------
+    def onDisconnect(self):
+        """断开回调"""
+        print 'onClose'
+        
+    #----------------------------------------------------------------------
+    def onError(self, error, reqid):
+        """错误回调"""
+        print 'onError', error
+    
     #----------------------------------------------------------------------
     def onGetInstruments(self, data, reqid):
         """查询合约代码回调"""
@@ -303,7 +311,7 @@ class FxcmApi(object):
         print data, reqid        
         
     #----------------------------------------------------------------------
-    def onSubscribe(self, data, reqid):
+    def onSubscribe(self, data, reqid):	
         """订阅行情回调"""
         print data, reqid    
     

vnpy/trader/gateway/fxcmGateway/FXCM_connect.json

@@ -0,0 +1,7 @@
+{
+	"account": "70934551",
+    "url": "https://api-demo.fxcm.com:443", 
+    "port": 443,
+    "token": "48055b5d9afac0a300143ac067ce04cd2430a434",
+    "proxy": "https://localhost:1080"
+}

vnpy/trader/gateway/fxcmGateway/__init__.py

@@ -0,0 +1,10 @@
+# encoding: UTF-8
+
+from vnpy.trader import vtConstant
+from fxcmGateway import FxcmGateway
+
+gatewayClass = FxcmGateway
+gatewayName = 'FXCM'
+gatewayDisplayName = u'福汇'
+gatewayType = vtConstant.GATEWAYTYPE_INTERNATIONAL
+gatewayQryEnabled = False

vnpy/trader/gateway/fxcmGateway/fxcmGateway.py

@@ -0,0 +1,290 @@
+# encoding: UTF-8
+
+import os
+import json
+from datetime import datetime
+
+from vnpy.api.fxcm import FxcmApi
+from vnpy.trader.vtGateway import *
+from vnpy.trader.vtConstant import *
+from vnpy.trader.vtFunction import getJsonPath
+
+# 价格类型映射
+priceTypeMap = {}
+priceTypeMap[PRICETYPE_LIMITPRICE] = 'limit'
+priceTypeMap[PRICETYPE_MARKETPRICE] = 'market'
+priceTypeMapReverse = {v: k for k, v in priceTypeMap.items()} 
+
+# 方向类型映射
+directionMap = {}
+directionMap[DIRECTION_LONG] = 'buy'
+directionMap[DIRECTION_SHORT] = 'sell'
+directionMapReverse = {v: k for k, v in directionMap.items()}
+
+
+########################################################################
+class FxcmGateway(VtGateway):
+    """FXCM接口"""
+
+    #----------------------------------------------------------------------
+    def __init__(self, eventEngine, gatewayName='FXCM'):
+        """Constructor"""
+        super(FxcmGateway, self).__init__(eventEngine, gatewayName)
+        
+        self.api = Api(self)     
+        
+        self.qryEnabled = False         # 是否要启动循环查询
+        
+        self.fileName = self.gatewayName + '_connect.json'
+        self.filePath = getJsonPath(self.fileName, __file__)             
+        
+    #----------------------------------------------------------------------
+    def connect(self):
+        """连接"""
+        # 载入json文件
+        try:
+            f = file(self.filePath)
+        except IOError:
+            log = VtLogData()
+            log.gatewayName = self.gatewayName
+            log.logContent = u'读取连接配置出错，请检查'
+            self.onLog(log)
+            return
+        
+        # 解析json文件
+        setting = json.load(f)
+        try:
+            account = str(setting['account'])
+            port = int(setting['port'])
+            url = str(setting['url'])
+            token = str(setting['token'])
+            proxy = str(setting['proxy'])
+        except KeyError:
+            log = VtLogData()
+            log.gatewayName = self.gatewayName
+            log.logContent = u'连接配置缺少字段，请检查'
+            self.onLog(log)
+            return            
+        
+        # 初始化接口
+        self.api.init(account, url, port, token, proxy)
+    
+    #----------------------------------------------------------------------
+    def subscribe(self, subscribeReq):
+        """订阅行情"""
+        self.api.subscribe(subscribeReq.symbol)
+        
+    #----------------------------------------------------------------------
+    def sendOrder(self, orderReq):
+        """发单"""
+        return self.api.sendOrder(orderReq)
+        
+    #----------------------------------------------------------------------
+    def cancelOrder(self, cancelOrderReq):
+        """撤单"""
+        self.api.cancelOrder(cancelOrderReq)
+        
+    #----------------------------------------------------------------------
+    def qryAccount(self):
+        """查询账户资金"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def qryPosition(self):
+        """查询持仓"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def close(self):
+        """关闭"""
+        self.api.stop()
+        
+    #----------------------------------------------------------------------
+    def initQuery(self):
+        """初始化连续查询"""
+        pass
+    
+    #----------------------------------------------------------------------
+    def query(self, event):
+        """注册到事件处理引擎上的查询函数"""
+        pass
+    
+    #----------------------------------------------------------------------
+    def startQuery(self):
+        """启动连续查询"""
+        pass
+    
+    #----------------------------------------------------------------------
+    def setQryEnabled(self, qryEnabled):
+        """设置是否要启动循环查询"""
+        pass
+    
+
+
+########################################################################
+class Api(FxcmApi):
+    """FXCM的API实现"""
+
+    #----------------------------------------------------------------------
+    def __init__(self, gateway):
+        """Constructor"""
+        super(Api, self).__init__()
+        
+        self.gateway = gateway                  # gateway对象
+        self.gatewayName = gateway.gatewayName  # gateway对象名称
+        
+        self.accout = ''
+        
+        self.orderDict = {}     # 缓存委托数据
+        
+    #----------------------------------------------------------------------
+    def onConnect(self):
+        """连接回调"""
+        self.writeLog(u'服务器连接成功')
+        
+    #----------------------------------------------------------------------
+    def onDisconnect(self):
+        """断开回调"""
+        self.writeLog(u'服务器连接断开')
+        
+    #----------------------------------------------------------------------
+    def onError(self, error, reqid):
+        """错误回调"""
+        err = VtErrorData()
+        err.gatewayName = self.gatewayNames
+        err.errorID = 0
+        err.errorTime = datetime.now().strftime('%H:%M:%S')
+        err.errorMsg = error
+        self.gateway.onError(err)
+    
+    #----------------------------------------------------------------------
+    def onGetInstruments(self, data, reqid):
+        """查询合约代码回调"""
+        for d in data['data']['instrument']:
+            contract = VtContractData()
+            contract.gatewayName = self.gatewayName
+            
+            contract.symbol = d['symbol']
+            contract.exchange = EXCHANGE_FXCM
+            contract.vtSymbol = '.'.join([contract.symbol, contract.exchange])
+            
+            contract.name = contract.symbol
+            contract.productClass = PRODUCT_FOREX
+            contract.priceTick = 0.0001
+            contract.size = data['order']
+            
+            self.gateway.onContract(contract)
+        
+    #----------------------------------------------------------------------
+    def onGetTable(self, data, reqid):
+        """查询表回调"""
+        print data, reqid        
+        
+    #----------------------------------------------------------------------
+    def onSubscribe(self, data, reqid):
+        """订阅行情回调"""
+        symbol = data['pairs']['symbol']
+        self.writeLog(u'%s行情订阅成功' %symbol)
+    
+    #----------------------------------------------------------------------
+    def onUnsubscribe(self, data, reqid):
+        """退订行情回调"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def onSubscribeModel(self, data, reqid):
+        """订阅表回调"""
+        pass
+    
+    #----------------------------------------------------------------------
+    def onUnsubscribeModel(self, data, reqid):
+        """退订表回调"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def onOpenTrade(self, data, reqid):
+        """开仓回调"""
+        print data, reqid
+        
+    #----------------------------------------------------------------------
+    def onCloseTrade(self, data, reqid):
+        """平仓回调"""
+        print data, reqid 
+        
+    #----------------------------------------------------------------------
+    def onChangeOrder(self, data, reqid):
+        """改单回调"""
+        print data, reqid       
+
+    #----------------------------------------------------------------------
+    def onDeleteOrder(self, data, reqid):
+        """撤单回调"""
+        print data, reqid       
+    
+    #----------------------------------------------------------------------
+    def onPriceUpdate(self, data):
+        """行情推送"""
+        print data
+        
+    #----------------------------------------------------------------------
+    def onModelUpdate(self, data):
+        """表推送"""
+        print data   
+        
+    #----------------------------------------------------------------------
+    def init(self, account, url, port, token, proxy):
+        """初始化"""
+        self.account = account
+        
+        self.connect(url, port, token, proxy)
+        
+    #----------------------------------------------------------------------
+    def writeLog(self, logContent):
+        """发出日志"""
+        log = VtLogData()
+        log.gatewayName = self.gatewayName
+        log.logContent = logContent
+        self.gateway.onLog(log)
+        
+    #----------------------------------------------------------------------
+    def qryInstruments(self):
+        """查询合约"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def qryOrders(self):
+        """查询委托"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def qryTrades(self):
+        """查询成交"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def sendOrder_(self, orderReq):
+        """发送委托"""
+        accountID = self.account
+        symbol = orderReq.symbol
+        
+        if orderReq.direction == DIRECTION_LONG:
+            isBuy = True
+        else:
+            isBuy = False
+            
+        amount = orderReq.volume
+        rate = orderReq.price
+        
+        self.openTrade(accountID, symbol, isBuy, amount, limit, isInPips, 
+                      atMarket, orderType, timeInForce)
+    
+    #----------------------------------------------------------------------
+    def cancelOrder(self, cancelOrderReq):
+        """撤销委托"""
+        pass
+    
+    
+#----------------------------------------------------------------------
+def getTime(t):
+    """把OANDA返回的时间格式转化为简单的时间字符串"""
+    return t[11:19]

vnpy/trader/language/chinese/constant.py

@@ -81,6 +81,7 @@ EXCHANGE_ICE = 'ICE'           # ICE交易所
 EXCHANGE_LME = 'LME'           # LME交易所
 
 EXCHANGE_OANDA = 'OANDA'       # OANDA外汇做市商
+EXCHANGE_FXCM = 'FXCM'         # FXCM外汇做市商
 
 EXCHANGE_OKCOIN = 'OKCOIN'       # OKCOIN比特币交易所
 EXCHANGE_HUOBI = 'HUOBI'         # 火币比特币交易所

vnpy/trader/language/english/constant.py

@@ -77,6 +77,7 @@ EXCHANGE_ICE = 'ICE'           # ICE交易所
 EXCHANGE_LME = 'LME'           # LME交易所
 
 EXCHANGE_OANDA = 'OANDA'       # OANDA外汇做市商
+EXCHANGE_FXCM = 'FXCM'         # FXCM外汇做市商
 
 EXCHANGE_OKCOIN = 'OKCOIN'       # OKCOIN比特币交易所
 EXCHANGE_HUOBI = 'HUOBI'         # 火币比特币交易所