获取环境接口

Administrator

2023-07-31 a3218ed469f147e229fd8923f20ad437cce244ae

获取环境接口

 third_data/data_server.py

@@ -40,6 +40,7 @@
    __jingxuan_cache_dict = {}
    __industry_cache_dict = {}


    def __get_limit_up_list(self):
        # 统计目前为止的代码涨停数量（分涨停原因）
        total_datas = KPLLimitUpDataRecordManager.total_datas

 third_data/kpl_data_manager.py

@@ -1,5 +1,6 @@
import json
import os
import time

import constant
from utils import tool
@@ -185,14 +186,17 @@

class KPLDataManager:
    __latest_datas = {}
    kpl_data_update_info = {}

    def __save_in_file(self, key, datas):
    @classmethod
    def __save_in_file(cls, key, datas):
        name = f"{tool.get_now_date_str()}_{key}.log"
        path = f"{constant.CACHE_PATH}/{name}"
        with open(path, 'w') as f:
            f.write(json.dumps(datas))

    def __get_from_file(self, key, day=tool.get_now_date_str()):
    @classmethod
    def __get_from_file(cls, key, day=tool.get_now_date_str()):
        name = f"{day}_{key}.log"
        path = f"{constant.CACHE_PATH}/{name}"
        if not os.path.exists(path):
@@ -203,7 +207,8 @@
                return json.loads(lines[0])
        return None

    def get_from_file(self, type, day):
    @classmethod
    def get_from_file(cls, type, day):
        name = f"{day}_{type.value}.log"
        path = f"{constant.CACHE_PATH}/{name}"
        if not os.path.exists(path):
@@ -214,8 +219,9 @@
                return json.loads(lines[0])
        return None

    @classmethod
    # 获取最近几天的数据，根据日期倒序返回
    def get_latest_from_file(self, type, count):
    def get_latest_from_file(cls, type, count):
        files = os.listdir(constant.CACHE_PATH)
        file_name_list = []
        for f in files:
@@ -236,17 +242,20 @@

        return fresults

    def save_data(self, type, datas):
        self.__latest_datas[type] = datas
        self.__save_in_file(type, datas)
    @classmethod
    def save_data(cls, type, datas):
        cls.kpl_data_update_info[type] = (tool.get_now_time_str(), len(datas))
        cls.__latest_datas[type] = datas
        cls.__save_in_file(type, datas)

    def get_data(self, type):
    @classmethod
    def get_data(cls, type):
        type = type.value
        if type in self.__latest_datas:
            return self.__latest_datas[type]
        result = self.__get_from_file(type)
        if type in cls.__latest_datas:
            return cls.__latest_datas[type]
        result = cls.__get_from_file(type)
        if result is not None:
            self.__latest_datas[type] = result
            cls.__latest_datas[type] = result
        return result



 trade/huaxin/trade_api_server.py

@@ -11,6 +11,8 @@
from l2.huaxin import huaxin_target_codes_manager
from third_data import block_info
from third_data.history_k_data_util import HistoryKDatasUtils, JueJinApi
from third_data.kpl_data_manager import KPLDataManager
from third_data.kpl_util import KPLDataType
from trade import trade_manager, trade_huaxin

from trade.huaxin import huaxin_trade_api, huaxin_trade_record_manager, \
@@ -305,6 +307,25 @@
                        except Exception as e:
                            return_str = json.dumps(
                                {"code": 0, "msg": str(e)})
                    elif type_ == 'get_env_info':
                        # 获取环境信息
                        fdata = {}
                        try:
                            date = JueJinApi.get_previous_trading_date(tool.get_now_date_str())
                            if date:
                                fdata["juejin"] = 1
                        except Exception as e:
                            fdata["juejin"] = 0
                        fdata["kpl"] = {}
                        # 获取开盘啦数据
                        kpl_types = [KPLDataType.LIMIT_UP.value, KPLDataType.JINGXUAN_RANK.value,
                                     KPLDataType.INDUSTRY_RANK.value]
                        for kpl_type in kpl_types:
                            if kpl_type in KPLDataManager.kpl_data_update_info:
                                fdata["kpl"][kpl_type] = KPLDataManager.kpl_data_update_info.get(kpl_type)

                        return_str = json.dumps(
                            {"code": 0, "data": fdata, "msg": ""})
                    # 查询委托列表
                    elif type_ == 'test':
                        # 卖出