bug修复

Administrator

2023-10-30 fb47d36048e94b9a506d5c153e3dd19a01e37df1

 third_data/kpl_util.py

@@ -1,3 +1,9 @@
import enum
import json

from db import mysql_data_delegate as mysql_data


def parse_kpl_datas(results):
    start_y = -1
    end_x = -1
@@ -23,3 +29,240 @@
            fdatas.append((temp[2][:6], temp[1]))
            temp = []
    return fdatas


# 涨停代码：
# (代码,名称,首次涨停时间,最近涨停时间,几板,涨停原因,板块,实际流通,主力净额,涨停原因代码,涨停原因代码数量)
# (0,1,6,25,9,16,11,15,12)
# 竞价代码：
# (代码,名称,涨停委买额,板块,竞价成交额,实际流通)
# (0,1,18,11,22,15)
# 炸板：
# (代码,名称,涨幅,板块,实际流通)
# (0,1,4,11,15)
# 跌停：
# (代码,名称,板块,实际流通)
# (0,1,11,15)
# 曾跌停：
# (代码,名称,涨幅,板块,实际流通)
# (0,1,4,11,15)

DABAN_TYPE_BIDDING = 8
DABAN_TYPE_LIMIT_UP = 1
DABAN_TYPE_OPEN_LIMIT_UP = 2
DABAN_TYPE_LIMIT_DOWN = 3
DABAN_TYPE_EVER_LIMIT_DOWN = 5


class KPLDataType(enum.Enum):
    BIDDING = "biddings"
    LIMIT_UP = "limit_up"
    OPEN_LIMIT_UP = "open_limit_up"
    LIMIT_DOWN = "limit_down"
    EVER_LIMIT_DOWN = "ever_limit_down"
    FENG_KOU = "feng_kou"
    BEST_FENG_KOU = "best_feng_kou"
    FENG_XIANG = "feng_xiang"
    INDUSTRY_RANK = "industry_rank"
    JINGXUAN_RANK = "jingxuan_rank"


def __parseDaBanItemData(data, type):
    if type == DABAN_TYPE_BIDDING:
        return data[0], data[1], data[18], data[11], data[22], data[15]
    elif type == DABAN_TYPE_LIMIT_UP:
        return data[0], data[1], data[6], data[25], data[9], data[16], data[11], data[15], data[12], data[26], data[27]
    elif type == DABAN_TYPE_OPEN_LIMIT_UP:
        return data[0], data[1], data[4], data[11], data[15]
    elif type == DABAN_TYPE_LIMIT_DOWN:
        return data[0], data[1], data[11], data[15]
    elif type == DABAN_TYPE_EVER_LIMIT_DOWN:
        return data[0], data[1], data[4], data[11], data[15]

    return None


# 最强风口
# (代码,名称,强度,涨幅,热门板块,所有板块)
def __parseBestFengKouItemData(data):
    return data[0], data[1], data[2], data[4], data[12], data[10]


# 市场风口
# (代码,名称,涨幅,主力净额,风口概念)
def __parseFengKouItemData(data):
    return data[0], data[1], data[3], data[7], data[11]


# 风向标
# (代码, 名称, 现价, 涨幅, 板块, 300万大单净额, 主力净额, 自由市值)
def __parseFengXiangBiaoItemData(data):
    return data[0], data[1], data[5], data[6], data[4], data[-3], data[13], data[10]


# 行业涨幅
# (代码,名称,主力净额,涨跌幅)
def __parseIndustry_rank(data):
    return data[0], data[1], data[6], data[3]


def parseDaBanData(data, type_):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["list"]
    fresult_ = []
    for d in list_:
        pdata = __parseDaBanItemData(d, type_)
        if pdata:
            fresult_.append(pdata)
    return fresult_


def parseFengKou(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["List"]
    fresult_ = []
    for d in list_:
        pdata = __parseFengKouItemData(d)
        if pdata:
            fresult_.append(pdata)
    return fresult_


def parseBestFengKou(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["List"]
    fresult_ = []
    for d in list_:
        pdata = __parseBestFengKouItemData(d)
        if pdata:
            fresult_.append(pdata)
    return fresult_


def parseFengXiang(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["list"]
    fresult_ = []
    for d in list_:
        pdata = __parseFengXiangBiaoItemData(d)
        if pdata:
            fresult_.append(pdata)
    return fresult_


def parseIndustryRank(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["list"]
    fresult_ = []
    for d in list_:
        pdata = __parseIndustry_rank(d)
        if pdata:
            fresult_.append(pdata)
    return fresult_


# 解析板块代码
def parsePlateCodes(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["list"]
    fresult_ = []
    for d in list_:
        # (代码,名称,现价,涨幅,自由流通,几板，龙几，主力净额,300w净额,机构增仓)
        fresult_.append((d[0], d[1], d[5], d[6], d[10], d[23], d[24], d[13], d[50], d[42]))
    return fresult_


# 解析概念中的板块强度
def parseSonPlat(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["List"]
    fresult_ = []
    for d in list_:
        # (代码,名称,强度)
        fresult_.append((d[0], d[1], d[2]))
    return fresult_


def __money_desc(money):
    if abs(money) > 100000000:
        return f"{round(money / 100000000, 2)}亿"
    else:
        return f"{round(money / 10000, 2)}万"


def parseMarketIndustry(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["list"]
    fresult_ = []
    for d in list_:
        # (代码,名称,涨幅,主力净额)
        fresult_.append((d[0], d[1], d[3], d[6]))
    return fresult_


def parseMarketJingXuan(data):
    if type(data) == str:
        data = json.loads(data)
    if int(data["errcode"]) != 0:
        raise Exception(f"解析数据出错，errcode：{data['errcode']}")
    list_ = data["list"]
    fresult_ = []
    for d in list_:
        # (代码,名称,强度,主力净额)
        fresult_.append((d[0], d[1], d[2], d[6]))
    return fresult_


class KPLPlatManager:
    def save_plat(self, _id, name):
        if not _id:
            return
        mysqldb = mysql_data.Mysqldb()
        key = f"{_id}-{name}"
        results = mysqldb.select_one(f"select * from kpl_plate where _name='{name}'")
        if not results:
            mysqldb.execute(f"insert into kpl_plate(_id,_name,_key) values({_id},'{name}','{key}')")

    def get_plat(self, name):
        mysqldb = mysql_data.Mysqldb()
        results = mysqldb.select_one(f"select * from kpl_plate where _name='{name}'")
        if results:
            return results[0]
        return None

    def get_same_plat_names(self, name):
        mysqldb = mysql_data.Mysqldb()
        plate = self.get_plat(name)
        if not plate:
            return {name}
        results = mysqldb.select_all(f"select _name from kpl_plate where _id='{plate}'")
        return set([r[0] for r in results])

    def get_same_plat_names_by_id(self, id_):
        mysqldb = mysql_data.Mysqldb()
        results = mysqldb.select_all(f"select _name from kpl_plate where _id='{id_}'")
        return set([r[0] for r in results])