异常保护

Administrator

4 天以前 48fb7a00951f91bdc707e5dd2d196e5bccb752c3

 l2/l2_data_util.py

@@ -12,13 +12,14 @@
import numpy

import constant
import gpcode_manager
import l2_data_util
from code_attribute import gpcode_manager
from db.redis_manager_delegate import RedisUtils
from l2 import l2_data_log, l2_data_source_util
import log
from db import redis_manager
import tool
from log_module import log, log_export, async_log_util
from db import redis_manager_delegate as redis_manager
from utils import tool

__db = 1
_redisManager = redis_manager.RedisManager(1)
# l2数据管理
# 本地最新一次上传的数据
@@ -29,22 +30,31 @@
# 用于加快数据处理，用空换时间
local_today_num_operate_map = {}

# 买入订单号映射,只有原生的L2数据才有
local_today_buyno_map = {}

def load_l2_data(code, force=False):
    redis = _redisManager.getRedis()
# 卖出订单号映射，只有原生的L2数据才有
local_today_sellno_map = {}

# 已经撤单的订单号
local_today_canceled_buyno_map = {}


def load_l2_data(code, load_latest=True, force=False):
    # 加载最近的l2数据
    if local_latest_datas.get(code) is None or force:
        # 获取最近的数据
        _data = redis.get("l2-data-latest-{}".format(code))
        if _data is not None:
            if code in local_latest_datas:
                local_latest_datas[code] = json.loads(_data)
            else:
                local_latest_datas.setdefault(code, json.loads(_data))
    if load_latest:
        if local_latest_datas.get(code) is None or force:
            # 获取最近的数据
            _data = RedisUtils.get(_redisManager.getRedis(), "l2-data-latest-{}".format(code))
            if _data is not None:
                if code in local_latest_datas:
                    local_latest_datas[code] = json.loads(_data)
                else:
                    local_latest_datas.setdefault(code, json.loads(_data))
        # 获取今日的数据

    if local_today_datas.get(code) is None or force:
        datas = log.load_l2_from_log()
        datas = log_export.load_l2_from_log()
        datas = datas.get(code)
        if datas is None:
            datas = []
@@ -66,8 +76,44 @@
        # local_today_datas[code] = new_datas
        # 根据今日数据加载
        load_num_operate_map(local_today_num_operate_map, code, local_today_datas.get(code), force)
        load_buy_no_map(local_today_buyno_map, code, local_today_datas.get(code), force)
        load_sell_no_map(local_today_sellno_map, code, local_today_datas.get(code), force)
        load_canceled_buy_no_map(local_today_canceled_buyno_map, code, local_today_datas.get(code), force)
        return data_normal
    return True


# L2数据是否正常
def l2_data_is_normal(code):
    datas = local_today_datas.get(code)
    if not datas:
        # 初始化
        local_today_datas[code] = []
        load_num_operate_map(local_today_num_operate_map, code, local_today_datas.get(code))
        load_buy_no_map(local_today_buyno_map, code, local_today_datas.get(code))
        load_sell_no_map(local_today_sellno_map, code, local_today_datas.get(code))
        load_canceled_buy_no_map(local_today_canceled_buyno_map, code, local_today_datas.get(code))

    if datas and len(datas) != datas[-1]["index"] + 1:
        return False
    return True


# 加载所有的l2数据
def load_l2_data_all(force=False):
    datas = log_export.load_l2_from_log()
    for code in datas:
        if force:
            local_today_datas[code] = datas[code]
        else:
            if code not in local_today_datas:
                local_today_datas[code] = datas[code]
        load_num_operate_map(local_today_num_operate_map, code, local_today_datas.get(code), force)
        load_buy_no_map(local_today_buyno_map, code, local_today_datas.get(code), force)
        load_sell_no_map(local_today_sellno_map, code, local_today_datas.get(code), force)
        load_canceled_buy_no_map(local_today_canceled_buyno_map, code, local_today_datas.get(code), force)

    constant.L2_DATA_IS_LOADED = True


# 将数据根据num-operate分类
@@ -84,6 +130,64 @@
        local_today_num_operate_map[code].get(key).append(data)


# 将数据根据orderNo分类,原生数据才有
def load_buy_no_map(local_today_buyno_map, code, source_datas, clear=False):
    # 只有原生L2数据才会有此操作
    if constant.L2_SOURCE_TYPE != constant.L2_SOURCE_TYPE_HUAXIN:
        return
    if local_today_buyno_map.get(code) is None:
        local_today_buyno_map[code] = {}
    if clear:
        local_today_buyno_map[code] = {}

    for data in source_datas:
        if data["val"]["operateType"] != 0:
            continue
        # 只填充买入数据
        key = "{}".format(data["val"]["orderNo"])
        if local_today_buyno_map[code].get(key) is None:
            local_today_buyno_map[code].setdefault(key, data)


# 将数据根据orderNo分类,原生数据才有
def load_sell_no_map(local_today_sellno_map, code, source_datas, clear=False):
    # 只有原生L2数据才会有此操作
    if constant.L2_SOURCE_TYPE != constant.L2_SOURCE_TYPE_HUAXIN:
        return
    if local_today_sellno_map.get(code) is None:
        local_today_sellno_map[code] = {}
    if clear:
        local_today_sellno_map[code] = {}

    for data in source_datas:
        if data["val"]["operateType"] != 2:
            continue
        # 只填充买入数据
        key = "{}".format(data["val"]["orderNo"])
        if local_today_sellno_map[code].get(key) is None:
            local_today_sellno_map[code].setdefault(key, data)


# 将数据根据orderNo分类已撤订单,原生数据才有
def load_canceled_buy_no_map(local_today_canceled_buyno_map, code, source_datas, clear=False):
    # 只有原生L2数据才会有此操作
    if constant.L2_SOURCE_TYPE != constant.L2_SOURCE_TYPE_HUAXIN:
        return
    if local_today_canceled_buyno_map.get(code) is None:
        local_today_canceled_buyno_map[code] = {}
    if clear:
        local_today_canceled_buyno_map[code] = {}

    for data in source_datas:
        # 只留下买撤
        if data["val"]["operateType"] != 1:
            continue
        # 只填充买入数据
        key = "{}".format(data["val"]["orderNo"])
        if local_today_canceled_buyno_map[code].get(key) is None:
            local_today_canceled_buyno_map[code].setdefault(key, data)


@tool.async_call
def saveL2Data(code, datas, msg=""):
    start_time = round(time.time() * 1000)
@@ -94,29 +198,29 @@
    redis_instance = _redisManager.getRedis()

    try:
        if redis_instance.setnx("l2-save-{}".format(code), "1") > 0:

        if RedisUtils.setnx(redis_instance, "l2-save-{}".format(code), "1", auto_free=False) > 0:
            # 计算保留的时间
            expire = tool.get_expire()
            i = 0
            for _data in datas:
                i += 1
                key = "l2-" + _data["key"]
                value = redis_instance.get(key)
                value = RedisUtils.get(redis_instance, key, auto_free=False)
                if value is None:
                    # 新增
                    try:
                        value = {"index": _data["index"], "re": _data["re"]}
                        redis_instance.setex(key, expire, json.dumps(value))
                        RedisUtils.setex(redis_instance, key, expire, json.dumps(value), auto_free=False)
                    except:
                        logging.error("更正L2数据出错：{} key:{}".format(code, key))
                else:
                    json_value = json.loads(value)
                    if json_value["re"] != _data["re"]:
                        json_value["re"] = _data["re"]
                        redis_instance.setex(key, expire, json.dumps(json_value))
                        RedisUtils.setex(redis_instance, key, expire, json.dumps(json_value), auto_free=False)
    finally:
        redis_instance.delete("l2-save-{}".format(code))
        RedisUtils.delete(redis_instance, "l2-save-{}".format(code), auto_free=False)
        RedisUtils.realse(redis_instance)

    print("保存新数据用时：", msg, "耗时：{}".format(round(time.time() * 1000) - start_time))
    return datas
@@ -124,28 +228,27 @@

# 保存l2数据
def save_l2_data(code, datas, add_datas):
    redis = _redisManager.getRedis()
    # 只有有新曾数据才需要保存
    if len(add_datas) > 0:
    if add_datas:
        # 保存最近的数据
        __start_time = round(time.time() * 1000)
        redis.setex("l2-data-latest-{}".format(code), tool.get_expire(), json.dumps(datas))
        l2_data_log.l2_time(code, round(time.time() * 1000) - __start_time, "保存最近l2数据用时")
        # 设置进内存
        local_latest_datas[code] = datas
        set_l2_data_latest_count(code, len(datas))
        if datas:
            RedisUtils.setex_async(__db, "l2-data-latest-{}".format(code), tool.get_expire(),
                                   json.dumps(datas))
            # l2_data_log.l2_time(code, round(time.time() * 1000) - __start_time, "保存最近l2数据用时")
            # 设置进内存
            local_latest_datas[code] = datas
            set_l2_data_latest_count(code, len(datas))
        try:
            log.logger_l2_data.info("{}-{}", code, add_datas)
            async_log_util.l2_data_log.info(log.logger_l2_data, f"{code}-{add_datas}")
        except Exception as e:
            logging.exception(e)
        saveL2Data(code, add_datas)


# 设置最新的l2数据采集的数量
def set_l2_data_latest_count(code, count):
    redis = _redisManager.getRedis()
    key = "latest-l2-count-{}".format(code)
    redis.setex(key, 2, count)
    RedisUtils.setex(_redisManager.getRedis(), key, 2, count)
    pass


@@ -153,10 +256,9 @@
def get_l2_data_latest_count(code):
    if code is None or len(code) < 1:
        return 0
    redis = _redisManager.getRedis()
    key = "latest-l2-count-{}".format(code)

    result = redis.get(key)
    result = RedisUtils.get(_redisManager.getRedis(), key)
    if result is None:
        return 0
    else:
@@ -175,7 +277,7 @@
    count = data["count"]
    data = data["data"]
    # 获取涨停价
    return day, client, channel, code, capture_time, process_time, data,count
    return day, client, channel, code, capture_time, process_time, data, count


# 元数据是否有差异
@@ -193,33 +295,12 @@
    return False


# 是否为大单
def is_big_money(val):
    price = float(val["price"])
    money = price * int(val["num"])
    if price > 3.0:
        if money >= 30000:
            return True
        else:
            return False
    else:
        max_money = price * 10000
        if money >= max_money * 0.95:
            return True
        else:
            return False


class L2DataUtil:
    @classmethod
    def is_same_time(cls, time1, time2):
        if constant.TEST:
            return True
        time1_s = time1.split(":")
        time1_second = int(time1_s[0]) * 3600 + int(time1_s[1]) * 60 + int(time1_s[2])
        time2_s = time2.split(":")
        time2_second = int(time2_s[0]) * 3600 + int(time2_s[1]) * 60 + int(time2_s[2])
        if abs(time2_second - time1_second) < 3:
        if abs(tool.trade_time_sub(time1, time2)) < 3:
            return True
        else:
            return False
@@ -280,7 +361,8 @@
                    # 保存到数据库，更新re的数据
                    save_list.append(_ldata)
        if len(save_list) > 0:
            saveL2Data(code, save_list, "保存纠正数据")
            # 暂时不将数据保存到redis
            # saveL2Data(code, save_list, "保存纠正数据")
            local_latest_datas[code] = latest_data
        return _datas

@@ -324,7 +406,7 @@
                # 数据重复次数默认为1
                datas.append({"key": key, "val": item, "re": 1})
                dataIndexs.setdefault(key, len(datas) - 1)
        # TODO 测试的时候开启，方便记录大单数据
        # 测试的时候开启，方便记录大单数据
        # l2_data_util.save_big_data(code, same_time_num, data)
        return datas

@@ -361,11 +443,16 @@

        if int(val["operateType"]) != 2:
            return False
        return True

        price = float(val["price"])
        num = int(val["num"])
        # if price * num * 100 < 50 * 10000:
        #     return False
    # 涨停卖撤
    @classmethod
    def is_limit_up_price_sell_cancel(cls, val):
        if int(val["limitPrice"]) != 1:
            return False

        if int(val["operateType"]) != 3:
            return False
        return True

    # 是否涨停买撤
@@ -377,10 +464,12 @@
        if int(val["operateType"]) != 1:
            return False

        price = float(val["price"])
        num = int(val["num"])
        # if price * num * 100 < 50 * 10000:
        #     return False
        return True

    @classmethod
    def is_buy_cancel(cls, val):
        if int(val["operateType"]) != 1:
            return False
        return True

    # 是否卖撤
@@ -397,6 +486,26 @@
            return True
        return False

    # 是否为买
    @classmethod
    def is_buy(cls, val):
        if int(val["operateType"]) == 0:
            return True
        return False

    # l2时间差值
    @classmethod
    def time_sub_as_ms(cls, val1, val2):
        # 计算时间差值
        sub_s = tool.trade_time_sub(val1["time"], val2["time"])
        sub_ms = int(val1["tms"]) - int(val2["tms"])
        fs = sub_s * 1000 + sub_ms
        return fs

    @classmethod
    def get_time_with_ms(cls, val):
        return val["time"] + "." + "{:0>3}".format(int(val["tms"]))


class L2TradeQueueUtils(object):
    # 买入数据是否已撤
@@ -408,8 +517,9 @@
        # 是否有买撤数据
        if cancel_datas:
            for cancel_data in cancel_datas:
                buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data(code, cancel_data,
                                                                                                 local_today_num_operate_map)
                buy_index = l2_data_source_util.L2DataSourceUtils.get_buy_index_with_cancel_data_v2(cancel_data,
                                                                                                    local_today_buyno_map.get(
                                                                                                        code))
                if buy_index == data["index"]:
                    return True
        return False
@@ -452,6 +562,10 @@
                start_index = len(temp_str.split(","))
                return new_index_list[start_index], new_index_list[start_index:start_index + len(queues)]
            return None, None

        # 3个数据以上的不需要判断最近的一次未涨停时间
        if len(queueList) >= 3:
            latest_not_limit_up_time = None

        # 判断匹配的位置是否可信
        def is_trust(indexes):
@@ -506,4 +620,4 @@


if __name__ == "__main__":
    print(load_l2_data("002235"))
    print(L2DataUtil.time_sub_as_ms({"time": "09:46:05", "tms": 480}, {"time": "09:46:04", "tms": 90}))