日志优化

Administrator

2023-08-02 b069d586b7ed4fbefd9bac22ae796f185962e7fe

 log_module/log.py

@@ -1,16 +1,9 @@
"""
日志
"""
import datetime
import os
import shutil
import sys

from loguru import logger

import constant
from code_attribute import gpcode_manager
from utils import tool


class MyLogger:
@@ -287,314 +280,3 @@
hx_logger_trade_callback = __mylogger.get_logger("hx_trade_callback")
hx_logger_trade_debug = __mylogger.get_logger("hx_trade_debug")
hx_logger_trade_loop = __mylogger.get_logger("hx_trade_loop")


class LogUtil:
    @classmethod
    def extract_log_from_key(cls, key, path, target_path):
        fw = open(target_path, mode='w', encoding="utf-8")
        try:
            with open(path, 'r', encoding="utf-8") as f:
                lines = f.readlines()
                for line in lines:
                    if line.find("{}".format(key)) > 0:
                        fw.write(line)
        finally:
            fw.close()


# 导出数据处理位置日志
def __export_l2_pos_range(code, date, dir):
    LogUtil.extract_log_from_key("{} 处理数据范围".format(code),
                                 "{}/logs/gp/l2/l2_process.{}.log".format(constant.get_path_prefix(), date),
                                 "{}/l2_process_{}.log".format(dir, date))


# 导出交易日志
def __export_l2_trade_log(code, date, dir):
    LogUtil.extract_log_from_key(code, "{}/logs/gp/l2/l2_trade.{}.log".format(constant.get_path_prefix(), date),
                                 "{}/l2_trade_{}.log".format(dir, date))


# 导出交易取消日志
def __export_l2_trade_cancel_log(code, date, dir):
    LogUtil.extract_log_from_key(code, "{}/logs/gp/l2/l2_trade_cancel.{}.log".format(constant.get_path_prefix(), date),
                                 "{}/l2_trade_cancel_{}.log".format(dir, date))


def __analyse_pricess_time():
    date = datetime.datetime.now().strftime("%Y-%m-%d")
    file_path = f"{constant.get_path_prefix()}/logs/gp/l2/l2_process.{date}.log"
    with open(file_path, encoding="utf-8") as f:
        line = f.readline()
        while line:
            time_ = line.split(":")[-1]
            if int(time_) > 150:
                print(line)
            line = f.readline()


def export_l2_log(code):
    if len(code) < 6:
        return
    date = datetime.datetime.now().strftime("%Y-%m-%d")
    dir_ = "{}/logs/gp/l2/{}".format(constant.get_path_prefix(), code)
    if not os.path.exists(dir_):
        os.mkdir(dir_)
    __export_l2_pos_range(code, date, dir_)
    __export_l2_trade_cancel_log(code, date, dir_)
    __export_l2_trade_log(code, date, dir_)


def compute_buy1_real_time(time_):
    ts = time_.split(":")
    s = int(ts[0]) * 3600 + int(ts[1]) * 60 + int(ts[2])
    cha = (s - 2) % 3
    return tool.time_seconds_format(s - 2 - cha)


def load_l2_from_log(date=None):
    today_data = {}
    if date is None:
        date = datetime.datetime.now().strftime("%Y-%m-%d")
    try:
        with open("{}/logs/gp/l2/l2_data.{}.log".format(constant.get_path_prefix(), date), mode='r') as f:
            while True:
                data = f.readline()
                if not data:
                    break
                index = data.find('save_l2_data:')
                index = data.find('-', index)
                data = data[index + 1:].strip()
                code = data[0:6]
                data = data[7:]
                dict_ = eval(data)
                if code not in today_data:
                    today_data[code] = dict_
                else:
                    today_data[code].extend(dict_)
        for key in today_data:
            news = sorted(today_data[key], key=lambda x: x["index"])
            today_data[key] = news
            print(key, len(today_data[key]) - 1, today_data[key][-1]["index"])
    except:
        pass
    return today_data


# 获取日志时间
def __get_log_time(line):
    time_ = line.split("|")[0].split(" ")[1].split(".")[0]
    return time_


# 获取L2每次批量处理数据的位置范围
def get_l2_process_position(code, date=None):
    if not date:
        date = datetime.datetime.now().strftime("%Y-%m-%d")
    pos_list = []
    with open("{}/logs/gp/l2/l2_process.{}.log".format(constant.get_path_prefix(), date), mode='r',
              encoding="utf-8") as f:
        while True:
            line = f.readline()
            if not line:
                break
            if line.find("code:{}".format(code)) < 0:
                continue
            time_ = __get_log_time(line)
            line = line[line.find("处理数据范围") + len("处理数据范围") + 1:line.find("处理时间")].strip()
            if len(pos_list) == 0 or pos_list[-1][1] < int(line.split("-")[0]):
                if int("093000") <= int(time_.replace(":", "")) <= int("150000"):
                    pos_list.append((int(line.split("-")[0]), int(line.split("-")[1])))
    return pos_list


# 获取L2每次批量处理数据的位置范围
def get_l2_trade_position(code, date=None):
    if not date:
        date = datetime.datetime.now().strftime("%Y-%m-%d")
    pos_list = []
    with open("{}/logs/gp/l2/l2_trade.{}.log".format(constant.get_path_prefix(), date), mode='r',
              encoding="utf-8") as f:
        while True:
            line = f.readline()
            if not line:
                break
            if line.find("code={}".format(code)) < 0:
                continue
            print(line)
            time_ = __get_log_time(line)
            if int("093000") > int(time_.replace(":", "")) or int(time_.replace(":", "")) > int("150000"):
                continue

            if line.find("获取到买入信号起始点") > 0:
                str_ = line.split("获取到买入信号起始点：")[1].strip()
                index = str_[0:str_.find(" ")].strip()
                # print("信号起始位置：", index)
                pos_list.append((0, int(index), ""))

            elif line.find("获取到买入执行位置") > 0:
                str_ = line.split("获取到买入执行位置：")[1].strip()
                index = str_[0:str_.find(" ")].strip()
                # print("买入执行位置：", index)
                pos_list.append((1, int(index), ""))
            elif line.find("触发撤单") > 0:
                str_ = line.split("触发撤单，撤单位置：")[1].strip()
                index = str_[0:str_.find(" ")].strip()
                # print("撤单位置：", index)
                pos_list.append((2, int(index), line.split("撤单原因：")[1]))
                pass
            else:
                continue
    return pos_list


# 获取交易进度
def get_trade_progress(code, date=None):
    if not date:
        date = datetime.datetime.now().strftime("%Y-%m-%d")
    index_list = []
    buy_queues = []
    with open("{}/logs/gp/l2/l2_trade_buy_queue.{}.log".format(constant.get_path_prefix(), date), mode='r',
              encoding="utf-8") as f:
        while True:
            line = f.readline()
            if not line:
                break
            time_ = __get_log_time(line).strip()
            if int(time_.replace(":", "")) > int("150000"):
                continue

            if line.find(f"{code}-[") >= 0:
                buy_queues.append((eval(line.split(f"{code}-")[1]), time_))

            if line.find("获取成交位置成功： code-{}".format(code)) < 0:
                continue
            try:
                index = int(line.split("index-")[1].split(" ")[0])
                index_list.append((index, time_))
            except:
                pass
    return index_list, buy_queues


# 获取H级撤单计算结果
def get_h_cancel_compute_info(code, date=None):
    if not date:
        date = datetime.datetime.now().strftime("%Y-%m-%d")
    path_str = f"{constant.get_path_prefix()}/logs/gp/l2/cancel/h_cancel.{date}.log"
    latest_info = None
    if os.path.exists(path_str):
        with open(path_str, mode='r', encoding="utf-8") as f:
            while True:
                line = f.readline()
                if not line:
                    break
                if line.find(f"code-{code}") < 0:
                    continue
                if line.find(f"H级撤单计算结果") < 0:
                    continue
                target_rate = line.split("目标比例：")[1].split(" ")[0].strip()
                cancel_num = line.split("取消计算结果")[1][1:].split("/")[0].strip()
                total_num = line.split("取消计算结果")[1][1:].split("/")[1].split(" ")[0].strip()
                latest_info = (target_rate, round(int(cancel_num) / int(total_num), 2), cancel_num, total_num,)
    return latest_info


# 读取看盘消息
def get_kp_msg_list(date=None):
    if not date:
        date = datetime.datetime.now().strftime("%Y-%m-%d")
    path_str = f"{constant.get_path_prefix()}/logs/gp/kp/kp_msg.{date}.log"
    msg_list = []
    if os.path.exists(path_str):
        with open(path_str, mode='r', encoding="utf-8") as f:
            while True:
                line = f.readline()
                if not line:
                    break
                msg_list.append(line)
    return msg_list


def export_logs(code):
    code_name = gpcode_manager.get_code_name(code)
    date = datetime.datetime.now().strftime("%Y-%m-%d")
    target_dir = f"{constant.get_path_prefix()}/logs/gp/l2/export/{code}_{code_name}_{date}"
    if os.path.exists(target_dir):
        shutil.rmtree(target_dir)
    os.makedirs(target_dir)
    log_names = ["l2_process", "l2_trade", "l2_trade_cancel", "l2_process_time", "l2_trade_buy",
                 "l2_trade_buy_progress", "cancel/h_cancel"]
    # 导出交易日志
    for log_name in log_names:
        key = f"code={code}"
        if log_name == "l2_process" or log_name == "l2_process_time" or log_name == "cancel/h_cancel" or log_name == "l2_trade_buy_progress":
            key = code
        target_path = f"{target_dir}/{log_name}.{code}_{code_name}.{date}.log"
        # 创建文件夹
        dir_path = "/".join(target_path.split("/")[:-1])
        if not os.path.exists(dir_path):
            os.makedirs(dir_path)
        LogUtil.extract_log_from_key(key, f"{constant.get_path_prefix()}/logs/gp/l2/{log_name}.{date}.log",
                                     target_path)


def export_trade_progress(code):
    path = f"{constant.get_path_prefix()}/logs/gp/l2/l2_trade_buy_progress.{tool.get_now_date_str()}.log"
    index_set = set()
    with open(path, mode='r', encoding="utf-8") as f:
        lines = f.readlines()
        for line in lines:
            if line.find(f"code-{code}") > -1 and line.find("确定交易进度成功") > -1:
                index = line.split("index-")[1].split(" ")[0]
                index_set.add(int(index))
    results = list(index_set)
    results.sort()
    return results


# 加载买入得分记录
def load_buy_score_recod(code):
    path = f"{constant.get_path_prefix()}/logs/gp/trade/trade_record.{tool.get_now_date_str()}.log"
    fdatas = []
    if os.path.exists(path):
        with open(path, 'r', encoding="utf-8") as f:
            lines = f.readlines()
            for line in lines:
                data_index = line.find(f"code={code}")
                if data_index > 0:
                    time_str = line[11:19]
                    data = line[line.find("data=") + 5:]
                    type = line[line.find("type=") + 5:line.find(" ", line.find("type="))]
                    fdatas.append((time_str, type, eval("{" + data + "}")))
    return fdatas


def load_kpl_reason_changes():
    path = f"{constant.get_path_prefix()}/logs/gp/kpl/kpl_limit_up_reason_change.{tool.get_now_date_str()}.log"
    fdatas = []
    if os.path.exists(path):
        with open(path, 'r', encoding="utf-8") as f:
            lines = f.readlines()
            for line in lines:
                data = line[line.find("code-") + 5:]
                code = data.split(":")[0]
                from_r = data.split(":")[1].split("-")[0]
                to_r = eval(data.split(":")[1].split("-")[1])
                fdatas.append((code, from_r, to_r))
    return fdatas


if __name__ == '__main__':
    logger_l2_process_time.info("test123")
    # print(get_h_cancel_compute_info("603912"))

    # logger_l2_h_cancel.info("test")
    # logger_l2_process_time.info("test123")
    # logger_buy_score.info("测试")
    # codes = ["603083"]
    # for code in codes:
    #     export_logs(code)

    # parse_l2_data()