Merge pull request #30 from lyic/master

增加了vn.trader的OANDA接入
This commit is contained in:
vn.py 2016-03-06 18:11:26 +08:00
commit 42b51f9eae
11 changed files with 1092 additions and 4 deletions

View File

@ -109,7 +109,7 @@
---
### 联系作者
作者知乎名用python的trader,想要联系作者可以通过知乎私信
作者知乎名用python的交易员,想要联系作者可以通过知乎私信
---
### License

View File

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

View File

@ -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

Binary file not shown.

View File

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

View File

View File

@ -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]

View File

@ -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)

View File

@ -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):
"""测试按钮用的函数"""

View File

@ -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' # 人民币

View File

@ -88,6 +88,13 @@ class MainEngine(object):
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):
"""创建接口"""