modules/datafact_generator/proportion_fact.py

from modules.datafact_generator.util import DataFact, DataFactGenerator
from modules.datafact_generator.value_fact import ValueFact
from typing import Any

class ProportionFact(DataFact):
    """ 单个 proportion fact """
    def __init__(self):
        super().__init__()
        self.type = "proportion"
        self.types = [
            "value_majority",
            "value_minority",
            "total_majority",
            "total_minority"
        ]

class ProportionFactGenerator(DataFactGenerator):
    """ 处理从数据提取 proportion facts 的问题 """
    def __init__(self, data: dict, value_facts: list[ValueFact]):
        super().__init__(data)

        # 使用计算好的 value facts 进行组合
        self.value_facts = value_facts

        self.total_facts: list[ValueFact] = []

        for fact in self.value_facts:
            if fact.subtype == "total":
                self.total_facts.append(fact)

    def extract_proportion_facts(self) -> list[ProportionFact]:
        proportion_facts: list[ProportionFact] = []

        # 如果数据的 y 中包含负数，那么 majoroty / minority 就失去意义了
        for single_data in self.tabular_data:
            if single_data[self.y_column] < 0:
                return []
        
        # 如果就一个 group 也不需要算了
        if self.group_column is None:
            return []
        
        # 对 value based 的操作
        # 随便找一个 x_list, 用来遍历
        any_group = next(iter(self.grouped_data.values()))
        x_list = any_group["x_list"]
        num_items = len(x_list)
        try:
            for i in range(num_items):
                # 对于某一个 x 值，把对应的 index, group_value, y 聚合起来
                info = {}
                for group_value, group_info in self.grouped_data.items():
                    y_list = group_info["y_list"]
                    indices = group_info["indices"]

                    info[group_value] = { # 构建一个 y 值和 index 的 dict 作传参
                        "y": y_list[i],
                        "index": indices[i]
                    }

                x_value = x_list[i]

                value_max_fact, value_min_fact = self._extract_value_based_facts(x_value, info)
                proportion_facts.append(value_max_fact)
                proportion_facts.append(value_min_fact)
        except Exception as e:
            print(e)

        # 直接依赖 value facts
        total_majority_fact, total_minority_fact = self._extract_total_based_facts(self.total_facts)

        proportion_facts.append(total_majority_fact)
        proportion_facts.append(total_minority_fact)
        
        return proportion_facts

    def _extract_value_based_facts(self, x_value: str, info: dict[str, dict[str, Any]]):
        """ 处理 value majority fact 和 value minority fact """
        max_proportion_fact, min_proportion_fact = ProportionFact(), ProportionFact()
        max_subtype, min_subtype = "value_majority", "value_minority"

        # 先把 y 值和 index 取出来，用 y 作索引取最大、最小值
        y_and_index = [(group_value, value["y"], value["index"]) for group_value, value in info.items()]
        y_values = [y for _, y, _ in y_and_index]
        sum_y = sum(y_values) or 1e-8

        max_y = max(y_values)
        min_y = min(y_values)

        max_proportion, min_proportion = max_y / sum_y, min_y / sum_y

        # 多个最大值/最小值情况
        max_group_values, max_indices = [], []
        min_group_values, min_indices = [], []

        for group_value, y, index in y_and_index:
            if y == max_y:
                max_group_values.append(group_value)
                max_indices.append(index)
            if y == min_y:
                min_group_values.append(group_value)
                min_indices.append(index)

        max_data_points = [self.tabular_data[index] for index in max_indices]
        min_data_points = [self.tabular_data[index] for index in min_indices]

        def generate_score():
            # 对于majority, 类别越多，分值肯定应该相对较高
            if 1 - max_proportion <= 1e-8:
                return 0.0, 0.0
            k_max = 1.0
            max_score = 1 / (1 + k_max * max_proportion / len(info))
            
            if min_proportion <= 1e-8:
                return max_score, 0.0
            k_min = 5.0
            min_score = 1 / (1 + k_min * min_proportion)
                        
            # 最小值用比例评分
            if min_proportion <= 1e-8:
                return max_score, 0.0 # 感觉一般来说有问题
            k_min = 5.0
            min_score = 1 / (1 + k_min * min_proportion)

            return max_score, min_score

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

            max_group_value_str = ", ".join(max_group_values)
            min_group_value_str = ", ".join(min_group_values)

            if len(max_group_values) == 1:
                if max_proportion > 0.5:
                    max_annotation = f"The {max_group_value_str} accounts for the majority in {x_value}."
                    max_reason = (
                        f"The {self.y_column} of {max_group_value_str} in {x_value} accounts for {max_proportion} (more than 50%) "
                        f"of the total of {self.y_column} of all {self.group_column}."
                    )
                else:
                    max_annotation = f"The {max_group_value_str} has the largest proportion in {x_value}."
                    max_reason = (
                        f"The {self.y_column} of {max_group_value_str} in {x_value} accounts for {max_proportion}, larger than all "
                        f"other {self.group_column}, of the total of {self.y_column} of all {self.group_column}."
                    )
            else:
                max_annotation = f"The {max_group_value_str} all have the largest proportion in {x_value}."
                max_reason = (
                    f"The {self.y_column} of {max_group_value_str} in {x_value} all account for {max_proportion}, larger than all "
                    f"other {self.group_column}, of the total of {self.y_column} of all {self.group_column}."
                )

            # 正常来说除非就两类，不然占"少数"应该是正常情况
            if len(min_group_values) == 1:
                min_annotation = f"The {min_group_value_str} has the smallest proportion in {x_value}."
                min_reason = (
                    f"The {self.y_column} of {min_group_value_str} in {x_value} accounts for {min_proportion}, smaller than all "
                    f"other {self.group_column}, of the total of {self.y_column} of all {self.group_column}."
                )
            else:
                min_annotation = f"The {min_group_value_str} all have the smallest proportion in {x_value}."
                min_reason = (
                    f"The {self.y_column} of {min_group_value_str} in {x_value} all account for {min_proportion}, smaller than all "
                    f"other {self.group_column}, of the total of {self.y_column} of all {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_proportion_fact.set_value(
            max_subtype, max_data_points, max_score, max_annotation, max_reason
        )

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

        return max_proportion_fact, min_proportion_fact

    def _extract_total_based_facts(self, total_facts: list[ValueFact]):
        """ 处理 total majority fact 和 total minority fact """
        max_proportion_fact, min_proportion_fact = ProportionFact(), ProportionFact()
        max_subtype, min_subtype = "value_majority", "value_minority"

        max_total = max(total_facts, key=lambda x: x.data_points[0][self.y_column]).data_points[0][self.y_column]
        min_total = min(total_facts, key=lambda x: x.data_points[0][self.y_column]).data_points[0][self.y_column]

        sum_total = sum([total_fact.data_points[0][self.y_column] for total_fact in total_facts]) or 1e-8

        max_proportion = max_total / sum_total
        min_proportion = min_total / sum_total

        max_total_facts: list[ValueFact] = []
        min_total_facts: list[ValueFact] = []

        for total_fact in total_facts:
            if total_fact.data_points[0][self.y_column] == max_total:
                max_total_facts.append(total_fact)
            if total_fact.data_points[0][self.y_column] == min_total:
                min_total_facts.append(total_fact)

        # total fact 的 data pooints 必定只有一项
        max_data_points = [total_fact.data_points[0] for total_fact in max_total_facts]
        min_data_points = [total_fact.data_points[0] for total_fact in max_total_facts]

        def generate_score():
            if 1 - max_proportion <= 1e-8:
                return 0.0, 0.0
            k_max = 1.0
            max_score = 1 / (1 + k_max * max_proportion / len(total_facts))
            
            if min_proportion <= 1e-8:
                return max_score, 0.0
            k_min = 5.0
            min_score = 1 / (1 + k_min * min_proportion)
            
            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:
                if max_proportion > 0.5:
                    max_annotation = f"The total value of {max_group_value_str} accounts for the majority in all {self.group_column}."
                    max_reason = (
                        f"The total value of {self.y_column} of {max_group_value_str} accounts for {max_proportion} (more than 50%) "
                        f"of the total of {self.y_column} of all {self.group_column}."
                    )
                else:
                    max_annotation = f"The total value of {max_group_value_str} has the largest proportion."
                    max_reason = (
                        f"The total value of {self.y_column} of {max_group_value_str} accounts for {max_proportion}, larger than all "
                        f"other {self.group_column}, of the total of {self.y_column} of all {self.group_column}."
                    )
            else:
                max_annotation = f"The total value of {max_group_value_str} all have the largest proportion."
                max_reason = (
                    f"The total value of {self.y_column} of {max_group_value_str} all account for {max_proportion}, larger than all "
                    f"other {self.group_column}, of the total of {self.y_column} of all {self.group_column}."
                )

            if len(min_data_points) == 1:
                min_annotation = f"The total value of {min_group_value_str} has the smallest proportion."
                min_reason = (
                    f"The total value of {self.y_column} of {min_group_value_str} accounts for {min_proportion}, smaller than all "
                    f"other {self.group_column}, of the total of {self.y_column} of all {self.group_column}."
                )
            else:
                min_annotation = f"The total value of {min_group_value_str} all have the smallest proportion."
                min_reason = (
                    f"The total value of {self.y_column} of {min_group_value_str} all account for {min_proportion}, smaller than all "
                    f"other {self.group_column}, of the total of {self.y_column} of all {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_proportion_fact.set_value(
            max_subtype, max_data_points, max_score, max_annotation, max_reason
        )

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

        return max_proportion_fact, min_proportion_fact