功能完善

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)

 strategy/env_info.py

@@ -1,6 +1,8 @@
"""
环境信息
"""
import json

from strategy.strategy_variable_factory import DataLoader
from third_data.history_k_data_manager import HistoryKDataManager
from utils import tool
@@ -22,6 +24,18 @@
        # Tick数据(更新时间, æ•°æ®æ•°é‡)
        self.ticks = ('', 0)

    def to_dict(self):
        d = self.__dict__
        return d

    @classmethod
    def to_obj(cls, json_str):
        result = json.loads(json_str)
        obj = RealTimeEnvInfo()
        for k in result:
            setattr(obj, k, result[k])
        return obj


def get_leading_limit_up_block_codes_count(day):
    """

 strategy/strategy_params_settings.py

@@ -49,7 +49,7 @@
        # æ˜¯å¦å¯ä¹°ä»Šæ—¥æ¶¨åœè¿‡çš„票
        self.can_buy_limited_up = False
        # æœ€ä½Žå¼€ç›˜æ¶¨å¹…
        self.min_open_rate = -0.02
        self.min_open_rate = -0.21
        # å¯ä¹°çš„涨幅比例
        self.avaiable_rates = (-0.03, 0.07)
        # ä»Šæ—¥æ¶¨åœä»·éœ€çªç ´XX日最高价,None表示此条数据不生效

 strategy/strategy_variable_factory.py

@@ -420,6 +420,8 @@
        """
        datas = self.load_all_codes_of_plates(is_for_buy=True)
        fdata = {}
        if not datas:
            return fdata
        for plate_code in datas:
            plate_name = datas[plate_code][0]
            codes_info = datas[plate_code][1]

 strategy/test.py

@@ -18,10 +18,8 @@


if __name__ == "__main__":
    print("======2个票涨停之后买")
    statistic_average(r"C:\Users\Administrator\Desktop\2个票涨停之后买.txt")
    print("======3个票涨停之后买")
    print("======3个票涨停之后买+开盘价>=-3")
    statistic_average(r"C:\Users\Administrator\Desktop\3个票涨停之后买.txt")
    print("======4个票涨停之后买")
    statistic_average(r"C:\Users\Administrator\Desktop\4个票涨停之后买.txt")
    print("======3个票涨停之后买+开盘价>=-2")
    statistic_average(r"C:\Users\Administrator\Desktop\3个票涨停之后买_开盘涨幅-2以上.txt")


 strategy/time_series_backtest.py

@@ -780,7 +780,7 @@
if __name__ == "__main__":
    back_test_dict = {}
    # days = ["2025-05-06", "2025-05-07", "2025-05-08", "2025-05-09", "2025-05-12", "2025-05-13", "2025-05-14",
    #         "2025-05-15", "2025-05-16"]
    #         "2025-05-15", "2025-05-16", "2025-05-19", "2025-05-20"]
    days = ["2025-05-12", "2025-05-13", "2025-05-14", "2025-05-15", "2025-05-16", "2025-05-19", "2025-05-20",
            "2025-05-21", "2025-05-22", "2025-05-23", "2025-05-26", "2025-05-27", "2025-05-28", "2025-05-29",
            "2025-05-30", "2025-06-03", "2025-06-04", "2025-06-05", "2025-06-06", "2025-06-09"]

 strategy/µÍÎü½Å±¾_±æʶ¶È_v6.py

@@ -142,7 +142,7 @@

    limit_up_codes_count = max([(p, len(sv.开盘啦最正板块涨停.get(p, []))) for p in can_buy_plates], key=lambda x: x[1])[1]

    threshold_money *= max(10 - limit_up_codes_count + 1, 5)/10
    threshold_money *= max(min(10 - limit_up_codes_count + 3, 10), 5)/10

    # print(target_code, sv.自由流通市值, threshold_money, limit_up_codes_count)


 third_data/history_k_data_manager.py

@@ -11,6 +11,7 @@
from huaxin_client import l1_subscript_codes_manager
from log_module.log import logger_debug
from third_data import history_k_data_util
from third_data.history_k_data_util import HistoryKDatasUtils
from utils import tool, init_data_util


@@ -186,5 +187,48 @@
        return codes


@tool.singleton
class TradeDateManager:
    # {"日期":日期列表}
    __date_list_dict = {}
    """
    äº¤æ˜“日期管理
    """

    def __load_dates(self, now_day):
        if self.__date_list_dict.get(now_day):
            return
        # èŽ·å–前后一个月的交易日期
        start_date = tool.date_sub(now_day, 30)
        end_date = tool.date_sub(now_day, -30)
        dates = HistoryKDatasUtils.get_trading_dates(start_date, end_date)
        self.__date_list_dict[now_day] = dates

    def get_next_trade_day(self, now_day):
        """
        èŽ·å–下一个交易日
        @param now_day:
        @return:
        """
        self.__load_dates(now_day)
        for day in self.__date_list_dict[now_day]:
            if day > now_day:
                return day
        return None

    def get_previous_trade_day(self, now_day):
        """
        èŽ·å–上一个交易日
        @param now_day:
        @return:
        """
        self.__load_dates(now_day)
        for day in reversed(self.__date_list_dict[now_day]):
            if day < now_day:
                return day
        return None


if __name__ == "__main__":
    print(HistoryKDataManager().get_history_bars_codes("2024-12-31"))
    print(TradeDateManager().get_next_trade_day("2025-06-09"))
    print(TradeDateManager().get_previous_trade_day("2025-06-09"))

 third_data/history_k_data_util.py

@@ -390,7 +390,10 @@

    @classmethod
    def get_trading_dates(cls, start_date, end_date):
        return hx_qc_value_util.get_trade_calendar(start_date, end_date)
        if constant.is_windows():
            return JueJinApi.get_trading_dates(start_date, end_date)
        else:
            return hx_qc_value_util.get_trade_calendar(start_date, end_date)

    @classmethod
    def get_now_price(cls, codes):