<reponame>lamter/slaveaccount<filename>slaveaccount/ctpgateway.py
# encoding: UTF-8

'''
vn.ctp的gateway接入

考虑到现阶段大部分CTP中的ExchangeID字段返回的都是空值
vtSymbol直接使用symbol
'''
import traceback
import pymongo.database
from pymongo import IndexModel, DESCENDING, ASCENDING
from pymongo.errors import OperationFailure
from threading import Event, Thread, Timer

import arrow
import tradingtime as tt

try:
    from queue import Queue, Empty
except ImportError:
    from queue import Queue, Empty
from slavem import Reporter

from slaveaccount.gateway import Gateway
from slaveaccount.constant import *
from slaveaccount.ctpmdapi import CtpMdApi
from slaveaccount.ctptdapi import CtpTdApi
from slaveaccount.object import VtQuery, VtSaveMongodbData


########################################################################
class CtpGateway(Gateway):
    """CTP接口"""

    # ----------------------------------------------------------------------
    def __init__(self, config, gatewayName, userID, client):
        """Constructor"""
        super(CtpGateway, self).__init__(config, gatewayName, userID, client)

        self.confg = config
        self.userID = userID
        self.password = config.get(userID, 'password')
        self.brokerID = config.get(userID, 'brokerID')
        self.tdAddress = config.get(userID, 'tdAddress')
        self.mdAddress = config.get(userID, 'mdAddress')
        self.db = None
        self.authCode = config.get(userID, 'authCode')
        self.appID = config.get(userID, 'appID')

        self.mdApi = CtpMdApi(self)  # 行情API
        self.tdApi = CtpTdApi(self)  # 交易API

        self.mdConnected = False  # 行情API连接状态，登录完成后为True
        self.tdConnected = False  # 交易API连接状态

        # 交易流水
        self.transferSerials = {}

        # 查询定时器
        self.qryTransferSerialTimer = None
        self.qryAccountTimer = None

        # 存库定时器
        self.saveAccountTimer = None

        # 查询队列
        self.qryQueue = Queue()  # VtQuery

        self.stoped = Event()
        self.queryForever = Thread(name='query', target=self.query)

        self.saveQueue = Queue()  # VtSaveMongodbData
        self.saveForever = Thread(name='save', target=self._save)

        self.slavemReport = Reporter(
            name=config.get('slavem', 'name'),
            type=config.get('slavem', 'type'),
            host=config.get('slavem', 'host'),
            port=config.getint('slavem', 'port'),
            dbn=config.get('slavem', 'dbn'),
            username=config.get('slavem', 'username'),
            password=config.get('slavem', 'password'),
            localhost=config.get('slavem', 'localhost'),
        )

    def run(self):
        self.dbconnect()
        self.setQryEnabled(True)
        self.connect()
        self.queryForever.start()
        self.saveForever.start()

    def dbconnect(self):
        """
        建立数据库链接
        :return:
        """
        self.logger.info('链接MongoDB')
        config = self.config
        self.db = self.clinet[config.get('mongodb', 'dbn')]
        assert isinstance(self.db, pymongo.database.Database)
        self.db.authenticate(
            config.get('mongodb', 'username'),
            config.get('mongodb', 'password')
        )

        # 检查链接是否成功
        self.clinet.server_info()

        self.balanceMinCol = self.db['ctp_balance_min']  # 权益分钟线
        self.balanceDailyCol = self.db['ctp_balance_daily']  # 收盘权益
        self.transferSerialCol = self.db['transfer_serial']  # 转账记录

        self.initContractCollection()

    def initContractCollection(self):
        # 需要建立的索引
        indexUserID = IndexModel([('userID', ASCENDING)], name='userID', background=True)
        indexTradingDay = IndexModel([('tradingDay', DESCENDING)], name='tradingDay', background=True)
        indexes = [indexUserID, indexTradingDay]

        self._initCollectionIndex(self.balanceMinCol, indexes)
        self._initCollectionIndex(self.balanceDailyCol, indexes)

        indexUserID = IndexModel([('accountID', ASCENDING)], name='accountID', background=True)
        indexTradingDay = IndexModel([('tradingDay', DESCENDING)], name='tradingDay', background=True)
        indexFutureSerial = IndexModel([('futureSerial', DESCENDING)], name='futureSerial', background=True)
        indexes = [indexUserID, indexTradingDay, indexFutureSerial]
        self._initCollectionIndex(self.transferSerialCol, indexes)

    def _initCollectionIndex(self, col, indexes):
        """
        初始化分钟线的 collection
        :return:
        """

        # 检查索引
        try:
            indexInformation = col.index_information()
            for indexModel in indexes:
                if indexModel.document['name'] not in indexInformation:
                    col.create_indexes(
                        [
                            indexModel,
                        ],
                    )
        except OperationFailure:
            # 有索引
            col.create_indexes(indexes)

    # ----------------------------------------------------------------------
    def connect(self):
        """连接"""

        # 如果json文件提供了验证码
        if self.authCode:
            self.tdApi.requireAuthentication = True

        # 创建行情和交易接口对象
        self.mdApi.connect(self.userID, self.password, self.brokerID, self.mdAddress)
        self.tdApi.connect(self.userID, self.password, self.brokerID, self.tdAddress, self.authCode, self.appID)
        self.logger.info('链接成功')

        # 初始化并启动查询
        self.initQuery()

    # ----------------------------------------------------------------------
    def subscribe(self, subscribeReq):
        """订阅行情"""
        self.mdApi.subscribe(subscribeReq)

    # ----------------------------------------------------------------------
    def sendOrder(self, orderReq):
        """发单"""
        return self.tdApi.sendOrder(orderReq)

    # ----------------------------------------------------------------------
    def cancelOrder(self, cancelOrderReq):
        """撤单"""
        self.tdApi.cancelOrder(cancelOrderReq)

    # ----------------------------------------------------------------------
    def qryAccount(self):
        """查询账户资金"""
        self.tdApi.qryAccount()
        self.setQryAccountTimer()

    def onAccount(self, account):
        # 时间到了就进行保存
        self.setQryAccountTimer()
        if not self.saveAccountTimer:
            # 还没有计时器
            self.setSaveAccountTimer()
        elif self.saveAccountTimer.isAlive():
            # 已经有计时器了
            pass
        else:
            # 已经有计时器了，但是已经触发了
            self.setSaveAccountTimer()

        # 收到了响应，汇报启动完成
        self.slavemReport.lanuchReport()

        return super(CtpGateway, self).onAccount(account)

    def setSaveAccountTimer(self):
        # 设置1分钟后存库
        def foo():
            document = self.account.__dict__.copy()
            now = arrow.now().replace(second=0, microsecond=0).datetime
            document['datetime'] = now
            is_tradingtime, tradeday = tt.get_tradingday(now)
            document['tradingDay'] = tradeday

            documentDaily = document.copy()

            # 分钟净值
            vtSave = VtSaveMongodbData(self.balanceMinCol.insert_one, (document,))
            self.saveQueue.put(vtSave)

            # 日线净值
            _filter = {
                'accountID': documentDaily['accountID'],
                'tradingDay': tradeday,
            }
            # 日净值
            vtSave = VtSaveMongodbData(self.balanceDailyCol.update_one, (_filter, {'$set': documentDaily},),
                                       {'upsert': True})
            self.saveQueue.put(vtSave)

            # 心跳
            self.slavemReport.heartBeat()

        moment = now = dt.datetime.now()
        wait = 0
        for wait in range(60):
            # 等到整分钟的时候才存库
            moment += dt.timedelta(seconds=1)
            if moment.minute != now.minute:
                break
        self.saveAccountTimer = t = Timer(wait, foo)
        t.daemon = True
        t.start()

    def setQryAccountTimer(self):
        def foo():
            q = VtQuery(self.qryAccount)
            self.qryQueue.put(q)

        if self.qryAccountTimer:
            self.qryAccountTimer.cancel()
        self.qryAccountTimer = t = Timer(10, foo)
        t.daemon = True
        t.start()

    # ----------------------------------------------------------------------
    def qryAccountregister(self):
        """查询签约银行"""
        self.tdApi.qryAccountregister()

    def onAccountRegister(self, bankAccount):
        return super(CtpGateway, self).onAccountRegister(bankAccount)

    # ----------------------------------------------------------------------
    def qryPosition(self):
        """查询持仓"""
        self.tdApi.qryPosition()

    def qryTransferSerial(self):
        """
        查询入金
        :return:
        """
        self.tdApi.qryTransferSerial()

        # 设置定时查询
        self.setQryTransferSerialTimer()

    def onTransferSerial(self, transferSerial):
        """
        响应转账流水
        :return:
        """
        r = super(CtpGateway, self).onTransferSerial(transferSerial)

        # 记录转账流水
        _id = transferSerial.futureSerial
        if _id not in self.transferSerials:
            self.transferSerials[_id] = transferSerial
            _filter = {
                'futureSerial': transferSerial.futureSerial,
                'tradingDay': transferSerial.tradingDay,
            }
            document = {'$set': transferSerial.__dict__.copy()}

            s = VtSaveMongodbData(
                self.transferSerialCol.update_one,
                (_filter, document,),
                {'upsert': True}
            )
            self.saveQueue.put(s)

        self.setQryTransferSerialTimer()

        return r

    def setQryTransferSerialTimer(self):
        # 每分钟查询一次
        def foo():
            try:
                q = VtQuery(self.qryTransferSerial)
                self.qryQueue.put(q)
            except Exception:
                self.logger.error(traceback.format_exc())

        if self.qryTransferSerialTimer:
            self.qryTransferSerialTimer.cancel()
        self.qryTransferSerialTimer = t = Timer(60, foo)
        t.daemon = True
        t.start()

    # ----------------------------------------------------------------------
    def close(self):
        """关闭"""
        self.stoped.set()

        for v in list(self.__dict__.values()):
            # 取消定时器
            if v.__class__ == Timer:
                v.cancel()

        for v in list(self.__dict__.values()):
            # 等待永驻循环结束
            if v.__class__ == Thread:
                if v.isAlive():
                    v.join(timeout=3)

        if self.mdConnected:
            self.mdApi.close()
        if self.tdConnected:
            self.tdApi.close()

        # 停止心跳
        self.slavemReport.endHeartBeat()
        self.logger.info('关闭心跳')

    # ----------------------------------------------------------------------
    def initQuery(self):
        """初始化连续查询"""

        queries = []

        # 签约银行
        q = VtQuery(self.qryAccountregister)
        queries.append(q)

        # 查询出入金
        q = VtQuery(self.qryTransferSerial)
        queries.append(q)

        q = VtQuery(self.qryAccount)
        queries.append(q)

        # self.qryPosition,  # TODO 查询持仓

        for q in queries:
            self.qryQueue.put(q)

    # ----------------------------------------------------------------------
    def query(self):
        """注册到事件处理引擎上的查询函数"""

        while not self.stoped.wait(3):
            # 查询间隔为3秒
            try:
                vtQuery = self.qryQueue.get_nowait()
                assert isinstance(vtQuery, VtQuery)
            except Empty:
                continue

            try:
                vtQuery.func(*vtQuery.args, **vtQuery.kwargs)
            except Exception:
                self.logger.error(traceback.format_exc())

    # ----------------------------------------------------------------------
    def setQryEnabled(self, qryEnabled):
        """设置是否要启动循环查询"""
        self.qryEnabled = qryEnabled

    def _save(self):
        """
        将数据存库
        :return:
        """

        def foo():
            vtSave = self.saveQueue.get(timeout=1)

            try:
                vtSave.func(*vtSave.args, **vtSave.kwargs)
            except Exception:
                self.logger.error(traceback.format_exc())

        while not self.stoped.wait(1):
            try:
                foo()
            except Empty:
                continue

        while True:
            try:
                foo()
            except Empty:
                break

# encoding: UTF-8

'''
vn.ctp的gateway接入

考虑到现阶段大部分CTP中的ExchangeID字段返回的都是空值
vtSymbol直接使用symbol
'''
import traceback
import pymongo.database
from pymongo import IndexModel, DESCENDING, ASCENDING
from pymongo.errors import OperationFailure
from threading import Event, Thread, Timer

import arrow
import tradingtime as tt

try:
    from queue import Queue, Empty
except ImportError:
    from queue import Queue, Empty
from slavem import Reporter

from slaveaccount.gateway import Gateway
from slaveaccount.constant import *
from slaveaccount.ctpmdapi import CtpMdApi
from slaveaccount.ctptdapi import CtpTdApi
from slaveaccount.object import VtQuery, VtSaveMongodbData


########################################################################
class CtpGateway(Gateway):
    """CTP接口"""

    # ----------------------------------------------------------------------
    def __init__(self, config, gatewayName, userID, client):
        """Constructor"""
        super(CtpGateway, self).__init__(config, gatewayName, userID, client)

        self.confg = config
        self.userID = userID
        self.password = config.get(userID, 'password')
        self.brokerID = config.get(userID, 'brokerID')
        self.tdAddress = config.get(userID, 'tdAddress')
        self.mdAddress = config.get(userID, 'mdAddress')
        self.db = None
        self.authCode = config.get(userID, 'authCode')
        self.appID = config.get(userID, 'appID')

        self.mdApi = CtpMdApi(self)  # 行情API
        self.tdApi = CtpTdApi(self)  # 交易API

        self.mdConnected = False  # 行情API连接状态，登录完成后为True
        self.tdConnected = False  # 交易API连接状态

        # 交易流水
        self.transferSerials = {}

        # 查询定时器
        self.qryTransferSerialTimer = None
        self.qryAccountTimer = None

        # 存库定时器
        self.saveAccountTimer = None

        # 查询队列
        self.qryQueue = Queue()  # VtQuery

        self.stoped = Event()
        self.queryForever = Thread(name='query', target=self.query)

        self.saveQueue = Queue()  # VtSaveMongodbData
        self.saveForever = Thread(name='save', target=self._save)

        self.slavemReport = Reporter(
            name=config.get('slavem', 'name'),
            type=config.get('slavem', 'type'),
            host=config.get('slavem', 'host'),
            port=config.getint('slavem', 'port'),
            dbn=config.get('slavem', 'dbn'),
            username=config.get('slavem', 'username'),
            password=config.get('slavem', 'password'),
            localhost=config.get('slavem', 'localhost'),
        )

    def run(self):
        self.dbconnect()
        self.setQryEnabled(True)
        self.connect()
        self.queryForever.start()
        self.saveForever.start()

    def dbconnect(self):
        """
        建立数据库链接
        :return:
        """
        self.logger.info('链接MongoDB')
        config = self.config
        self.db = self.clinet[config.get('mongodb', 'dbn')]
        assert isinstance(self.db, pymongo.database.Database)
        self.db.authenticate(
            config.get('mongodb', 'username'),
            config.get('mongodb', 'password')
        )

        # 检查链接是否成功
        self.clinet.server_info()

        self.balanceMinCol = self.db['ctp_balance_min']  # 权益分钟线
        self.balanceDailyCol = self.db['ctp_balance_daily']  # 收盘权益
        self.transferSerialCol = self.db['transfer_serial']  # 转账记录

        self.initContractCollection()

    def initContractCollection(self):
        # 需要建立的索引
        indexUserID = IndexModel([('userID', ASCENDING)], name='userID', background=True)
        indexTradingDay = IndexModel([('tradingDay', DESCENDING)], name='tradingDay', background=True)
        indexes = [indexUserID, indexTradingDay]

        self._initCollectionIndex(self.balanceMinCol, indexes)
        self._initCollectionIndex(self.balanceDailyCol, indexes)

        indexUserID = IndexModel([('accountID', ASCENDING)], name='accountID', background=True)
        indexTradingDay = IndexModel([('tradingDay', DESCENDING)], name='tradingDay', background=True)
        indexFutureSerial = IndexModel([('futureSerial', DESCENDING)], name='futureSerial', background=True)
        indexes = [indexUserID, indexTradingDay, indexFutureSerial]
        self._initCollectionIndex(self.transferSerialCol, indexes)

    def _initCollectionIndex(self, col, indexes):
        """
        初始化分钟线的 collection
        :return:
        """

        # 检查索引
        try:
            indexInformation = col.index_information()
            for indexModel in indexes:
                if indexModel.document['name'] not in indexInformation:
                    col.create_indexes(
                        [
                            indexModel,
                        ],
                    )
        except OperationFailure:
            # 有索引
            col.create_indexes(indexes)

    # ----------------------------------------------------------------------
    def connect(self):
        """连接"""

        # 如果json文件提供了验证码
        if self.authCode:
            self.tdApi.requireAuthentication = True

        # 创建行情和交易接口对象
        self.mdApi.connect(self.userID, self.password, self.brokerID, self.mdAddress)
        self.tdApi.connect(self.userID, self.password, self.brokerID, self.tdAddress, self.authCode, self.appID)
        self.logger.info('链接成功')

        # 初始化并启动查询
        self.initQuery()

    # ----------------------------------------------------------------------
    def subscribe(self, subscribeReq):
        """订阅行情"""
        self.mdApi.subscribe(subscribeReq)

    # ----------------------------------------------------------------------
    def sendOrder(self, orderReq):
        """发单"""
        return self.tdApi.sendOrder(orderReq)

    # ----------------------------------------------------------------------
    def cancelOrder(self, cancelOrderReq):
        """撤单"""
        self.tdApi.cancelOrder(cancelOrderReq)

    # ----------------------------------------------------------------------
    def qryAccount(self):
        """查询账户资金"""
        self.tdApi.qryAccount()
        self.setQryAccountTimer()

    def onAccount(self, account):
        # 时间到了就进行保存
        self.setQryAccountTimer()
        if not self.saveAccountTimer:
            # 还没有计时器
            self.setSaveAccountTimer()
        elif self.saveAccountTimer.isAlive():
            # 已经有计时器了
            pass
        else:
            # 已经有计时器了，但是已经触发了
            self.setSaveAccountTimer()

        # 收到了响应，汇报启动完成
        self.slavemReport.lanuchReport()

        return super(CtpGateway, self).onAccount(account)

    def setSaveAccountTimer(self):
        # 设置1分钟后存库
        def foo():
            document = self.account.__dict__.copy()
            now = arrow.now().replace(second=0, microsecond=0).datetime
            document['datetime'] = now
            is_tradingtime, tradeday = tt.get_tradingday(now)
            document['tradingDay'] = tradeday

            documentDaily = document.copy()

            # 分钟净值
            vtSave = VtSaveMongodbData(self.balanceMinCol.insert_one, (document,))
            self.saveQueue.put(vtSave)

            # 日线净值
            _filter = {
                'accountID': documentDaily['accountID'],
                'tradingDay': tradeday,
            }
            # 日净值
            vtSave = VtSaveMongodbData(self.balanceDailyCol.update_one, (_filter, {'$set': documentDaily},),
                                       {'upsert': True})
            self.saveQueue.put(vtSave)

            # 心跳
            self.slavemReport.heartBeat()

        moment = now = dt.datetime.now()
        wait = 0
        for wait in range(60):
            # 等到整分钟的时候才存库
            moment += dt.timedelta(seconds=1)
            if moment.minute != now.minute:
                break
        self.saveAccountTimer = t = Timer(wait, foo)
        t.daemon = True
        t.start()

    def setQryAccountTimer(self):
        def foo():
            q = VtQuery(self.qryAccount)
            self.qryQueue.put(q)

        if self.qryAccountTimer:
            self.qryAccountTimer.cancel()
        self.qryAccountTimer = t = Timer(10, foo)
        t.daemon = True
        t.start()

    # ----------------------------------------------------------------------
    def qryAccountregister(self):
        """查询签约银行"""
        self.tdApi.qryAccountregister()

    def onAccountRegister(self, bankAccount):
        return super(CtpGateway, self).onAccountRegister(bankAccount)

    # ----------------------------------------------------------------------
    def qryPosition(self):
        """查询持仓"""
        self.tdApi.qryPosition()

    def qryTransferSerial(self):
        """
        查询入金
        :return:
        """
        self.tdApi.qryTransferSerial()

        # 设置定时查询
        self.setQryTransferSerialTimer()

    def onTransferSerial(self, transferSerial):
        """
        响应转账流水
        :return:
        """
        r = super(CtpGateway, self).onTransferSerial(transferSerial)

        # 记录转账流水
        _id = transferSerial.futureSerial
        if _id not in self.transferSerials:
            self.transferSerials[_id] = transferSerial
            _filter = {
                'futureSerial': transferSerial.futureSerial,
                'tradingDay': transferSerial.tradingDay,
            }
            document = {'$set': transferSerial.__dict__.copy()}

            s = VtSaveMongodbData(
                self.transferSerialCol.update_one,
                (_filter, document,),
                {'upsert': True}
            )
            self.saveQueue.put(s)

        self.setQryTransferSerialTimer()

        return r

    def setQryTransferSerialTimer(self):
        # 每分钟查询一次
        def foo():
            try:
                q = VtQuery(self.qryTransferSerial)
                self.qryQueue.put(q)
            except Exception:
                self.logger.error(traceback.format_exc())

        if self.qryTransferSerialTimer:
            self.qryTransferSerialTimer.cancel()
        self.qryTransferSerialTimer = t = Timer(60, foo)
        t.daemon = True
        t.start()

    # ----------------------------------------------------------------------
    def close(self):
        """关闭"""
        self.stoped.set()

        for v in list(self.__dict__.values()):
            # 取消定时器
            if v.__class__ == Timer:
                v.cancel()

        for v in list(self.__dict__.values()):
            # 等待永驻循环结束
            if v.__class__ == Thread:
                if v.isAlive():
                    v.join(timeout=3)

        if self.mdConnected:
            self.mdApi.close()
        if self.tdConnected:
            self.tdApi.close()

        # 停止心跳
        self.slavemReport.endHeartBeat()
        self.logger.info('关闭心跳')

    # ----------------------------------------------------------------------
    def initQuery(self):
        """初始化连续查询"""

        queries = []

        # 签约银行
        q = VtQuery(self.qryAccountregister)
        queries.append(q)

        # 查询出入金
        q = VtQuery(self.qryTransferSerial)
        queries.append(q)

        q = VtQuery(self.qryAccount)
        queries.append(q)

        # self.qryPosition,  # TODO 查询持仓

        for q in queries:
            self.qryQueue.put(q)

    # ----------------------------------------------------------------------
    def query(self):
        """注册到事件处理引擎上的查询函数"""

        while not self.stoped.wait(3):
            # 查询间隔为3秒
            try:
                vtQuery = self.qryQueue.get_nowait()
                assert isinstance(vtQuery, VtQuery)
            except Empty:
                continue

            try:
                vtQuery.func(*vtQuery.args, **vtQuery.kwargs)
            except Exception:
                self.logger.error(traceback.format_exc())

    # ----------------------------------------------------------------------
    def setQryEnabled(self, qryEnabled):
        """设置是否要启动循环查询"""
        self.qryEnabled = qryEnabled

    def _save(self):
        """
        将数据存库
        :return:
        """

        def foo():
            vtSave = self.saveQueue.get(timeout=1)

            try:
                vtSave.func(*vtSave.args, **vtSave.kwargs)
            except Exception:
                self.logger.error(traceback.format_exc())

        while not self.stoped.wait(1):
            try:
                foo()
            except Empty:
                continue

        while True:
            try:
                foo()
            except Empty:
                break
# encoding: UTF-8 vn.ctp的gateway接入  考虑到现阶段大部分CTP中的ExchangeID字段返回的都是空值 vtSymbol直接使用symbol ######################################################################## CTP接口 # ---------------------------------------------------------------------- Constructor # 行情API # 交易API # 行情API连接状态，登录完成后为True # 交易API连接状态 # 交易流水 # 查询定时器 # 存库定时器 # 查询队列 # VtQuery # VtSaveMongodbData 建立数据库链接         :return: # 检查链接是否成功 # 权益分钟线 # 收盘权益 # 转账记录 # 需要建立的索引 初始化分钟线的 collection         :return: # 检查索引 # 有索引 # ---------------------------------------------------------------------- 连接 # 如果json文件提供了验证码 # 创建行情和交易接口对象 # 初始化并启动查询 # ---------------------------------------------------------------------- 订阅行情 # ---------------------------------------------------------------------- 发单 # ---------------------------------------------------------------------- 撤单 # ---------------------------------------------------------------------- 查询账户资金 # 时间到了就进行保存 # 还没有计时器 # 已经有计时器了 # 已经有计时器了，但是已经触发了 # 收到了响应，汇报启动完成 # 设置1分钟后存库 # 分钟净值 # 日线净值 # 日净值 # 心跳 # 等到整分钟的时候才存库 # ---------------------------------------------------------------------- 查询签约银行 # ---------------------------------------------------------------------- 查询持仓 查询入金         :return: # 设置定时查询 响应转账流水         :return: # 记录转账流水 # 每分钟查询一次 # ---------------------------------------------------------------------- 关闭 # 取消定时器 # 等待永驻循环结束 # 停止心跳 # ---------------------------------------------------------------------- 初始化连续查询 # 签约银行 # 查询出入金 # self.qryPosition,  # TODO 查询持仓 # ---------------------------------------------------------------------- 注册到事件处理引擎上的查询函数 # 查询间隔为3秒 # ---------------------------------------------------------------------- 设置是否要启动循环查询 将数据存库         :return: