import enum
|
import json
|
|
from db import mysql_data_delegate as mysql_data
|
|
|
def parse_kpl_datas(results):
|
start_y = -1
|
end_x = -1
|
index = 0
|
datas = []
|
for result in results:
|
text = result[1]
|
if text.find("股票名称") > -1:
|
start_y = result[0][0][1]
|
|
if text.find("竞价涨幅") > -1:
|
end_x = result[0][0][0]
|
if start_y > 0 and end_x > 0:
|
if result[0][0][0] < end_x and result[0][0][1] > start_y and (result[0][1][0] - result[0][0][0]) > 30:
|
datas.append(text)
|
index += 1
|
datas = datas[:3 * 5]
|
fdatas = []
|
temp = []
|
for data in datas:
|
temp.append(data)
|
if len(temp) == 3:
|
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"
|
JINGXUAN_RANK_OUT = "jingxuan_rank_out"
|
MARKET_STRONG = "market_strong"
|
|
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 __parseLimitUpItemData(data):
|
# (代码, 名称, 首次涨停时间, 最近涨停时间, 几板, 涨停原因, 板块, 实际流通, 主力净额, 涨停原因代码, 涨停原因代码数量)
|
return data[0], data[1], data[4], data[4], data[18], data[5], data[12], data[13], data[8], data[19], data[20]
|
|
|
# 最强风口
|
# (代码,名称,强度,涨幅,热门板块,所有板块)
|
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 parseLimitUpData(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 = __parseLimitUpItemData(d)
|
if pdata:
|
fresult_.append(pdata)
|
return fresult_, data.get("day")
|
|
|
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])
|
|
|
# 获取高位板的数量
|
def get_high_level_count(key):
|
if key.find("连板") >= 0:
|
# 形式如: 3连板
|
return int(key.replace("连板", ""))
|
elif key.find("天") >= 0 and key.find("板") >= 0:
|
# 形式如：5天4板
|
return int(key.split("天")[1].replace("板", ""))
|
# 形式如：首板
|
return 1
|
|
|
# 获取实时涨停列表中的涨停原因
|
def get_current_limit_up_reasons(item) -> set:
|
b = {item[5]}
|
if item[6]:
|
b |= set(item[6].split("、"))
|
return b
|
|
|
# 获取目前的涨停原因
|
def get_current_limit_up_reason(item) -> str:
|
return item[5]
|
|
|
def filter_block(block):
|
if not block:
|
return block
|
return block.replace("概念", "")
|
|
|
if __name__ == "__main__":
|
print(get_high_level_count("首板"))
|
print(get_high_level_count("5天4板"))
|
print(get_high_level_count("2连板"))
|
print(get_high_level_count("4连板"))
|
