bug修复/日志添加

admin

2024-01-08 1110af9cc42cbf6a3ebbb953f18585cb37ba5b8c

 middle_api_server.py

@@ -1,6 +1,7 @@
import hashlib
import json
import logging
import random
import socket
import socketserver
import threading
@@ -11,6 +12,7 @@
import trade_manager
from db import mysql_data, redis_manager
from db.redis_manager import RedisUtils
from log import logger_request_debug
from utils import socket_util, hosting_api_util, huaxin_trade_record_manager, huaxin_util, tool, global_data_cache_util
from utils.history_k_data_util import HistoryKDatasUtils, JueJinApi
from utils.huaxin_trade_record_manager import PositionManager
@@ -63,253 +65,278 @@
                    # print("收到数据------", f"{data_str[:20]}......{data_str[-20:]}")
                    data_json = json.loads(data_str)
                    type_ = data_json['type']
                    if type(type_) == int:
                        # 处理数字型TYPE
                        return_str = self.process_num_type(sk, type_, data_str)
                        break
                    thread_id = random.randint(0, 1000000)
                    try:
                        logger_request_debug.info(f"middle_api_server 请求开始({thread_id})：{type_}")
                        if type(type_) == int:
                            # 处理数字型TYPE
                            return_str = self.process_num_type(sk, type_, data_str)
                            break

                    is_sign_right = socket_util.is_client_params_sign_right(data_json)
                    # ------客户端请求接口-------
                    if type_ == 'buy':
                        # 验证签名
                        if not is_sign_right:
                            raise Exception("签名错误")
                        codes_data = data_json["data"]
                        code = codes_data["code"]
                        volume = codes_data["volume"]
                        price = codes_data["price"]
                        try:
                            if not code:
                                raise Exception("请上传code")
                            if not volume:
                                raise Exception("请上传volume")
                        is_sign_right = socket_util.is_client_params_sign_right(data_json)
                        # ------客户端请求接口-------
                        if type_ == 'buy':
                            # 验证签名
                            if not is_sign_right:
                                raise Exception("签名错误")
                            codes_data = data_json["data"]
                            code = codes_data["code"]
                            volume = codes_data["volume"]
                            price = codes_data["price"]
                            try:
                                if not code:
                                    raise Exception("请上传code")
                                if not volume:
                                    raise Exception("请上传volume")

                            if round(float(price), 2) <= 0:
                                prices = HistoryKDatasUtils.get_now_price([code])
                                if not prices:
                                    raise Exception("现价获取失败")
                                price = prices[0][1]
                            # 下单
                            result = hosting_api_util.trade_order(hosting_api_util.TRADE_DIRECTION_BUY, code, volume,
                                                                  round(float(price), 2))
                            if result:
                                resultJSON = result
                                print("下单结果：", resultJSON)
                                if resultJSON['code'] == 0:
                                if round(float(price), 2) <= 0:
                                    prices = HistoryKDatasUtils.get_now_price([code])
                                    if not prices:
                                        raise Exception("现价获取失败")
                                    price = prices[0][1]
                                # 下单
                                result = hosting_api_util.trade_order(hosting_api_util.TRADE_DIRECTION_BUY, code, volume,
                                                                      round(float(price), 2))
                                if result:
                                    resultJSON = result
                                    print("下单结果：", resultJSON)
                                    if resultJSON['code'] == 0:
                                        return_str = json.dumps({"code": 0})
                                    else:
                                        raise Exception(resultJSON['msg'])
                                break
                            except Exception as e:
                                raise e
                        elif type_ == 'cancel_order':
                            # 验证签名
                            if not is_sign_right:
                                raise Exception("签名错误")
                            codes_data = data_json["data"]
                            code = codes_data["code"]
                            orderSysID = codes_data.get("orderSysID")
                            accountId = codes_data.get("accountId")
                            if code:
                                result = hosting_api_util.trade_cancel_order(hosting_api_util.TRADE_DIRECTION_BUY, code,
                                                                             accountId,
                                                                             orderSysID, True)
                                print("---撤单结果----")
                                print(result)
                                if result["code"] == 0:
                                    return_str = json.dumps({"code": 0})
                                else:
                                    raise Exception(resultJSON['msg'])
                                    raise Exception(result["msg"])
                            else:
                                return_str = json.dumps({"code": 1, "msg": "请上传代码"})
                            break
                        except Exception as e:
                            raise e
                    elif type_ == 'cancel_order':
                        # 验证签名
                        if not is_sign_right:
                            raise Exception("签名错误")
                        codes_data = data_json["data"]
                        code = codes_data["code"]
                        orderSysID = codes_data.get("orderSysID")
                        accountId = codes_data.get("accountId")
                        if code:
                            result = hosting_api_util.trade_cancel_order(hosting_api_util.TRADE_DIRECTION_BUY, code,
                                                                         accountId,
                                                                         orderSysID, True)
                            print("---撤单结果----")
                            print(result)

                        elif type_ == 'sell':
                            # 验证签名
                            if not is_sign_right:
                                raise Exception("签名错误")
                            codes_data = data_json["data"]
                            code = codes_data["code"]
                            volume = codes_data["volume"]
                            price_type = codes_data["price_type"]
                            result = hosting_api_util.trade_order(hosting_api_util.TRADE_DIRECTION_SELL, code, volume,
                                                                  '', price_type=price_type)
                            if result["code"] == 0:
                                return_str = json.dumps({"code": 0})
                                return_str = json.dumps(result)
                            else:
                                raise Exception(result["msg"])
                        else:
                            return_str = json.dumps({"code": 1, "msg": "请上传代码"})
                        break

                    elif type_ == 'sell':
                        # 验证签名
                        if not is_sign_right:
                            raise Exception("签名错误")
                        codes_data = data_json["data"]
                        code = codes_data["code"]
                        volume = codes_data["volume"]
                        price_type = codes_data["price_type"]
                        result = hosting_api_util.trade_order(hosting_api_util.TRADE_DIRECTION_SELL, code, volume,
                                                              '',price_type=price_type)
                        if result["code"] == 0:
                            print("---卖出结果----")
                            print(result)
                            break
                        elif type_ == 'get_code_position_info':
                            # 验证签名
                            if not is_sign_right:
                                raise Exception("签名错误")
                            codes_data = data_json["data"]
                            code = codes_data["code"]
                            result = hosting_api_util.get_code_position_info(code)
                            return_str = json.dumps(result)
                        else:
                            raise Exception(result["msg"])
                        print("---卖出结果----")
                        print(result)
                        break
                    elif type_ == 'get_code_position_info':
                        # 验证签名
                        if not is_sign_right:
                            raise Exception("签名错误")
                        codes_data = data_json["data"]
                        code = codes_data["code"]
                        result = hosting_api_util.get_code_position_info(code)
                        return_str = json.dumps(result)
                        break
                            break

                    elif type_ == 'common':
                        # 验证签名
                        if not is_sign_right:
                            raise Exception("签名错误")
                        params = data_json["data"]
                        result = hosting_api_util.common_request(params)
                        return_str = json.dumps(result)
                        break
                        elif type_ == 'common':
                            # 验证签名
                            if not is_sign_right:
                                raise Exception("签名错误")
                            params = data_json["data"]
                            result = hosting_api_util.common_request(params)
                            return_str = json.dumps(result)
                            break

                    elif type_ == 'get_cost_price':
                        # 获取成本价
                        codes_data = data_json["data"]
                        code = codes_data["code"]
                        try:
                            price = PositionManager.get_cost_price(code)
                            return_str = json.dumps({"code": 0, "data": {"price": price}})
                        except Exception as e:
                            return_str = json.dumps({"code": 1, "msg": str(e)})
                        break
                    elif type_ == 'delegate_list':
                        # 委托列表
                        update_time = data_json["data"]["update_time"]
                        # 是否可撤 0/1
                        can_cancel = data_json["data"]["can_cancel"]
                        results, update_time = None, None
                        if can_cancel:
                            results, update_time = huaxin_trade_record_manager.DelegateRecordManager.list_by_day(
                                tool.get_now_date_str("%Y%m%d"), None,
                                [huaxin_util.TORA_TSTP_OST_Accepted, huaxin_util.TORA_TSTP_OST_PartTraded])
                        else:
                            results, update_time = huaxin_trade_record_manager.DelegateRecordManager.list_by_day(
                                tool.get_now_date_str("%Y%m%d"), update_time)
                        return_str = json.dumps(
                            {"code": 0, "data": {"list": results, "updateTime": update_time}, "msg": "请上传代码"})
                        break
                    elif type_ == 'deal_list':
                        # 成交列表
                        results = huaxin_trade_record_manager.DealRecordManager.list_by_day(
                            tool.get_now_date_str("%Y%m%d"))
                        return_str = json.dumps(
                            {"code": 0, "data": {"list": results}, "msg": ""})
                    elif type_ == 'position_list':
                        # 持仓股列表
                        results, update_time = huaxin_trade_record_manager.PositionManager.list_by_day(
                            tool.get_now_date_str("%Y%m%d"))
                        return_str = json.dumps(
                            {"code": 0, "data": {"list": results}, "msg": ""})
                    elif type_ == 'money_list':
                        # 资金详情
                        money_data = huaxin_trade_record_manager.MoneyManager.get_data()
                        return_str = json.dumps(
                            {"code": 0, "data": money_data, "msg": ""})
                    elif type_ == 'sync_trade_data':
                        # 同步交易数据
                        sync_type = data_json["data"]["type"]
                        hosting_api_util.refresh_trade_data(sync_type)
                        return_str = json.dumps(
                            {"code": 0, "data": {}, "msg": ""})
                    elif type_ == "get_huaxin_subscript_codes":
                        # 获取华鑫订阅的代码
                        fresults = global_data_cache_util.huaxin_subscript_codes
                        update_time = global_data_cache_util.huaxin_subscript_codes_update_time
                        if update_time is None:
                            update_time = ''
                        return_str = json.dumps(
                            {"code": 0, "data": {"count": len(fresults), "list": fresults, "update_time": update_time},
                             "msg": ""})
                        pass
                    elif type_ == "export_l2_data":
                        # 导出L2数据
                        code = data_json["data"]["code"]
                        hosting_api_util.export_l2_data(code)
                        return_str = json.dumps(
                            {"code": 0, "data": {}, "msg": ""})
                    elif type_ == 'everyday_init':
                        # 每日初始化
                        hosting_api_util.everyday_init()
                        return_str = json.dumps(
                            {"code": 0, "data": {}, "msg": ""})
                    elif type_ == 'huaxin_channel_state':
                        # 华鑫通道状态
                        types = []
                        fdata = {}
                        return_str = json.dumps(
                            {"code": 0, "data": fdata, "msg": ""})
                    elif type_ == 'juejin_is_valid':
                        # 掘金是否可用
                        try:
                            date = JueJinApi.get_previous_trading_date(tool.get_now_date_str())
                            if date:
                        elif type_ == 'get_cost_price':
                            # 获取成本价
                            codes_data = data_json["data"]
                            code = codes_data["code"]
                            try:
                                price = PositionManager.get_cost_price(code)
                                return_str = json.dumps({"code": 0, "data": {"price": price}})
                            except Exception as e:
                                return_str = json.dumps({"code": 1, "msg": str(e)})
                            break
                        elif type_ == 'delegate_list':
                            # 委托列表
                            update_time = data_json["data"]["update_time"]
                            # 是否可撤 0/1
                            can_cancel = data_json["data"]["can_cancel"]
                            results, update_time = None, None
                            if can_cancel:
                                results, update_time = huaxin_trade_record_manager.DelegateRecordManager.list_by_day(
                                    tool.get_now_date_str("%Y%m%d"), None,
                                    [huaxin_util.TORA_TSTP_OST_Accepted, huaxin_util.TORA_TSTP_OST_PartTraded])
                            else:
                                results, update_time = huaxin_trade_record_manager.DelegateRecordManager.list_by_day(
                                    tool.get_now_date_str("%Y%m%d"), update_time)
                            return_str = json.dumps(
                                {"code": 0, "data": {"list": results, "updateTime": update_time}, "msg": "请上传代码"})
                            break
                        elif type_ == 'deal_list':
                            # 成交列表
                            results = huaxin_trade_record_manager.DealRecordManager.list_by_day(
                                tool.get_now_date_str("%Y%m%d"))
                            return_str = json.dumps(
                                {"code": 0, "data": {"list": results}, "msg": ""})
                        elif type_ == 'position_list':
                            # 持仓股列表
                            results, update_time = huaxin_trade_record_manager.PositionManager.list_by_day(
                                tool.get_now_date_str("%Y%m%d"))
                            return_str = json.dumps(
                                {"code": 0, "data": {"list": results}, "msg": ""})
                        elif type_ == 'money_list':
                            # 资金详情
                            money_data = huaxin_trade_record_manager.MoneyManager.get_data()
                            return_str = json.dumps(
                                {"code": 0, "data": money_data, "msg": ""})
                        elif type_ == 'sync_trade_data':
                            # 同步交易数据
                            sync_type = data_json["data"]["type"]
                            hosting_api_util.refresh_trade_data(sync_type)
                            return_str = json.dumps(
                                {"code": 0, "data": {}, "msg": ""})
                        elif type_ == "get_huaxin_subscript_codes":
                            # 获取华鑫订阅的代码
                            fresults = global_data_cache_util.huaxin_subscript_codes
                            update_time = global_data_cache_util.huaxin_subscript_codes_update_time
                            if update_time is None:
                                update_time = ''
                            return_str = json.dumps(
                                {"code": 0, "data": {"count": len(fresults), "list": fresults, "update_time": update_time},
                                 "msg": ""})
                            pass
                        elif type_ == "export_l2_data":
                            # 导出L2数据
                            code = data_json["data"]["code"]
                            hosting_api_util.export_l2_data(code)
                            return_str = json.dumps(
                                {"code": 0, "data": {}, "msg": ""})
                        elif type_ == 'everyday_init':
                            # 每日初始化
                            hosting_api_util.everyday_init()
                            return_str = json.dumps(
                                {"code": 0, "data": {}, "msg": ""})
                        elif type_ == 'huaxin_channel_state':
                            # 华鑫通道状态
                            types = []
                            fdata = {}
                            return_str = json.dumps(
                                {"code": 0, "data": fdata, "msg": ""})
                        elif type_ == 'juejin_is_valid':
                            # 掘金是否可用
                            try:
                                date = JueJinApi.get_previous_trading_date(tool.get_now_date_str())
                                if date:
                                    return_str = json.dumps(
                                        {"code": 0, "msg": ""})
                            except Exception as e:
                                return_str = json.dumps(
                                    {"code": 0, "msg": ""})
                        except Exception as e:
                            return_str = json.dumps(
                                {"code": 0, "msg": str(e)})
                    elif type_ == 'get_env_info':
                        # 获取环境信息
                        result = hosting_api_util.get_env_info()
                        return_str = json.dumps(result)
                    elif type_ == 'sync_l1_subscript_codes':
                        # 获取环境信息
                        result = hosting_api_util.sync_l1_subscript_codes()
                        return_str = json.dumps(result)
                                    {"code": 0, "msg": str(e)})
                        elif type_ == 'get_env_info':
                            # 获取环境信息
                            result = hosting_api_util.get_env_info()
                            return_str = json.dumps(result)
                        elif type_ == 'sync_l1_subscript_codes':
                            # 获取环境信息
                            result = hosting_api_util.sync_l1_subscript_codes()
                            return_str = json.dumps(result)

                    elif type_ == 'get_system_logs':
                        # 获取环境信息
                        start_index = data_json["data"]["start_index"]
                        count = data_json["data"]["count"]
                        result = hosting_api_util.get_system_logs(start_index, count)
                        return_str = json.dumps(result)
                    elif type_ == 'test_redis':
                        redis = redis_manager.RedisManager(5).getRedisNoPool()
                        try:
                            _start_time = time.time()
                            times = []
                            for i in range(0, 100):
                                RedisUtils.sadd(redis, "test_set", f"000000:{i}", auto_free=False)
                            times.append(time.time() - _start_time)
                            _start_time = time.time()
                            for i in range(0, 20):
                                RedisUtils.smembers(redis, "test_set", auto_free=False)
                            times.append(time.time() - _start_time)
                            return_str = json.dumps(
                                {"code": 0, "data": times, "msg": ""})
                        finally:
                            redis.close()
                    elif type_ == 'get_code_trade_info':
                        # 获取环境信息
                        code = data_json["data"]["code"]
                        result = hosting_api_util.get_code_trade_info(code)
                        return_str = json.dumps(result)
                    elif type_ == 'get_l2_listen_active_count':
                        result = hosting_api_util.get_l2_listen_active_count()
                        return_str = json.dumps(result)
                    elif type_ == "trade_server_channels":
                        channels = socket_manager.ClientSocketManager.list_client()
                        return_str = json.dumps({"code": 0, "data": channels})
                    elif type_ == "save_running_data":
                        result = hosting_api_util.save_running_data()
                        return_str = json.dumps(result)
                    elif type_ == "add_sell_rule":
                        result = hosting_api_util.sell_rule(hosting_api_util.OPERRATE_ADD, data=data_json["data"])
                        return_str = json.dumps(result)
                    elif type_ == "del_sell_rule":
                        id_ = data_json["data"]["id"]
                        result = hosting_api_util.sell_rule(hosting_api_util.OPERRATE_DELETE, data={"id": id_})
                        return_str = json.dumps(result)
                    elif type_ == "list_sell_rule":
                        result = hosting_api_util.sell_rule(hosting_api_util.OPERRATE_GET, data={})
                        return_str = json.dumps(result)
                    elif type_ == "get_code_position_info":
                        code = data_json["data"]["code"]
                        result = hosting_api_util.get_code_position_info(code)
                        return_str = json.dumps(result)
                    elif type_ == "common":
                        params = data_json["data"]
                        result = hosting_api_util.common_request(params)
                        return_str = json.dumps(result)

                        elif type_ == 'get_system_logs':
                            # 获取环境信息
                            start_index = data_json["data"]["start_index"]
                            count = data_json["data"]["count"]
                            result = hosting_api_util.get_system_logs(start_index, count)
                            return_str = json.dumps(result)
                        elif type_ == 'test_redis':
                            redis = redis_manager.RedisManager(5).getRedisNoPool()
                            try:
                                _start_time = time.time()
                                times = []
                                for i in range(0, 100):
                                    RedisUtils.sadd(redis, "test_set", f"000000:{i}", auto_free=False)
                                times.append(time.time() - _start_time)
                                _start_time = time.time()
                                for i in range(0, 20):
                                    RedisUtils.smembers(redis, "test_set", auto_free=False)
                                times.append(time.time() - _start_time)
                                return_str = json.dumps(
                                    {"code": 0, "data": times, "msg": ""})
                            finally:
                                redis.close()
                        elif type_ == 'get_code_trade_info':
                            # 获取环境信息
                            code = data_json["data"]["code"]
                            result = hosting_api_util.get_code_trade_info(code)
                            return_str = json.dumps(result)
                        elif type_ == 'get_l2_listen_active_count':
                            result = hosting_api_util.get_l2_listen_active_count()
                            return_str = json.dumps(result)
                        elif type_ == "trade_server_channels":
                            trade_channels = socket_manager.ClientSocketManager.list_client(
                                socket_manager.ClientSocketManager.CLIENT_TYPE_TRADE)
                            common_channels = socket_manager.ClientSocketManager.list_client(
                                socket_manager.ClientSocketManager.CLIENT_TYPE_COMMON)
                            data = {}
                            available_count = 0
                            active_count = 0
                            now_time_str = tool.get_now_time_str()
                            for t in trade_channels:
                                if not t[1]:
                                    available_count += 1
                                if tool.trade_time_sub(now_time_str, t[2]) < 60:
                                    active_count += 1
                            data["trade"] = (len(trade_channels), available_count, active_count)
                            available_count = 0
                            active_count = 0
                            for t in common_channels:
                                if not t[1]:
                                    available_count += 1
                                if tool.trade_time_sub(now_time_str, t[2]) < 60:
                                    active_count += 1
                            data["common"] = (len(common_channels), available_count, active_count)
                            return_str = json.dumps({"code": 0, "data": data})
                        elif type_ == "save_running_data":
                            result = hosting_api_util.save_running_data()
                            return_str = json.dumps(result)
                        elif type_ == "add_sell_rule":
                            result = hosting_api_util.sell_rule(hosting_api_util.OPERRATE_ADD, data=data_json["data"])
                            return_str = json.dumps(result)
                        elif type_ == "del_sell_rule":
                            id_ = data_json["data"]["id"]
                            result = hosting_api_util.sell_rule(hosting_api_util.OPERRATE_DELETE, data={"id": id_})
                            return_str = json.dumps(result)
                        elif type_ == "list_sell_rule":
                            result = hosting_api_util.sell_rule(hosting_api_util.OPERRATE_GET, data={})
                            return_str = json.dumps(result)
                        elif type_ == "get_code_position_info":
                            code = data_json["data"]["code"]
                            result = hosting_api_util.get_code_position_info(code)
                            return_str = json.dumps(result)
                        elif type_ == "common":
                            params = data_json["data"]
                            result = hosting_api_util.common_request(params)
                            return_str = json.dumps(result)
                    finally:
                        logger_request_debug.info(f"middle_api_server 请求结束({thread_id})：{type}")
                break
                # sk.close()
            except Exception as e:
@@ -420,6 +447,28 @@
                else:
                    return_str = json.dumps({"code": 1, "msg": "不可以取消"})

            elif type == 421:
                # 加入暂不买
                data = json.loads(_str)
                codes = data["data"]["codes"]
                for code in codes:
                    hosting_api_util.add_code_list(code, hosting_api_util.CODE_LIST_MUST_BUY)
                return_str = json.dumps({"code": 0})

            elif type == 422:
                # 移除暂不买
                data = json.loads(_str)
                codes = data["data"]["codes"]
                for code in codes:
                    hosting_api_util.remove_code_list(code, hosting_api_util.CODE_LIST_MUST_BUY)
                return_str = json.dumps({"code": 0})

            elif type == 423:
                # 暂不买列表
                result = hosting_api_util.get_code_list(hosting_api_util.CODE_LIST_MUST_BUY)
                return_str = json.dumps(result)


            elif type == 430:
                # 查询代码属性
                data = json.loads(_str)