vnpy/examples/Services/runServer.py

# encoding: utf-8

# 该文件，为无界面启动文件，以vtServer为容器，加载MainEngine
# 配置：
#  self.gateway_name  ，gateway 的连接名称，在vtEngine.initGateway()里面定义，对应的配置文件是 "连接名称_connect.json"，
#  self.strategies：启动的策略实例，须在catStrategy/CtaSetting.json 里面定义  [u'S28_RB1001', u'S28_TFT', u'S28_HCRB',u'atr_rsi']
#  vtServer的ZMQ端口： 从VT_Setting.json中配置，根据AUTO_CONNCET_GW找到对应得端口配置

import os
import sys

import ctypes
from datetime import datetime, timedelta, date
from time import sleep
from threading import Thread

# 将repostory的目录i，作为根目录，添加到系统环境中。
ROOT_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
sys.path.append(ROOT_PATH)

from vnpy.trader.vtEvent import *
from vnpy.rpc import RpcServer
from vnpy.trader.vtEngine import MainEngine
from vnpy.trader.gateway import ctpGateway
from vnpy.trader.setup_logger import setup_logger,get_logger
from vnpy.trader.util_monitor import *
from vnpy.trader.vtGlobal import globalSetting
from vnpy.trader.util_gpid import *
from vnpy.trader.app import ctaStrategy,riskManager

AUTO_CONNCET_GW = 'CTP'
########################################################################
class VtServer(RpcServer):
    """vn.trader 无界面服务器"""

    # ----------------------------------------------------------------------
    def __init__(self, repAddress, pubAddress):
        """Constructor"""
        super(VtServer, self).__init__(repAddress, pubAddress)
        #self.usePickle()

        # gateway 是否连接
        self.connected = False
        # gateway 的连接名称，在vtEngine.initGateway()里面定义，对应的配置文件是 "连接名称_connect.json"，
        self.gateway_name = AUTO_CONNCET_GW
        # 启动的策略实例，须在catStrategy/CtaSetting.json 里面定义  [u'S28_RB1001', u'S28_TFT', u'S28_HCRB',u'atr_rsi']
        self.strategies = [u'S30_RB0510', u'S30_HCRB05']
        self.g_count = 0
        self.disconnect_signal = 0
        self.last_dt = datetime.now()

        # 创建事件引擎
        ee = EventEngine2()

        # 创建主引擎对象
        print( u'instance mainengine')
        self.engine = MainEngine(ee)

        # 添加CTP Gateway,配置文件为 CTP_Post
        self.engine.addGateway(ctpGateway, self.gateway_name)

        # 添加应用
        self.engine.addApp(ctaStrategy)
        self.engine.addApp(riskManager)


        # 注册主引擎的方法到服务器的RPC函数
        self.register(self.engine.connect)
        self.register(self.engine.disconnect)
        self.register(self.engine.subscribe)
        self.register(self.engine.sendOrder)
        self.register(self.engine.cancelOrder)
        self.register(self.engine.qryAccount)
        self.register(self.engine.qryPosition)
        self.register(self.engine.checkGatewayStatus)  # 检测gateway的连接状态
        self.register(self.engine.qryStatus)  # 检测ctaEngine的状态
        self.register(self.engine.exit)
        self.register(self.engine.writeLog)
        self.register(self.engine.dbConnect)
        self.register(self.engine.dbInsert)
        self.register(self.engine.dbQuery)
        self.register(self.engine.dbUpdate)
        self.register(self.engine.getContract)
        self.register(self.engine.getAllContracts)
        self.register(self.engine.getOrder)
        self.register(self.engine.getAllWorkingOrders)
        self.register(self.engine.getAllGatewayNames)
        self.register(self.engine.saveData)
        self.register(self.engine.initStrategy)
        self.register(self.engine.startStrategy)
        self.register(self.engine.stopStrategy)

        # 注册事件引擎发送的事件处理监听
        self.engine.eventEngine.registerGeneralHandler(self.eventHandler)


    def trade_off(self):
        """检查现在是否为非交易时间"""
        now = datetime.now()
        a = datetime.now().replace(hour=2, minute=35, second=0, microsecond=0)
        b = datetime.now().replace(hour=8, minute=55, second=0, microsecond=0)
        c = datetime.now().replace(hour=15, minute=30, second=0, microsecond=0)
        d = datetime.now().replace(hour=20, minute=55, second=0, microsecond=0)
        weekend = (now.isoweekday() == 6 and now >= a) or (now.isoweekday() == 7) or (now.isoweekday() == 1 and now <=b)
        off = (a <= now <= b) or (c <= now <= d) or weekend
        return off

    def disconnect(self):
        """"断开底层gateway的连接"""
        if self.engine:
            self.engine.disconnect(self.gateway_name)
            self.connected = False

    def onTimer(self, event):
        """定时器执行逻辑，每十秒执行一次"""

        # 十秒才执行一次检查
        self.g_count += 1
        if self.g_count <= 30:
            return
        self.g_count = 0
        dt = datetime.now()
        self.engine.qryStatus()
        if dt.hour != self.last_dt.hour:
            self.last_dt = dt
            print(u'noUiMain.py checkpoint:{0}'.format(dt))
            self.engine.writeLog( u'noUiMain.py checkpoint:{0}'.format(dt))

        # 定时断开
        if self.trade_off():
            """非交易时间"""
            if self.connected:
                self.engine.writeLog(u'断开连接{0}'.format(self.gateway_name))
                self.disconnect()
                self.engine.writeLog(u'清空数据引擎')
                self.engine.clearData()
                self.connected = False
                self.engine.writeNotification(u'非交易时间{0}，断开连接{1}'.format(dt, self.gateway_name))
            return

        # 交易时间内，定时重连和检查
        if not self.connected:
            self.engine.writeLog(u'启动连接{0}'.format(self.gateway_name))
            self.engine.writeLog(u'清空数据引擎')
            self.engine.clearData()
            self.engine.writeLog(u'重新连接{0}'.format(self.gateway_name))
            self.engine.connect(self.gateway_name)
            self.connected = True
            self.disconnect_signal = 0
            self.engine.writeNotification(u'{0}，重新连接{1}'.format(dt, self.gateway_name))
            return
        else:
            if not self.engine.checkGatewayStatus(self.gateway_name):
                self.disconnect_signal += 1

                if self.disconnect_signal >= 5:
                    self.engine.writeWarning(u'检查连接{0}异常，超过{1}次'.format(self.gateway_name,self.disconnect_signal))
                    sys.exit(0)
            else:
                self.disconnect_signal = 0

    def start(self):
        """启动"""
        super(VtServer, self).start()

        # 若需要连接数据库，则启动
        # self.mainEngine.dbConnect()

        # 加载cta的配置
        print( u'load cta setting')
        self.engine.ctaEngine.loadSetting()

        print(u'initialize all strategies')
        # 初始化策略，如果多个，则需要逐一初始化多个
        for s in self.strategies:
            print( 'init trategy {0}'.format(s))
            self.engine.ctaEngine.initStrategy(s)
            # 逐一启动策略
            print( 'start strategy {0}'.format(s))
            self.engine.ctaEngine.startStrategy(s)

        # 指定的连接配置
        if not self.trade_off():
            print( u'connect gateway:{0}'.format(self.gateway_name))
            self.engine.connect(self.gateway_name)
            self.connected = True

        # 注册定时器，用于判断重连
        self.engine.eventEngine.register(EVENT_TIMER, self.onTimer)

        # 所有的日志监控
        self.logM = LogMonitor(self.engine.eventEngine)
        self.errorM = ErrorMonitor(self.engine.eventEngine)
        self.tradeM = TradeMonitor(self.engine.eventEngine)
        self.orderM = OrderMonitor(self.engine.eventEngine, self.engine)
        self.positionM = PositionMonitor(self.engine.eventEngine)
        self.accountM = AccountMonitor(self.engine.eventEngine)

        self.engine.writeNotification(u'{0}，服务启动{1}'.format(datetime.now(),self. gateway_name))

    # ----------------------------------------------------------------------
    def eventHandler(self, event):
        """事件处理"""
        try:
            # 调用RpcServer.publish()
            if isinstance(event.type_, str):
                self.publish(event.type_, event)
            else:
                self.publish(event.type_.encode('utf-8'), event)

        except Exception as ex:
            print( u'event Handler exception:{0}'.format(str(ex)))

    # ----------------------------------------------------------------------
    def stopServer(self):
        """停止服务器"""
        print( 'stopServer')
        # 关闭引擎
        self.engine.exit()
        # 停止服务器线程
        self.stop()

# ----------------------------------------------------------------------
def printLog(content):
    """打印日志"""
    print( datetime.now().strftime("%H:%M:%S"), '\t', content)

# ----------------------------------------------------------------------
def runServer():
    """运行服务器"""

    try:
        log_file_name = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                                 'logs', u'noUiMain.log'))
    except Exception as ex:
        print( u'Use local dict:{0}'.format(os.getcwd()))
        log_file_name = os.path.abspath(os.path.join(os.getcwd(), 'logs', u'noUiMain.log'))

    setup_logger(filename=log_file_name, debug=False)

    # Req/Publish端口
    try:
        zmqAddressDict = globalSetting['ZMQ']
        zmqAddress = zmqAddressDict[AUTO_CONNCET_GW]
        reqAddress = zmqAddress['ReqAddress']
        pubAddress = zmqAddress['PubAddress']
    except:
        reqAddress = 'tcp://*:2014'
        pubAddress = 'tcp://*:2016'

    # 创建并启动服务器
    server = VtServer(reqAddress, pubAddress)
    server.start()

    printLog('-' * 50)
    printLog(u'Request端口:{0}'.format(reqAddress))
    printLog(u'Publish端口:{0}'.format(pubAddress))
    printLog(u'Trader服务器已启动')


if __name__ == '__main__':

    # 主程序
    thread = Thread(target=runServer, args=())
    thread.start()