import json import logging import os import threading import constant from api.outside_api_command_manager import ActionCallback from code_attribute import gpcode_manager from huaxin_client import l1_subscript_codes_manager from huaxin_client.client_network import SendResponseSkManager from log_module import async_log_util from strategy import strategy_params_settings, env_info, strategy_manager from strategy.env_info import RealTimeEnvInfo from strategy.place_order_queue_manager import PlaceOrderRecordManager from strategy.strategy_manager import PlateWhiteListManager from strategy.strategy_params_settings import StrategyParamsSettingsManager, StrategyParamsSettings from strategy.strategy_variable import StockVariables from strategy.strategy_variable_factory import DataLoader from third_data.history_k_data_manager import TradeDateManager from third_data.kpl_block_manager import KPLCodeJXBlocksManager from trade import trade_record_log_util from trade.trade_manager import TradeStateManager, DealCodesManager, logger_trade from utils import socket_util, middle_api_protocol, tool, huaxin_util OPERRATE_SET = 1 # 设置 OPERRATE_DELETE = 2 # 删除 OPERRATE_GET = 3 # 获取 OPERRATE_ADD = 4 # 新增 class MyAPICallback(ActionCallback): @classmethod def __send_response(cls, data_bytes): sk = SendResponseSkManager.create_send_response_sk(addr=middle_api_protocol.SERVER_HOST, port=middle_api_protocol.SERVER_PORT) try: data_bytes = socket_util.load_header(data_bytes) sk.sendall(data_bytes) result, header_str = socket_util.recv_data(sk) result = json.loads(result) if result["code"] != 0: raise Exception(result['msg']) finally: sk.close() def send_response(self, data, _client_id, _request_id): data_bytes = json.dumps({"type": "response", "data": data, "client_id": _client_id, "request_id": _request_id}).encode('utf-8') for i in range(3): try: self.__send_response(data_bytes) # print("发送数据成功") break except Exception as e1: logging.exception(e1) def __on_get_settings(self): """ 获取交易参数 @return: """ result = strategy_params_settings.StrategyParamsSettingsManager().get_settings().to_json_str() result = json.loads(result) result["trade_state"] = 1 if TradeStateManager().is_can_buy_cache() else 0 return {"code": 0, "data": result} def __on_set_settings(self, data): """ 设置交易参数 @return: """ settings = strategy_params_settings.StrategyParamsSettingsManager().get_settings() settings_vars = vars(settings) for k in data: if k == "trade_state": if data[k]: TradeStateManager().open_buy() else: TradeStateManager().close_buy() continue if k not in settings_vars: # 没有在属性里面 continue settings.__setattr__(k, data[k]) strategy_params_settings.StrategyParamsSettingsManager().set_settings(settings) # 同步参数设置 return {"code": 0, "data": {}} def __on_get_env(self, need_hsitory_data): """ 获取环境信息 @param need_hsitory_data: 是否需要历史数据 @return: """ fdata = {} # 实时数据 fdata["real_time_data"] = RealTimeEnvInfo().to_dict() # 历史数据 fdata["history_data"] = {} print("获取环境", os.getpid()) if need_hsitory_data: if tool.get_now_time_str() < '16:00:00': # 如果在16:00之前采用当前日期 day = tool.get_now_date_str() else: # 如果在16:00之后采用下一个交易日 day = TradeDateManager().get_next_trade_day(tool.get_now_date_str()) fdata["history_data"][ "leading_limit_up_block_codes_count"] = env_info.get_leading_limit_up_block_codes_count( day) if tool.get_now_time_str() < '16:00:00': # 如果在16:00之前采用当前日期 day = TradeDateManager().get_previous_trade_day(tool.get_now_date_str()) else: # 如果在16:00之后采用下一个交易日 day = tool.get_now_date_str() fdata["history_data"]["k_bars_count"] = env_info.get_history_k_bars(day) day = tool.get_now_date_str() fdata["history_data"]["kpl_code_jx_blocks_count"] = env_info.get_kpl_code_jx_blocks(day) return {"code": 0, "data": fdata, "msg": "测试结果"} def __on_update_leading_limit_up_datas(self): """ 更新领涨代码信息 @return: """ def update(): plates = __DataLoader.get_limit_up_reasons_with_plate_code() for p in plates: print(p) __DataLoader.load_plate_codes(p[0], p[1]) if tool.get_now_time_str() < '16:00:00': # 如果在16:00之前采用当前日期 day = tool.get_now_date_str() else: # 如果在16:00之后采用下一个交易日 day = TradeDateManager().get_next_trade_day(tool.get_now_date_str()) __DataLoader = DataLoader(day) threading.Thread(target=lambda: update(), daemon=True).start() return {"code": 0} def __on_update_kpl_code_jx_blocks_datas(self): """ 更新开盘啦精选板块数据 @return: """ def update(): codes = set() codes_sh, codes_sz = l1_subscript_codes_manager.get_codes() codes |= set([x.decode() for x in codes_sh]) codes |= set([x.decode() for x in codes_sz]) KPLCodeJXBlocksManager(day, codes).start_download_blocks() # 如果在16:00之前采用当前日期 day = tool.get_now_date_str() threading.Thread(target=lambda: update(), daemon=True).start() return {"code": 0} def __on_init_data(self): try: strategy_manager.low_suction_strtegy.load_data() return {"code": 0} except Exception as e: logging.exception(e) return {"code": 1, "msg": str(e)} def __on_plate_white_list(self, data): operate = data["operate"] if operate == OPERRATE_GET: plates = PlateWhiteListManager().get_plates() return {"code": 0, "data": list(plates)} elif operate == OPERRATE_ADD: plate = data["plate"] PlateWhiteListManager().add_plate(plate) return {"code": 0} elif operate == OPERRATE_DELETE: plate = data["plate"] PlateWhiteListManager().remove_plate(plate) return {"code": 0} def OnCommonRequest(self, client_id, request_id, data): ctype = data["ctype"] result_json = {} if ctype == "get_settings": result_json = self.__on_get_settings() elif ctype == 'set_settings': del data["ctype"] result_json = self.__on_set_settings(data) elif ctype == 'get_env': # 获取环境数据 result_json = self.__on_get_env(data.get("history")) elif ctype == 'update_leading_limit_up_datas': # 更新领涨数据 result_json = self.__on_update_leading_limit_up_datas() elif ctype == 'update_kpl_code_jx_blocks_datas': # 更新开盘啦精选数据 result_json = self.__on_update_kpl_code_jx_blocks_datas() elif ctype == 'init_datas': # 初始化数据 result_json = self.__on_init_data() elif ctype == 'get_place_order_records': # 获取下单记录 datas = trade_record_log_util.get_trade_records(trade_record_log_util.TYPE_PLACE_ORDER) for data in datas: data[3]["code_name"] = gpcode_manager.CodesNameManager().get_code_name(data[3]["code"]) result_json = {"code": 0, "data": datas} elif ctype == 'get_can_buy_plates': # 获取可以买的板块 settings: StrategyParamsSettings = StrategyParamsSettingsManager().get_settings() try: current_limit_up_plate_codes = strategy_manager.low_suction_strtegy.current_limit_up_plate_codes plates = [plate for plate, codes in current_limit_up_plate_codes.items() if len(codes) >= settings.limit_up_count_of_new_plate] except: plates = [] result_json = {"code": 0, "data": plates} elif ctype == 'plate_white_list': result_json = self.__on_plate_white_list(data) elif ctype == 'get_not_process_place_order_record': # 获取未处理的下单记录 record = PlaceOrderRecordManager(tool.get_now_date_str()).get_not_process_record() if not record: result_json = {"code": 1, "msg": '暂无记录'} else: # (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅) fdata = { "id": record[0], "code": record[1], "name": gpcode_manager.CodesNameManager().get_code_name(record[1]), "platesInfo": {k: [(xx, gpcode_manager.CodesNameManager().get_code_name(xx)) for xx in v] for k, v in record[2].items()}, "time_str": record[4], "bigOrderInfo": record[3] } result_json = {"code": 0, "data": fdata} elif ctype == 'set_place_order_buy': id_ = data.get("id") place_order_record_manager = PlaceOrderRecordManager(tool.get_now_date_str()) # (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅) record = place_order_record_manager.get_not_process_record_by_id(id_) if not record: result_json = {"code": 1, "msg": '记录不存在'} else: code = record[1] sv: StockVariables = strategy_manager.low_suction_strtegy.stock_variables_dict.get(code) if sv is None: result_json = {"code": 1, "msg": '代码变量对象不存在'} else: place_order_record_manager.set_buy(record[0]) # 可以下单 # 判断是否可以买 order_ref = huaxin_util.create_order_ref() price = tool.get_buy_max_price(sv.当前价) volume = 100 DealCodesManager().place_order(set(record[2].keys()), record[1], order_ref, price, volume) trade_record_log_util.add_place_order_log(code, trade_record_log_util.PlaceOrderInfo(code=code, time_str= record[4], price=sv.当前价, rate=round(( sv.当前价 - sv.昨日收盘价) * 100 / sv.昨日收盘价, 2), plates=record[ 2].keys(), plates_info= record[2], info=record[3] )) async_log_util.info(logger_trade, f"{code}下单，板块：{record[2].keys()}") result_json = {"code": 0, "msg": '处理成功'} elif ctype == 'set_place_order_not_buy': id_ = data.get("id") place_order_record_manager = PlaceOrderRecordManager(tool.get_now_date_str()) # (ID,代码,板块信息, 大单信息, 时间, 价格, 涨幅) record = place_order_record_manager.get_not_process_record_by_id(id_) if not record: result_json = {"code": 1, "msg": '记录不存在'} else: place_order_record_manager.set_not_buy(record[0], "手动驳回") result_json = {"code": 0, "msg": '处理成功'} elif ctype == 'add_black_list': code = data.get("code") gpcode_manager.BlackListCodeManager().add_code(code) result_json = {"code": 0, "msg": '添加成功'} self.send_response(result_json, client_id, request_id)