modules/datafact_generator/difference_fact.py

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

class DifferenceFact(DataFact):
    """ 单个 difference fact """
    def __init__(self):
        super().__init__()
        self.type = "difference"
        self.types = [
            # 不同 group 之间 value 比较
            "maximum_large",
            "maximum_small",
            "minimum_large",
            "minimum_small",
            "average_large",
            "average_small",

            # temporal
            "sudden_increase", # 相邻时间相差很多
            "sudden_decrease",

            # categorical
            "sudden_change" # 按照值排序后，相邻的相差很多
        ]

class DifferenceFactGenerator(DataFactGenerator):
    def __init__(self, data: dict, value_facts: list[ValueFact]):
        super().__init__(data)

        self.value_facts = value_facts

        self.max_facts: list[ValueFact] = []
        self.min_facts: list[ValueFact] = []
        self.avg_facts: list[ValueFact] = []

        for fact in self.value_facts:
            if fact.subtype == "max":
                self.max_facts.append(fact)
            if fact.subtype == "min":
                self.min_facts.append(fact)
            if fact.subtype == "avg":
                self.avg_facts.append(fact)

    def extract_difference_facts(self) -> list[DifferenceFact]:
        proportion_facts: list[DifferenceFact] = []

        # 不同 group 之间 value 比较
        if len(self.max_facts) > 1:
            max_large_fact, max_small_fact = self._extract_max_based_facts(self.max_facts)
            min_large_fact, min_small_fact = self._extract_min_based_facts(self.min_facts)
            avg_large_fact, avg_small_fact = self._extract_avg_based_facts(self.avg_facts)

            proportion_facts.extend([max_large_fact, max_small_fact, min_large_fact, min_small_fact, avg_large_fact, avg_small_fact])

        # sudden
        for group_value in self.grouped_data.keys():
            indices = self.grouped_data[group_value]["indices"]
            y_list = self.grouped_data[group_value]["y_list"]
            if self.is_temporal:
                increase_difference_fact, decrease_difference_fact = self._extract_temporal_sudden(group_value, indices, y_list)
                proportion_facts.append(increase_difference_fact)
                proportion_facts.append(decrease_difference_fact)
            else:
                difference_fact = self._extract_categorical_sudden(group_value, indices, y_list)
                proportion_facts.append(difference_fact)

        return proportion_facts

    def _extract_facts_base(self, fact_type: str, facts: list[ValueFact]):
        """ 模版函数 """
        max_differencen_fact, min_difference_fact = DifferenceFact(), DifferenceFact()
        max_subtype, min_subtype = f"{fact_type}_large", f"{fact_type}_small"

        # value fact 的处理中, max, min, avg 统一地会把那个值放在 data_points[*][y_column] 中
        max_val = max(facts, key=lambda x: x.data_points[0][self.y_column]).data_points[0][self.y_column] 
        min_val = min(facts, key=lambda x: x.data_points[0][self.y_column]).data_points[0][self.y_column] 

        all_val = [fact.data_points[0][self.y_column] for fact in facts]

        max_facts: list[ValueFact] = []
        min_facts: list[ValueFact] = []

        max_data_points, min_data_points = [], []

        for fact in facts:
            if fact.data_points[0][self.y_column] == max_val:
                max_facts.append(fact)
                max_data_points.extend(fact.data_points)
            if fact.data_points[0][self.y_column] == min_val:
                min_facts.append(fact)
                min_data_points.extend(fact.data_points)

        def generate_score():
            """ 统一使用单边 z 检验 """
            mu = mean(all_val)
            sigma = stdev(all_val)
            if sigma == 0:
                max_score = 1.0 if max_val > mu else 0.0
                min_score = 1.0 if min_val < mu else 0.0
                return max_score, min_score
            k = 2.0
            z0 = 0.8
            
            z = (max_val - mu) / sigma
            max_score = expit(k * (z - z0))

            z = (mu - min_val) / sigma  
            min_score = expit(k * (z - z0))

            return max_score, min_score
        
        def generate_annotation_and_reason():
            max_annotation, max_reason = "", ""
            min_annotation, min_reason = "", ""

            max_group_value_str = ", ".join([max_data_point[self.group_column] for max_data_point in max_data_points])
            min_group_value_str = ", ".join([min_data_point[self.group_column] for min_data_point in min_data_points])
            
            if len(max_data_points) == 1:
                max_annotation = f"The {fact_type} value of {max_group_value_str} is the largest in all groups."
                max_reason = (
                    f"The {fact_type} value of {self.y_column} of {max_group_value_str} has a value of {max_val}, "
                    f"which is larger than all other {self.group_column}."
                )
            else:
                max_annotation = f"The {fact_type} value of {max_group_value_str} are all the largest in all groups."
                max_reason = (
                    f"The {fact_type} value of {self.y_column} of {max_group_value_str} all have a value of {max_val}, "
                    f"which is larger than all other {self.group_column}."
                )

            if len(min_data_points) == 1:
                min_annotation = f"The {fact_type} value of {min_group_value_str} is the smallest in all groups."
                min_reason = (
                    f"The {fact_type} value of {self.y_column} of {min_group_value_str} has a value of {min_val}, "
                    f"which is smaller than all other {self.group_column}."
                )
            else:
                min_annotation = f"The {fact_type} value of {min_group_value_str} are all the smallest in all groups."
                min_reason = (
                    f"The {fact_type} value of {self.y_column} of {min_group_value_str} all have a value of {max_val}, "
                    f"which is smaller than all other {self.group_column}."
                )

            return max_annotation, max_reason, min_annotation, min_reason

        max_score, min_score = generate_score()
        max_annotation, max_reason, min_annotation, min_reason = generate_annotation_and_reason()

        max_differencen_fact.set_value(
            max_subtype, max_data_points, max_score, max_annotation, max_reason
        )

        min_difference_fact.set_value(
            min_subtype, min_data_points, min_score, min_annotation, min_reason
        )

        return max_differencen_fact, min_difference_fact

    def _extract_max_based_facts(self, max_facts: list[ValueFact]):
        return self._extract_facts_base("maximum", max_facts)

    def _extract_min_based_facts(self, min_facts: list[ValueFact]):
        return self._extract_facts_base("minimum", min_facts)

    def _extract_avg_based_facts(self, avg_facts: list[ValueFact]):
        return self._extract_facts_base("average", avg_facts)

    def _extract_temporal_sudden(self, group_value: str, indices: list[int], y_list: list):
        """ 选择一个 group 中最显著的上升 / 下降 """

        # 找到相邻值中绝对值相差最大的

        max_diff_idx_increase = []
        max_diff_increase = 0
        
        max_diff_idx_decrease = []
        max_diff_decrease = 0
        
        for idx in range(len(y_list)-1):
            diff = abs(y_list[idx] - y_list[idx+1])

            if y_list[idx] < y_list[idx+1]: # sudden increase
                if diff > max_diff_increase:
                    max_diff_increase = diff
                    max_diff_idx_increase = [idx]
                elif diff == max_diff_increase:
                    max_diff_idx_increase.append(idx)
            
            else: # sudden decrease
                if diff < max_diff_decrease:
                    max_diff_decrease = diff
                    max_diff_idx_decrease = [idx]
                elif diff == max_diff_decrease:
                    max_diff_idx_decrease.append(idx)
        
        increase_difference_fact, decrease_difference_fact = DifferenceFact(), DifferenceFact()
        increase_subtype, decrease_subtype = "sudden_increase", "sudden_decrease"

        before_increase_data_points = [self.tabular_data[indices[i]] for i in max_diff_idx_increase]
        before_decrease_data_points = [self.tabular_data[indices[i]] for i in max_diff_idx_decrease]

        after_increase_data_points = [self.tabular_data[indices[i+1]] for i in max_diff_idx_increase]
        after_decrease_data_points = [self.tabular_data[indices[i+1]] for i in max_diff_idx_decrease]

        def generate_score():
            max_val = max(y_list)
            min_val = min(y_list)

            k = 4.0
            z0 = 0.1

            increase_ratio = max_diff_increase / (max_val - min_val)
            decrease_ratio = max_diff_decrease / (max_val - min_val)

            increase_score = expit(k * (increase_ratio - z0)) if max_diff_increase else 0.0
            decrease_score = expit(k * (decrease_ratio - z0)) if max_diff_decrease else 0.0

            return increase_score, decrease_score

        def generate_annotation_and_reason():
            increase_annotation, increase_reason = "", ""
            decrease_annotation, decrease_reason = "", ""

            increase_positions = [data_point.get(self.x_column) for data_point in after_increase_data_points]
            increase_positions_str = ", ".join(increase_positions)
            after_increase_values = [str(data_point.get(self.y_column)) for data_point in after_increase_data_points]
            after_increase_values_str = ", ".join(after_increase_values)
            before_increase_values = [str(data_point.get(self.y_column)) for data_point in before_increase_data_points]
            before_increase_values_str = ", ".join(before_increase_values)

            decrease_positions = [data_point.get(self.x_column) for data_point in after_decrease_data_points]
            decrease_positions_str = ", ".join(decrease_positions)
            after_decrease_values = [str(data_point.get(self.y_column)) for data_point in after_decrease_data_points]
            after_decrease_values_str = ", ".join(after_decrease_values)
            before_decrease_values = [str(data_point.get(self.y_column)) for data_point in before_decrease_data_points]
            before_decrease_values_str = ", ".join(before_decrease_values)
            
            if len(max_diff_idx_increase) == 1:
                increase_annotation = f"The {group_value} shows a sudden increase at {increase_positions_str}."
                increase_reason = (
                    f"The {self.y_column} for {group_value} at {increase_positions_str} is {after_increase_values_str}, "
                    f"which is significantly higher than the previous value of {before_increase_values_str}."
                )
            else:
                increase_annotation = f"The {group_value} exhibits sudden increases at multiple points: {increase_positions_str}."
                increase_reason = (
                    f"At these positions, the {self.y_column} values for {group_value} are {after_increase_values_str}, "
                    f"which are significantly higher than the preceding values of {before_increase_values_str}."
                )

            if len(max_diff_idx_decrease) == 1:
                decrease_annotation = f"The {group_value} shows a sudden decrease at {decrease_positions_str}."
                decrease_reason = (
                    f"The {self.y_column} for {group_value} at {decrease_positions_str} is {after_decrease_values_str}, "
                    f"which is significantly lower than the previous value of {before_decrease_values_str}."
                )
            else:
                decrease_annotation = f"The {group_value} exhibits sudden decreases at multiple points: {decrease_positions_str}."
                decrease_reason = (
                    f"At these positions, the {self.y_column} values for {group_value} are {after_decrease_values_str}, "
                    f"which are significantly lower than the preceding values of {before_decrease_values_str}."
                )

            return increase_annotation, increase_reason, decrease_annotation, decrease_reason
        
        increase_score, decrease_score = generate_score()
        increase_annotation, increase_reason, decrease_annotation, decrease_reason = generate_annotation_and_reason()

        increase_difference_fact.set_value(
            increase_subtype, after_increase_data_points, increase_score, increase_annotation, increase_reason
        )

        decrease_difference_fact.set_value(
            decrease_subtype, after_decrease_data_points, decrease_score, decrease_annotation, decrease_reason
        )

        return increase_difference_fact, decrease_difference_fact

    def _extract_categorical_sudden(self, group_value: str, indices: list[int], y_list: list):
        difference_fact = DifferenceFact()
        subtype = "sudden_change"

        sorted_pairs = sorted(zip(y_list, indices), key=lambda x: x[0]) 
        sorted_y_list, sorted_indices = zip(*sorted_pairs)

        y_list = list(sorted_y_list)
        indices = list(sorted_indices)

        max_diff_idx = []
        max_diff = 0
        
        for idx in range(len(y_list)-1):
            diff = y_list[idx+1] - y_list[idx]

            if diff > max_diff:
                max_diff = diff
                max_diff_idx = [idx]
            elif diff == max_diff:
                max_diff_idx.append(idx)

        before_change_data_points = [self.tabular_data[indices[i]] for i in max_diff_idx]
        after_change_data_points = [self.tabular_data[indices[i+1]] for i in max_diff_idx]

        def generate_score():
            max_val = max(y_list)
            min_val = min(y_list)

            k = 4.0
            z0 = 0.1

            ratio = max_diff / (max_val - min_val)
            score = expit(k * (ratio - z0)) if max_diff else 0

            return score
        
        def generate_annotation_and_reason():
            annotation, reason = "", ""

            before_change_positions = [data_point.get(self.x_column) for data_point in before_change_data_points]
            after_change_positions = [data_point.get(self.x_column) for data_point in after_change_data_points]
            before_change_positions_str = ", ".join(before_change_positions)
            after_change_positions_str = ", ".join(after_change_positions)

            after_change_values = [str(data_point.get(self.y_column)) for data_point in after_change_data_points]
            after_change_values_str = ", ".join(after_change_values)
            before_change_values = [str(data_point.get(self.y_column)) for data_point in before_change_data_points]
            before_change_values_str = ", ".join(before_change_values)
            
            if len(max_diff_idx) == 1:
                annotation = (
                    f"The {group_value} shows a sudden jump in {self.y_column} from {before_change_positions_str} "
                    f"to {after_change_positions_str}."
                )
                reason = (
                    f"The {self.y_column} for {group_value} differs significantly between {before_change_values_str} at "
                    f"{before_change_positions_str} and {after_change_values_str} at {after_change_positions_str}."
                )
            else:
                annotation = (
                    f"The {group_value} exhibits multiple sudden jumps in {self.y_column}, transitioning from "
                    f"{before_change_positions_str} to {after_change_positions_str}."
                )
                reason = (
                    f"Across these points, the {self.y_column} for {group_value} differs significantly between "
                    f"{before_change_values_str} at {before_change_positions_str} and "
                    f"{after_change_values_str} at {after_change_positions_str}."
                )

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

        difference_fact.set_value(
            subtype, after_change_data_points, score, annotation, reason
        )

        return difference_fact