data-service/clacRestGroupData.py

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

import time
import datetime
from StockRestPojo import StockRestPojo
from dbOperation import dbOperation
from dbOperationStock import dbOperationStock
import db_config
from threading import Thread
from time import sleep
import asyncio
import json

# from CsqDataPojo import CsqDataPojo


async def updateStockInfo(code, stockRestPojo):
    db = dbOperationStock(db_config.db_stock)
    sql = "update stock set `now` = '" + str(stockRestPojo.now) + "', `change` = '" + str(stockRestPojo.change) + "', `pct_change` = '" + str(stockRestPojo.pctchange) + "' where stock_code = '" + str(code) + "'"
    db.update(sql)
    db.close()

async def analysis(dbConfig):
    db = dbOperation(dbConfig)
    sql = "select id id, data_content dataContent from t_csq_data where status = 0 limit 1"
    csqDataPojoList = db.query_list(sql)
    for csqDataPojo in csqDataPojoList:
        db.update("update t_csq_data set status = 1 where id = '" + str(csqDataPojo['id']) + "'")
        restData = eval(csqDataPojo['dataContent'])
        for key,value in restData.items():
            if "None".strip() == str(value[0]).strip():
                print("无效数据跳过")
                break
            else:
                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))

                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_" + str(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)
                if stockRestPojo.now != 0:
                    await asyncio.gather(
                        updateStockInfo(key, stockRestPojo)
                    )
        db.dele("delete from t_csq_data where id = '" + str(csqDataPojo['id']) + "'")
    db.close()

async def main():
    await asyncio.gather(
        analysis(db_config.db_gupiao)
    )

try:
    #定时任务 每3秒钟触发一次
    asyncio.run(main())

except Exception as ee:
    print("error >>>",ee)