功能完善

Administrator

2025-06-09 fd088787bc52e93bbce8f6aeb080fa8f472015aa

 api/outside_api_callback.py

@@ -1,10 +1,14 @@
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
from utils import socket_util, middle_api_protocol
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):
@@ -40,21 +44,65 @@
        @return:
        """
        result = strategy_params_settings.settings.to_json_str()
        return json.dumps({"code": 0, "data": result})
        return {"code": 0, "data": result}

    def __on_get_env(self):
        """
        获取环境信息
        @return:
        """
        return json.dumps({"code": 0, "data": {}, "msg": "测试结果"})
        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_str = ''
        result_json = {}
        if ctype == "get_settings":
            result_str = self.__on_get_settings()
            result_json = self.__on_get_settings()
        elif ctype == 'get_env':
            result_str = self.__on_get_env()

        self.send_response(result_str, client_id, request_id)
            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)