vnpy/beta/quantos/tkproGateway/tkproGateway.py

# encoding: UTF-8

'''
quantOS的TkPro系统接入
'''

import sys
import os
import json
import traceback
from datetime import datetime

from vnpy.trader.vtConstant import *
from vnpy.trader.vtObject import *
from vnpy.trader.vtGateway import VtGateway
from vnpy.trader.vtFunction import getJsonPath
from vnpy.trader.vtEvent import EVENT_TIMER

from .DataApi import DataApi
from .TradeApi import TradeApi
from collections import namedtuple



# 以下为一些VT类型和TkPro类型的映射字典

# 动作印射
actionMap = {}
actionMap[(DIRECTION_LONG, OFFSET_OPEN)] = "Buy"
actionMap[(DIRECTION_SHORT, OFFSET_OPEN)] = "Short"
actionMap[(DIRECTION_LONG, OFFSET_CLOSE)] = "Cover"
actionMap[(DIRECTION_SHORT, OFFSET_CLOSE)] = "Sell"
actionMap[(DIRECTION_LONG, OFFSET_CLOSEYESTERDAY)] = "CoverYesterday"
actionMap[(DIRECTION_SHORT, OFFSET_CLOSEYESTERDAY)] = "SellYesterday"
actionMap[(DIRECTION_LONG, OFFSET_CLOSETODAY)] = "CoverToday"
actionMap[(DIRECTION_SHORT, OFFSET_CLOSETODAY)] = "SellToday"
actionMapReverse = {v: k for k, v in actionMap.items()}

# 交易所类型映射
exchangeMap = {}
exchangeMap[EXCHANGE_CFFEX] = 'CFE'
exchangeMap[EXCHANGE_SHFE] = 'SHF'
exchangeMap[EXCHANGE_CZCE] = 'CZC'
exchangeMap[EXCHANGE_DCE] = 'DCE'
exchangeMap[EXCHANGE_SSE] = 'SH'
exchangeMap[EXCHANGE_SZSE] = 'SZ'
exchangeMapReverse = {v:k for k,v in exchangeMap.items()}


# 持仓类型映射
sideMap = {}
sideMap[DIRECTION_LONG] = 'Long'
sideMap[DIRECTION_SHORT] = 'Short'
sideMapReverse = {v:k for k,v in sideMap.items()}

# 产品类型映射
productClassMapReverse = {}
productClassMapReverse[1] = PRODUCT_EQUITY
productClassMapReverse[3] = PRODUCT_EQUITY
productClassMapReverse[4] = PRODUCT_EQUITY
productClassMapReverse[5] = PRODUCT_EQUITY
productClassMapReverse[8] = PRODUCT_BOND
productClassMapReverse[17] = PRODUCT_BOND
productClassMapReverse[101] = PRODUCT_FUTURES
productClassMapReverse[102] = PRODUCT_FUTURES
productClassMapReverse[103] = PRODUCT_FUTURES

# 委托状态映射
statusMapReverse = {}
statusMapReverse['New'] = STATUS_UNKNOWN
statusMapReverse['Accepted'] = STATUS_NOTTRADED
statusMapReverse['Cancelled'] = STATUS_CANCELLED
statusMapReverse['Filled'] = STATUS_ALLTRADED
statusMapReverse['Rejected'] = STATUS_REJECTED



########################################################################
class TkproGateway(VtGateway):
    """TkPro接口"""

    #----------------------------------------------------------------------
    def __init__(self, eventengine, gatewayName='TKPRO'):
        """Constructor"""
        super(TkproGateway, self).__init__(eventengine, gatewayName)
        
        self.dataApi = TkproDataApi(self)      # 行情
        self.tradeApi = TkproTradeApi(self)    # 交易
        
        self.qryEnabled = False         # 是否要启动循环查询
        
        self.fileName = self.gatewayName + '_connect.json'
        self.filePath = getJsonPath(self.fileName, __file__)          
    
    #---------------------------------------------------------------------- 
    def connect(self):
        """连接"""
        try:
            f = file(self.filePath)
        except IOError:
            log = VtLogData()
            log.gatewayName = self.gatewayName
            log.logContent = u'无法加载配置'
            self.onLog(log)
            return
        
        setting = json.load(f)
        try:
            username = str(setting['username'])
            token = str(setting['token'])
            strategy = int(setting['strategy'])
            tradeAddress = str(setting['tradeAddress'])
            dataAddress = str(setting['dataAddress'])
        except KeyError:
            log = VtLogData()
            log.gatewayName = self.gatewayName
            log.logContent = u'连接配置缺少字段，请检查'
            self.onLog(log)
            return

        # 创建行情和交易接口对象
        self.dataApi.connect(dataAddress, username, token)
        self.tradeApi.connect(tradeAddress, username, token, strategy)
        
        # 初始化并启动查询
        self.initQuery()
    
    #----------------------------------------------------------------------
    def subscribe(self, subscribeReq):
        """订阅行情"""
        self.dataApi.subscribe(subscribeReq)
        
    #----------------------------------------------------------------------
    def sendOrder(self, orderReq):
        """发单"""
        self.tradeApi.sendOrder(orderReq)
     
    #----------------------------------------------------------------------
    def cancelOrder(self, cancelOrderReq):
        """撤单"""
        self.tradeApi.cancelOrder(cancelOrderReq)
 
    #----------------------------------------------------------------------
    def qryAccount(self):
        """查询账户资金"""
        self.tradeApi.qryAccount()
        
    #----------------------------------------------------------------------
    def qryPosition(self):
        """查询持仓"""
        self.tradeApi.qryPosition()
        
    #----------------------------------------------------------------------
    def close(self):
        """关闭"""
        pass
        
    #----------------------------------------------------------------------
    def initQuery(self):
        """初始化连续查询"""
        if self.qryEnabled:
            # 需要循环的查询函数列表
            self.qryFunctionList = [self.qryPosition, self.qryAccount]
            
            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 TkproTradeApi(object):
    """TkPro交易API实现"""

    #----------------------------------------------------------------------
    def __init__(self, gateway):
        """Constructor"""
        super(TkproTradeApi, self).__init__()
        
        self.gateway = gateway                  # gateway对象
        self.gatewayName = gateway.gatewayName  # gateway对象名称
        
        self.api = None
    
    #----------------------------------------------------------------------
    def onOrderStatus(self, data):
        """委托信息推送"""
        if isinstance(data, dict):
            data = namedtuple('Order', list(data.keys()))(*list(data.values()))
        
        order = VtOrderData()
        order.gatewayName = self.gatewayName
    
        symbol, exchange = data.security.split('.')
        order.symbol = symbol
        order.exchange = exchangeMapReverse[exchange]
        order.vtSymbol = '.'.join([order.symbol, order.exchange])
    
        order.orderID = str(data.entrust_no)
        order.taskID = str(data.task_id)
        order.vtOrderID = order.orderID
        
        order.direction, order.offset = actionMapReverse.get(data.entrust_action, (DIRECTION_UNKNOWN, OFFSET_UNKNOWN))
        order.totalVolume = data.entrust_size
        order.tradedVolume = data.fill_size
        order.price = data.entrust_price
        order.status = statusMapReverse.get(data.order_status)
        order.tradePrice = data.fill_price 

        t = str(data.entrust_time)
        t = t.rjust(6, '0')
        order.orderTime = '%s:%s:%s' %(t[0:2],t[2:4],t[4:])   
        
        self.gateway.onOrder(order)
    
    #----------------------------------------------------------------------
    def onTaskStatus(self, data):
        """"""
        pass
    
    #----------------------------------------------------------------------
    def onTrade(self, data):
        """成交信息推送"""
        if isinstance(data, dict):
            data = namedtuple('Trade', list(data.keys()))(*list(data.values()))
            
        trade = VtTradeData()
        trade.gatewayName = self.gatewayName
        
        symbol, exchange = data.security.split('.')
        trade.symbol = symbol
        trade.exchange = exchangeMapReverse[exchange]
        trade.vtSymbol = '.'.join([trade.symbol, trade.exchange])
        
        trade.direction, trade.offset = actionMapReverse.get(data.entrust_action, (DIRECTION_UNKNOWN, OFFSET_UNKNOWN))
        
        trade.tradeID = str(data.fill_no)
        trade.vtTradeID = str(data.fill_no)
        
        trade.orderID = str(data.entrust_no)
        trade.vtOrderID = trade.orderID
        trade.taskID = str(data.task_id)
        
        trade.price = data.fill_price
        trade.volume = data.fill_size
        
        t = str(data.fill_time)
        t = t.rjust(6, '0')
        trade.tradeTime = '%s:%s:%s' %(t[0:2],t[2:4],t[4:])        
        
        self.gateway.onTrade(trade)
        
    #----------------------------------------------------------------------
    def onConnection(self, data):
        """"""
        self.writeLog(u'连接状态更新：%s' %data)
        
        if data:
            self.qryInstrument()
            self.qryOrder()
            self.qryTrade()        

    #----------------------------------------------------------------------
    def connect(self, tradeAddress, username, token, strategy):
        """初始化连接"""
        if self.api:
            self.writeLog(u'交易已经连接')
            return
        
        self.api = TradeApi(tradeAddress)
        self.api.set_data_format('obj')
        
        # 登录
        result, msg = self.api.login(username, token)
        
        if not result:
            self.writeLog(u'交易登录失败，错误信息：%s' %msg)
            return
        
        result, msg = self.api.use_strategy(strategy)
        
        if result:
            self.writeLog(u'选定策略号：%s' %strategy)
        else:
            self.writeLog(u'选定策略号失败')
        
        self.api.set_ordstatus_callback(self.onOrderStatus)
        self.api.set_trade_callback(self.onTrade)
        self.api.set_task_callback(self.onTaskStatus)
        self.api.set_connection_callback(self.onConnection)

    #----------------------------------------------------------------------
    def close(self):
        """关闭"""
        pass
            
    #----------------------------------------------------------------------
    def writeLog(self, logContent):
        """记录日志"""
        log = VtLogData()
        log.gatewayName = self.gatewayName
        log.logContent = logContent
        self.gateway.onLog(log)
    
    #----------------------------------------------------------------------
    def sendOrder(self, orderReq):
        """发单"""
        if not self.api:
            return
        
        exchange = exchangeMap.get(orderReq.exchange, '')
        security = '.'.join([orderReq.symbol, exchange])        
        action = actionMap.get((orderReq.direction, orderReq.offset), '')
        
        taskid, msg = self.api.place_order(security, action, orderReq.price, int(orderReq.volume))
        
        if taskid is 0:
            self.writeLog(u'委托失败，错误信息：%s' %msg)
    
    #----------------------------------------------------------------------
    def cancelOrder(self, cancelOrderReq):
        """撤单"""
        if not self.api:
            return

        result, msg = self.api.cancel_order(cancelOrderReq.orderID)
        if result is 0:
            self.writeLog(u'撤单失败，错误信息：%s' %msg)             
    
    #----------------------------------------------------------------------
    def qryPosition(self):
        """查询持仓"""
        l, msg = self.api.query_position()
        
        if l is None:
            self.writeLog(u'查询持仓失败，错误信息：%s' %msg)  
            return False
        
        for data in l:
            position = VtPositionData()
            position.gatewayName = self.gatewayName
            
            symbol, exchange = data.security.split('.')
            position.symbol = symbol
            position.exchange = exchangeMapReverse[exchange]
            position.vtSymbol = '.'.join([position.symbol, position.exchange])
                        
            position.direction = sideMapReverse.get(data.side, DIRECTION_UNKNOWN)
            position.vtPositionName = '.'.join([position.vtSymbol, position.direction])
            
            position.price = data.cost_price
            position.ydPosition = data.pre_size
            position.tdPosition = data.today_size
            position.position = data.current_size
            position.frozen = data.frozen_size
            
            position.commission = data.commission
            position.enable = data.enable_size  
            position.want = data.want_size
            position.initPosition = data.init_size
            position.trading = data.trading_pnl
            position.holding = data.holding_pnl
            position.last = data.last_price
        
            self.gateway.onPosition(position)
        
        return True
    
    #----------------------------------------------------------------------
    def qryAccount(self):
        """查询资金"""
        l, msg = self.api.query_account()
        
        if l is None:
            self.writeLog(u'查询资金失败，错误信息：%s' %msg)  
            return False
        
        for data in l:
            account = VtAccountData()
            account.gatewayName = self.gatewayName
                
            account.accountID = '_'.join([str(data.id), data.type])
            account.vtAccountID = '.'.join([account.accountID, account.gatewayName])
            account.available = data.enable_balance
            account.balance = account.available + data.frozen_balance
            account.closeProfit = data.close_pnl
            account.commission = data.commission
            account.margin = data.margin
            account.positionProfit = data.holding_pnl
            account.preBalance = data.init_balance
            
            self.gateway.onAccount(account)
    
    #----------------------------------------------------------------------
    def qryOrder(self):
        """查询委托"""
        l, msg = self.api.query_order()
        
        if l is None:
            self.writeLog(u'查询委托失败，错误信息：%s' %msg)
        else:
            for data in l:
                self.onOrderStatus(data)
                
            self.writeLog(u'查询委托完成')
    
    #----------------------------------------------------------------------
    def qryTrade(self):
        """查询成交"""
        l, msg = self.api.query_trade()
        
        if l is None:
            self.writeLog(u'查询成交失败，错误信息：%s' %msg)
            return False
        
        for data in l:
            self.onTrade(data)
            
        self.writeLog(u'查询成交完成')
        return True
    
    #----------------------------------------------------------------------
    def qryInstrument(self):
        """查询合约"""
        # 通过DataAPI查询所有信息
        df, msg = self.gateway.dataApi.api.query(
            view='jz.instrumentInfo',
            fields='symbol,name,inst_type,buylot,pricetick,multiplier',
            filter='inst_type=1',
            data_format='pandas'
            )
        
        d = {}
        for n, row in df.iterrows():
            d[row.symbol] = row
        
        # 查询所有信息
        l, msg = self.api.query_universe()
        
        if l is None:
            self.writeLog(u'查询合约失败，错误信息：%s' %msg)
            return False
        
        for data in l:
            row = d[data.security]
            
            contract = VtContractData()
            contract.gatewayName = self.gatewayName
            
            symbol, exchange = data.security.split('.')
            contract.symbol = symbol
            contract.exchange = exchangeMapReverse[exchange]
            contract.vtSymbol = '.'.join([contract.symbol, contract.exchange])
            contract.productClass = PRODUCT_EQUITY
            contract.name = unicode(row['name'])
            contract.priceTick = float(row['pricetick'])
            contract.size = int(row['multiplier'])
            
            self.gateway.onContract(contract)
            
        self.writeLog(u'查询合约完成')
        return True
    
    
########################################################################
class TkproDataApi(object):
    """TkPro行情API实现"""

    #----------------------------------------------------------------------
    def __init__(self, gateway):
        """Constructor"""
        super(TkproDataApi, self).__init__()
        
        self.gateway = gateway
        self.gatewayName = gateway.gatewayName
        
        self.api = None
        
        self.fields = "open,close,high,low,last,\
        volume,turnover,oi,preclose,time,date,\
        askprice1,askprice2,askprice3,askprice4,askprice5,\
        bidprice1,bidprice2,bidprice3,bidprice4,bidprice5,\
        askvolume1,askvolume2,askvolume3,askvolume4,askvolume5,\
        bidvolume1,bidvolume2,bidvolume3,bidvolume4,bidvolume5,\
        limit_up,limit_down"
        
    #----------------------------------------------------------------------
    def onMarketData(self, k, data):	
        """行情推送"""
        tick = VtTickData()
        tick.gatewayName = self.gatewayName
        
        try:
            l = data['symbol'].split('.')
            tick.symbol = l[0]
            tick.exchange = exchangeMapReverse[l[1]]
            tick.vtSymbol = '.'.join([tick.symbol, tick.exchange])
            
            tick.openPrice = data['open']
            tick.highPrice = data['high']
            tick.lowPrice = data['low']
            tick.volume = data['volume']
            tick.turnover = data['turnover']
            tick.lastPrice = data['last']
            
            tick.openInterest = data['oi']
            tick.preClosePrice = data['preclose']
            tick.date = str(data['date'])
            
            t = str(data['time'])
            t = t.rjust(9, '0')
            tick.time = '%s:%s:%s.%s' %(t[0:2],t[2:4],t[4:6],t[6:])
            
            tick.bidPrice1 = data['bidprice1']
            tick.askPrice1 = data['askprice1']
            tick.bidVolume1 = data['bidvolume1']
            tick.askVolume1 = data['askvolume1']
            
            if 'bidprice2' in data:
                tick.bidPrice2 = data['bidprice2']
                tick.bidPrice3 = data['bidprice3']
                tick.bidPrice4 = data['bidprice4']
                tick.bidPrice5 = data['bidprice5']
                
                tick.askPrice2 = data['askprice2']
                tick.askPrice3 = data['askprice3']
                tick.askPrice4 = data['askprice4']
                tick.askPrice5 = data['askprice5']
        
                tick.bidVolume2 = data['bidvolume2']
                tick.bidVolume3 = data['bidvolume3']
                tick.bidVolume4 = data['bidvolume4']
                tick.bidVolume5 = data['bidvolume5']
        
                tick.askVolume2 = data['askvolume2']
                tick.askVolume3 = data['askvolume3']
                tick.askVolume4 = data['askvolume4']
                tick.askVolume5 = data['askvolume5']
            
            tick.upperLimit = data['limit_up']
            tick.lowerLimit = data['limit_down']
    
            self.gateway.onTick(tick)
        except Exception as e:
            self.writeLog(u'行情更新失败，错误信息：%s' % str(e))

    #----------------------------------------------------------------------
    def connect(self, dataAddress, username, token):
        """连接"""
        if self.api:
            self.writeLog(u'行情已经连接')
            return
        
        self.api = DataApi(dataAddress)
        
        result, msg = self.api.login(username, token)
        
        if not result:
            self.writeLog(u'行情登录失败，错误信息：%sa' %str(msg))
            return
        
        self.writeLog(u'行情连接成功')

    #----------------------------------------------------------------------
    def subscribe(self, subscribeReq):
        """订阅行情"""
        exchange = exchangeMap.get(subscribeReq.exchange, '')
        security = '.'.join([subscribeReq.symbol, exchange])
        
        subscribed, msg = self.api.subscribe(security, fields=self.fields, func=self.onMarketData)
        
        if not subscribed:
            self.writeLog(u'行情订阅失败，错误信息：%s' %str(msg))
    
    #----------------------------------------------------------------------
    def writeLog(self, logContent):
        """记录日志"""
        log = VtLogData()
        log.gatewayName = self.gatewayName
        log.logContent = logContent
        self.gateway.onLog(log)