modules/datafact_generator/value_fact.py

from modules.datafact_generator.util import DataFact, DataFactGenerator
from statistics import mean, stdev
from scipy.special import expit

class ValueFact(DataFact):
    """
    单个 value_fact.
    NOTE 这里 avg, total 均并无实际意义, score 统一设为 0, annotation, reason 统一设为 "",
        其目的是后续不同 group 的 total, avg 比较得到组合 facts
    """
    def __init__(self):
        super().__init__()
        self.type: str = "value"
        self.types = ["max", "min", "avg", "total"] # 所有可选的 value_fact

class ValueFactGenerator(DataFactGenerator):
    """ 处理从数据提取 value_facts 的问题 """
    def __init__(self, data: dict):
        super().__init__(data)

    def extract_value_facts(self) -> list[ValueFact]:
        """ 暴露的接口，提取数据中所有 value_facts """
        value_facts: list[ValueFact] = []
        
        for group_value in self.grouped_data.keys():
            indices = self.grouped_data[group_value]["indices"]
            y_list = self.grouped_data[group_value]["y_list"]

            max_fact = self._extract_max(group_value, indices, y_list)
            min_fact = self._extract_min(group_value, indices, y_list)
            avg_fact = self._extract_avg(group_value, indices, y_list)
            total_fact = self._extract_total(group_value, indices, y_list)

            value_facts.extend([max_fact, min_fact, avg_fact, total_fact])

        return value_facts

    def _extract_max(self, group_value: str, indices: list[int], y_list: list):
        """ 提取单个 group 中 subtype 为 max 的 facts """
        value_fact = ValueFact()
        subtype = "max"

        # 先找到所有最大值在这组内的序号，再用每个组内序号索引全局序号
        max_val = max(y_list)
        all_max_indices = [i for i, v in enumerate(y_list) if v == max_val]
        data_points = [self.tabular_data[indices[i]] for i in all_max_indices]

        def generate_score():
            """ 计算最大值评分 """    
            mu = mean(y_list)
            sigma = stdev(y_list)

            if sigma == 0:
                return 1.0 if max_val > mu else 0.0

            z = (max_val - mu) / sigma

            # 套一个 sigmoid, 控制一下
            k = 2.0
            z0 = 0.8
            score = expit(k * (z - z0))

            return score

        def generate_annotation_and_reason():
            """ 生成注释 """
            max_positions = [data_points[i].get(self.x_column) for i in range(len(data_points))]
            max_positions_str = ", ".join(max_positions)

            annotation, reason = "", ""

            if len(data_points) > 1:
                annotation = f"The {group_value} has maximum values at {max_positions_str}"
            else:
                annotation = f"The {group_value} has a maximum value at {max_positions_str}"

            # 如果是时序的，我们说它是范围内最大的；如果不是，我们说它是所有类别中最大的
            if self.is_temporal:
                # 我们假设数据是按照时序排好的
                temporal_begin = self.tabular_data[0][self.x_column]
                temporal_end = self.tabular_data[-1][self.x_column]

                if len(data_points) > 1:
                    reason = f"The {self.y_column} of {group_value} have maximum values of {max_val}, which is the largest from {temporal_begin} to {temporal_end}."
                else:
                    reason = f"The {self.y_column} of {group_value} has a maximum value of {max_val}, which is the largest from {temporal_begin} to {temporal_end}."
            else:
                if len(data_points) > 1:
                    reason = f"The {self.y_column} of {group_value} have maximum values of {max_val}, which is the largest in all categories."
                else:
                    reason = f"The {self.y_column} of {group_value} has a maximum value of {max_val}, which is the largest in all categories."

            return annotation, reason

        score = generate_score()
        annotation, reason = generate_annotation_and_reason()

        value_fact.set_value(subtype, data_points, score, annotation, reason)

        return value_fact

    def _extract_min(self, group_value: str, indices: list[int], y_list: list):
        """ 提取单个 group 中 subtype 为 min 的 facts """
        value_fact = ValueFact()
        subtype = "min"

        # 先找到所有最大值在这组内的序号，再用每个组内序号索引全局序号
        min_val = min(y_list)
        all_min_indices = [i for i, v in enumerate(y_list) if v == min_val]
        data_points = [self.tabular_data[indices[i]] for i in all_min_indices]

        def generate_score():
            """ 计算最小值评分（值越小、越异常，分数越高） """    
            min_val = min(y_list)
            mu = mean(y_list)
            sigma = stdev(y_list)

            if sigma == 0:
                return 1.0 if min_val < mu else 0.0

            z = (mu - min_val) / sigma
            
            k = 2.0
            z0 = 0.8
            score = expit(k * (z - z0))

            return score

        def generate_annotation_and_reason():
            """ 生成注释 """
            min_positions = [data_points[i].get(self.x_column) for i in range(len(data_points))]
            min_positions_str = ", ".join(min_positions)

            annotation, reason = "", ""

            if len(data_points) > 1:
                annotation = f"The {group_value} has minimum values at {min_positions_str}"
            else:
                annotation = f"The {group_value} has a minimum value at {min_positions_str}"

            # 如果是时序的，我们说它是范围内最大的；如果不是，我们说它是所有类别中最大的
            if self.is_temporal:
                # 我们假设数据是按照时序排好的
                temporal_begin = self.tabular_data[0][self.x_column]
                temporal_end = self.tabular_data[-1][self.x_column]

                if len(data_points) > 1:
                    reason = f"The {self.y_column} of {group_value} have minimum values of {min_val}, which is the largest from {temporal_begin} to {temporal_end}."
                else:
                    reason = f"The {self.y_column} of {group_value} has a minimum value of {min_val}, which is the largest from {temporal_begin} to {temporal_end}."
            else:
                if len(data_points) > 1:
                    reason = f"The {self.y_column} of {group_value} have minimum values of {min_val}, which is the largest in all categories."
                else:
                    reason = f"The {self.y_column} of {group_value} has a mainimum value of {min_val}, which is the largest in all categories."

            return annotation, reason

        score = generate_score()
        annotation, reason = generate_annotation_and_reason()

        value_fact.set_value(subtype, data_points, score, annotation, reason)

        return value_fact

    def _extract_avg(self, group_value: str, indices: list[int], y_list: list):
        """ 提取提取单个 group 中数据中 subtype 为 avg 的 facts """
        # 参见 ValueFact 说明，这里无实际意义

        # data_points 中的 x 值用 "avg" 替换
        value_fact = ValueFact()
        subtype = "avg"

        # 先找到所有最大值在这组内的序号，再用每个组内序号索引全局序号
        avg_val = sum(y_list) / len(y_list)
        
        data_points = {}
        if self.group_column:
            data_points = [{
                self.group_column: group_value,
                self.x_column: "avg",
                self.y_column: avg_val
            }]
        else:
            data_points = [{
                self.x_column: "avg",
                self.y_column: avg_val
            }]

        value_fact.set_value(subtype, data_points, 0, "", "")

        return value_fact

    def _extract_total(self, group_value: str, indices: list[int], y_list: list):
        """ 提取单个 group 中 subtype 为 total 的 facts """
        value_fact = ValueFact()
        subtype = "total"

        # 先找到所有最大值在这组内的序号，再用每个组内序号索引全局序号
        total_val = sum(y_list)
        
        data_points = {}
        if self.group_column:
            data_points = [{
                self.group_column: group_value,
                self.x_column: "total",
                self.y_column: total_val
            }]
        else:
            data_points = [{
                self.x_column: "total",
                self.y_column: total_val
            }]

        value_fact.set_value(subtype, data_points, 0, "", "")

        return value_fact