redis异步数据提交

Administrator

2023-08-07 f366d71c79fa6a17ecc6d994ad6f57e645b84209

 l2/l2_data_manager_new.py

@@ -59,6 +59,8 @@
# m值大单处理
m_big_money_begin_cache={}
m_big_money_process_index_cache={}


class L2BigNumForMProcessor:

    def __init__(self):
@@ -73,7 +75,7 @@
            tool.CodeDataCacheUtil.set_cache(m_big_money_begin_cache, code, index)
            # 保存位置
            key = "m_big_money_begin-{}".format(code)
            RedisUtils.setex(self.__get_redis(), key, tool.get_expire(), index)
            RedisUtils.setex_async(self.__get_redis(), key, tool.get_expire(), index)

    # 获取计算开始位置
    def __get_begin_pos(self, code):
@@ -91,7 +93,6 @@
        tool.CodeDataCacheUtil.set_cache(m_big_money_begin_cache,code,val)
        return val


    # 清除已经处理的数据
    def clear_processed_end_index(self, code):
        tool.CodeDataCacheUtil.clear_cache(m_big_money_process_index_cache, code)
@@ -102,7 +103,7 @@
    def __set_processed_end_index(self, code, index):
        tool.CodeDataCacheUtil.set_cache(m_big_money_process_index_cache,code,index)
        key = "m_big_money_process_index-{}".format(code)
        RedisUtils.setex(self.__get_redis(), key, tool.get_expire(), index)
        RedisUtils.setex_async(self.__get_redis(), key, tool.get_expire(), index)

    # 是否已经处理过
    def __get_processed_end_index(self, code):
@@ -119,7 +120,6 @@
        val = self.__get_processed_end_index(code)
        tool.CodeDataCacheUtil.set_cache(m_big_money_process_index_cache, code, val)
        return val


    # 处理大单
    def process(self, code, start_index, end_index, limit_up_price):
@@ -343,7 +343,7 @@
            # 时间差不能太大才能处理
            if not l2_trade_util.is_in_forbidden_trade_codes(code):
                # 判断是否已经挂单
                state = trade_manager.get_trade_state_cache(code)
                state = trade_manager.CodesTradeStateManager().get_trade_state_cache(code)
                start_index = len(total_datas) - len(add_datas)
                end_index = len(total_datas) - 1
                if state == trade_manager.TRADE_STATE_BUY_DELEGATED or state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER or state == trade_manager.TRADE_STATE_BUY_SUCCESS:
@@ -828,7 +828,9 @@
                for i in range(trade_index + 1, total_data[-1]["index"] + 1):
                    if L2DataUtil.is_limit_up_price_buy(total_data[i]["val"]):
                        left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count(code,
                                                                                                              total_data[i]["index"],
                                                                                                              total_data[
                                                                                                                  i][
                                                                                                                  "index"],
                                                                                                              total_data,
                                                                                                              num_operate_map)
                        if left_count > 0: