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)