import json
|
import time
|
|
import requests
|
|
from utils import tool
|
|
|
class XGBApi:
|
|
@classmethod
|
def __base_request_get(cls, url):
|
response = requests.get(url)
|
return response.text
|
|
@classmethod
|
def get_limit_up_codes_infos(cls):
|
url = f"https://flash-api.xuangubao.cn/api/surge_stock/stocks?date={tool.get_now_date_str('%Y%m%d')}&normal=true&uplimit=true"
|
result = cls.__base_request_get(url)
|
result = json.loads(result)
|
if result["code"] == 20000:
|
items = result["data"]["items"]
|
return items
|
return None
|
|
@classmethod
|
def get_limit_up_reasons(cls):
|
url = f"https://flash-api.xuangubao.cn/api/surge_stock/plates"
|
result = cls.__base_request_get(url)
|
result = json.loads(result)
|
if result["code"] == 20000:
|
items = result["data"]["items"]
|
return items
|
return None
|
|
@classmethod
|
def get_code_blocks(cls, code):
|
symbol = code
|
if code.find('00') or code.find('30'):
|
symbol += ".SZ"
|
elif code.find('60') or code.find('68'):
|
symbol += ".SS"
|
|
url = f"https://flash-api.xuangubao.cn/api/stage2/plates_by_any_stock?symbol={symbol}&fields=core_avg_pcp,plate_name"
|
result = cls.__base_request_get(url)
|
result = json.loads(result)
|
if result["code"] == 20000:
|
data = result["data"]
|
return [data[k]['plate_name'] for k in data]
|
return None
|
|
|
__limit_up_codes_infos_cache = None
|
__limit_up_reasons_dict_cache = {}
|
|
|
def run():
|
global __limit_up_codes_infos_cache
|
global __limit_up_reasons_dict_cache
|
while True:
|
try:
|
results = XGBApi.get_limit_up_codes_infos()
|
if results:
|
__limit_up_codes_infos_cache = (results, time.time())
|
results = XGBApi.get_limit_up_reasons()
|
if results:
|
__limit_up_reasons_dict_cache = {k["name"]: k for k in results}
|
except:
|
pass
|
finally:
|
time.sleep(3)
|
|
|
# 加载涨停代码
|
def __load_limit_up_codes():
|
global __limit_up_codes_infos_cache
|
global __limit_up_reasons_dict_cache
|
if __limit_up_codes_infos_cache is None or time.time() - __limit_up_codes_infos_cache[1] > 10:
|
items = XGBApi.get_limit_up_codes_infos()
|
if items:
|
__limit_up_codes_infos_cache = (items, time.time())
|
items = XGBApi.get_limit_up_reasons()
|
__limit_up_reasons_dict_cache = {k["name"]: k for k in items}
|
|
|
# 获取涨停列表的原因
|
def get_code_limit_up_reasons(code):
|
__load_limit_up_codes()
|
if __limit_up_codes_infos_cache:
|
__limit_up_codes_infos = __limit_up_codes_infos_cache[0]
|
for item in __limit_up_codes_infos:
|
if item[0].split(".")[0] == code:
|
# 涨停原因：(板块,原因,['成分股名称1','成分股名称2'])
|
reasons = [[x["name"], __limit_up_reasons_dict_cache.get(x["name"]).get("description") if x[
|
"name"] in __limit_up_reasons_dict_cache else '']
|
for x in item[8]]
|
for r in reasons:
|
code_names = []
|
for l in __limit_up_codes_infos:
|
if r[0] in [ll["name"] for ll in l[8]]:
|
code_names.append(l[1])
|
r.append(code_names)
|
# 涨停原因
|
return reasons, item[5], XGBApi.get_code_blocks(code), round(item[4] / 100000000, 2)
|
return None
|
|
|
if __name__ == '__main__':
|
print(XGBApi.get_limit_up_codes_infos())
|
