From 48fb7a00951f91bdc707e5dd2d196e5bccb752c3 Mon Sep 17 00:00:00 2001 From: Administrator <admin@example.com> Date: 星期三, 18 六月 2025 18:41:30 +0800 Subject: [PATCH] 异常保护 --- servers/data_server.py | 220 ++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 files changed, 200 insertions(+), 20 deletions(-) diff --git a/servers/data_server.py b/servers/data_server.py index d0055e2..a548267 100644 --- a/servers/data_server.py +++ b/servers/data_server.py @@ -8,21 +8,26 @@ import dask import requests -from code_attribute.gpcode_manager import BlackListCodeManager +import constant +import inited_data +from api import low_suction_data_pusher +from code_attribute.gpcode_manager import BlackListCodeManager, HumanRemoveForbiddenManager from l2.huaxin import huaxin_target_codes_manager from l2.l2_transaction_data_manager import HuaXinBuyOrderManager -from log_module.log import logger_system, logger_debug, logger_kpl_limit_up, logger_request_api +from log_module.log import logger_system, logger_debug, logger_kpl_limit_up, logger_request_api, \ + logger_kpl_market_strong, logger_kpl_new_blocks from third_data.custom_block_in_money_manager import CodeInMoneyManager from third_data.kpl_data_constant import LimitUpCodesBlockRecordManager, LimitUpDataConstant, \ ContainsLimitupCodesBlocksManager from third_data.kpl_limit_up_data_manager import LatestLimitUpBlockManager, CodeLimitUpSequenceManager from third_data.third_blocks_manager import BlockMapManager -from trade.buy_radical import radical_buy_data_manager +from trade.buy_radical import radical_buy_data_manager, new_block_processor from trade.buy_radical.block_special_codes_manager import BlockSpecialCodesManager +from trade.buy_radical.new_block_processor import BeforeBlocksComputer from trade.buy_strategy import OpenLimitUpGoodBlocksBuyStrategy from trade.buy_radical.radical_buy_data_manager import RadicalBuyBlockManager, BeforeSubDealBigOrderManager -from utils import global_util, tool, data_export_util -from code_attribute import gpcode_manager +from utils import global_util, tool, data_export_util, init_data_util +from code_attribute import gpcode_manager, code_nature_analyse from log_module import log_analyse, log_export, async_log_util from l2 import code_price_manager, l2_data_util, transaction_progress from cancel_strategy.s_l_h_cancel_strategy import HourCancelBigNumComputer, LCancelRateManager @@ -61,6 +66,8 @@ __industry_cache_dict = {} __latest_limit_up_codes_set = set() __data_process_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10) + # 鏂伴鏉愯姹� + __new_blocks_codes_request_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=5) # 浠g爜鐨勬定骞� __code_limit_rate_dict = {} @@ -74,6 +81,8 @@ records = LimitUpDataConstant.history_limit_up_datas if not currents: currents = self.__kplDataManager.get_data(KPLDataType.LIMIT_UP) + if currents is None: + currents = [] # 鑾峰彇鍘嗗彶娑ㄥ仠 if not records: KPLLimitUpDataRecordManager.load_total_datas() @@ -470,6 +479,12 @@ # 涓嶈兘涔扮殑鏉垮潡鍒楄〃 results = KPLPlateForbiddenManager().list_all_cache() response_data = json.dumps({"code": 0, "data": list(results)}) + elif url.path == "/kpl/list_deleted_forbidden_plate": + # 鑾峰彇宸茬粡鍒犻櫎鐨勬澘鍧� + results = KPLPlateForbiddenManager().list_all_deleted_cache() + if results: + results -= KPLPlateForbiddenManager().list_all_cache() + response_data = json.dumps({"code": 0, "data": list(results)}) elif url.path == "/kpl/get_plate_codes": # 鑾峰彇娑ㄥ仠鍘熷洜涓嬮潰鐨勪唬鐮� @@ -586,7 +601,15 @@ code_info[0]) != trade_constant.TRADE_STATE_NOT_TRADE: code_info[5] = 1 - response_data = json.dumps({"code": 0, "data": codes_info}) + # 娑ㄥ仠鏁版嵁 + fdatas = {"limit_up_list": codes_info} + # 杈ㄨ瘑搴︾エ + fdatas["speical_codes"] = [(x, gpcode_manager.get_code_name(x)) for x in special_codes] + forbidden_refer_codes = KPLPlateForbiddenManager().get_watch_high_codes_by_block(plate) + if forbidden_refer_codes is None: + forbidden_refer_codes = set() + fdatas["forbidden_refer_codes"] = [(x, gpcode_manager.get_code_name(x)) for x in forbidden_refer_codes] + response_data = json.dumps({"code": 0, "data": fdatas}) elif url.path == "/kpl/get_open_limit_up_count_rank": # 鑾峰彇鐐告澘娆℃暟鎺掕 ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()]) @@ -788,8 +811,9 @@ deals_month = trade_data_manager.AccountMoneyManager().get_deal_count_info(start_date, end_date) cost_month = sum([round(0.1 * x[1], 2) for x in delegates_month]) make_month = 0 - make_month += max(1 * deals_month[0][1], deals_month[0][2] * 1.854 / 10000) + 1 * deals_month[1][ - 1] + 0 * deals_month[2][1] + make_month += max(1 * deals_month[0][1] if deals_month[0][1] else 0, + deals_month[0][2] * 1.854 / 10000 if deals_month[0][2] else 0) + 1 * deals_month[1][ + 1] + 0 * deals_month[2][1] fdata["month_commission"] = round(make_month - cost_month, 2) # 璁$畻褰撴棩鎵嬬画璐硅鎯� delegates = trade_data_manager.AccountMoneyManager().get_delegated_count_info() @@ -918,29 +942,118 @@ deal_big_order_detail_info = None try: # 鑾峰彇鎴愪氦澶у崟锛�(鍙傝�冨ぇ鍗曢噾棰�,宸叉垚浜ゅぇ鍗曢噾棰�,澶у崟瑕佹眰閲戦, 宸叉垚浜ゆ定鍋滀拱閲戦, 宸叉垚浜ゆ定鍋滃崠閲戦) - th_buy = BeforeSubDealBigOrderManager().get_big_order_threshold(code) - th_sell = BeforeSubDealBigOrderManager().get_big_sell_order_threshold(code) + th_temp_buy_info = BeforeSubDealBigOrderManager().get_temp_deal_big_order_threshold_info( + code) + th_buy, th_buy_default = BeforeSubDealBigOrderManager().get_big_order_threshold(code) + try: + th_sell = BeforeSubDealBigOrderManager().get_big_sell_order_threshold(code) + except: + th_sell = 0 + + # (缂哄皯鐨勮祫閲�, 鍑�鎴愪氦閲戦, 瑕佹眰鐨勫ぇ鍗曢噾棰�, 璁$畻寰楀埌鐨勫ぇ鍗曢槇鍊奸噾棰�, 浜轰负璁剧疆鐨勫ぇ鍗�) deal_big_money_info = radical_buy_data_manager.get_total_deal_big_order_info( code, gpcode_manager.get_limit_up_price_as_num(code)) + if deal_big_money_info[1] == 0 and len(codes) == 1: + # 鎬绘垚浜ら噾棰濅负0 + deal_big_money_info = list(deal_big_money_info) + # 娌℃湁璁㈤槄L2浼氬嚭鐜版病鏈夊�肩殑鎯呭喌锛屽鏋滄定鍋滆繃灏辨媺鍙栦箣鍓嶇殑娑ㄥ仠涔�/鍗栧ぇ鍗� + deal_big_orders_result = radical_buy_data_manager.request_deal_big_orders(code) + if deal_big_orders_result: + buy_datas, sell_datas = deal_big_orders_result[0], deal_big_orders_result[1] + limit_up_price = gpcode_manager.get_limit_up_price_as_num(code) + limit_up_price_money_list = [x[0] for x in buy_datas if x[1] == limit_up_price] + if limit_up_price_money_list: + threshold_money = BeforeSubDealBigOrderManager().compute_re_limit_up_big_money_threshold( + limit_up_price_money_list) + else: + threshold_money = 299e4 + logger_debug.info(f"{code}-涓存椂鍥炲皝鍧囧ぇ鍗曪細{threshold_money}") + # 璁剧疆涔板崟闃堝�� + th_buy = threshold_money + buy_money = sum(limit_up_price_money_list) + sell_money = sum([x[0] for x in sell_datas if x[1] == limit_up_price]) + # 娑ㄥ仠涔板噣鎴愪氦閲戦鍑�涔伴 + deal_big_money_info[1] = buy_money - sell_money + # 瑕佹眰鐨勫ぇ鍗曠疮璁¢噾棰� + deal_big_money_info[ + 2] = radical_buy_data_manager.compute_total_deal_big_order_threshold_money(code, + limit_up_price, + threshold_money) + # logger_debug.info(f"{code}-绱澶у崟闃堝�硷細{deal_big_money_info[2]}") + # logger_debug.info(f"{code}闈炶闃呭ぇ鍗曪紝 buy_money-{buy_money}, sell_money-{sell_money}") + + big_money_rate = radical_buy_data_manager.TotalDealBigOrderInfoManager.get_big_order_rate( + code) + if not big_money_rate: + big_money_rate = 0 # 澶у崟鎴愪氦淇℃伅 deal_big_order_info = [ - (output_util.money_desc(th_buy), output_util.money_desc(th_sell)), output_util.money_desc(deal_big_money_info[1]), - output_util.money_desc(deal_big_money_info[2])] + (output_util.money_desc(th_temp_buy_info[0] if th_temp_buy_info else 0), # 涓存椂涔板ぇ鍗曢槇鍊� + output_util.money_desc(th_buy), # 涔板ぇ鍗曢槇鍊� + output_util.money_desc(th_sell), # 鍗栧崟闃堝�� + big_money_rate * 100 # 澶у崟鎴愪氦姣� + ), + # 娑ㄥ仠澶у崟鍑�涔板叆 + output_util.money_desc(deal_big_money_info[1]), + # 绱鎬诲ぇ鍗曢槇鍊� + output_util.money_desc(deal_big_money_info[2]), + # 鍘熺疮璁″ぇ鍗曢槇鍊硷紙闈炰汉涓鸿缃級 + output_util.money_desc(deal_big_money_info[3]), + # 浜轰负璁剧疆鐨勫ぇ鍗� + output_util.money_desc(deal_big_money_info[4]) if deal_big_money_info[4] else '', + ] if len(codes) == 1: # 鍔犺浇澶у崟璇︽儏 deal_big_order_detail_info = radical_buy_data_manager.get_l2_big_order_deal_info(code) # 鍔犺浇娑ㄥ仠澶у崟璇︽儏 - limit_up_big_order_detail = radical_buy_data_manager.get_total_detal_big_order_details(code) - deal_big_order_info.append(output_util.money_desc(limit_up_big_order_detail[0]+limit_up_big_order_detail[1])) + # 涔板ぇ鍗�, 涓婃澘鍓嶄拱澶у崟, 鍗栧ぇ鍗�, 涓婃澘鍓嶅崠澶у崟 + limit_up_big_order_detail = radical_buy_data_manager.get_total_detal_big_order_details( + code) + # 绱娑ㄥ仠涔伴噾棰� + buy_money = output_util.money_desc(limit_up_big_order_detail[0] + limit_up_big_order_detail[1]) + sell_money = output_util.money_desc(limit_up_big_order_detail[2] + limit_up_big_order_detail[3]) + deal_big_order_info.append(buy_money) + # 绱娑ㄥ仠鍗栭噾棰� + deal_big_order_info.append(sell_money) deal_big_order_info.append( - output_util.money_desc(limit_up_big_order_detail[2] + limit_up_big_order_detail[3])) - except: + radical_buy_data_manager.TotalDealBigOrderInfoManager().get_big_order_rate(code)) + + except Exception as e: + logger_debug.error(f"鍙兘娌℃湁鑾峰彇鍒版定鍋滀环锛歿code}") + if not gpcode_manager.get_limit_up_price(code): + init_data_util.re_set_price_pre(code) + logger_debug.exception(e) deal_big_order_info = None code_name = gpcode_manager.get_code_name(code) - fresults.append((code, code_name, deal_big_order_info, deal_big_order_detail_info)) + + big_order_enough = 0 + try: + total_deal_big_order_result = radical_buy_data_manager.get_total_deal_big_order_info(code, + gpcode_manager.get_limit_up_price_as_num( + code)) + if total_deal_big_order_result and total_deal_big_order_result[0]<=0: + big_order_enough = 1 + except: + pass + fresults.append((code, code_name, deal_big_order_info, deal_big_order_detail_info, big_order_enough)) response_data = json.dumps({"code": 0, "data": fresults}) - except Exception as e: + except Exception as e: response_data = json.dumps({"code": 1, "data": str(1)}) + + elif url.path == "/get_all_special_codes": + # 鑾峰彇鎵�鏈夎鲸璇嗗害鐨勪唬鐮� + code_blocks_dict = BlockSpecialCodesManager().get_code_blocks_dict() + fdata = {} + for k in code_blocks_dict: + fdata[k] = list(code_blocks_dict[k]) + response_data = json.dumps({"code": 0, "data": fdata}) + elif url.path == "/get_new_blocks_special_codes": + # 鑾峰彇鎵�鏈夎鲸璇嗗害鐨勪唬鐮� + code_blocks_dict = BlockSpecialCodesManager().get_temp_code_blocks_dict() + fdata = {} + for k in code_blocks_dict: + fdata[k] = list(code_blocks_dict[k]) + response_data = json.dumps({"code": 0, "data": fdata}) async_log_util.info(logger_request_api, f"缁撴潫璇锋眰{tool.get_thread_id()}-{url}") self.send_response(200) @@ -971,7 +1084,42 @@ self.__send_response(result_str) def __process_kpl_data(self, data_origin): + def do_limit_up(result_list_): + + def request_new_blocks_codes(blocks_info, all_new_blocks): + """ + 璇锋眰鏂版澘鍧楃殑浠g爜 + @param blocks_info:[(鏉垮潡鍚嶇О,鏉垮潡浠g爜)] + @return: + """ + yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes() + blocks = set() + for bi in blocks_info: + if bi[0] in blocks: + continue + blocks.add(bi[0]) + result = kpl_api.getCodesByPlate(bi[1]) + result = json.loads(result) + code_info_list = [] + for d in result["list"]: + if d[0] in yesterday_codes: + continue + # 娑ㄥ箙瑕佸ぇ浜�5% + rate = d[6] / int(round((tool.get_limit_up_rate(d[0]) - 1) * 10)) + if rate / ((tool.get_limit_up_rate(d[0]) - 1) * 10) < 5: + continue + # 鏍煎紡锛�(浠g爜,娑ㄥ箙) + code_info_list.append((d[0], d[6])) + # 淇濆瓨鏂伴鏉� + datas = [(d[0], d[6]) for d in result["list"]] + async_log_util.info(logger_kpl_new_blocks, f"{(tool.get_thread_id(), bi, datas)}") + if code_info_list: + # 灏嗕唬鐮佸姞鍏ユ柊棰樻潗 + new_block_processor.process_new_block_by_component_codes(bi[0], + set([x[0] for x in code_info_list]), + all_new_blocks) + try: if result_list_: # 淇濆瓨娑ㄥ仠鏃堕棿 @@ -1041,6 +1189,39 @@ try: OpenLimitUpGoodBlocksBuyStrategy.set_current_limit_up_data(result_list_) RadicalBuyBlockManager.set_current_limit_up_datas(result_list_) + except: + pass + + try: + # 鏂伴鏉� + new_block_processor.process_limit_up_list({x[0]: x[5] for x in result_list_}) + new_block_codes = new_block_processor.screen_new_blocks_with_limit_up_datas( + [(x[0], x[5]) for x in result_list_]) + if new_block_codes: + # 缁熻鏉垮潡鐨勪唬鐮� + records = KPLLimitUpDataRecordManager.total_datas + block_plate_code_dict = {} + for x in records: + block_plate_code_dict[kpl_util.filter_block(x[2])] = x[15] + # 鏂版澘鍧� + update_new_block_plates = [] + for b in new_block_codes: + for c in new_block_codes[b]: + new_block_processor.process_new_block_by_limit_up_list(c, b) + + for r in new_block_codes: + if r in block_plate_code_dict: + update_new_block_plates.append((r, block_plate_code_dict[r])) + if update_new_block_plates: + # 闇�瑕佽幏鍙栨澘鍧椾笅鐨勪唬鐮� + self.__new_blocks_codes_request_thread_pool.submit( + lambda: request_new_blocks_codes(update_new_block_plates, new_block_codes.keys())) + except Exception as e: + logger_debug.exception(e) + + # 灏嗘暟鎹帹閫佽嚦鍏朵粬椤圭洰 + try: + low_suction_data_pusher.push_limit_up_list(result_list_) except: pass @@ -1123,11 +1304,10 @@ RealTimeKplMarketData.set_market_jingxuan_out_blocks(result_list) elif type_ == KPLDataType.MARKET_STRONG.value: strong = data["data"] - logger_debug.debug("寮�鐩樺暒甯傚満寮哄害锛歿}", strong) + logger_kpl_market_strong.info(strong) # 淇濆瓨甯傚満鐑害 if strong is not None: RealTimeKplMarketData.set_market_strong(strong) - return json.dumps({"code": 0}) def __send_response(self, data): -- Gitblit v1.8.0