添加低吸中间服务器

admin

2025-01-16 9d2b0ab3967761b2d4b4e1c92034c3c0ea6705f5

 code_attribute/code_price_manager.py

@@ -2,9 +2,10 @@
代码价格管理
"""
import json
import time

from db.redis_manager import RedisUtils
from utils import tool
from utils import tool, history_k_data_util
from db import redis_manager as redis_manager


@@ -152,5 +153,58 @@
            self.__save_buy1_price_info(code, time_str, None)


class CodesLimitRateManager:
    __pre_close_dict = {}
    __current_price_dict = {}
    """
    涨幅管理
    """

    @classmethod
    def __load_pre_close_prices(cls, codes):
        """
        获取昨天的收盘价格
        :param codes:
        :return:
        """
        fcodes = set()
        for code in codes:
            if code in cls.__pre_close_dict:
                continue
            fcodes.add(code)
        if not fcodes:
            return
        results = history_k_data_util.HistoryKDatasUtils.get_gp_latest_info(list(fcodes), fields="sec_id, pre_close")
        for result in results:
            cls.__pre_close_dict[result["sec_id"]] = round(result["pre_close"], 2)

    @classmethod
    def get_price_rates(cls, codes):
        """
        获取代码的价格涨幅
        :param codes:
        :return:
        """
        cls.__load_pre_close_prices(codes)
        # 获取现价
        now_time = time.time()
        price_codes = set()
        for code in codes:
            if code not in cls.__current_price_dict or now_time - cls.__current_price_dict[code][1] > 5:
                price_codes.add(code)
        if price_codes:
            results = history_k_data_util.HistoryKDatasUtils.get_now_price(price_codes)
            for r in results:
                cls.__current_price_dict[r[0]] = (r[1], time.time())
        rate_dict = {}
        for code in codes:
            pre_close = cls.__pre_close_dict.get(code)
            now_price_info = cls.__current_price_dict.get(code)
            if pre_close and now_price_info:
                rate = round((now_price_info[0] - pre_close) * 100 / pre_close, 2)
                rate_dict[code] = rate
        return rate_dict


if __name__ == "__main__":
    print(Buy1PriceManager().get_limit_up_info("002777"))