import json import logging import threading from api.outside_api_command_manager import ActionCallback from huaxin_client.client_network import SendResponseSkManager from strategy import strategy_params_settings, env_info from strategy.env_info import RealTimeEnvInfo from strategy.strategy_variable_factory import DataLoader from third_data.history_k_data_manager import TradeDateManager from utils import socket_util, middle_api_protocol, tool 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.settings.to_json_str() return {"code": 0, "data": result} def __on_get_env(self): """ 获取环境信息 @return: """ fdata = {} # 实时数据 fdata["real_time_data"] = RealTimeEnvInfo().to_dict() # 历史数据 fdata["history_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) 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 OnCommonRequest(self, client_id, request_id, data): ctype = data["ctype"] result_json = {} if ctype == "get_settings": result_json = self.__on_get_settings() elif ctype == 'get_env': result_json = self.__on_get_env() elif ctype == 'update_leading_limit_up_datas': result_json = self.__on_update_leading_limit_up_datas() self.send_response(result_json, client_id, request_id)