data-service/getRestData.py

# -*- coding:utf-8 -*-

__author__ = 'weijie'

from EmQuantAPI import *
from datetime import timedelta, datetime
import time
import traceback
import json
from StockRestPojo import StockRestPojo
from dbOperation import dbOperation
from threading import Thread
import db_config

print("开始了")
def mainCallback(quantdata):
    """
    mainCallback 是主回调函数，可捕捉如下错误
    在start函数第三个参数位传入，该函数只有一个为c.EmQuantData类型的参数quantdata
    :param quantdata:c.EmQuantData
    :return:
    """
    print ("mainCallback",str(quantdata))
    print("开始了2222")
    #登录掉线或者 登陆数达到上线（即登录被踢下线） 这时所有的服务都会停止
    if str(quantdata.ErrorCode) == "10001011" or str(quantdata.ErrorCode) == "10001009":
        print ("Your account is disconnect. You can force login automatically here if you need.")
    #行情登录验证失败（每次连接行情服务器时需要登录验证）或者行情流量验证失败时，会取消所有订阅，用户需根据具体情况处理
    elif str(quantdata.ErrorCode) == "10001021" or str(quantdata.ErrorCode) == "10001022":
        print ("Your all csq subscribe have stopped.")
    #行情服务器断线自动重连连续6次失败（1分钟左右）不过重连尝试还会继续进行直到成功为止，遇到这种情况需要确认两边的网络状况
    elif str(quantdata.ErrorCode) == "10002009":
        print ("Your all csq subscribe have stopped, reconnect 6 times fail.")
        # 行情订阅遇到一些错误(这些错误会导致重连，错误原因通过日志输出，统一转换成EQERR_QUOTE_RECONNECT在这里通知)，正自动重连并重新订阅,可以做个监控
    elif str(quantdata.ErrorCode) == "10002012":
        print ("csq subscribe break on some error, reconnect and request automatically.")
        # 资讯服务器断线自动重连连续6次失败（1分钟左右）不过重连尝试还会继续进行直到成功为止，遇到这种情况需要确认两边的网络状况
    elif str(quantdata.ErrorCode) == "10002014":
        print ("Your all cnq subscribe have stopped, reconnect 6 times fail.")
    # 资讯订阅遇到一些错误(这些错误会导致重连，错误原因通过日志输出，统一转换成EQERR_INFO_RECONNECT在这里通知)，正自动重连并重新订阅,可以做个监控
    elif str(quantdata.ErrorCode) == "10002013":
        print ("cnq subscribe break on some error, reconnect and request automatically.")
    # 资讯登录验证失败（每次连接资讯服务器时需要登录验证）或者资讯流量验证失败时，会取消所有订阅，用户需根据具体情况处理
    elif str(quantdata.ErrorCode) == "10001024" or str(quantdata.ErrorCode) == "10001025":
        print("Your all cnq subscribe have stopped.")
    else:
        pass

def work(code):
    if(code != ""):
        #实时行情订阅使用范例
        data = c.csq(code, 'date,time,now,high,low,open,preclose,roundlot,change,pctchange,volume,amount,volumeratio,commissionratio,commissiondiff,tradestatus,outvolume,involume,highlimit,lowlimit,speed,averageprice,buyprice1,buyprice2,buyprice3,buyprice4,buyprice5,buyvolume1,buyvolume2,buyvolume3,buyvolume4,buyvolume5,sellprice1,sellprice2,sellprice3,sellprice4,sellprice5,sellvolume1,sellvolume2,sellvolume3,sellvolume4,sellvolume5,closedtime,closedvolume,closedamount','Pushtype=2',csqCallback)
        # print("csq返回" + str(data))
        if(data.ErrorCode != 0):
            print("request csq Error, ", data.ErrorMsg)
        else:
            print("csq输出结果======分隔线======")
            time.sleep(2)
            # text = input("press any key to cancel csq \r\n")
            #取消订阅
            # data = c.csqcancel(data.SerialID)
def startCallback(message):
    print("[EmQuantAPI Python]", message)
    return 1

def csqCallback(quantdata):
    """
    csqCallback 是csq订阅时提供的回调函数模板。该函数只有一个为c.EmQuantData类型的参数quantdata
    :param quantdata:c.EmQuantData
    :return:
    """
    serialID  = quantdata.SerialID
    # 设置线程序列号，标识正在抓取数据中
    db = dbOperation(db_config.db_gupiao)
    restData = quantdata.Data
    # print(restData)
    for key,value in restData.items():
        if "None".strip() == str(value[0]).strip():
            print("无效数据跳过")
            break
        else:
            db.update("update t_stock_base_info set seralid = '"+str(serialID)+"' where code = '" + str(key) + "'")
            stockRestPojo = StockRestPojo()
            stockRestPojo.date = str(value[0])
            stockRestPojo.time = str(value[1])
            stockRestPojo.now = str(value[2])
            stockRestPojo.high = str(value[3])
            stockRestPojo.low = str(value[4])
            stockRestPojo.open = str(value[5])
            stockRestPojo.preclose = str(value[6])
            stockRestPojo.roundlot = str(value[7])
            stockRestPojo.change = str(value[8])
            stockRestPojo.pctchange = str(value[9])
            stockRestPojo.volume = str(value[10])
            stockRestPojo.amount = str(value[11])
            stockRestPojo.volumeratio = str(value[12])
            stockRestPojo.commissionratio = str(value[13])
            stockRestPojo.commissiondiff = str(value[14])
            stockRestPojo.tradestatus = str(value[15])
            stockRestPojo.outvolume = str(value[16])
            stockRestPojo.involume = str(value[17])
            stockRestPojo.highlimit = str(value[18])
            stockRestPojo.lowlimit = str(value[19])
            stockRestPojo.speed = str(value[20])
            stockRestPojo.averageprice = str(value[21])
            stockRestPojo.buyprice1 = str(value[22])
            stockRestPojo.buyprice2 = str(value[23])
            stockRestPojo.buyprice3 = str(value[24])
            stockRestPojo.buyprice4 = str(value[25])
            stockRestPojo.buyprice5 = str(value[26])
            stockRestPojo.buyvolume1 = str(value[27])
            stockRestPojo.buyvolume2 = str(value[28])
            stockRestPojo.buyvolume3 = str(value[29])
            stockRestPojo.buyvolume4 = str(value[30])
            stockRestPojo.buyvolume5 = str(value[31])
            stockRestPojo.sellprice1 = str(value[32])
            stockRestPojo.sellprice2 = str(value[33])
            stockRestPojo.sellprice3 = str(value[34])
            stockRestPojo.sellprice4 = str(value[35])
            stockRestPojo.sellprice5 = str(value[36])
            stockRestPojo.sellvolume1 = str(value[37])
            stockRestPojo.sellvolume2 = str(value[38])
            stockRestPojo.sellvolume3 = str(value[39])
            stockRestPojo.sellvolume4 = str(value[40])
            stockRestPojo.sellvolume5 = str(value[41])
            stockRestPojo.closedtime = str(value[42])
            stockRestPojo.closedvolume = str(value[43])
            stockRestPojo.closedamount = str(value[44])

            datTime = time.mktime(time.strptime(str(value[0]) + str(value[1]), "%Y%m%d%H%M%S"))
            stockRestPojo.realTime = str(int(datTime))
            stockRestPojo.dateOne = "0"
            stockRestPojo.dateFive = "0"
            stockRestPojo.dateFifteen = "0"
            stockRestPojo.dateThirty = "0"
            stockRestPojo.dateSixty = "0"

            min_ = time.localtime(datTime).tm_min # 获取分钟
            sec_ = time.localtime(datTime).tm_sec # 获取秒数

            # 1分钟数据
            if sec_ != 0:
                tempDatTime = datTime + 60 - sec_
                stockRestPojo.dateOne = str(int(tempDatTime))
            else:
                stockRestPojo.dateOne = str(int(datTime))

            # 5分钟数据
            if sec_ != 0 or min_%5 != 0:
                tempDatTime = datTime + (5 * 60) - sec_ - (min_%5 * 60)
                stockRestPojo.dateFive = str(int(tempDatTime))
            else:
                stockRestPojo.dateFive = str(int(datTime))

            # 15分钟数据
            if sec_ != 0 or min_%15 != 0:
                tempDatTime = datTime + (15 * 60) - sec_ - (min_%15 * 60)
                stockRestPojo.dateFifteen = str(int(tempDatTime))
            else:
                stockRestPojo.dateFifteen = str(int(datTime))

            # 30分钟数据
            if sec_ != 0 or min_%30 != 0:
                tempDatTime = datTime + (30 * 60) - sec_ - (min_%30 * 60)
                stockRestPojo.dateThirty = str(int(tempDatTime))
            else:
                stockRestPojo.dateThirty = str(int(datTime))

            # 60分钟数据
            if sec_ != 0 or min_%60 != 0:
                tempDatTime = datTime + (60 * 60) - sec_ - (min_%60 * 60)
                stockRestPojo.dateSixty = str(int(tempDatTime))
            else:
                stockRestPojo.dateSixty = str(int(datTime))

            sql = "insert into data_rt_" + key.replace(".", "_").lower() + "(`realTime`,`date`,`time`,`now`,`high`,`low`,`open`,`preclose`,`roundlot`,`change`,`pctchange`,`volume`,`amount`,`volumeratio`,`commissionratio`,`commissiondiff`,`tradestatus`,`outvolume`,`involume`,`highlimit`,`lowlimit`,`speed`,`averageprice`,`buyprice1`,`buyprice2`,`buyprice3`,`buyprice4`,`buyprice5`,`buyvolume1`,`buyvolume2`,`buyvolume3`,`buyvolume4`,`buyvolume5`,`sellprice1`,`sellprice2`,`sellprice3`,`sellprice4`,`sellprice5`,`sellvolume1`,`sellvolume2`,`sellvolume3`,`sellvolume4`,`sellvolume5`,`closedtime`,`closedvolume`,`closedamount`,`dateOne`,`dateFive`,`dateFifteen`,`dateThirty`,`dateSixty`,`addTime`) values('" + str(stockRestPojo.realTime) + "','" + str(stockRestPojo.date) + "','" + str(stockRestPojo.time) + "','" + str(stockRestPojo.now) + "','" + str(stockRestPojo.high) + "','" + str(stockRestPojo.low) + "','" + str(stockRestPojo.open) + "','" + str(stockRestPojo.preclose) + "','" + str(stockRestPojo.roundlot) + "','" + str(stockRestPojo.change) + "','" + str(stockRestPojo.pctchange) + "','" + str(stockRestPojo.volume) + "','" + str(stockRestPojo.amount) + "','" + str(stockRestPojo.volumeratio) + "','" + str(stockRestPojo.commissionratio) + "','" + str(stockRestPojo.commissiondiff) + "','" + str(stockRestPojo.tradestatus) + "','" + str(stockRestPojo.outvolume) + "','" + str(stockRestPojo.involume) + "','" + str(stockRestPojo.highlimit) + "','" + str(stockRestPojo.lowlimit) + "','" + str(stockRestPojo.speed) + "','" + str(stockRestPojo.averageprice) + "','" + str(stockRestPojo.buyprice1) + "','" + str(stockRestPojo.buyprice2) + "','" + str(stockRestPojo.buyprice3) + "','" + str(stockRestPojo.buyprice4) + "','" + str(stockRestPojo.buyprice5) + "','" + str(stockRestPojo.buyvolume1) + "','" + str(stockRestPojo.buyvolume2) + "','" + str(stockRestPojo.buyvolume3) + "','" + str(stockRestPojo.buyvolume4) + "','" + str(stockRestPojo.buyvolume5) + "','" + str(stockRestPojo.sellprice1) + "','" + str(stockRestPojo.sellprice2) + "','" + str(stockRestPojo.sellprice3) + "','" + str(stockRestPojo.sellprice4) + "','" + str(stockRestPojo.sellprice5) + "','" + str(stockRestPojo.sellvolume1) + "','" + str(stockRestPojo.sellvolume2) + "','" + str(stockRestPojo.sellvolume3) + "','" + str(stockRestPojo.sellvolume4) + "','" + str(stockRestPojo.sellvolume5) + "','" + str(stockRestPojo.closedtime) + "','" + str(stockRestPojo.closedvolume) + "','" + str(stockRestPojo.closedamount) + "','" + str(stockRestPojo.dateOne) + "','" + str(stockRestPojo.dateFive) + "','" + str(stockRestPojo.dateFifteen) + "','" + str(stockRestPojo.dateThirty) + "','" + str(stockRestPojo.dateSixty) +"', now());"
            db.insert(sql)
    db.close()

try:
    #调用登录函数（激活后使用，不需要用户名密码）
    loginResult = c.start("ForceLogin=1", '', mainCallback)
    if(loginResult.ErrorCode != 0):
        print("login in fail")
        exit()
    db = dbOperation(db_config.db_gupiao)
    stockPojoList = db.query_list("select id, code, name, list_date listDate, seralid from t_stock_base_info where is_rest_add = 0 order by id asc")
    # 拼接证券代码
    codesStr = ""
    # 单线程
    i = 1
    for stockPojo in stockPojoList:
        if 0 != i%500 :
            codesStr = codesStr + stockPojo['code'] + ','
        if 0 == i%500 or str(i) == str(len(stockPojoList)):
            if(codesStr != ""):
                #实时行情订阅使用范例
                codesStr = codesStr[:-1]
                data = c.csq(codesStr, 'date,time,now,high,low,open,preclose,roundlot,change,pctchange,volume,amount,volumeratio,commissionratio,commissiondiff,tradestatus,outvolume,involume,highlimit,lowlimit,speed,averageprice,buyprice1,buyprice2,buyprice3,buyprice4,buyprice5,buyvolume1,buyvolume2,buyvolume3,buyvolume4,buyvolume5,sellprice1,sellprice2,sellprice3,sellprice4,sellprice5,sellvolume1,sellvolume2,sellvolume3,sellvolume4,sellvolume5,closedtime,closedvolume,closedamount','Pushtype=2',csqCallback)
            codesStr = ""
        i = i + 1
    db.close()
    text = input("press any key to cancel csq \r\n")
#退出
    data = logoutResult = c.stop()
except Exception as ee:
    print("error >>>",ee)
    traceback.print_exc()
else:
    print("demo end")