easyquant/test.py

import logging
import os
from datetime import timedelta
import datetime

import requests
from typing import Any, Optional, List, Dict

import pandas as pd



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()


class strategy_config:

    def __init__(self,
                 strategy_name: str,
                 strategy_template_id: str,
                 config_save: network_config,
                 log):
        """
        基础必要初始化策略配置
        """
        # 策略名
        self.strategy_name = strategy_name
        # 配置模版id
        self.template_id = strategy_template_id
        # 配置变更保存方式
        self.connection = config_save
        # 日志记录器
        self.logger = Logger(log)

        """
        实例化strategy_config类时，初始化必要的属性
        """
        # 策略ID
        self.strategy_id = None
        # 配置对应主键
        self.configkey_id = {}
        # 配置信息
        self.config = {}

        # 核心缓存
        self.strategy_state = state()

        # 初始化配置文件
        self.init_from_db()

    def init_from_db(self):
        self.strategy_id = self.connection.register_strategy(self.strategy_name, self.template_id)

        self.logger.info("初始化策略配置", self.strategy_id)
        self.logger.info('  注册策略序列:' + str(self.strategy_id), self.strategy_id)
        config_list = self.connection.get_data_list(self.strategy_id, 'config')
        self.config = {item['name']: item for item in config_list}
        self.configkey_id = {item['name']: item['id'] for item in config_list}

        self.logger.info('确认配置参数:', self.strategy_id)
        for config in config_list:
            self.logger.info(f"  参数名: {config['name']}, 值: {config['new_value']}, 备注: {config['memo']}", self.strategy_id)
        self.logger.info('确认配置参数完毕, 如有错误请立刻停止执行', self.strategy_id)

    def get_all_configs(self) -> dict:
        """
        获取所有策略配置参数
        :return: 包含所有配置参数的字典
        """
        return self.config

    def set_config(self, key: str, value: str) -> None:
        """
        通过字典语法设置配置项，并自动触发数据库更新

        Args:
            key (str): 配置项名称
            value (Any): 配置项的新值
        """
        id = self.configkey_id.get(key)
        self.connection.update_config(id, value)
        self.config[key]['new_value'] = value
    def get_config(self, key: str):
        func = conf.config_type(key)
        return func(key)





class hold_data:
    def __init__(self, data: dict):
        self.stock = None
        self.price = None
        self.amount = None
        self.cost = None
        self.profit = None
        self.profit_rate = None


class state:
    def __init__(self):
        self.data = {}
        self.init()

    # 初始化
    def init(self):
        self.data = {
            "create_date": "",
            "position_list": {},
            "sell_request_list": {},
            "buy_request_list": {},
            "high_stock_list": {},
            "low_stock_list": {},
            "reason_to_sell": "",
            "history_buy_list": {},
            "history_sell_list": {},
        }

    def get_buy_requests(self):
        return self.data["buy_request_list"]

    def clear_buy_requests(self):
        """
        清空买入请求列表
        """
        self.data["buy_request_list"] = {}

    def reset(self):
        self.init()

    def set_state_datetime(self, key: str, value: datetime) -> None:
        """
        设置状态值
        :param key: 状态键
        :param value: 状态值
        """
        self.data[key] = value

    def set_hold_stock(self, stock: str, data: dict) -> None:
        """
        设置持仓股票数据
        :param stock: 股票代码
        :param data: hold_data 对象
        """
        hd = hold_data(data)
        if not isinstance(hd, hold_data):
            raise ValueError("data must be an instance of hold_data")
        self.data["position_list"][stock] = hd

    def set_high_stock(self, stock: str, data: dict) -> None:
        """
        设置高位股票数据
        :param stock: 股票代码
        :param data: hold_data 对象
        """
        hd = hold_data(data)
        if not isinstance(hd, hold_data):
            raise ValueError("data must be an instance of hold_data")
        self.data["high_stock_list"][stock] = hd

    def set_low_stock(self, stock: str, data: dict) -> None:
        """
        设置低位股票数据
        :param stock: 股票代码
        :param data: hold_data 对象
        """
        hd = hold_data(data)
        if not isinstance(hd, hold_data):
            raise ValueError("data must be an instance of hold_data")
        self.data["low_stock_list"][stock] = hd

    def set_sell_request(self, stock: str, data: dict, reason: str) -> None:
        """
        设置卖出请求数据
        :param stock: 股票代码
        :param data: hold_data 对象
        """
        hd = hold_data(data)
        if not isinstance(hd, hold_data):
            raise ValueError("data must be an instance of hold_data")
        self.data["sell_request_list"][stock + "_" + reason] = hd

    def set_buy_request(self, stock: str, data: dict, reason: str) -> None:
        """
        设置买入请求数据
        :param stock: 股票代码
        :param data: hold_data 对象
        """
        hd = hold_data(data)
        if not isinstance(hd, hold_data):
            raise ValueError("data must be an instance of hold_data")
        self.data["buy_request_list"][stock + "_" + reason] = hd

    def clear_buys_requests(self):
        """
        清空买入请求列表
        """
        self.data["buy_request_list"] = {}

    def get_sell_reason(self):
        return self.data["reason_to_sell"]

    def set_sell_reason(self, reason: str):
        """
        设置卖出原因
        :param reason: 卖出原因
        """
        self.data["reason_to_sell"] = reason

    def get_sell_requests(self):
        """
        获取卖出请求列表
        :return: 卖出请求列表
        """
        return self.data["sell_request_list"]

    def get_yesterday_high_list(self):
        """
        获取昨日涨停股票列表
        :return: 昨日涨停股票列表
        """
        return self.data["high_stock_list"]

    def get_hold_list(self):
        return self.data["position_list"]


class conf:

    @staticmethod
    def to_int(value: str, params, separator: str = '_'):
        return int(value)

    @staticmethod
    def to_float(value: str, params, separator: str = '_'):
        return float(value)

    @staticmethod
    def to_str(value: str, params, separator: str = '_'):
        return str(value)

    @staticmethod
    def to_bool(value: str, params, separator: str = '_'):
        return bool(value)

    @staticmethod
    def to_list(value: str, separator: str = '_'):
        return value.split(separator)

    @staticmethod
    def config_type(value: str):
        data = {
            "int": conf.to_int,
            "float": conf.to_float,
            "str": conf.to_str,
            "bool": conf.to_bool,
            "list": conf.to_list,
        }

        return data[value]


class DataHelper:
    """
    数据操作辅助类
    封装了数据接口的调用，包括 get_price 与 history 函数，
    并在内部捕获异常、输出中文错误日志，避免重复编写 try/except 代码。
    """

    @staticmethod
    def get_price_safe(
            security: Any,
            end_date: Any,
            frequency: str,
            fields: List[str],
            count: int,
            panel: bool = False,
            skip_paused: bool = True,
            fq: Optional[str] = None,
            fill_paused: bool = False
    ) -> Optional[pd.DataFrame]:
        """
        安全调用 get_price 数据接口
        参数:
            security: 单只股票代码或股票代码列表
            end_date: 数据截止日期
            frequency: 数据频率，如 "daily" 或 "1m"
            fields: 需要获取的数据字段列表（例如 ['open', 'close']）
            count: 请求数据的记录数
            panel: 是否返回面板数据（默认为 False）
            skip_paused: 是否跳过停牌股票（默认为 True）
            fq: 复权方式（例如"pre"或"post"，默认 None）
            fill_paused: 是否填充停牌数据（默认为 False）
        返回:
            返回包含数据的 DataFrame，如果出错则返回 None
        """
        try:
            # 调用聚宽提供的 get_price 获取数据
            df = get_price(
                security,
                end_date=end_date,
                frequency=frequency,
                fields=fields,
                count=count,
                panel=panel,
                skip_paused=skip_paused,
                fq=fq,
                fill_paused=fill_paused
            )
            return df
        except Exception as e:
            # 输出中文错误日志，并返回 None
            # logger.error(f"获取 {security} 的价格数据时出错: {e}")
            return None

    @staticmethod
    def get_current_data():
        """
        获取当前市场数据
        返回:
            当前市场数据的字典，包含股票代码到数据对象的映射
        """
        try:
            # 调用聚宽提供的 get_current_data 获取当前数据
            return get_current_data()
        except Exception as e:
            # Logger.error(f"获取当前市场数据时出错: {e}")
            return {}

    @staticmethod
    def get_history_safe(
            security: Any,
            unit: str,
            field: str,
            count: int,
    ) -> Optional[Dict[str, List[float]]]:
        """
        安全调用 history 数据接口，批量获取历史数据

        参数:
            security: 单只或多只股票代码
            unit: 数据单位，例如 "1m" 表示1分钟数据
            field: 请求数据字段名称，如 "close"（收盘价）
            count: 请求历史数据记录数

        返回:
            返回一个字典，映射股票代码到对应的数据列表；出错则返回 None
        """
        try:
            # 调用聚宽的 history 函数获取数据
            data = history(count, unit=unit, field=field, security_list=security)
            return data
        except Exception as e:
            return None

    @staticmethod
    def get_index_stocks(index_code: str) -> List[str]:
        """
        获取指定指数的成分股列表

        参数:
            index_code: 指数代码，例如 "000300.XSHG" 表示沪深300指数
            date: 可选参数，指定查询日期，默认为 None，表示查询最新成分股

        返回:
            成分股代码列表
        """
        try:
            # 调用聚宽的 get_index_stocks 函数获取成分股
            return get_index_stocks(index_code)
        except Exception as e:
            return []

    @staticmethod
    def get_security_info(security: str) -> Optional[Any]:
        """
        获取指定股票的安全信息

        参数:
            security: 股票代码，例如 "000001.XSHE"

        返回:
            股票的安全信息对象，如果出错则返回 None
        """
        try:
            # 调用聚宽的 get_security_info 函数获取股票信息
            return get_security_info(security)
        except Exception as e:
            return None

    @staticmethod
    def get_stock_list(context: Any, index_stocks: str, stock_num: int) -> List[str]:
        """
        选股模块：
        1. 从指定股票池（如 399101.XSHE 指数成分股）中获取初步股票列表；
        2. 应用多个过滤器筛选股票（次新股、科创股、ST、停牌、涨跌停等）；
        3. 基于基本面数据（EPS、市值）排序后返回候选股票列表。
        参数:
            context: 聚宽平台传入的交易上下文对象
        返回:
            筛选后的候选股票代码列表
        """
        # 从指定指数中获取初步股票列表
        initial_list: List[str] = DataHelper.get_index_stocks(index_stocks)

        # 依次应用过滤器，筛去不符合条件的股票
        initial_list = DataHelper.filter_new_stock(context, initial_list)  # 过滤次新股
        initial_list = DataHelper.filter_kcbj_stock(initial_list)  # 过滤科创/北交股票
        initial_list = DataHelper.filter_st_stock(initial_list)  # 过滤ST或风险股票
        initial_list = DataHelper.filter_paused_stock(initial_list)  # 过滤停牌股票
        initial_list = DataHelper.filter_limitup_stock(context, initial_list)  # 过滤当日涨停（未持仓时）的股票
        initial_list = DataHelper.filter_limitdown_stock(context, initial_list)  # 过滤当日跌停（未持仓时）的股票

        # 利用基本面查询获取股票代码和EPS数据，并按照市值升序排序
        q = query(valuation.code, indicator.eps) \
            .filter(valuation.code.in_(initial_list)) \
            .order_by(valuation.market_cap.asc())
        df = get_fundamentals(q)
        stock_list: List[str] = list(df.code)
        stock_list = stock_list[:50]  # 限制数据规模，防止一次处理数据过大
        # 取前2倍目标持仓股票数作为候选池
        final_list: List[str] = stock_list[:2 * int(stock_num)]

        # 查询并输出候选股票的财务信息（如财报日期、营业收入、EPS）
        if final_list:
            info_query = query(
                valuation.code,
                income.pubDate,
                income.statDate,
                income.operating_revenue,
                indicator.eps
            ).filter(valuation.code.in_(final_list))
            df_info = get_fundamentals(info_query)
            # for _, row in df_info.iterrows():
            #     log.info(
            #         f"股票 {row['code']}：报告日期 {row.get('pubDate', 'N/A')}，统计日期 {row.get('statDate', 'N/A')}，营业收入 {row.get('operating_revenue', 'N/A')}，EPS {row.get('eps', 'N/A')}")
        return final_list

    @staticmethod
    def filter_paused_stock(stock_list: List[str]) -> List[str]:
        """
        过滤停牌的股票
        参数:
            stock_list: 待过滤的股票代码列表
        返回:
            未停牌的股票代码列表
        """
        current_data = DataHelper.get_current_data()
        return [stock for stock in stock_list if not current_data[stock].paused]

    @staticmethod
    def filter_st_stock(stock_list: List[str]) -> List[str]:
        """
        过滤带有 ST 或其他风险标识的股票
        参数:
            stock_list: 待过滤的股票代码列表
        返回:
            无 ST 或风险标识的股票代码列表
        """
        current_data = DataHelper.get_current_data()
        return [stock for stock in stock_list if (not current_data[stock].is_st) and
                ('ST' not in current_data[stock].name) and
                ('*' not in current_data[stock].name) and
                ('退' not in current_data[stock].name)]

    @staticmethod
    def filter_kcbj_stock(stock_list: List[str]) -> List[str]:
        """
        过滤科创、北交股票
        参数:
            stock_list: 待过滤的股票代码列表
        返回:
            过滤后的股票代码列表（排除以 '4'、'8' 开头以及以 '68' 起始的股票）
        """
        return [stock for stock in stock_list if stock[0] not in ('4', '8') and not stock.startswith('68')]

    @staticmethod
    def filter_limitup_stock(context: Any, stock_list: List[str]) -> List[str]:
        """
        过滤当天已经涨停的股票（若未持仓则过滤）
        参数:
            context: 交易上下文对象
            stock_list: 待过滤的股票代码列表
        返回:
            过滤后的股票代码列表
        """
        history_data = DataHelper.get_history_safe(stock_list, unit='1m', field='close', count=1)
        current_data = DataHelper.get_current_data()
        if history_data is None:
            return stock_list
        return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or
                (history_data.get(stock, [0])[-1] < current_data[stock].high_limit)]

    @staticmethod
    def filter_limitdown_stock(context: Any, stock_list: List[str]) -> List[str]:
        """
        过滤当天已经跌停的股票（若未持仓则过滤）
        参数:
            context: 交易上下文对象
            stock_list: 待过滤的股票代码列表
        返回:
            过滤后的股票代码列表
        """
        history_data = DataHelper.get_history_safe(stock_list, unit='1m', field='close', count=1)
        current_data = DataHelper.get_current_data()
        if history_data is None:
            return stock_list
        return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or
                (history_data.get(stock, [float('inf')])[-1] > current_data[stock].low_limit)]

    @staticmethod
    def filter_new_stock(context: Any, stock_list: List[str]) -> List[str]:
        """
        过滤次新股：排除上市时间不足375天的股票
        参数:
            context: 交易上下文对象
            stock_list: 待过滤的股票代码列表
        返回:
            过滤后的股票代码列表
        """
        yesterday = context.previous_date
        return [stock for stock in stock_list if not (yesterday - DataHelper.get_security_info(stock).start_date < timedelta(days=375))]

    @staticmethod
    def filter_highprice_stock(self, context: Any, stock_list: List[str]) -> List[str]:
        """
        过滤股价高于设定上限（up_price）的股票（非持仓股票参与过滤）
        参数:
            context: 交易上下文对象
            stock_list: 待过滤的股票代码列表
        返回:
            过滤后的股票代码列表
        """
        history_data = DataHelper.get_history_safe(stock_list, unit='1m', field='close', count=1)
        if history_data is None:
            return stock_list
        return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or
                history_data.get(stock, [self.up_price + 1])[-1] <= self.up_price]

    @staticmethod
    def filter_not_buy_again(stock_list: List[str], not_buy_again) -> List[str]:
        """
        过滤掉当日已买入的股票，避免重复下单
        参数:
            stock_list: 待过滤的股票代码列表
        返回:
            未买入的股票代码列表
        """
        return [stock for stock in stock_list if stock not in not_buy_again]



class Logger:
    """
    简单的日志操作类，支持控制台和文件输出
    可以使用现有的logger对象或创建新的实例
    """
    _instance = None

    def __new__(cls, existing_logger=None, *args, **kwargs):
        if existing_logger is not None:
            # 如果提供了现有的logger对象，创建新实例并使用该对象
            instance = super().__new__(cls)
            instance.logger = existing_logger
            return instance

        # 否则使用单例模式
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, existing_logger=None, log_file="logs/app.log"):
        # 如果使用现有logger，直接返回
        if existing_logger is not None:
            return

        # 如果已经初始化过，直接返回
        if hasattr(self, 'logger'):
            return

        # 创建logs目录（如果不存在）
        os.makedirs(os.path.dirname(log_file), exist_ok=True)

        # 初始化logger
        self.logger = logging.getLogger('AppLogger')
        self.logger.setLevel(logging.DEBUG)

        # 清除可能存在的处理器
        self.logger.handlers.clear()

        # 设置日志格式
        formatter = logging.Formatter(
            '%(asctime)s [%(levelname)s] %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )

        # 添加控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)

        # 添加文件处理器
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)

    def debug(self, message, strategy_id=None):
        """记录调试级别的日志"""
        self.logger.debug(message)
        try:
            network_config.insert_logs(strategy_id, "debug", message)
        except:
            pass

    def info(self, message, strategy_id=None):
        """记录信息级别的日志"""
        self.logger.info(message)
        try:
            network_config.insert_logs(strategy_id, "info ", message)
        except:
            pass

    def warning(self, message, strategy_id=None):
        """记录警告级别的日志"""
        self.logger.warning(message)
        try:
            network_config.insert_logs(strategy_id, "warn", message)
        except:
            pass

    def error(self, message, strategy_id=None):
        """记录错误级别的日志"""
        self.logger.error(message)
        try:
            network_config.insert_logs(strategy_id, "error", message)
        except:
            pass

    def critical(self, message, strategy_id=None):
        """记录严重错误级别的日志"""
        self.logger.critical(message)
        try:
            network_config.insert_logs(strategy_id, "critical", message)
        except:
            pass


class exchange:

    def close_positions(self, context: Any, position: Any):
        """
        关闭所有持仓
        """
        raise NotImplementedError("This method should be implemented by subclasses")

    def buy_security(self, context: Any, target_list: List[str]) -> None:
        """
        买入操作：对目标股票执行买入，下单资金均摊分配
        """
        raise NotImplementedError("This method should be implemented by subclasses")

    def open_position(self, security: str, value: float) -> bool:
        """
        开仓操作：尝试买入指定股票
        参数:
            security: 股票代码
            value: 分配给该股票的资金
        返回:
            若下单成功（部分或全部成交）返回 True，否则返回 False
        """
        raise NotImplementedError("This method should be implemented by subclasses")


class jq_channel(exchange):

    def __init__(self):
        pass

    def order_target_value_(self, stock: str, value: float) -> Any:
        """
        封装 order_target_value 函数进行下单，同时记录中文日志和异常信息
        参数:
            security: 股票代码
            value: 下单目标金额
        返回:
            下单后生成的订单对象；若失败返回 None
        """
        try:
            order = order_target_value(stock, value)
            return order
        except Exception as e:
            return None

    def close_positions(self, context: Any, position: Any):
        """
        平仓操作：尽可能将指定股票仓位全部卖出
        参数:
            position: 持仓对象
        返回:
            若下单后订单全部成交返回 True，否则返回 False
        """
        security = position.security
        order = self.order_target_value_(security, 0)

        if order is None:
            # log.error(f"股票 {security} 下单失败，可能是API调用错误")
            return False

        if order.status == OrderStatus.held and order.filled == order.amount:
            # log.info(f"股票 {security} 卖出订单完全成交")
            return True
        else:
            # log.warning(f"股票 {security} 卖出订单部分成交或未成交，状态: {order.status}, 已成交: {order.filled}/{order.amount}")
            return False

    def buy_security(self, context: Any, target_list: List[str]) -> None:
        """
        买入操作：对目标股票执行买入，下单资金均摊分配

        参数:
            context: 聚宽平台传入的交易上下文对象
            target_list: 目标股票代码列表
        """
        position_count = len(context.portfolio.positions)
        target_num = len(target_list)
        if target_num > position_count:
            try:
                value = context.portfolio.cash / (target_num - position_count)
            except ZeroDivisionError as e:
                # log.error(f"资金分摊时除零错误: {e}")
                return
            for stock in target_list:
                if context.portfolio.positions[stock].total_amount == 0:
                    if self.open_position(stock, value):
                        # log.info(f"已买入股票 {stock}，分配资金 {value:.2f}")
                        # self.not_buy_again.append(stock)
                        if len(context.portfolio.positions) == target_num:
                            break

    def open_position(self, security: str, value: float) -> bool:
        """
        开仓操作：尝试买入指定股票
        参数:
            security: 股票代码
            value: 分配给该股票的资金
        返回:
            若下单成功（部分或全部成交）返回 True，否则返回 False
        """
        order = self.order_target_value_(security, value)
        if order is not None and order.filled > 0:
            return True
        return False


class trade_strategy:

    def __init__(self, strategy_name, strategy_template_id, config_save, log, channel: exchange):
        # -------必要参数-------
        self.log = Logger(log)  # 日志记录器
        self.strategy_config = strategy_config(strategy_name,
                                               strategy_template_id,
                                               config_save,
                                               self.log)

        self.state = state()  # 策略状态管理器，跟踪策略运行状态
        self.channel = channel

        # 添加持仓监控和管理器

    def init_strategy(self) -> None:
        """
        策略初始化函数
        配置交易环境参数，包括防未来数据、基准、滑点、订单成本以及日志输出等级。
        参数:
            context: 聚宽平台传入的交易上下文对象
        """
        # 启用防未来数据以确保历史回测的严谨性
        self.log.info("初始化策略TradingStrategy配置...")

        set_option('avoid_future_data', True)
        # 设置策略基准为上证指数
        # set_benchmark(self.strategy_config.config["benchmark_stock"])
        set_benchmark("000001.XSHG")
        # 使用真实市场价格，模拟实际交易
        set_option('use_real_price', True)
        # 设置固定滑点，确保价格偏差不大
        set_slippage(FixedSlippage(3 / 10000))
        # 设置订单成本，包括印花税和佣金
        set_order_cost(OrderCost(
            open_tax=0,
            close_tax=0.001,  # 卖出时0.1%的印花税
            open_commission=2.5 / 10000,
            close_commission=2.5 / 10000,
            close_today_commission=0,
            min_commission=5  # 最低佣金为5元
        ), type='stock')
        # 设置日志输出等级（中文日志输出）
        self.log.logger.set_level('order', 'error')
        self.log.logger.set_level('system', 'error')
        self.log.logger.set_level('strategy', 'debug')

    def set_target_list(self, target_list):
        self.target_list = target_list.copy()

    def get_target_list(self):
        """获取本次调仓候选股票列表"""
        return self.target_list

    def get_configs(self) -> dict:
        """
        获取策略配置
        :return: strategy_config 对象
        """
        return self.strategy_config.get_all_configs()

    def get_strategy_config(self) -> strategy_config:
        """
        获取策略配置字典
        :return: 包含所有配置参数的字典
        """
        return self.strategy_config

    def close_position(self, context, stock: str):
        """
        平仓操作：尽可能将指定股票仓位全部卖出
        参数:
            position: 持仓对象
        返回:
            若下单后订单全部成交返回 True，否则返回 False
        """
        self.channel.close_positions(context, stock)

    def open_position(self, security: str, value: float) -> bool:
        """
        开仓操作：尝试买入指定股票
        参数:
            security: 股票代码
            value: 分配给该股票的资金
        返回:
            若下单成功（部分或全部成交）返回 True，否则返回 False
        """
        return self.channel.open_position(security, value)


class base_task:
    def __init__(self, strategy: trade_strategy, sub_task=None):
        self.strategy = strategy
        self.name = "base_task"
        self.remark = "base_task"
        self.memo = "base_task"
        self.sub_task = sub_task

        self.strategy_config = self.strategy.get_strategy_config()
        self.configs = self.strategy.get_configs()
        self.log = self.strategy.log

    def init(self, context: Any):
        pass

    def config(self, context: Any):
        pass

    def run(self, context: Any):
        pass

    def handle(self, context: Any):
        pass

    def end(self, context: Any):
        pass

    def process(self, context: Any):
        self.init(context)
        self.config(context)
        self.run(context)
        self.handle(context)
        self.end(context)

    def get_config(self, key: str):
        configs = self.configs.get(key)
        data_type = configs['data_type']
        data_params = configs['data_param']
        func = conf.config_type(data_type)
        return func(self.configs[key]['new_value'], data_params)




class check_state_before_task(base_task):
    """
    重置state信息
    最前置信息
    """

    def __init__(self, strategy: trade_strategy, sub_task=None):
        super().__init__(strategy, sub_task)

    def init(self, context):
        self.log.info("--------------------------")
        self.log.info("重置运行中环境数据")

    def config(self, context):
        # 重置state信息
        self.strategy.state.reset()
        self.hold_stock = [position.security for position in list(context.portfolio.positions.values())]
        self.yesterday_high_list = []
        self.yesterday_low_list = []

    def run(self, context):
        positions = context.portfolio.positions
        for stock in positions:
            self.strategy.state.set_hold_stock(stock.security, stock.__dict__)

        if self.hold_stock:
            # 获取持仓股票昨日数据（包括收盘价、涨停价、跌停价）
            df = DataHelper.get_price_safe(
                self.hold_stock,
                end_date=context.previous_date,
                frequency='daily',
                fields=['close', 'high_limit', 'low_limit'],
                count=1,
                panel=False,
                fill_paused=False
            )
            if df is not None and not df.empty:
                # 过滤出收盘价等于涨停价的股票，作为昨日涨停股票
                self.yesterday_high_list = list(df[df['close'] == df['high_limit']]['code'])
                self.yesterday_low_list = list(df[df['close'] == df['low_limit']]['code'])

    def handle(self, context):
        for stock in self.yesterday_high_list:
            self.strategy.state.set_high_stock(stock['code'], stock)
        for stock in self.yesterday_low_list:
            self.strategy.state.set_low_stock(stock['code'], stock)


class check_market_env_task(base_task):
    """
    交易环境确认
    1、判断是否为特殊月份（1月/4月）
    2、如果是特殊月份且通过了清仓检查，则清仓所有持仓
    3、如果不是特殊月份且没有交易信号，则重置交易信号
    """

    def __init__(self, strategy: trade_strategy):
        super().__init__(strategy)




    def config(self, context: Any):

        self.current_date = context.current_dt.date()
        self.current_month = self.current_date.month
        self.current_day = self.current_date.day

        # 判断是否执行空仓策略
        self.filter_month_clear_switch = self.get_config('filter_month_clear_switch')
        self.filer_month_data = self.get_config('filter_month_data')
        # 判断是否为特殊月 = True时 等于 执行1月/4月 全部清仓
        self.in_special_month = str(self.current_month) in self.filer_month_data

        self.today_trade_switch = self.get_config('today_trade_switch')  # 是否为不交易信号


    def run(self, context: Any):
        #
        if self.filter_month_clear_switch and self.in_special_month:
            # 进入特殊月份，应清仓
            for stock in self.strategy.state.data['position_list'].keys():
                self.strategy.state.set_sell_request(stock, self.strategy.state.data['position_list'][stock], f"enter_{self.current_month}_month")
            # 清仓后就不交易
            self.today_trade_switch = False
        elif not self.filter_month_clear_switch:
            # 不在特殊月 或 不执行空仓策略  可以交易
            self.today_trade_switch = True
        else:
            # 如果是特殊月份但没有清仓条件，则继续交易
            self.today_trade_switch = True


    def handle(self, context: Any):
        # 保存状态
        old_today_trade_switch = self.get_config("today_trade_switch")
        if old_today_trade_switch != self.today_trade_switch:
            self.strategy_config.set_config("today_trade_switch", self.today_trade_switch) # 临时变量
            self.log.info(f"交易环境更新: today_trade_switch 状态更新为 {self.today_trade_switch}")


    def end(self, context: Any):

        self.log.info("-----------------")
        self.log.info("一.今日环境确认:")
        self.log.info(" 当前日期: {}".format(self.current_date))
        self.log.info(" 空仓策略: {}".format("开启" if self.filter_month_clear_switch else "关闭"))
        self.log.info(" 今日交易: {}".format("交易" if self.today_trade_switch else "清仓"))
        self.log.info(" 特殊月份: {}".format(",".join(str(item) for item in self.filer_month_data)))
        self.log.info(" 当前月份: {}".format(self.current_month))

        # 今日清仓

class check_positions_stop_loss_task(base_task):

    def __init__(self, strategy: trade_strategy):
        super().__init__(strategy)

    def config(self, context: Any):

        self.stoploss_switch = self.get_config('stoploss_switch')  # 是否启用止损策略
        self.hold_list = self.strategy.state.data['position_list']  # 持仓列表
        self.today_trade_switch = self.get_config('today_trade_switch')  # 今日交易开关
        self.stoploss_strategy = self.get_config('stoploss_strategy')  # 止损策略
        self.stoploss_stock_rate_data = self.get_config('stoploss_stock_rate_data')  # 个股止损阀值
        self.stoploss_market_rate_data = self.get_config('stoploss_market_rate_data')  # 市场止损阀值
        # self.reason_to_sell = self.strategy.state.get_sell_reason()  # 卖出原因
        self.index_stocks = self.get_config('index_stocks')  # 指数代码
        self.temp_sells_list = {}

    def run(self, context: Any):
        if len(self.hold_list) == 0:
            self.log.debug("当前无持仓，跳过止损检查。")
            return

        if self.stoploss_switch and self.today_trade_switch:
            if self.stoploss_strategy == 1 or self.stoploss_strategy == 3:
                # 个股止盈或止损判断
                for stock in list(context.portfolio.positions.keys()):
                    pos = context.portfolio.positions[stock]
                    if pos.price >= pos.avg_cost * 2:
                        # 通过持仓监控器注册卖出请求，而不是直接卖出
                        # self.strategy.state.set_sell_request(stock, pos, 'take_profit')
                        self.sell_request(stock, pos, 'take_profit')
                        self.log.debug(f"股票 {stock} 实现100%盈利，执行止盈卖出。")
                    elif pos.price < pos.avg_cost * self.stoploss_stock_rate_data:
                        # 通过持仓监控器注册卖出请求，而不是直接卖出
                        self.strategy.state.set_sell_request(stock, pos, 'stop_loss')
                        self.sell_request(stock, pos, 'take_profit')
                        self.log.debug(f"股票 {stock} 触及止损阈值，执行卖出。")
                        # self.strategy.state.set_sell_reason("stoploss")

            if self.stoploss_strategy == 2 or self.stoploss_strategy == 3:
                # 大盘止损判断，若整体市场跌幅过大则平仓所有股票
                stock_list = DataHelper.get_index_stocks(self.index_stocks)
                df = DataHelper.get_price_safe(
                    stock_list,
                    end_date=context.previous_date,
                    frequency='daily',
                    fields=['close', 'open'],
                    count=1,
                    panel=False
                )
                if df is not None and not df.empty:
                    down_ratio = (df['close'] / df['open']).mean()
                    if down_ratio <= self.stoploss_market_rate_data:
                        # self.strategy.state.set_sell_reason("stoploss")
                        self.log.debug(f"市场检测到跌幅（平均跌幅 {down_ratio:.2%}），卖出所有持仓。")
                        for stock in list(context.portfolio.positions.keys()):
                            pos = context.portfolio.positions[stock]
                            # 通过持仓监控器注册卖出请求，而不是直接卖出
                            self.sell_request(stock, pos, 'market_stop_loss')

    def handle(self, context: Any):
        # 保存状态
        for key, value in self.temp_sells_list.items():
            stock, reason = key.split('_')
            self.strategy.state.set_sell_request(stock, value, reason)
        pass

    def end(self, context: Any):
        self.log.info("-----------------")
        self.log.info("止损策略开关:{}".format("启用" if self.stoploss_switch else "不启用"))
        self.log.info("今日交易开关:{}".format("交易" if self.today_trade_switch else "不交易"))
        self.log.info("止损策略方式:{}".format("个股止损" if self.stoploss_strategy == 1 else "大盘止损" if self.stoploss_strategy == 2 else "联合止损" if
        self.stoploss_strategy == 3 else "异常"))
        self.log.info("当前卖出请求")
        for key, value in self.temp_sells_list.items():
            self.log.info(f"  {key}: {value}")

        # 总卖出请求
        sells_list = self.strategy.state.get_sell_requests()
        if len(sells_list) > 0:
            self.log.info("总卖出请求:")
            for stock, data in sells_list.items():
                self.log.info(f"  {stock}: {data}")
        pass

    def sell_request(self, stock, data, reason):
        self.temp_sells_list[f"{stock}_{reason}"] = data

class buy_stocks_func_task(base_task):
    """
    每周调仓策略：
    如果非空仓日，先选股得到目标股票列表，再卖出当前持仓中不在目标列表且昨日未涨停的股票，
    最后买入目标股票，同时记录当天买入情况避免重复下单。
    参数:
        context: 聚宽平台传入的交易上下文对象
    """

    def __init__(self, strategy: trade_strategy):
        super().__init__(strategy)

    def config(self, context: Any):

        # 每次调仓的目标股票数量
        self.stock_num = self.get_config('stock_num')
        # 获取当前持仓列表
        self.hold_list = self.strategy.state.get_hold_list()
        self.index_stocks = self.get_config('index_stocks')
        self.target_list = DataHelper.get_stock_list(context, self.index_stocks, self.stock_num)
        self.temp_target_list = self.target_list[:self.stock_num]
        # 昨日涨停股票列表
        self.yesterday_HL_list: List[str] = self.strategy.state.get_yesterday_high_list()
        self.today_trade_switch = self.get_config('today_trade_switch')
        self.temp_sells_list = {}
        self.temp_buys_list = {}

    def run(self, context: Any):

        if not self.today_trade_switch:
            self.log.info("今日非交易日，跳过调仓操作。")
            return

            # 重置当天已买入记录
            # self.position_manager.reset_not_buy_again()
            # 取目标持仓数以内的股票作为调仓目标
        self.log.info(f"每周调仓目标股票: {self.temp_target_list}")

        # 遍历当前持仓，若股票不在目标列表，则执行卖出操作
        for stock in self.hold_list:
            if stock not in self.temp_target_list:
                pos = context.portfolio.positions.get(stock)
                self.log.info(f"调仓决策：卖出股票 {stock}")
                # 通过持仓监控器注册卖出请求，而不是直接卖出
                self.sell_request(stock, pos, 'rebalance')
                # self.strategy.state.set_sell_request(stock, pos, 'rebalance')
                self.sell_request(stock, pos, 'rebalance')
            else:
                self.log.info(f"调仓决策：继续持有股票 {stock}")

        # 对目标股票执行买入操作前，先将其注册到待买入队列
        buy_targets = [stock for stock in self.temp_target_list if stock not in self.hold_list]
        if buy_targets:
            for stock in buy_targets:
                pos = context.portfolio.positions.get(stock)
                # self.strategy.state.set_buy_request(stock, pos, 'buy')
                self.buys_request(stock, pos, 'buy')
                self.log.info(f"调仓决策：将 {len(buy_targets)} 只股票加入待买入队列: {buy_targets}")

    def handle(self, context: Any):
        self.strategy.set_target_list(self.temp_target_list)
        pass

    def end(self, context: Any):
        pass
        # 这里可以添加任何必要的清理或总结操作

    def sell_request(self, stock, data, reason):
        self.temp_sells_list[f"{stock}_{reason}"] = data

    def buys_request(self, stock, data, reason):
        self.temp_buys_list[f"{stock}_{reason}"] = data


strategy = trade_strategy("测试策略", 10000000, network_config(), log, jq_channel())


def check_state_before_func(context: Any):
    task = check_state_before_task(strategy)
    task.process(context)

def check_market_env_func(context: Any):
    task = check_market_env_task(strategy)
    task.process(context)

def check_positions_stop_loss_func(context: Any):
    task = check_positions_stop_loss_task(strategy)
    task.process(context)

def buy_stocks_func(context: Any):
    task = buy_stocks_func_task(strategy)
    task.process(context)
def initialize(context: Any) -> None:
    # # 初始化策略参数
    strategy.init_strategy()  # 策略初始化函数

    # # 注册调度任务
    run_daily(check_state_before_func, time='9:01')  # 开盘前先检查持仓状态
    run_daily(check_market_env_func, time='9:05')
    run_daily(check_positions_stop_loss_func, time='9:10')
    run_daily(buy_stocks_func, time='9:15')  # 每周调仓
    #
    #
    # run_daily(process_pending_sells_func, time='10:15')  # 处理待卖出股票
    # run_weekly(buy_stocks_func, 2, time='10:30')  # 周二进行调仓
    # run_daily(process_pending_buys_func, time='10:20')  # 处理待买入股票
    # run_daily(process_pending_sells_func, time='13:05')  # 午盘开盘后也处理一次卖出
    # run_daily(process_pending_buys_func, time='13:15')  # 午盘开盘后也处理一次买入
    # run_daily(trade_afternoon_func, time='14:30')
    # run_daily(process_pending_sells_func, time='14:40')  # 收盘前再处理一次卖出
    # run_daily(process_pending_buys_func, time='14:45')  # 收盘前再处理一次买入
    # run_daily(close_account_func, time='14:50')
    # run_weekly(print_position_info_func, 5, time='15:10')  # 周五收盘后打印持仓信息
    pass
# initialize(None)  # 调用初始化函数，开始策略执行