系统日志记录

Administrator

2023-08-28 bb868482186f05f70e92dd17e57c80e98bd7d09f

系统日志记录

 db/redis_manager_delegate.py

@@ -8,7 +8,7 @@
import redis

import constant
from log_module.log import logger_redis_debug
from log_module.log import logger_redis_debug, logger_system
from utils import tool, middle_api_protocol

config = constant.REDIS_CONFIG
@@ -172,6 +172,7 @@
    # 运行异步任务
    @classmethod
    def run_loop(cls):
        logger_system.info("启动Redis数据同步服务")
        while True:
            try:
                data = cls.__async_task_queue.get()

 huaxin_client/l1_client.py

@@ -146,12 +146,13 @@
    for i in range(15):
        try:
            codes_sh, codes_sz = l1_subscript_codes_manager.get_codes()

            logger_local_huaxin_l1.info(f"获取上证，深证代码数量：sh-{len(codes_sh)} sz-{len(codes_sz)}")
            break
        except Exception as e:
            logger_local_huaxin_l1.exception(e)
            time.sleep(4)

    logger_system.info(f"获取L1订阅目标票数量：sh-{len(codes_sh)} sz-{len(codes_sz)}")
    # 打印接口版本号
    print(xmdapi.CTORATstpXMdApi_GetApiVersion())

@@ -176,6 +177,8 @@
    # 启动接口
    api.Init()

    logger_system.info("L1订阅服务启动成功")

    # 等待程序结束
    while True:
        print("数量", len(level1_data_dict))

 huaxin_client/l1_subscript_codes_manager.py

@@ -88,10 +88,10 @@


def get_codes():
    # codes_sh, codes_sz = get_codes_from_file()
    # if not codes_sh or not codes_sz:
    return request_l1_subscript_target_codes()
    # return codes_sh, codes_sz
    codes_sh, codes_sz = get_codes_from_file()
    if not codes_sh or not codes_sz:
        return request_l1_subscript_target_codes()
    return codes_sh, codes_sz


if __name__ == '__main__':

 huaxin_client/l2_client.py

@@ -577,7 +577,7 @@
    global l2CommandManager
    l2CommandManager = command_manager.L2CommandManager()
    l2CommandManager.init(MyL2ActionCallback())
    print("l2_client启动成功")
    logger_system.info("l2_client启动成功")
    while True:
        time.sleep(2)


 huaxin_client/trade_client.py

@@ -1034,7 +1034,7 @@
    tradeCommandManager = command_manager.TradeCommandManager()
    tradeCommandManager.init(MyTradeActionCallback(), l2pipe, pipe_strategy)
    tradeCommandManager.run()
    print("trade_client启动成功")
    logger_system.info("trade_client启动成功")
    while True:
        time.sleep(2)


 log_module/log_export.py

@@ -323,6 +323,22 @@
    return fdatas


# 读取系统日志
def load_system_log():
    path = f"{constant.get_path_prefix()}/logs/gp/system/system.{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:
                if line:
                    time_str = line.split("|")[0].strip()
                    level = line.split("|")[1].strip()
                    data = line.split("|")[2].split(" - ")[1].strip()
                    fdatas.append((time_str, level, data))
    return fdatas


if __name__ == '__main__':
    load_huaxin_local_buy_no()
    # print(get_h_cancel_compute_info("603912"))

 main.py

@@ -47,7 +47,7 @@

# 主服务
def createServer(pipe):
    print("create Server")
    logger_system.info("create Server")
    laddr = "", 9001
    tcpserver = MyThreadingTCPServer(laddr, MyBaseRequestHandle, pipe_trade=pipe)  # 注意：参数是MyBaseRequestHandle
    try:
@@ -58,7 +58,7 @@


def createDataServer():
    print("create DataServer")
    logger_system.info("create DataServer")
    tcpserver = data_server.run("", 9004)
    tcpserver.serve_forever()
    try:
@@ -70,6 +70,7 @@

if __name__ == '__main__':
    logger_l2_trade.info("启动程序")
    logger_system.info("启动程序")
    log.close_print()
    # 策略与server间的通信
    pss_server, pss_strategy = multiprocessing.Pipe()

 outside_api_command_manager.py

@@ -52,6 +52,7 @@
API_TYPE_CODE_TRADE_STATE = "code_trade_state"  # 代码交易状态
API_TYPE_GET_ENV = "get_env"  # 获取环境信息
API_TYPE_SYNC_L1_TARGET_CODES = "sync_l1_subscript_codes"  # 同步L1需要订阅的代码
API_TYPE_SYSTEM_LOG = "system_log"  # 系统日志


class ActionCallback(object):
@@ -187,6 +188,8 @@
                            cls.action_callback.OnGetEnvInfo(client_id, request_id, data)
                        elif content_type == API_TYPE_SYNC_L1_TARGET_CODES:
                            cls.action_callback.OnSyncL2SubscriptCodes(client_id, request_id)
                        elif content_type == API_TYPE_SYSTEM_LOG:
                            cls.action_callback.OnSystemLog(client_id, request_id, data)
                    except Exception as e:
                        logging.exception(e)
                        pass

 trade/huaxin/huaxin_trade_data_update.py

@@ -6,7 +6,7 @@
import threading
import time

from log_module.log import hx_logger_trade_debug
from log_module.log import hx_logger_trade_debug, logger_system
from trade import trade_huaxin, trade_manager
from trade.huaxin import huaxin_trade_api, huaxin_trade_record_manager
from trade.huaxin.huaxin_trade_record_manager import TradeOrderIdManager
@@ -16,6 +16,7 @@


def __read_trade_data_queue():
    logger_system.info("启动读取交易数据更新队列")
    while True:
        try:
            data = trade_data_request_queue.get()

 trade/huaxin/trade_api_server.py

@@ -14,6 +14,7 @@
from db.redis_manager_delegate import RedisUtils
from l2 import l2_data_manager_new
from l2.huaxin import huaxin_target_codes_manager
from log_module.log import logger_system
from third_data import block_info
from third_data.history_k_data_util import HistoryKDatasUtils, JueJinApi
from third_data.kpl_data_manager import KPLDataManager
@@ -420,6 +421,7 @@


def __set_target_codes(pipe_l2):
    logger_system.info("启动读取L2订阅代码队列")
    while True:
        try:
            _datas = huaxin_target_codes_manager.HuaXinL2SubscriptCodesManager.pop()
@@ -446,6 +448,7 @@


def __read_sync_task(pipe):
    logger_system.info("启动读取数据同步服务")
    while True:
        try:
            if pipe:
@@ -491,7 +494,7 @@


def run(pipe_server, pipe_l2):
    print("create TradeApiServer")
    logger_system.info("create TradeApiServer")
    # 拉取交易信息
    huaxin_trade_data_update.run()
    #

 trade/huaxin/trade_server.py

@@ -29,7 +29,7 @@
from l2.huaxin import huaxin_target_codes_manager
from l2.huaxin.huaxin_target_codes_manager import HuaXinL1TargetCodesManager
from l2.l2_data_manager_new import L2TradeDataProcessor
from log_module import async_log_util
from log_module import async_log_util, log_export
from log_module.log import hx_logger_l2_upload, hx_logger_contact_debug, hx_logger_trade_callback, \
    hx_logger_l2_orderdetail, hx_logger_l2_transaction, hx_logger_l2_market_data, logger_l2_trade_buy_queue, \
    logger_l2_g_cancel, logger_debug, logger_system
@@ -362,7 +362,8 @@
                            is_normal = l2_data_util.load_l2_data(code, load_latest=False)
                            volume_rate = code_volumn_manager.get_volume_rate(code)
                            volume_rate_index = code_volumn_manager.get_volume_rate_index(volume_rate)
                            m_val = L2PlaceOrderParamsManager(code, True, volume_rate, volume_rate_index, None).get_m_val()[0]
                            m_val = \
                            L2PlaceOrderParamsManager(code, True, volume_rate, volume_rate_index, None).get_m_val()[0]
                            limit_up_price = gpcode_manager.get_limit_up_price(code)
                            m_val_num = int(m_val / (float(limit_up_price) * 100))

@@ -734,6 +735,21 @@
        except Exception as e:
            logger_debug.error(e)

    def OnSystemLog(self, client_id, request_id, data):
        logger_debug.debug("OnSystemLog")
        try:
            start_index = data["start_index"]
            count = data["count"]
            # 读取系统日志
            logs_data = log_export.load_system_log()
            total_count = len(logs_data)
            if start_index > 0:
                logs_data = logs_data[start_index:start_index+count]
            result = {"code": 0, "data": {"total_count": total_count, "list": logs_data}}
            self.send_response(result, client_id, request_id)
        except Exception as e:
            logger_debug.error(e)


def run(pipe_trade, pipe_l1):
    # 执行一些初始化数据
@@ -757,7 +773,7 @@
    t1 = threading.Thread(target=lambda: async_log_util.run_sync(), daemon=True)
    t1.start()

    print("create TradeServer")
    logger_system.info("create TradeServer")
    t1 = threading.Thread(target=lambda: clear_invalid_client(), daemon=True)
    t1.start()