增加vn.trader的oandaGateway

vn.oanda/test.py

@@ -1,6 +1,6 @@
 # encoding: utf-8
 
-from api import OandaApi
+from vnoanda import OandaApi
 
     
 if __name__ == '__main__':

vn.oanda/vnoanda.py

@@ -230,9 +230,9 @@ class OandaApi(object):
                             print callback.__name__                        
                         callback(data, reqID)    
                     except Exception, e:                  
-                        self.onError(error)                      
+                        self.onError(str(e), reqID)                      
                 else:                
-                    self.onError(error)
+                    self.onError(error, reqID)
             except Empty:
                 pass
             

vn.trader/oandaGateway/OANDA_connect.json

@@ -0,0 +1,5 @@
+{
+    "token": "请在OANDA网站申请", 
+    "accountId": "请在OANDA网站申请",
+    "settingName": "practice"
+}

vn.trader/oandaGateway/oandaGateway.py

@@ -0,0 +1,456 @@
+# encoding: UTF-8
+
+'''
+vn.oanda的gateway接入
+
+由于OANDA采用的是外汇做市商的交易模式，因此和国内接口方面有若干区别，具体如下：
+
+* 行情数据反映的是OANDA的报价变化，因此只有买卖价，而没有成交价
+
+* OANDA的持仓管理分为单笔成交持仓（Trade数据，国内没有）
+  和单一资产汇总持仓（Position数据）
+  
+* OANDA系统中的所有时间都采用UTC时间（世界协调时，中国是UTC+8）
+
+* 由于采用的是外汇做市商的模式，用户的限价委托当价格被触及时就会立即全部成交，
+  不会出现部分成交的情况，因此委托状态只有已报、成交、撤销三种
+  
+* 外汇市场采用24小时交易，因此OANDA的委托不像国内收盘后自动失效，需要用户指定
+  失效时间，本接口中默认设置为24个小时候失效
+'''
+
+
+import os
+import json
+import datetime
+
+from vnoanda import OandaApi
+from vtGateway import *
+
+# 价格类型映射
+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 OandaGateway(VtGateway):
+    """OANDA接口"""
+
+    #----------------------------------------------------------------------
+    def __init__(self, eventEngine, gatewayName='OANDA'):
+        """Constructor"""
+        super(OandaGateway, self).__init__(eventEngine, gatewayName)
+        
+        self.api = Api(self)     
+        
+        self.qryEnabled = False         # 是否要启动循环查询
+        
+    #----------------------------------------------------------------------
+    def connect(self):
+        """连接"""
+        # 载入json文件
+        fileName = self.gatewayName + '_connect.json'
+        fileName = os.getcwd() + '\\oandaGateway\\' + fileName
+        
+        try:
+            f = file(fileName)
+        except IOError:
+            log = VtLogData()
+            log.gatewayName = self.gatewayName
+            log.logContent = u'读取连接配置出错，请检查'
+            self.onLog(log)
+            return
+        
+        # 解析json文件
+        setting = json.load(f)
+        try:
+            token = str(setting['token'])
+            accountId = str(setting['accountId'])
+            settingName = str(setting['settingName'])
+        except KeyError:
+            log = VtLogData()
+            log.gatewayName = self.gatewayName
+            log.logContent = u'连接配置缺少字段，请检查'
+            self.onLog(log)
+            return            
+        
+        # 初始化接口
+        self.api.init(settingName, token, accountId)
+        log = VtLogData()
+        log.gatewayName = self.gatewayName
+        log.logContent = u'接口初始化成功'
+        self.onLog(log)
+
+        # 查询信息
+        self.api.qryInstruments()
+        self.api.qryOrders()
+        self.api.qryTrades()
+        
+        # 初始化并启动查询
+        self.initQuery()
+    
+    #----------------------------------------------------------------------
+    def subscribe(self, subscribeReq):
+        """订阅行情"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def sendOrder(self, orderReq):
+        """发单"""
+        return self.api.sendOrder_(orderReq)
+        
+    #----------------------------------------------------------------------
+    def cancelOrder(self, cancelOrderReq):
+        """撤单"""
+        self.api.cancelOrder_(cancelOrderReq)
+        
+    #----------------------------------------------------------------------
+    def qryAccount(self):
+        """查询账户资金"""
+        self.api.getAccountInfo()
+        
+    #----------------------------------------------------------------------
+    def qryPosition(self):
+        """查询持仓"""
+        self.api.getPositions()
+        
+    #----------------------------------------------------------------------
+    def close(self):
+        """关闭"""
+        self.api.exit()
+        
+    #----------------------------------------------------------------------
+    def initQuery(self):
+        """初始化连续查询"""
+        if self.qryEnabled:
+            # 需要循环的查询函数列表
+            self.qryFunctionList = [self.qryAccount, self.qryPosition]
+            
+            self.qryCount = 0           # 查询触发倒计时
+            self.qryTrigger = 2         # 查询触发点
+            self.qryNextFunction = 0    # 上次运行的查询函数索引
+            
+            self.startQuery()
+    
+    #----------------------------------------------------------------------
+    def query(self, event):
+        """注册到事件处理引擎上的查询函数"""
+        self.qryCount += 1
+        
+        if self.qryCount > self.qryTrigger:
+            # 清空倒计时
+            self.qryCount = 0
+            
+            # 执行查询函数
+            function = self.qryFunctionList[self.qryNextFunction]
+            function()
+            
+            # 计算下次查询函数的索引，如果超过了列表长度，则重新设为0
+            self.qryNextFunction += 1
+            if self.qryNextFunction == len(self.qryFunctionList):
+                self.qryNextFunction = 0
+    
+    #----------------------------------------------------------------------
+    def startQuery(self):
+        """启动连续查询"""
+        self.eventEngine.register(EVENT_TIMER, self.query)
+    
+    #----------------------------------------------------------------------
+    def setQryEnabled(self, qryEnabled):
+        """设置是否要启动循环查询"""
+        self.qryEnabled = qryEnabled
+    
+
+
+########################################################################
+class Api(OandaApi):
+    """OANDA的API实现"""
+
+    #----------------------------------------------------------------------
+    def __init__(self, gateway):
+        """Constructor"""
+        super(Api, self).__init__()
+        
+        self.gateway = gateway                  # gateway对象
+        self.gatewayName = gateway.gatewayName  # gateway对象名称
+        
+        self.orderDict = {}     # 缓存委托数据
+        
+    #----------------------------------------------------------------------
+    def onError(self, error, reqID):
+        """错误信息回调"""
+        err = VtErrorData()
+        err.gatewayName = self.gatewayName
+        err.errorMsg = error
+        self.gateway.onError(err)
+
+    #----------------------------------------------------------------------
+    def onGetInstruments(self, data, reqID):
+        """回调函数"""
+        if not 'instruments' in data:
+            return
+        l = data['instruments']
+        for d in l:
+            contract = VtContractData()
+            contract.gatewayName = self.gatewayName
+            
+            contract.symbol = d['instrument']
+            contract.name = d['displayName']
+            contract.exchange = EXCHANGE_OANDA
+            contract.vtSymbol = '.'.join([contract.symbol, contract.exchange])
+            contract.priceTick = float(d['pip'])
+            contract.size = 1
+            contract.productClass = PRODUCT_FOREX
+            self.gateway.onContract(contract)
+        
+        self.writeLog(u'交易合约信息查询完成')
+
+    #----------------------------------------------------------------------
+    def onGetAccountInfo(self, data, reqID):
+        """回调函数"""
+        account = VtAccountData()
+        account.gatewayName = self.gatewayName
+        
+        account.accountID = str(data['accountId'])
+        account.vtAccountID = '.'.join([self.gatewayName, account.accountID])
+        
+        account.available = data['marginAvail']
+        account.margin = data['marginUsed']
+        account.closeProfit = data['realizedPl']
+        account.positionProfit = data['unrealizedPl']
+        account.balance = data['balance']
+        
+        self.gateway.onAccount(account)
+        
+    #----------------------------------------------------------------------
+    def onGetOrders(self, data, reqID):
+        """回调函数"""
+        if not 'orders' in data:
+            return
+        l = data['orders']  
+        
+        for d in l:
+            order = VtOrderData()
+            order.gatewayName = self.gatewayName
+            
+            order.symbol = d['instrument']
+            order.exchange = EXCHANGE_OANDA
+            order.vtSymbol = '.'.join([order.symbol, order.exchange])
+            order.orderID = str(d['id'])
+            
+            order.direction = directionMapReverse.get(d['side'], DIRECTION_UNKNOWN)
+            order.offset = OFFSET_NONE
+            order.status = STATUS_NOTTRADED     # OANDA查询到的订单都是活动委托
+            
+            order.price = d['price']
+            order.totalVolume = d['units']
+            order.orderTime = getTime(d['time'])
+            
+            order.vtOrderID = '.'.join([self.gatewayName , order.orderID])
+            
+            self.gateway.onOrder(order)
+            
+            self.orderDict[order.orderID] = order
+            
+        self.writeLog(u'委托信息查询完成')
+    
+    #----------------------------------------------------------------------
+    def onGetPositions(self, data, reqID):
+        """回调函数"""
+        if not 'positions' in data:
+            return
+        l = data['positions']
+        
+        for d in l:
+            pos = VtPositionData()
+            pos.gatewayName = self.gatewayName
+            
+            pos.symbol = d['instrument']
+            pos.exchange = EXCHANGE_OANDA
+            pos.vtSymbol = '.'.join([pos.symbol, pos.exchange])
+            pos.direction = directionMapReverse.get(d['side'], DIRECTION_UNKNOWN)
+            pos.position = d['units']
+            pos.price = d['avgPrice']
+            pos.vtPositionName = '.'.join([pos.vtSymbol, pos.direction])
+            
+            self.gateway.onPosition(pos)
+
+    #----------------------------------------------------------------------
+    def onGetTransactions(self, data, reqID):
+        """回调函数"""
+        if not 'transactions' in data:
+            return
+        l = data['transactions']
+        
+        for d in l:
+            # 这里我们只关心委托成交
+            if d['type'] == 'ORDER_FILLED':
+                trade = VtTradeData()
+                trade.gatewayName = self.gatewayName
+                
+                trade.symbol = d['instrument']
+                trade.exchange = EXCHANGE_OANDA
+                trade.vtSymbol = '.'.join([trade.symbol, trade.exchange])
+                trade.tradeID = str(d['id'])
+                trade.vtTradeID = '.'.join([self.gatewayName, trade.tradeID])
+                trade.orderID = str(d['orderId'])
+                trade.vtOrderID = '.'.join([self.gatewayName, trade.orderID])                 
+                trade.direction = directionMapReverse.get(d['side'], DIRECTION_UNKNOWN)
+                trade.offset = OFFSET_NONE
+                trade.price = d['price']
+                trade.volume = d['units']
+                trade.tradeTime = getTime(d['time'])
+                
+                self.gateway.onTrade(trade)
+                
+        self.writeLog(u'成交信息查询完成')
+        
+    #----------------------------------------------------------------------
+    def onPrice(self, data):
+        """行情推送"""
+        if 'tick' not in data:
+            return
+        d = data['tick']
+        
+        tick = VtTickData()
+        tick.gatewayName = self.gatewayName
+    
+        tick.symbol = d['instrument']
+        tick.exchange = EXCHANGE_OANDA
+        tick.vtSymbol = '.'.join([tick.symbol, tick.exchange])    
+        tick.bidPrice1 = d['bid']
+        tick.askPrice1 = d['ask']
+        tick.time = getTime(d['time'])
+        
+        # 做市商的TICK数据只有买卖的报价，因此最新价格选用中间价代替
+        tick.lastPrice = (tick.bidPrice1 + tick.askPrice1)/2        
+        
+        self.gateway.onTick(tick)
+        
+    #----------------------------------------------------------------------
+    def onEvent(self, data):
+        """事件推送（成交等）"""
+        if 'transaction' not in data:
+            return
+        
+        d = data['transaction']
+        
+        # 委托成交
+        if d['type'] == 'ORDER_FILLED':
+            # 推送成交事件
+            trade = VtTradeData()
+            trade.gatewayName = self.gatewayName
+            
+            trade.symbol = d['instrument']
+            trade.exchange = EXCHANGE_OANDA
+            trade.vtSymbol = '.'.join([trade.symbol, trade.exchange])
+            
+            trade.tradeID = str(d['id'])
+            trade.vtTradeID = '.'.join([self.gatewayName, trade.tradeID])
+            
+            trade.orderID = str(d['orderId'])
+            trade.vtOrderID = '.'.join([self.gatewayName, trade.orderID])                    
+
+            trade.direction = directionMapReverse.get(d['side'], DIRECTION_UNKNOWN)
+            trade.offset = OFFSET_NONE
+            
+            trade.price = d['price']
+            trade.volume = d['units']
+            trade.tradeTime = getTime(d['time'])
+            
+            self.gateway.onTrade(trade)      
+            
+            # 推送委托事件
+            order = self.orderDict.get(str(d['orderId']), None)
+            if not order:
+                return
+            order.status = STATUS_ALLTRADED
+            self.gateway.onOrder(order)             
+        
+        # 委托下达
+        elif d['type'] in ['MARKET_ORDER_CREATE', 'LIMIT_ORDER_CREATE']:
+            order = VtOrderData()
+            order.gatewayName = self.gatewayName
+    
+            order.symbol = d['instrument']
+            order.exchange = EXCHANGE_OANDA
+            order.vtSymbol = '.'.join([order.symbol, order.exchange])
+            order.orderID = str(d['id'])
+            order.direction = directionMapReverse.get(d['side'], DIRECTION_UNKNOWN)
+            order.offset = OFFSET_NONE
+            order.status = STATUS_NOTTRADED    
+            order.price = d['price']
+            order.totalVolume = d['units']
+            order.orderTime = getTime(d['time'])
+            order.vtOrderID = '.'.join([self.gatewayName , order.orderID])
+    
+            self.gateway.onOrder(order)   
+            self.orderDict[order.orderID] = order
+            
+        # 委托撤销
+        elif d['type'] == 'ORDER_CANCEL':
+            order = self.orderDict.get(str(d['orderId']), None)
+            if not order:
+                return
+            order.status = STATUS_CANCELLED
+            self.gateway.onOrder(order)
+    
+    #----------------------------------------------------------------------
+    def writeLog(self, logContent):
+        """发出日志"""
+        log = VtLogData()
+        log.gatewayName = self.gatewayName
+        log.logContent = logContent
+        self.gateway.onLog(log)
+        
+    #----------------------------------------------------------------------
+    def qryInstruments(self):
+        """查询合约"""
+        params = {'accountId': self.accountId}
+        self.getInstruments(params)
+        
+    #----------------------------------------------------------------------
+    def qryOrders(self):
+        """查询委托"""
+        self.getOrders({})
+        
+    #----------------------------------------------------------------------
+    def qryTrades(self):
+        """查询成交"""
+        # 最多查询100条记录
+        self.getTransactions({'count': 100})
+        
+    #----------------------------------------------------------------------
+    def sendOrder_(self, orderReq):
+        """发送委托"""
+        params = {}
+        params['instrument'] = orderReq.symbol
+        params['units'] = orderReq.volume
+        params['side'] = directionMap.get(orderReq.direction, '')
+        params['price'] = orderReq.price
+        params['type'] = priceTypeMap.get(orderReq.priceType, '')
+        
+        # 委托有效期24小时
+        expire = datetime.datetime.now() + datetime.timedelta(days=1)
+        params['expiry'] = expire.isoformat('T') + 'Z'
+        
+        self.sendOrder(params)
+    
+    #----------------------------------------------------------------------
+    def cancelOrder_(self, cancelOrderReq):
+        """撤销委托"""
+        self.cancelOrder(cancelOrderReq.orderID)
+    
+    
+#----------------------------------------------------------------------
+def getTime(t):
+    """把OANDA返回的时间格式转化为简单的时间字符串"""
+    return t[11:19]

vn.trader/oandaGateway/vnoanda.py

@@ -0,0 +1,608 @@
+# encoding: utf-8
+
+import json
+import requests
+from Queue import Queue, Empty
+from threading import Thread
+
+
+API_SETTING = {}
+API_SETTING['practice'] = {'rest': 'https://api-fxpractice.oanda.com',
+                           'stream': 'https://stream-fxpractice.oanda.com'}
+API_SETTING['trade'] = {'rest': 'https://api-fxpractice.oanda.com',
+                        'stream': 'https://stream-fxtrade.oanda.com/'}
+
+
+FUNCTIONCODE_GETINSTRUMENTS = 0
+FUNCTIONCODE_GETPRICES = 1
+FUNCTIONCODE_GETPRICEHISTORY = 2
+FUNCTIONCODE_GETACCOUNTS = 3
+FUNCTIONCODE_GETACCOUNTINFO = 4
+FUNCTIONCODE_GETORDERS = 5
+FUNCTIONCODE_SENDORDER = 6
+FUNCTIONCODE_GETORDERINFO = 7
+FUNCTIONCODE_MODIFYORDER = 8
+FUNCTIONCODE_CANCELORDER = 9
+FUNCTIONCODE_GETTRADES = 10
+FUNCTIONCODE_GETTRADEINFO = 11
+FUNCTIONCODE_MODIFYTRADE= 12
+FUNCTIONCODE_CLOSETRADE = 13
+FUNCTIONCODE_GETPOSITIONS = 14
+FUNCTIONCODE_GETPOSITIONINFO= 15
+FUNCTIONCODE_CLOSEPOSITION = 16
+FUNCTIONCODE_GETTRANSACTIONS = 17
+FUNCTIONCODE_GETTRANSACTIONINFO = 18
+FUNCTIONCODE_GETACCOUNTHISTORY = 19
+FUNCTIONCODE_GETCALENDAR = 20
+FUNCTIONCODE_GETPOSITIONRATIOS = 21
+FUNCTIONCODE_GETSPREADS = 22
+FUNCTIONCODE_GETCOMMIMENTS = 23
+FUNCTIONCODE_GETORDERBOOK = 24
+FUNCTIONCODE_GETAUTOCHARTIST = 25
+FUNCTIONCODE_STREAMPRICES = 26
+FUNCTIONCODE_STREAMEVENTS = 27
+
+
+########################################################################
+class OandaApi(object):
+    """"""
+    DEBUG = False
+
+    #----------------------------------------------------------------------
+    def __init__(self):
+        """Constructor"""
+        self.token = ''
+        self.accountId = ''
+        self.headers = {}
+        self.restDomain = ''
+        self.streamDomain = ''
+        self.session = None
+        
+        self.functionSetting = {}
+        
+        self.active = False         # API的工作状态
+        
+        self.reqID = 0              # 请求编号
+        self.reqQueue = Queue()     # 请求队列
+        self.reqThread = Thread(target=self.processQueue)   # 请求处理线程
+        
+        self.streamPricesThread = Thread(target=self.processStreamPrices)   # 实时行情线程
+        self.streamEventsThread = Thread(target=self.processStreamEvents)   # 实时事件线程（成交等）
+        
+    #----------------------------------------------------------------------
+    def init(self, settingName, token, accountId):
+        """初始化接口"""
+        self.restDomain = API_SETTING[settingName]['rest']
+        self.streamDomain = API_SETTING[settingName]['stream']
+        self.session = requests.Session()
+        
+        self.token = token
+        self.accountId = accountId
+        
+        self.headers['Authorization'] = 'Bearer ' + self.token
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETINSTRUMENTS, {'path': '/v1/instruments',
+                                                          'method': 'GET'})
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETPRICES, {'path': '/v1/prices',
+                                                          'method': 'GET'})        
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETPRICEHISTORY, {'path': 'v1/candles',
+                                                                'method': 'GET'})
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETACCOUNTS, {'path': '/v1/accounts',
+                                                          'method': 'GET'})  
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETACCOUNTINFO, {'path': '/v1/accounts/%s' %self.accountId,
+                                                          'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETORDERS, {'path': '/v1/accounts/%s/orders' %self.accountId,
+                                                          'method': 'GET'})        
+        
+        self.initFunctionSetting(FUNCTIONCODE_SENDORDER, {'path': '/v1/accounts/%s/orders' %self.accountId,
+                                                          'method': 'POST'})     
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETORDERINFO, {'path': '/v1/accounts/%s/orders' %self.accountId,
+                                                          'method': 'GET'})     
+        
+        self.initFunctionSetting(FUNCTIONCODE_MODIFYORDER, {'path': '/v1/accounts/%s/orders' %self.accountId,
+                                                          'method': 'PATCH'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_CANCELORDER, {'path': '/v1/accounts/%s/orders' %self.accountId,
+                                                          'method': 'DELETE'})             
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETTRADES, {'path': '/v1/accounts/%s/trades' %self.accountId,
+                                                          'method': 'GET'})        
+
+        self.initFunctionSetting(FUNCTIONCODE_GETTRADEINFO, {'path': '/v1/accounts/%s/trades' %self.accountId,
+                                                             'method': 'GET'})     
+        
+        self.initFunctionSetting(FUNCTIONCODE_MODIFYTRADE, {'path': '/v1/accounts/%s/trades' %self.accountId,
+                                                            'method': 'PATCH'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_CLOSETRADE, {'path': '/v1/accounts/%s/trades' %self.accountId,
+                                                           'method': 'DELETE'})     
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETPOSITIONS, {'path': '/v1/accounts/%s/positions' %self.accountId,
+                                                             'method': 'GET'})     
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETPOSITIONINFO, {'path': '/v1/accounts/%s/positions' %self.accountId,
+                                                                'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_CLOSEPOSITION, {'path': '/v1/accounts/%s/positions' %self.accountId,
+                                                              'method': 'DELETE'})            
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETTRANSACTIONS, {'path': '/v1/accounts/%s/transactions' %self.accountId,
+                                                                'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETTRANSACTIONINFO, {'path': '/v1/accounts/%s/transactions' %self.accountId,
+                                                                   'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETACCOUNTHISTORY, {'path': '/v1/accounts/%s/alltransactions' %self.accountId,
+                                                                  'method': 'GET'})     
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETCALENDAR, {'path': '/labs/v1/calendar',
+                                                            'method': 'GET'})           
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETPOSITIONRATIOS, {'path': '/labs/v1/historical_position_ratios',
+                                                                  'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETSPREADS, {'path': '/labs/v1/spreads',
+                                                           'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETCOMMIMENTS, {'path': '/labs/v1/commitments',
+                                                              'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETORDERBOOK, {'path': '/labs/v1/orderbook_data',
+                                                             'method': 'GET'})    
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETAUTOCHARTIST, {'path': '/labs/v1/autochartist',
+                                                                'method': 'GET'})  
+        
+        self.initFunctionSetting(FUNCTIONCODE_GETAUTOCHARTIST, {'path': '/labs/v1/autochartist',
+                                                                'method': 'GET'})  
+        
+        self.initFunctionSetting(FUNCTIONCODE_STREAMPRICES, {'path': '/v1/prices',
+                                                             'method': 'GET'})  
+        
+        self.initFunctionSetting(FUNCTIONCODE_STREAMEVENTS, {'path': '/v1/events',
+                                                             'method': 'GET'})          
+        
+        
+        self.active = True
+        self.reqThread.start()
+        self.streamEventsThread.start()
+        self.streamPricesThread.start()
+        
+    #----------------------------------------------------------------------
+    def exit(self):
+        """退出接口"""
+        self.active = False
+        self.reqThread.join()
+        
+    #----------------------------------------------------------------------
+    def initFunctionSetting(self, code, setting):
+        """初始化API功能字典"""
+        self.functionSetting[code] = setting
+        
+    #----------------------------------------------------------------------
+    def processRequest(self, req):
+        """发送请求并通过回调函数推送数据结果"""
+        url = req['url']
+        method = req['method']
+        params = req['params']
+        
+        stream = False
+        if 'stream' in req:
+            stream = req['stream']
+
+        if method in ['GET', 'DELETE']:
+            myreq = requests.Request(method, url, headers=self.headers, params=params)
+        elif method in ['POST', 'PATCH']:
+            myreq = requests.Request(method, url, headers=self.headers, data=params)
+        pre = myreq.prepare()
+
+        r = None
+        error = None
+        
+        try:
+            r = self.session.send(pre, stream=stream)
+        except Exception, e:
+            error = e
+
+        return r, error
+    
+    #----------------------------------------------------------------------
+    def processQueue(self):
+        """处理请求队列中的请求"""
+        while self.active:
+            try:
+                req = self.reqQueue.get(block=True, timeout=1)  # 获取请求的阻塞为一秒
+                callback = req['callback']
+                reqID = req['reqID']
+                
+                r, error = self.processRequest(req)
+                
+                if r:
+                    try:
+                        data = r.json()
+                        if self.DEBUG:
+                            print callback.__name__                        
+                        callback(data, reqID)    
+                    except Exception, e:                  
+                        self.onError(str(e), reqID)                      
+                else:                
+                    self.onError(error, reqID)
+            except Empty:
+                pass
+            
+    #----------------------------------------------------------------------
+    def sendRequest(self, code, params, callback, optional=''):
+        """发送请求"""
+        setting = self.functionSetting[code]
+        
+        url = self.restDomain + setting['path']
+        if optional:
+            url = url + '/' + optional
+            
+        self.reqID += 1
+            
+        req = {'url': url,
+               'method': setting['method'],
+               'params': params,
+               'callback': callback,
+               'reqID': self.reqID}
+        self.reqQueue.put(req)
+        
+        return self.reqID
+    
+    #----------------------------------------------------------------------
+    def onError(self, error, reqID):
+        """错误信息回调"""
+        print error, reqID
+            
+    #----------------------------------------------------------------------
+    def getInstruments(self, params):
+        """查询可交易的合约列表"""
+        return self.sendRequest(FUNCTIONCODE_GETINSTRUMENTS, params, self.onGetInstruments)
+        
+    #----------------------------------------------------------------------
+    def onGetInstruments(self, data, reqID):
+        """回调函数"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def getPrices(self, params):
+        """查询价格"""
+        return self.sendRequest(FUNCTIONCODE_GETPRICES, params, self.onGetPrices)
+        
+    #----------------------------------------------------------------------
+    def onGetPrices(self, data, reqID):
+        """回调函数"""
+        pass
+        
+    #----------------------------------------------------------------------
+    def getPriceHisory(self, params):
+        """查询历史价格数据"""
+        return self.sendRequest(FUNCTIONCODE_GETPRICEHISTORY, params, self.onGetPriceHistory)
+        
+    #----------------------------------------------------------------------
+    def onGetPriceHistory(self, data, reqID):
+        """回调函数"""
+        pass    
+        
+    #----------------------------------------------------------------------
+    def getAccounts(self):
+        """查询用户的所有账户"""
+        return self.sendRequest(FUNCTIONCODE_GETACCOUNTS, {}, self.onGetAccounts)
+    
+    #----------------------------------------------------------------------
+    def onGetAccounts(self, data, reqID):
+        """回调函数"""
+        pass   
+        
+    #----------------------------------------------------------------------
+    def getAccountInfo(self):
+        """查询账户数据"""
+        return self.sendRequest(FUNCTIONCODE_GETACCOUNTINFO, {}, self.onGetAccountInfo)
+    
+    #----------------------------------------------------------------------
+    def onGetAccountInfo(self, data, reqID):
+        """回调函数"""
+        pass   
+        
+    #----------------------------------------------------------------------
+    def getOrders(self, params):
+        """查询所有委托"""
+        return self.sendRequest(FUNCTIONCODE_GETORDERS, params, self.onGetOrders)
+    
+    #----------------------------------------------------------------------
+    def onGetOrders(self, data, reqID):
+        """回调函数"""
+        pass     
+
+    #----------------------------------------------------------------------
+    def sendOrder(self, params):
+        """发送委托"""
+        return self.sendRequest(FUNCTIONCODE_SENDORDER, params, self.onSendOrder)
+    
+    #----------------------------------------------------------------------
+    def onSendOrder(self, data, reqID):
+        """回调函数"""
+        pass         
+    
+    #----------------------------------------------------------------------
+    def getOrderInfo(self, optional):
+        """查询委托信息"""
+        return self.sendRequest(FUNCTIONCODE_GETORDERINFO, {}, self.onGetOrderInfo, optional)
+    
+    #----------------------------------------------------------------------
+    def onGetOrderInfo(self, data, reqID):
+        """回调函数"""
+        pass     
+        
+    #----------------------------------------------------------------------
+    def modifyOrder(self, params, optional):
+        """修改委托"""
+        return self.sendRequest(FUNCTIONCODE_MODIFYORDER, params, self.onModifyOrder, optional)
+    
+    #----------------------------------------------------------------------
+    def onModifyOrder(self, data, reqID):
+        """回调函数"""
+        pass      
+    
+    #----------------------------------------------------------------------
+    def cancelOrder(self, optional):
+        """查询委托信息"""
+        return self.sendRequest(FUNCTIONCODE_CANCELORDER, {}, self.onCancelOrder, optional)
+    
+    #----------------------------------------------------------------------
+    def onCancelOrder(self, data, reqID):
+        """回调函数"""
+        pass     
+    
+    #----------------------------------------------------------------------
+    def getTrades(self, params):
+        """查询所有仓位"""
+        return self.sendRequest(FUNCTIONCODE_GETTRADES, params, self.onGetTrades)
+    
+    #----------------------------------------------------------------------
+    def onGetTrades(self, data, reqID):
+        """回调函数"""
+        pass     
+    
+    #----------------------------------------------------------------------
+    def getTradeInfo(self, optional):
+        """查询仓位信息"""
+        return self.sendRequest(FUNCTIONCODE_GETTRADEINFO, {}, self.onGetTradeInfo, optional)
+    
+    #----------------------------------------------------------------------
+    def onGetTradeInfo(self, data, reqID):
+        """回调函数"""
+        pass     
+        
+    #----------------------------------------------------------------------
+    def modifyTrade(self, params, optional):
+        """修改仓位"""
+        return self.sendRequest(FUNCTIONCODE_MODIFYTRADE, params, self.onModifyTrade, optional)
+    
+    #----------------------------------------------------------------------
+    def onModifyTrade(self, data, reqID):
+        """回调函数"""
+        pass      
+    
+    #----------------------------------------------------------------------
+    def closeTrade(self, optional):
+        """平仓"""
+        return self.sendRequest(FUNCTIONCODE_CLOSETRADE, {}, self.onCloseTrade, optional)
+    
+    #----------------------------------------------------------------------
+    def onCloseTrade(self, data, reqID):
+        """回调函数"""
+        pass         
+
+    #----------------------------------------------------------------------
+    def getPositions(self):
+        """查询所有汇总仓位"""
+        return self.sendRequest(FUNCTIONCODE_GETPOSITIONS, {}, self.onGetPositions)
+    
+    #----------------------------------------------------------------------
+    def onGetPositions(self, data, reqID):
+        """回调函数"""
+        pass  
+        
+    #----------------------------------------------------------------------
+    def getPositionInfo(self, optional):
+        """查询汇总仓位信息"""
+        return self.sendRequest(FUNCTIONCODE_GETPOSITIONINFO, {}, self.onGetPositionInfo, optional)
+    
+    #----------------------------------------------------------------------
+    def onGetPositionInfo(self, data, reqID):
+        """回调函数"""
+        pass         
+
+    #----------------------------------------------------------------------
+    def closePosition(self, optional):
+        """平仓汇总仓位信息"""
+        return self.sendRequest(FUNCTIONCODE_CLOSEPOSITION, {}, self.onClosePosition, optional)
+    
+    #----------------------------------------------------------------------
+    def onClosePosition(self, data, reqID):
+        """回调函数"""
+        pass      
+    
+    
+    #----------------------------------------------------------------------
+    def getTransactions(self, params):
+        """查询所有资金变动"""
+        return self.sendRequest(FUNCTIONCODE_GETTRANSACTIONS, params, self.onGetTransactions)
+    
+    #----------------------------------------------------------------------
+    def onGetTransactions(self, data, reqID):
+        """回调函数"""
+        pass  
+        
+    #----------------------------------------------------------------------
+    def getTransactionInfo(self, optional):
+        """查询资金变动信息"""
+        return self.sendRequest(FUNCTIONCODE_GETTRANSACTIONINFO, {}, self.onGetTransactionInfo, optional)
+    
+    #----------------------------------------------------------------------
+    def onGetTransactionInfo(self, data, reqID):
+        """回调函数"""
+        pass   
+        
+    #----------------------------------------------------------------------
+    def getAccountHistory(self):
+        """查询账户资金变动历史"""
+        return self.sendRequest(FUNCTIONCODE_GETACCOUNTHISTORY, {}, self.onGetAccountHistory)
+    
+    #----------------------------------------------------------------------
+    def onGetAccountHistory(self, data, reqID):
+        """回调函数"""
+        pass      
+    
+    #----------------------------------------------------------------------
+    def getCalendar(self, params):
+        """查询日历"""
+        return self.sendRequest(FUNCTIONCODE_GETCALENDAR, params, self.onGetCalendar)
+    
+    #----------------------------------------------------------------------
+    def onGetCalendar(self, data, reqID):
+        """回调函数"""
+        pass       
+        
+    #----------------------------------------------------------------------
+    def getPositionRatios(self, params):
+        """查询持仓比例"""
+        return self.sendRequest(FUNCTIONCODE_GETPOSITIONRATIOS, params, self.onGetPositionRatios)
+    
+    #----------------------------------------------------------------------
+    def onGetPositionRatios(self, data, reqID):
+        """回调函数"""
+        pass  
+        
+    #----------------------------------------------------------------------
+    def getSpreads(self, params):
+        """查询所有仓位"""
+        return self.sendRequest(FUNCTIONCODE_GETSPREADS, params, self.onGetSpreads)
+    
+    #----------------------------------------------------------------------
+    def onGetSpreads(self, data, reqID):
+        """回调函数"""
+        pass       
+        
+    #----------------------------------------------------------------------
+    def getCommitments(self, params):
+        """查询交易商持仓情况"""
+        return self.sendRequest(FUNCTIONCODE_GETCOMMIMENTS, params, self.onGetCommitments)
+    
+    #----------------------------------------------------------------------
+    def onGetCommitments(self, data, reqID):
+        """回调函数"""
+        pass       
+    
+    #----------------------------------------------------------------------
+    def getOrderbook(self, params):
+        """查询订单簿"""
+        return self.sendRequest(FUNCTIONCODE_GETORDERBOOK, params, self.onGetOrderbook)
+    
+    #----------------------------------------------------------------------
+    def onGetOrderbook(self, data, reqID):
+        """回调函数"""
+        pass       
+        
+    #----------------------------------------------------------------------
+    def getAutochartist(self, params):
+        """查询Autochartist识别的模式"""
+        return self.sendRequest(FUNCTIONCODE_GETAUTOCHARTIST, params, self.onGetAutochartist)
+    
+    #----------------------------------------------------------------------
+    def onGetAutochartist(self, data, reqID):
+        """回调函数"""
+        pass   
+        
+    #----------------------------------------------------------------------
+    def onPrice(self, data):
+        """行情推送"""
+        print data
+        
+    #----------------------------------------------------------------------
+    def onEvent(self, data):
+        """事件推送（成交等）"""
+        print data
+        
+    #----------------------------------------------------------------------
+    def processStreamPrices(self):
+        """获取价格推送"""
+        # 首先获取所有合约的代码
+        setting = self.functionSetting[FUNCTIONCODE_GETINSTRUMENTS]
+        req = {'url': self.restDomain + setting['path'],
+               'method': setting['method'],
+               'params': {'accountId': self.accountId}}
+        r, error = self.processRequest(req)
+        if r:
+            try:
+                data = r.json()
+                symbols = [d['instrument'] for d in data['instruments']]   
+            except Exception, e:
+                self.onError(e, -1)
+                return
+        else:
+            self.onError(error, -1)
+            return
+
+        # 然后订阅所有的合约行情
+        setting = self.functionSetting[FUNCTIONCODE_STREAMPRICES]
+        params = {'accountId': self.accountId,
+                  'instruments': ','.join(symbols)}
+        req = {'url': self.streamDomain + setting['path'],
+               'method': setting['method'],
+               'params': params,
+               'stream': True}      
+        r, error = self.processRequest(req)
+        
+        if r:
+            for line in r.iter_lines():
+                if line:
+                    try:
+                        msg = json.loads(line)
+                        
+                        if self.DEBUG:
+                            print self.onPrice.__name__
+                            
+                        self.onPrice(msg)
+                    except Exception, e:
+                        self.onError(e, -1)
+                
+                if not self.active:
+                    break
+        else:
+            self.onError(error, -1)
+    
+    #----------------------------------------------------------------------
+    def processStreamEvents(self):
+        """获取事件推送"""
+        setting = self.functionSetting[FUNCTIONCODE_STREAMEVENTS]
+        req = {'url': self.streamDomain + setting['path'],
+               'method': setting['method'],
+               'params': {},
+               'stream': True}
+        r, error = self.processRequest(req)
+        if r:
+            for line in r.iter_lines():
+                if line:
+                    try:
+                        msg = json.loads(line)
+                        
+                        if self.DEBUG:
+                            print self.onEvent.__name__
+                            
+                        self.onEvent(msg)
+                    except Exception, e:
+                        self.onError(e, -1)
+                
+                if not self.active:
+                    break
+        else:
+            self.onError(error, -1)

vn.trader/uiMainWindow.py

@@ -91,6 +91,9 @@ class MainWindow(QtGui.QMainWindow):
         connectIbAction = QtGui.QAction(u'连接IB', self)
         connectIbAction.triggered.connect(self.connectIb) 
         
+        connectOandaAction = QtGui.QAction(u'连接OANDA', self)
+        connectOandaAction.triggered.connect(self.connectOanda)
+        
         connectDbAction = QtGui.QAction(u'连接数据库', self)
         connectDbAction.triggered.connect(self.mainEngine.dbConnect)
         
@@ -118,7 +121,9 @@ class MainWindow(QtGui.QMainWindow):
         sysMenu.addAction(connectFemasAction)
         sysMenu.addAction(connectKsotpAction)
         sysMenu.addAction(connectKsgoldAction)
+        sysMenu.addSeparator()
         sysMenu.addAction(connectIbAction)    
+        sysMenu.addAction(connectOandaAction)
         sysMenu.addSeparator()
         sysMenu.addAction(connectWindAction)
         sysMenu.addSeparator()
@@ -202,6 +207,11 @@ class MainWindow(QtGui.QMainWindow):
         """连接Ib"""
         self.mainEngine.connect('IB')
         
+    #----------------------------------------------------------------------
+    def connectOanda(self):
+        """连接OANDA"""
+        self.mainEngine.connect('OANDA')
+        
     #----------------------------------------------------------------------
     def test(self):
         """测试按钮用的函数"""

vn.trader/vtConstant.py

@@ -66,6 +66,8 @@ EXCHANGE_SMART = 'SMART'       # IB智能路由（股票、期权）
 EXCHANGE_GLOBEX = 'GLOBEX'     # CME电子交易平台
 EXCHANGE_IDEALPRO = 'IDEALPRO' # IB外汇ECN
 
+EXCHANGE_OANDA = 'OANDA'       # OANDA外汇做市商
+
 # 货币类型
 CURRENCY_USD = 'USD'            # 美元
 CURRENCY_CNY = 'CNY'            # 人民币

vn.trader/vtEngine.py

@@ -87,6 +87,13 @@ class MainEngine(object):
             self.addGateway(IbGateway, 'IB')
         except Exception, e:
             print e
+            
+        try:
+            from oandaGateway.oandaGateway import OandaGateway
+            self.addGateway(OandaGateway, 'OANDA')
+            self.gatewayDict['OANDA'].setQryEnabled(True)
+        except Exception, e:
+            print e
 
     #----------------------------------------------------------------------
     def addGateway(self, gateway, gatewayName=None):