""" kpl API数据获取与处理 """ import json import os.path import time import datetime import dask import requests import constant from log_module.log import logger_common, logger_kpl_jingxuan_in, logger_Overall_market_strength_score, \ logger_stock_of_markets_plate, logger_debug from strategy import data_cache from strategy import basic_methods from strategy.kpl_data_manager import KPLMarketsSiftPlateLogManager from trade import middle_api_protocol from utils import hx_qc_value_util, tool # 获取logger实例 logger = logger_common now = time.time() print(f"kpl_api开始运行--{now}") # 竞价 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 def __base_request(url, data, timeout=10): DELEGATE = True if not DELEGATE: headers = { "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "User-Agent": "Dalvik / 2.1.0(Linux;U;Android 6.0.1;MuMu Build/V417IR)" } # proxies={'https': '192.168.3.251:9002'} # 禁止代理，不然会走本地代理 response = requests.post(url, data=data, headers=headers, proxies={"http": None, "https": None}, timeout=timeout) if response.status_code != 200: raise Exception("请求出错") return response.text else: fdata = middle_api_protocol.load_kpl(url, data, timeout) return middle_api_protocol.request(fdata) def daBanList(pidType): data = "Order=1&a=DaBanList&st=100&c=HomeDingPan&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23" \ f"&VerSion=5.8.0.2&Index=0&Is_st=1&PidType={pidType}&apiv=w32&Type=4&FilterMotherboard=0&Filter=0&FilterTIB=0" \ "&FilterGem=0 " result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 市场行情-行业 主力净额由高到低排序 实时获取 def getMarketIndustryRealRankingInfo(orderJingE_DESC=True): data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=20&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=4&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 市场行情-精选 主力净额由高到低排序 实时获取 st=20 请求地址中 st=20 代表获取前20（排序方式为主力净额降序排列） # def getMarketJingXuanRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=10&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=7&" # result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # return result # def getMarketJingXuanRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=3&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=7&" # result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # return result # #市场行情-精选 （排序方式为强度降序排列） st=20 请求地址中 st=20 代表获取前20 def getMarketJingXuanRealRankingInfo(): data = f"Order=1&a=RealRankingInfo&st=20&a=RealRankingInfo&apiv=w35&Type=1&Index=0&c=ZhiShuRanking&VerSion=5.13.0.0&Order=1&PhoneOSNew=1&ZSType=7&DeviceID=d6f20ce9-fa08-31c9-a493-536ebb8e9773&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 获取代码的板块 def getStockIDPlate(code): data = f"a=GetStockIDPlate_New&apiv=w32&c=StockL2Data&StockID={code}&PhoneOSNew=1&UserID=0&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Token=0&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) result = json.loads(result) if int(result["errcode"]) != 0: return None return result["ListJX"] if result["ListJX"] else result["List"] # IDPlate = getStockIDPlate(600550) # print(f"IDPlate===={IDPlate}") # 获取代码的精选板块 # 返回格式：[(板块代码,板块名称,涨幅百分比)] def getCodeJingXuanBlocks(code): data = f"a=GetStockIDPlate&apiv=w32&Type=2&c=StockL2Data&StockID={code}&PhoneOSNew=1&UserID=0&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Token=0&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) result = json.loads(result) return result.get("ListJX") # 获取该概念下的个股代码及其他 st=100 获取前排100只股票 【获取数量】由于这里直接控制强度的数值数量，暂不轻易修改。目标设定为全部，或100 def getCodesByPlate(plate_code): data = f"Order=1&a=ZhiShuStockList_W8&st=100&c=ZhiShuRanking&PhoneOSNew=1&old=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&IsZZ=0&Token=0&Index=0&apiv=w32&Type=6&IsKZZType=0&UserID=0&PlateID={plate_code}&" return __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) # 获取概念中的板块中的子板块 def getSonPlate(plate_code): data = f"a=SonPlate_Info&apiv=w32&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&PlateID={plate_code}&" return __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) # 市场行情-行业 # def getMarketIndustryRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=80&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=4&" # return __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # # # 市场行情-精选 # def getMarketJingXuanRealRankingInfo(orderJingE_DESC=True): # data = f"Order={1 if orderJingE_DESC else 0}&a=RealRankingInfo&st=80&apiv=w32&Type=5&c=ZhiShuRanking&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&VerSion=5.8.0.2&Index=0&ZSType=7&" # return __base_request("https://apphq.longhuvip.com/w1/api/index.php", # data=data) # 获取自由流通市值 def getZYLTAmount(code): data = f"a=GetStockPanKou_Narrow&apiv=w32&c=StockL2Data&VerSion=5.8.0.2&State=1&PhoneOSNew=1&DeviceID=a38adabd-99ef-3116-8bb9-6d893c846e23&StockID={code}&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) result = json.loads(result) if "real" in result: return result["real"].get("actualcirculation_value") return None # 获取涨停列表 def __getLimitUpInfo(pidType, page, pageSize): data = f"Order=0&a=DailyLimitPerformance&st={pageSize}&apiv=w35&Type=4&c=HomeDingPan&PhoneOSNew=1&DeviceID=a38adabb-99ef-3116-8bb9-6d893c846e24&VerSion=5.13.0.0&Index={(page - 1) * pageSize}&PidType={pidType}&" result = __base_request("https://apphq.longhuvip.com/w1/api/index.php", data=data) return result # 整理涨停列表的数据 def getLimitUpInfoNew(): pids = [(1, "首板"), (2, "2连板"), (3, "3连板"), (4, "4连板"), (5, "")] fresults = [] for pid_info in pids: results = [] for i in range(10): # start_time = time.time() result = __getLimitUpInfo(pid_info[0], i + 1, 20) # print("请求用时", time.time() - start_time) result = json.loads(result) datas = result["info"][0] results.extend(datas) # day = result["info"][1] # 连板天数？ if len(datas) < 20: break for r in results: if not r[18] and pid_info[1]: r[18] = pid_info[1] # # 替换掉板块名称 机器人概念/机器人 # for i in range(len(r)): # if type(r[i]) == str: # r[i] = kpl_util.filter_block(r[i]) fresults.extend(results) return json.dumps({"errcode": 0, "list": fresults}) # 获取涨停信息数据 def get_limit_up_info(): # 获取涨停信息列表 limit_up_info = json.loads(getLimitUpInfoNew())['list'] return limit_up_info # 获取市场情绪综合强度【完整】 def changeStatistics(): """ 获取市场强度 :return: """ result = __base_request("https://apphwhq.longhuvip.com/w1/api/index.php", f"a=ChangeStatistics&apiv=w35&c=HomeDingPan&PhoneOSNew=1&UserID=0&DeviceID=d6f20ce9-fa08-31c9-a493-536ebb8e9774&VerSion=5.13.0.0&Token=0&") # data = result.text data = json.loads(result) return data["info"][0] # 获取市场情绪综合强度 def get_market_strong(): """ 获取市场强度 :return: """ result = __base_request("https://apphwhq.longhuvip.com/w1/api/index.php", f"a=DiskReview&apiv=w35&c=HomeDingPan&VerSion=5.13.0.0&PhoneOSNew=1&DeviceID=d6f20ce9-fa08-31c9-a493-536ebb8e9773&") data = json.loads(result) return int(data["info"]["strong"]) # 市场情绪--涨跌统计 # 数据格式： # SJZT:实际涨停 SJDT:实际跌停 SZJS:涨数量 ZT:涨停 DT:跌停 XDJS:跌数量 sign:人气概述 def getMarketFelling(): result = __base_request("https://apphwhq.longhuvip.com/w1/api/index.php", f"a=ZhangFuDetail&apiv=w35&c=HomeDingPan&PhoneOSNew=1&DeviceID=d6f20ce9-fa08-31c9-a493-536ebb8e9774&VerSion=5.13.0.0&") data = json.loads(result) return data["info"] # market_strong = get_market_strong() # print(f"market_strong==={market_strong}") if __name__ == "__main__": MarketFelling = getMarketFelling() print(f"MarketFelling==={MarketFelling}") changeStatistics = changeStatistics() print(f"changeStatistics==={changeStatistics}")