bug修复

Administrator

2023-12-22 9039563a3c2e1349b0f71003f8a3f0b793924e82

bug修复

 code_attribute/code_nature_analyse.py

@@ -338,6 +338,32 @@
        return False


# 暂时不使用
# 从最近一次涨停开始，是否涨幅过高
def is_up_too_high_from_latest_limit_up(record_datas):
    datas = copy.deepcopy(record_datas)
    datas.sort(key=lambda x: x["bob"])
    datas = datas[-20:]
    datas.reverse()
    today_limit_up_price = round(float(gpcode_manager.get_limit_up_price_by_preprice(datas[0]["close"])), 2)
    max_price = 0
    limit_up_price = None
    for i in range(0, len(datas)):
        item = datas[i]
        if item['high'] > max_price:
            max_price = item['high']
        if __is_limited_up(item):
            limit_up_price = item['high']
            break
    if not limit_up_price:
        return False
    if today_limit_up_price < max_price:
        return False
    if (today_limit_up_price - limit_up_price) / limit_up_price > 0.25:
        return True
    return False


# 最近几天是否有最大量
def is_have_latest_max_volume(record_datas, day_count):
    datas = copy.deepcopy(record_datas)

 test/test_code_attribute.py

@@ -5,19 +5,21 @@
def is_too_high(datas):
    limit_up_price = round(datas[0]["close"] * 1.1, 2)
    datas.reverse()
    is_new_high = code_nature_analyse.is_price_too_high_in_days(datas, limit_up_price)
    is_new_high = code_nature_analyse.is_up_too_high_from_latest_limit_up(datas)
    return is_new_high


if __name__ == "__main__":
    code_str = "002915"
    # code_str = "002103,002108,002176,002189,002328,002397,002457,002495,002578,002903,002933,003028,003040,600234,600293,600764,603090,603162,603193,603270,603332,603380,603615,603711"
    code_str = "002933"
    codes = code_str.split(",")
    for code in codes:
        if not tool.is_shsz_code(code):
            continue
        try:
            limit_up_price = 16.96
            # limit_up_price = 16.96
            volumes_data = init_data_util.get_volumns_by_code(code, 150)
            is_too_high(volumes_data)
            # volumes_data = volumes_data[1:]
            print(code, is_too_high(volumes_data))
        except:
            print(code, "出错")

 third_data/code_plate_key_manager.py

@@ -9,7 +9,7 @@

import constant
from db.redis_manager_delegate import RedisUtils
from third_data import kpl_block_util, kpl_api
from third_data import kpl_block_util, kpl_api, kpl_util
from utils import global_util, tool
from log_module import log, async_log_util
from db import redis_manager_delegate as redis_manager
@@ -518,10 +518,14 @@
                if not general_blocks or block not in general_blocks:
                    # 没在泛化板块中
                    continue
            if d[4].find("连板") > 0:
                if d[4].replace("连板", "").isdigit():
                    count = int(d[4].replace("连板", ""))
            count = kpl_util.get_high_level_count(d[4])
                    if count >= 3:
                if d[4].find("连板") > 0:
                    is_strong_block = True
                    break
                elif d[0] in yesterday_current_limit_up_codes and len(block_codes) >= 2:
                    # 几天几板，且最近2连板
                    # 看是否有首板后排
                        is_strong_block = True
                        break

@@ -776,10 +780,8 @@
        high_level_general_code_blocks = {}
        # 是否是3板及以上的高位板
        for r in current_limit_up_datas:
            if r[4].find("连板") > 0:
                if r[4].replace("连板", "").isdigit():
                    count = int(r[4].replace("连板", ""))
                    if count >= 3:
            count = kpl_util.get_high_level_count(r[4])
            if count >= 3 and r[0] in yesterday_current_limit_up_codes:
                        latest_datas = latest_current_limit_up_records[:count-1]
                        # 是高位板
                        # 当日精选

 third_data/data_server.py

@@ -635,7 +635,8 @@
                                logger_debug.exception(e)
                kpl_data_manager.KPLLimitUpDataRecordManager.save_record(tool.get_now_date_str(), result_list_)
                self.__kplDataManager.save_data(type_, result_list_)

        # 将"概念"二字替换掉
        data = json.loads(json.dumps(data).replace("概念", ""))
        type_ = data["type"]
        print("开盘啦type:", type_)
        if type_ == KPLDataType.BIDDING.value:

 third_data/kpl_api.py

@@ -142,7 +142,7 @@
    result = json.loads(result)
    if result:
        if "List" in result:
            names = [x["CName"] for x in result["List"]]
            names = [x["CName"].replace("概念", "") for x in result["List"]]
            return names
    return []

@@ -154,7 +154,7 @@
    result = json.loads(result)
    if result:
        if "List" in result:
            names = [x["CName"] for x in result["List"]]
            names = [x["CName"].replace("概念", "") for x in result["List"]]
            return names
    return []

@@ -176,7 +176,6 @@
    except:
        pass
    return list(set(blocks))



if __name__ == "__main__":

 third_data/kpl_util.py

@@ -266,3 +266,22 @@
        mysqldb = mysql_data.Mysqldb()
        results = mysqldb.select_all(f"select _name from kpl_plate where _id='{id_}'")
        return set([r[0] for r in results])


# 获取高位板的数量
def get_high_level_count(key):
    if key.find("连板") >= 0:
        # 形式如: 3连板
        return int(key.replace("连板", ""))
    elif key.find("天") >= 0 and key.find("板") >= 0:
        # 形式如：5天4板
        return int(key.split("天")[1].replace("板", ""))
    # 形式如：首板
    return 1


if __name__ == "__main__":
    print(get_high_level_count("首板"))
    print(get_high_level_count("5天4板"))
    print(get_high_level_count("2连板"))
    print(get_high_level_count("4连板"))