data_server.py
@@ -5,17 +5,19 @@
import json
import logging
import socketserver
import time
from http.server import BaseHTTPRequestHandler
import urllib.parse as urlparse
import psutil
import constant
from db import redis_manager_delegate as redis_manager, mysql_data_delegate as mysql_data
from db.redis_manager_delegate import RedisUtils
from log_module import log_export
from log_module import log_export, async_log_util
from log_module.log import hx_logger_l2_transaction, logger_debug, logger_request_api
from strategy import data_cache
from strategy.kpl_data_manager import KPLStockOfMarketsPlateLogManager
from strategy.kpl_data_manager import KPLMarketsSiftPlateLogManager, KPLMarketStockHeatLogManager
from strategy.trade_setting import TradeSetting
from trade import huaxin_trade_api, huaxin_trade_data_update
from trade.huaxin_trade_record_manager import DelegateRecordManager, DealRecordManager, MoneyManager, PositionManager
@@ -95,7 +97,7 @@
            fdatas = []
            for code in codes:
                data = data_cache.latest_code_market_info_dict.get(code)
                logger_debug.info(f"获取L1行情接口:{code}-{data}")
                # logger_debug.info(f"获取L1行情接口:{code}-{data}")
                if data:
                    fdatas.append(data)
            response_data = json.dumps({"code": 0, "data": fdatas})
@@ -121,7 +123,10 @@
            response_data = json.dumps({"code": 0, "data": {}})
        elif url.path == "/get_env":
            request_id = params_dict.get("request_id")
            use_time_list = []
            try:
                __start_time = time.time()
                fdata = {}
                # try:
                #     date = HistoryKDatasUtils.get_trading_dates(tool.date_sub(tool.get_now_date_str(), 10),
@@ -144,6 +149,7 @@
                    fdata["redis"] = 1
                except:
                    fdata["redis"] = 0
                use_time_list.append(("验证redis", time.time() - __start_time))
                try:
                    # 验证mysql
@@ -151,12 +157,14 @@
                    fdata["mysql"] = 1
                except:
                    fdata["mysql"] = 0
                use_time_list.append(("验证mysql", time.time() - __start_time))
                try:
                    # redis异步任务数量
                    fdata["redis_async_task_count"] = redis_manager.RedisUtils.get_async_task_count()
                except:
                    pass
                use_time_list.append(("验证异步任务数量", time.time() - __start_time))
                # 获取交易通道
                try:
@@ -165,29 +173,60 @@
                except Exception as e:
                    logger_debug.exception(e)
                    fdata["trade_channel_access"] = 0
                use_time_list.append(("验证交易通道", time.time() - __start_time))
                # 获取CPU与内存适用情况
                memory_info = psutil.virtual_memory()
                cpu_percent = psutil.cpu_percent(interval=1)
                fdata["device"] = {"cpu": cpu_percent, "memery": memory_info.percent}
                use_time_list.append(("获取设备资源占用", time.time() - __start_time))
                # 获取交易通道
                result = {"code": 0, "data": fdata, "msg": ""}
                # print("OnGetEnvInfo 成功")
                response_data = json.dumps(result)
            except Exception as e:
                response_data = json.dumps({"code": 1, "msg": str(e)})
        elif url.path == "/load_kpl_stock_of_markets_plate":
                logger_debug.error(f"环境获取异常:{request_id}")
                logger_debug.exception(e)
            finally:
                if use_time_list and use_time_list[-1][1] > 10:
                    logger_debug.warning(f"环境获取时间大于10s({request_id}):{use_time_list}")
        # 获取板块强度数据
        elif url.path == "/load_get_kpl_market_sift_plate":
            # 加载数据
            KPLStockOfMarketsPlateLogManager().load_data()
            KPLMarketsSiftPlateLogManager().load_data()
            response_data = json.dumps({"code": 0, "msg": "暂无内容"})
        elif url.path == "/get_kpl_stock_of_markets_plate":
        elif url.path == "/get_kpl_market_sift_plate":
            # 获取开盘啦流入板块详细信息
            print("==========get_kpl_market_sift_plate==========")
            try:
                time_str = params_dict.get("time")
                if not time_str:
                    time_str = tool.get_now_time_str()
                fdatas = KPLMarketsSiftPlateLogManager().get_filter_log_datas()
                response_data = json.dumps({"code": 1, "msg": "暂无内容"})
                for i in range(len(fdatas) - 1, -1, -1):
                    if fdatas[i][0] <= time_str:
                        response_data = json.dumps({"code": 0, "data": fdatas[i]})
                        break
            except Exception as e:
                logging.exception(e)
                response_data = json.dumps({"code": 1, "msg": str(e)})
        # 获取个股强度数据
        elif url.path == "/load_kpl_market_stock_heat":
            # 加载数据
            KPLMarketStockHeatLogManager().load_data()
            response_data = json.dumps({"code": 0, "msg": "暂无内容"})
        elif url.path == "/get_kpl_market_stock_heat":
            # 获取开盘啦流入板块详细信息
            print("==========get_kpl_stock_of_markets_plate==========")
            try:
                time_str = params_dict.get("time")
                if not time_str:
                    time_str = tool.get_now_time_str()
                fdatas = KPLStockOfMarketsPlateLogManager().get_filter_log_datas()
                fdatas = KPLMarketStockHeatLogManager().get_filter_log_datas()
                response_data = json.dumps({"code": 1, "msg": "暂无内容"})
                for i in range(len(fdatas) - 1, -1, -1):
                    if fdatas[i][0] <= time_str:
@@ -239,12 +278,13 @@
            print("接收到POST请求:", str(path))
            url = urlparse.urlparse(path)
            if url.path == "/trade_callback":
                # 接受开盘啦数据
                body = self.__parse_request()
                if type(body) != str:
                    huaxin_trade_api.add_trade_callback_data(json.dumps(body))
                else:
                    huaxin_trade_api.add_trade_callback_data(body)
                if constant.IS_SIMULATED_TRADE:
                    # 接受开盘啦数据
                    body = self.__parse_request()
                    if type(body) != str:
                        huaxin_trade_api.add_trade_callback_data(json.dumps(body))
                    else:
                        huaxin_trade_api.add_trade_callback_data(body)
                result_str = json.dumps({"code": 0})
            elif url.path == "/buy":
                # 签名验证
@@ -290,6 +330,12 @@
                            raise Exception("没有获取到L1数据")
                        pre_price = data[1]
                        current_price = data[2] if data[2] else data[5][0][0]
                        # 获取最新成交价格
                        latest_deal_price = data_cache.latest_deal_price_dict.get(code)
                        if latest_deal_price:
                            current_price = round(float(latest_deal_price), 2)
                            async_log_util.info(logger_debug, f"根据成交价卖出:{code}-{latest_deal_price}")
                        price = tool.get_buy_min_price(current_price)
                        price = max(price, tool.get_limit_down_price(code, pre_price))
                    else:
@@ -313,6 +359,8 @@
                    result_str = json.dumps({"code": 1, "msg": "未上传金额"})
                    return
                money = int(money)
                logger_debug.info(f"设置开仓金额:{money}")
                data_cache.BUY_MONEY_PER_CODE = money
                result_str = json.dumps({"code": 0})
@@ -354,6 +402,16 @@
                # 成交大单传递
                datas = self.rfile.read(int(self.headers['content-length']))
                _str = str(datas, encoding="gbk")
                datas = json.loads(_str)
                for d in datas:
                    if d[1] != 0:
                        continue
                    code, data = d[0], d[2]
                    if code not in data_cache.big_order_deal_dict:
                        data_cache.big_order_deal_dict[code] = []
                    data_cache.big_order_deal_dict[code].append(d)
                    # 获取买大单数量
                    len(data_cache.big_order_deal_dict.get(code, []))
                hx_logger_l2_transaction.info(_str)
                # 记录日志
                result_str = json.dumps({"code": 0})