easyquant/config/network_config.py

import datetime

import requests


class network_config:

    def __init__(self):
        """
        初始化网络配置
        """
        self.BASE_URL = "http://101.34.79.70:8081"  # 替换为你的 NocoDB 地址
        API_TOKEN = "mgIUTo_BATQnVPE3czsHLOLLgcQmnVv1qs7gJTIO"  # 在用户设置中创建
        self.headers = {
            "xc-token": API_TOKEN
        }

        self.project_table = {
            "strategy_config": "mw1zi3dr5sy03em",  # 查看项目URL获取
            "strategy_base": "mtvgo6pbrlmbgjc",
            "strategy_config_template": "m4nty4o5t6sxzts",
            "strategy_logs": "m8ebfypvbfpk38p",
            "today_state": "mosii8hqry8xx9b"
        }

        pass

    def register_strategy(self, strategy_name: str, template_id: str = ""):
        """
        注册策略
        :param strategy_name: 策略名称
        :return: None
        """
        PROJECT_ID = self.project_table['strategy_base']
        url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
        data = {

            "text": strategy_name,
            "memo": template_id,
            "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),  # 使用当前时间
            "modify_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),  # 使用当前时间

        }
        response = requests.post(url, headers=self.headers, data=data)
        response.raise_for_status()  # 检查错误
        result = response.json()
        template_list = self.get_template_list(template_id)
        for template in template_list:
            self.insert_config(result["id"], template)

        return result["id"]

    def get_strategy_id(self, strategy_name: str):
        """
        获取策略ID
        :param strategy_name: 策略名称
        :return: 策略ID
        """
        PROJECT_ID = self.project_table['strategy_base']
        response = requests.get(f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records", headers=self.headers)
        response.raise_for_status()
        data = response.json()
        records = data["list"]
        return next((str(record["id"]) for record in records if record.get("text") == strategy_name), None)

    def get_template_list(self, strategy_id: str) -> any:
        """
        获取网络配置参数
        :param type:
        :param strategy_id:
        :param key: 配置参数的键
        :return: 配置参数的值
        """
        PROJECT_ID = self.project_table['strategy_config_template']
        url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
        params = dict(limit=100, offset=0, where=f"(template_id,eq,{strategy_id})")
        try:
            return requests.get(url, headers=self.headers, params=params).json()['list']
        except Exception as e:
            return []

    def get_data_list(self, strategy_id: str, type: str) -> any:
        """
        获取网络配置参数
        :param type:
        :param strategy_id:
        :param key: 配置参数的键
        :return: 配置参数的值
        """
        # 这里可以添加获取配置逻辑
        PROJECT_ID = self.project_table['strategy_config']
        url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
        params = dict(limit=100, offset=0, where=f"(strategy_id,eq,{strategy_id})")
        try:
            result = []
            records = requests.get(url, headers=self.headers, params=params).json()['list']
            for record in records:
                if record.get("strategy_id") == str(strategy_id) and record.get("type") == type:
                    result.append(record)
            return result
        except Exception as e:
            return []

    def update_config(self, id: str, new_value: str):
        """
        更新网络配置参数

        :return:
        """
        PROJECT_ID = self.project_table['strategy_config']
        url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
        new_value = {
            "id": id,
            "new_value": new_value,
            "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }
        update_response = requests.patch(url, headers=self.headers, json=new_value)
        update_response.raise_for_status()
        return update_response.json()

    def insert_config(self, strategy_id: str, template: dict):
        """
        插入网络配置参数
        """
        PROJECT_ID = self.project_table['strategy_config']
        url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
        data = {
            "strategy_id": strategy_id,
            "type": "config",
            "name": template['name'],
            "data_type": template['data_type'],
            "data_param": template['data_param'],
            "memo": template['memo'],
            "new_value": template['value'],
            "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }
        response = requests.post(url, headers=self.headers, data=data)
        response.raise_for_status()  # 检查错误

    # def update_check(self, id: str, old_value: str, new_value: str):
    #     """
    #     更新网络配置参数
    #
    #     :return:
    #     """
    #     PROJECT_ID = self.project_table['strategy_config']
    #     url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records/"
    #     new_value = {
    #         "id": id,
    #         "old_value": old_value,
    #         "new_value": new_value,
    #         "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    #     }
    #     update_response = requests.patch(url, headers=self.headers, json=new_value)
    #     update_response.raise_for_status()
    #     return update_response.json()
    #
    # def insert_check(self, strategy_id: str, key: str, value: str):
    #     """
    #     插入网络配置参数
    #     """
    #     PROJECT_ID = self.project_table['strategy_config']
    #     url = f"{self.BASE_URL}/api/v2/tables/{PROJECT_ID}/records"
    #     data = {
    #         "strategy_id": strategy_id,
    #         "type": "check",
    #         "name": key,
    #         "new_value": value,
    #         "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),  # 使用当前时间
    #
    #     }
    #     response = requests.post(url, headers=self.headers, data=data)
    #     response.raise_for_status()  # 检查错误

    @staticmethod
    def insert_logs(strategy_id: str, log_type: str, log_content: str):
        """
        插入日志记录
        :param strategy_id: 策略ID
        :param log_type: 日志类型
        :param log_content: 日志内容
        :return: None
        """
        PROJECT_ID = "m8ebfypvbfpk38p"
        url = f"http://101.34.79.70:8081/api/v2/tables/{PROJECT_ID}/records"
        data = {
            "strategy_id": strategy_id,
            "msg_type": log_type,
            "message": log_content,
            "create_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),  # 使用当前时间
        }
        API_TOKEN = "mgIUTo_BATQnVPE3czsHLOLLgcQmnVv1qs7gJTIO"  # 在用户设置中创建
        headers = {
            "xc-token": API_TOKEN
        }
        response = requests.post(url, headers=headers, data=data)
        response.raise_for_status()

    @staticmethod
    def insert_state(strategy_id: str,state_str:str,trade_date:str):
        """

        :param strategy_id:
        :param state_str:
        :param trade_date:
        :return:
        """
        PROJECT_ID = "mosii8hqry8xx9b"
        url = f"http://101.34.79.70:8081/api/v2/tables/{PROJECT_ID}/records"
        data = {
            "strategy_id": strategy_id,
            "trade_date": trade_date,  # 使用当前时间
            "state": state_str
        }
        API_TOKEN = "mgIUTo_BATQnVPE3czsHLOLLgcQmnVv1qs7gJTIO"  # 在用户设置中创建
        headers = {
            "xc-token": API_TOKEN
        }
        response = requests.post(url, headers=headers, data=data)
        response.raise_for_status()



# aaa = network_config()
# ddd = aaa.get_data_list("10000113","config")
# print(ddd)