L撤快速成交触发撤单修改

Administrator

2024-02-05 8c7f84cbd0afde7600b0c67a8032fa19c57d8c41

 code_attribute/code_nature_analyse.py

@@ -218,17 +218,19 @@
    # 是否具有辨识度
    p9 = is_special(record_datas)
    p10 = is_latest_10d_max_volume_at_latest_2d(record_datas)
    # 最近5天是否跌停/炸板
    p11 = __is_latest_open_limit_up_or_limit_down(record_datas, 5)
    # 最近5天是否炸板
    p11 = __is_latest_open_limit_up(record_datas, 5)
    # 30天内是否有涨停
    p12 = __has_limit_up(record_datas, 30)
    # 最近5天是否跌停
    p13 = __is_latest_limit_down(record_datas, 5)

    return p1, p2, p3, p4, p5, p6, p7, p8, p9, p10, p11, p12
    return p1, p2, p3, p4, p5, p6, p7, p8, p9, p10, p11, p12, p13


# 是否具有K线形态
def is_has_k_format(limit_up_price, record_datas):
    is_too_high, is_new_top, is_lowest, is_near_new_top, is_n, is_v, has_format, volume_info, is_special, has_max_volume, open_limit_up, is_limit_up_in_30days = get_k_format(
    is_too_high, is_new_top, is_lowest, is_near_new_top, is_n, is_v, has_format, volume_info, is_special, has_max_volume, open_limit_up, is_limit_up_in_30days, is_latest_limit_down = get_k_format(
        float(limit_up_price), record_datas)
    if not has_format:
        return False, "不满足K线形态"
@@ -338,6 +340,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)
@@ -372,9 +400,55 @@
    #     return False
    rate = (float(limit_up_price) - min_price) / min_price
    # print(rate)
    if rate >= 0.28:
    if rate >= 0.319:
        return True
    return False


# 连续涨停后是否回调不足够
def is_continue_limit_up_not_enough_fall_dwon(record_datas):
    # 10 天内是否有连续3板
    datas = copy.deepcopy(record_datas)
    datas.sort(key=lambda x: x["bob"])
    datas = datas[0 - 10:]
    limit_up_continue_count_info = None
    max_limit_up_continue_count_info_list = []  # [连续涨停次数,涨停起点]

    for i in range(len(datas)):
        item = datas[i]
        if __is_limit_up(item):
            if not limit_up_continue_count_info:
                limit_up_continue_count_info = [1, i]
            else:
                limit_up_continue_count_info[0] += 1
        else:
            if limit_up_continue_count_info:
                max_limit_up_continue_count_info_list.append(limit_up_continue_count_info)
                limit_up_continue_count_info = None
    max_limit_up_info = None
    for x in max_limit_up_continue_count_info_list:
        if max_limit_up_info is None:
            max_limit_up_info = x
        if max_limit_up_info[0] <=  x[0]:
            max_limit_up_info = x

    if not max_limit_up_info or max_limit_up_info[0] < 3:
        print("无3连板")
        return False
    start_index = max_limit_up_info[1]
    max_price_info = [0, None]
    for i in range(start_index, len(datas)):
        item = datas[i]
        if item["high"] > max_price_info[0]:
            max_price_info = [item["high"], i]
    # 计算回踩价格
    lowest_price_threhhold = round((1-0.28) * max_price_info[0], 2)
    for i in range(max_price_info[1] + 1, len(datas)):
        item = datas[i]
        if item["low"] < lowest_price_threhhold:
            print("回踩足够")
            return False
    return True


# 是否有涨停
@@ -478,15 +552,23 @@


# 最近几天是否有炸板或跌停
def __is_latest_open_limit_up_or_limit_down(datas, day_count):
def __is_latest_open_limit_up(datas, day_count):
    datas = copy.deepcopy(datas)
    datas.sort(key=lambda x: x["bob"])
    items = datas[0-day_count:]
    items = datas[0 - day_count:]
    for item in items:
        limit_up_price = float(gpcode_manager.get_limit_up_price_by_preprice(item["pre_close"]))
        if abs(limit_up_price - item["high"]) < 0.001 and abs(limit_up_price - item["close"]) > 0.001:
            # 炸板
            return True
    return False


def __is_latest_limit_down(datas, day_count):
    datas = copy.deepcopy(datas)
    datas.sort(key=lambda x: x["bob"])
    items = datas[0 - day_count:]
    for item in items:
        # 是否有跌停
        limit_down_price = float(gpcode_manager.get_limit_down_price_by_preprice(item["pre_close"]))
        if abs(limit_down_price - item["close"]) < 0.001:
@@ -608,22 +690,39 @@

# 是否具有辨识度
def is_special(datas):
    # 30个交易日内有≥5天曾涨停且连续涨停数或曾涨停≥2天
    if len(datas) > 30:
        datas = datas[-30:]
    datas = copy.deepcopy(datas)
    datas.sort(key=lambda x: x["bob"])
    datas_30 = datas[-30:]
    datas_90 = datas[-90:]
    count = 0
    # 30个交易日内累计涨停次数≥4次
    continue_count = 0
    has_continue = False
    for item in datas_30:
        if __is_limit_up(item):
            continue_count += 1
            count += 1
            if continue_count >= 4:
                has_continue = True
        else:
            continue_count = 0
    if count >= 5 and has_continue:
        return True, "短期辨识度"

    count = 0
    continue_count = 0
    last_index = -1
    for i in range(len(datas)):
        if __is_limited_up(datas[i]):
            if last_index >= 0 and i - last_index == 1:
                continue_count += 1
    has_continue = False
    # 90个交易日内涨停次数≥6次
    for item in datas_90:
        if __is_limit_up(item):
            continue_count += 1
            count += 1
            last_index = i
    if count >= 5 and continue_count > 0:
        return True, ''
    return False, ''
            if continue_count >= 4:
                has_continue = True
        else:
            continue_count = 0
    if count >= 6 and has_continue:
        return True, "长期辨识度"
    return False, ""


if __name__ == "__main__":