bug修复/策略完善

Administrator

2025-07-01 bccc14e6d890cefb5309c904a6413bea1270ff03

 api/outside_api_callback.py

@@ -1,14 +1,29 @@
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 strategy import strategy_params_settings, env_info
from strategy import strategy_params_settings, env_info, strategy_manager
from strategy.env_info import RealTimeEnvInfo
from strategy.strategy_manager import PlateWhiteListManager
from strategy.strategy_params_settings import StrategyParamsSettingsManager, StrategyParamsSettings
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
from utils import socket_util, middle_api_protocol, tool

OPERRATE_SET = 1  # 设置
OPERRATE_DELETE = 2  # 删除
OPERRATE_GET = 3  # 获取
OPERRATE_ADD = 4  # 新增


class MyAPICallback(ActionCallback):
@@ -44,7 +59,10 @@
        @return:
        """
        result = strategy_params_settings.StrategyParamsSettingsManager().get_settings().to_json_str()
        return {"code": 0, "data": json.loads(result)}
        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):
        """
@@ -52,7 +70,17 @@
        @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)
        # 同步参数设置
@@ -69,6 +97,7 @@
        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之前采用当前日期
@@ -87,6 +116,10 @@
                # 如果在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):
@@ -111,6 +144,46 @@
        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 = {}
@@ -120,7 +193,36 @@
            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)
        self.send_response(result_json, client_id, request_id)