redis异步数据提交

Administrator

2023-08-07 f366d71c79fa6a17ecc6d994ad6f57e645b84209

 l2/l2_data_manager_new.py

@@ -57,8 +57,10 @@


# m值大单处理
m_big_money_begin_cache={}
m_big_money_process_index_cache={}
m_big_money_begin_cache = {}
m_big_money_process_index_cache = {}


class L2BigNumForMProcessor:

    def __init__(self):
@@ -73,24 +75,23 @@
            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):
        key = "m_big_money_begin-{}".format(code)
        val = RedisUtils.get(self.__get_redis(),key)
        val = RedisUtils.get(self.__get_redis(), key)
        if val is None:
            return None
        return int(val)

    def __get_begin_pos_cache(self, code):
        cache_result = tool.CodeDataCacheUtil.get_cache(m_big_money_begin_cache,code)
        cache_result = tool.CodeDataCacheUtil.get_cache(m_big_money_begin_cache, code)
        if cache_result[0]:
            return cache_result[1]
        val =   self.__get_begin_pos(code)
        tool.CodeDataCacheUtil.set_cache(m_big_money_begin_cache,code,val)
        val = self.__get_begin_pos(code)
        tool.CodeDataCacheUtil.set_cache(m_big_money_begin_cache, code, val)
        return val


    # 清除已经处理的数据
    def clear_processed_end_index(self, code):
@@ -100,9 +101,9 @@

    # 添加已经处理过的单
    def __set_processed_end_index(self, code, index):
        tool.CodeDataCacheUtil.set_cache(m_big_money_process_index_cache,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: