vnpy/vn.trader/gateway/lhangGateway/lhangGateway.py

# encoding: UTF-8

'''
vn.lhang的gateway接入
'''


import os
import json
from datetime import datetime
from time import sleep

import vnlhang
from vtGateway import *


SYMBOL_BTCCNY = 'BTCCNY'
SYMBOL_ZECCNY = 'ZECCNY'

SYMBOL_MAP = {}
SYMBOL_MAP['btc_cny'] = SYMBOL_BTCCNY
SYMBOL_MAP['zec_cny'] = SYMBOL_ZECCNY
SYMBOL_MAP_REVERSE = {v: k for k, v in SYMBOL_MAP.items()}


DIRECTION_MAP = {}
DIRECTION_MAP['buy'] = DIRECTION_LONG
DIRECTION_MAP['sell'] = DIRECTION_SHORT

STATUS_MAP = {}
STATUS_MAP[0] = STATUS_NOTTRADED
STATUS_MAP[1] = STATUS_PARTTRADED
STATUS_MAP[2] = STATUS_ALLTRADED
STATUS_MAP[4] = STATUS_UNKNOWN
STATUS_MAP[-1] = STATUS_CANCELLED


########################################################################
class LhangGateway(VtGateway):
    """链行接口"""

    #----------------------------------------------------------------------
    def __init__(self, eventEngine, gatewayName='LHANG'):
        """Constructor"""
        super(LhangGateway, self).__init__(eventEngine, gatewayName)
        
        self.api = LhangApi(self)
        
    #----------------------------------------------------------------------
    def connect(self):
        """连接"""
        # 载入json文件
        fileName = self.gatewayName + '_connect.json'
        path = os.path.abspath(os.path.dirname(__file__))
        fileName = os.path.join(path, 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:
            accessKey = str(setting['apiKey'])
            secretKey = str(setting['secretKey'])
            interval = setting['interval']
            debug = setting['debug']
        except KeyError:
            log = VtLogData()
            log.gatewayName = self.gatewayName
            log.logContent = u'连接配置缺少字段，请检查'
            self.onLog(log)
            return            
        
        # 初始化接口
        self.api.connect(accessKey, secretKey, interval, debug)
        self.writeLog(u'接口初始化成功')
        
        # 启动查询
        self.initQuery()
        self.startQuery()
        
    #----------------------------------------------------------------------
    def writeLog(self, content):
        """发出日志"""
        log = VtLogData()
        log.gatewayName = self.gatewayName
        log.logContent = content
        self.onLog(log)        
    
    #----------------------------------------------------------------------
    def subscribe(self, subscribeReq):
        """订阅行情，自动订阅全部行情，无需实现"""
        pass
        
    #----------------------------------------------------------------------
    def sendOrder(self, orderReq):
        """发单"""
        self.api.sendOrder(orderReq)
        
    #----------------------------------------------------------------------
    def cancelOrder(self, cancelOrderReq):
        """撤单"""
        self.api.cancel(cancelOrderReq)
        
    #----------------------------------------------------------------------
    def qryAccount(self):
        """查询账户资金"""
        pass
        
    #----------------------------------------------------------------------
    def qryPosition(self):
        """查询持仓"""
        pass
        
    #----------------------------------------------------------------------
    def close(self):
        """关闭"""
        self.api.exit()
        
    #----------------------------------------------------------------------
    def initQuery(self):
        """初始化连续查询"""
        if self.qryEnabled:
            self.qryFunctionList = [self.api.queryPrice,
                                    self.api.queryWorkingOrders,
                                    self.api.queryAccount]
            self.startQuery()  
    
    #----------------------------------------------------------------------
    def query(self, event):
        """注册到事件处理引擎上的查询函数"""
        for function in self.qryFunctionList:
            function()
                
    #----------------------------------------------------------------------
    def startQuery(self):
        """启动连续查询"""
        self.eventEngine.register(EVENT_TIMER, self.query)
    
    #----------------------------------------------------------------------
    def setQryEnabled(self, qryEnabled):
        """设置是否要启动循环查询"""
        self.qryEnabled = qryEnabled
    

########################################################################
class LhangApi(vnlhang.LhangApi):
    """"""

    #----------------------------------------------------------------------
    def __init__(self, gateway):
        """Constructor"""
        super(LhangApi, self).__init__()
        
        self.gateway = gateway
        self.gatewayName = gateway.gatewayName
        
        self.interval = 1

        self.localID = 0            # 本地委托号
        self.localSystemDict = {}   # key:localID, value:systemID
        self.systemLocalDict = {}   # key:systemID, value:localID
        self.workingOrderDict = {}  # key:localID, value:order
        self.reqLocalDict = {}      # key:reqID, value:localID
        self.cancelDict = {}        # key:localID, value:cancelOrderReq

        self.tradeID = 0

        self.tickDict = {}      # key:symbol, value:tick

    #----------------------------------------------------------------------
    def onError(self, error, req, reqID):
        """错误推送"""
        err = VtErrorData()
        err.gatewayName = self.gatewayName
        err.errorMsg = str(error)
        err.errorTime = datetime.now().strftime('%H:%M:%S')
        self.gateway.onError(err)

    #----------------------------------------------------------------------
    def onGetTicker(self, data, req, reqID):
        """查询行情回调"""
        ticker = data['ticker']
        params = req['params']
        symbol = SYMBOL_MAP[params['symbol']]

        if symbol not in self.tickDict:
            tick = VtTickData()
            tick.gatewayName = self.gatewayName

            tick.symbol = symbol
            tick.exchange = EXCHANGE_LHANG
            tick.vtSymbol = '.'.join([tick.symbol, tick.exchange])
            self.tickDict[symbol] = tick
        else:
            tick = self.tickDict[symbol]

        tick.highPrice = float(ticker['high'])
        tick.lowPrice = float(ticker['low'])
        tick.lastPrice = float(ticker['latest'])
        tick.openPrice = tick.lastPrice - float(ticker['change'])
        tick.volume = ticker['vol']

    # ----------------------------------------------------------------------
    def onGetDepth(self, data, req, reqID):
        """查询深度回调"""
        params = req['params']
        symbol = SYMBOL_MAP[params['symbol']]
        if symbol not in self.tickDict:
            tick = VtTickData()
            tick.gatewayName = self.gatewayName

            tick.symbol = symbol
            tick.exchange = EXCHANGE_LHANG
            tick.vtSymbol = '.'.join([tick.symbol, tick.exchange])
            self.tickDict[symbol] = tick
        else:
            tick = self.tickDict[symbol]

        tick.bidPrice1, tick.bidVolume1 = data['bids'][0]
        tick.bidPrice2, tick.bidVolume2 = data['bids'][1]
        tick.bidPrice3, tick.bidVolume3 = data['bids'][2]
        tick.bidPrice4, tick.bidVolume4 = data['bids'][3]
        tick.bidPrice5, tick.bidVolume5 = data['bids'][4]

        tick.askPrice1, tick.askVolume1 = data['asks'][0]
        tick.askPrice2, tick.askVolume2 = data['asks'][1]
        tick.askPrice3, tick.askVolume3 = data['asks'][2]
        tick.askPrice4, tick.askVolume4 = data['asks'][3]
        tick.askPrice5, tick.askVolume5 = data['asks'][4]

        now = datetime.now()
        tick.time = now.strftime('%H:%M:%S')
        tick.date = now.strftime('%Y%m%d')

        self.gateway.onTick(tick)

    # ----------------------------------------------------------------------
    def onGetTrades(self, data, req, reqID):
        """查询历史成交"""
        print(data, reqID)

    # ----------------------------------------------------------------------
    def onGetKline(self, data, req, reqID):
        print(data, reqID)
    
    # ----------------------------------------------------------------------
    def onGetUserInfo(self, data, req, reqID):
        """查询Ｋ线回报"""
        d = data['info']
        account = VtAccountData()
        account.gatewayName = self.gatewayName
        account.accountID = self.gatewayName
        account.vtAccountID = '.'.join([account.accountID, self.gatewayName])
        account.balance = d['asset']['net']
        self.gateway.onAccount(account)

        # 推送持仓数据
        posCny = VtPositionData()
        posCny.gatewayName = self.gatewayName
        posCny.symbol = 'CNY'
        posCny.exchange = EXCHANGE_LHANG
        posCny.vtSymbol = '.'.join([posCny.symbol, posCny.exchange])
        posCny.vtPositionName = posCny.vtSymbol
        posCny.frozen = d['freeze']['cny']
        posCny.position = posCny.frozen + d['free']['cny']
        self.gateway.onPosition(posCny)

        posBtc = VtPositionData()
        posBtc.gatewayName = self.gatewayName
        posBtc.symbol = 'BTC'
        posBtc.exchange = EXCHANGE_LHANG
        posBtc.vtSymbol = '.'.join([posBtc.symbol, posBtc.exchange])
        posBtc.vtPositionName = posBtc.vtSymbol
        posBtc.frozen = d['freeze']['btc']
        posBtc.position = posBtc.frozen + d['free']['btc']
        self.gateway.onPosition(posBtc)

        posZec = VtPositionData()
        posZec.gatewayName = self.gatewayName
        posZec.symbol = 'ZEC'
        posZec.exchange = EXCHANGE_LHANG
        posZec.vtSymbol = '.'.join([posZec.symbol, posZec.exchange])
        posZec.vtPositionName = posZec.vtSymbol
        posZec.frozen = d['freeze']['zec']
        posZec.position = posZec.frozen + d['free']['zec']
        self.gateway.onPosition(posZec)

        # 查询历史委托
        self.queryOrders()

    # ----------------------------------------------------------------------
    def onCreateOrder(self, data, req, reqID):
        """发单回调"""
        localID = self.reqLocalDict[reqID]
        systemID = data['id']
        self.localSystemDict[localID] = systemID
        self.systemLocalDict[systemID] = localID

        # 撤单
        if localID in self.cancelDict:
            req = self.cancelDict[localID]
            self.cancel(req)
            del self.cancelDict[localID]

        # 推送委托信息
        order = self.workingOrderDict[localID]
        if data['result'] == 'success':
            order.status = STATUS_NOTTRADED
        self.gateway.onOrder(order)

    # ----------------------------------------------------------------------
    def onCancelOrder(self, data, req, reqID):
        """撤单回调"""
        if data['result'] == 'success':
            systemID = req['params']['id']
            localID = self.systemLocalDict[systemID]

            order = self.workingOrderDict[localID]
            order.status = STATUS_CANCELLED

            del self.workingOrderDict[localID]
            self.gateway.onOrder(order)

    # ----------------------------------------------------------------------
    def onGetOrdersInfo(self, data, req, reqID):
        """查询委托回报"""
        if 'orders' in data:
            for d in data['orders']:
                systemID = d['order_id']
                localID = self.systemLocalDict[systemID]
                order = self.workingOrderDict.get(localID, None)
                if not order:
                    return

                # 记录最新成交的金额
                newTradeVolume = float(d['deal_amount']) - order.tradedVolume
                if newTradeVolume:
                    trade = VtTradeData()
                    trade.gatewayName = self.gatewayName
                    trade.symbol = order.symbol
                    trade.vtSymbol = order.vtSymbol

                    self.tradeID += 1
                    trade.tradeID = str(self.tradeID)
                    trade.vtTradeID = '.'.join([trade.tradeID, trade.gatewayName])

                    trade.volume = newTradeVolume
                    trade.price = d['avg_price']
                    trade.direction = order.direction
                    trade.offset = order.offset
                    trade.exchange = order.exchange
                    trade.tradeTime = datetime.now().strftime('%H:%M:%S')

                    self.gateway.onTrade(trade)

                # 更新委托状态
                order.tradedVolume = float(d['deal_amount'])
                order.status = STATUS_MAP.get(d['status'], STATUS_UNKNOWN)

                if newTradeVolume:
                    self.gateway.onOrder(order)

                if order.status == STATUS_ALLTRADED or order.status == STATUS_CANCELLED:
                    del self.workingOrderDict[order.orderID]

    # ----------------------------------------------------------------------
    def onGetOrdersInfoHistory(self, data, req, reqID):
        """撤单回报"""
        if 'orders' in data:
            for d in data['orders']:
                order = VtOrderData()
                order.gatewayName = self.gatewayName

                order.symbol = SYMBOL_MAP[data['symbol']]
                order.exchange = EXCHANGE_LHANG
                order.vtSymbol = '.'.join([order.symbol, order.exchange])

                systemID = d['order_id']
                self.localID += 1
                localID = str(self.localID)
                self.systemLocalDict[systemID] = localID
                self.localSystemDict[localID] = systemID
                order.orderID = localID
                order.vtOrderID = '.'.join([order.orderID, order.gatewayName])

                order.totalVolume = float(d['amount'])
                order.tradedVolume = float(d['deal_amount'])
                order.price = float(d['price'])
                order.direction = DIRECTION_MAP[d['type']]
                order.offset = OFFSET_NONE
                order.orderTime = datetime.fromtimestamp(d['create_time'], '%H:%M:%S')

                # 委托状态
                if order.tradedVolume == 0:
                    order.status = STATUS_NOTTRADED
                else:
                    order.status = STATUS_PARTTRADED

                # 缓存病推送
                self.workingOrderDict[localID] = order
                self.gateway.onOrder(order)
        
    #----------------------------------------------------------------------
    def connect(self, apiKey, secretKey, interval, debug):
        """初始化"""
        self.interval = interval
        self.DEBUG = debug

        self.init(apiKey, secretKey, self.interval)

        # 推送合约信息
        contract = VtContractData()
        contract.gatewayName = self.gatewayName
        contract.symbol = SYMBOL_BTCCNY
        contract.exchange = EXCHANGE_LHANG
        contract.vtSymbol = '.'.join([contract.symbol, contract.exchange])
        contract.name = u'人民币现货BTC'
        contract.size = 1
        contract.priceTick = 0.01
        contract.productClass = PRODUCT_SPOT
        self.gateway.onContract(contract)

        contract = VtContractData()
        contract.gatewayName = self.gatewayName
        contract.symbol = SYMBOL_ZECCNY
        contract.exchange = EXCHANGE_LHANG
        contract.vtSymbol = '.'.join([contract.symbol, contract.exchange])
        contract.name = u'人民币现货ZEC'
        contract.size = 1
        contract.priceTick = 0.01
        contract.productClass = PRODUCT_SPOT
        self.gateway.onContract(contract)
        
    #----------------------------------------------------------------------
    def sendOrder(self, req):
        """发单"""
        """发送委托"""
        # 检查是否填入了价格，禁止市价委托
        if req.priceType != PRICETYPE_LIMITPRICE:
            err = VtErrorData()
            err.gatewayName = self.gatewayName
            err.errorMsg = u'链行接口仅支持限价单'
            err.errorTime = datetime.now().strftime('%H:%M:%S')
            self.gateway.onError(err)
            return None

        # 发送限价委托
        s = SYMBOL_MAP_REVERSE[req.symbol]

        if req.direction == DIRECTION_LONG:
            type_ = 'buy'
        else:
            type_ = 'sell'

        reqID = self.createOrder(s, type_, req.price, req.volume)

        self.localID += 1
        localID = str(self.localID)
        self.reqLocalDict[reqID] = localID

        # 推送委托信息
        order = VtOrderData()
        order.gatewayName = self.gatewayName

        order.symbol = req.symbol
        order.exchange = EXCHANGE_LHANG
        order.vtSymbol = '.'.join([order.symbol, order.exchange])

        order.orderID = localID
        order.vtOrderID = '.'.join([order.orderID, order.gatewayName])

        order.direction = req.direction
        order.offset = OFFSET_UNKNOWN
        order.price = req.price
        order.volume = req.volume
        order.orderTime = datetime.now().strftime('%H:%M:%S')
        order.status = STATUS_UNKNOWN

        self.workingOrderDict[localID] = order
        self.gateway.onOrder(order)

        # 返回委托号
        return order.vtOrderID
    
    #----------------------------------------------------------------------
    def cancel(self, req):
        """撤单"""
        localID = req.orderID
        if localID in self.localSystemDict:
            systemID = self.localSystemDict[localID]
            s = SYMBOL_MAP_REVERSE[req.symbol]
            self.cancelOrder(s, systemID)
        else:
            self.cancelDict[localID] = req

    #----------------------------------------------------------------------
    def queryOrders(self):
        """查询委托"""
        for s in SYMBOL_MAP.keys():
            self.getOrdersInfoHistory(s, '0', '1', '200')

    #----------------------------------------------------------------------
    def queryWorkingOrders(self):
        """查询活动委托"""
        for localID, order in self.workingOrderDict.items():
            if localID in self.localSystemDict:
                systemID = self.localSystemDict[localID]
                s = SYMBOL_MAP_REVERSE[order.symbol]
                self.getOrdersInfo(s, systemID)

    #----------------------------------------------------------------------
    def queryPrice(self):
        """查询行情"""
        for s in SYMBOL_MAP.keys():
            self.getTicker(s)
            self.getDepth(s, 5, 0)

    #----------------------------------------------------------------------
    def queryAccount(self):
        """查询资金和资产"""
        self.getUserInfo()