fix(roi): persist fixed cohort display coefficient

examples/auto_put_ad_mini/roi_control/fission_multiplier.py

@@ -17,11 +17,14 @@ from typing import Mapping
 import pandas as pd
 
 
-DEFAULT_FISSION_PARAMETER_VERSION = "20260712_A0-A15_v1"
+LEGACY_FISSION_PARAMETER_VERSION = "20260712_A0-A15_v1"
+DEFAULT_FISSION_PARAMETER_VERSION = "20260712_A0-A15_v2"
 FISSION_HORIZON_DAYS = 15
 PARAMETER_ROOT = Path(__file__).resolve().parent / "data" / "fission_multiplier"
 SOURCE_MULTIPLIER_COLUMN = "T15成熟系数（相对T0）"
 DISPLAY_MULTIPLIER_COLUMN = "传播裂变系数-对T0裂变"
+SOURCE_TOTAL_TO_FIRST_COLUMN = "每首层累计裂变活跃人次"
+DISPLAY_TOTAL_TO_FIRST_COLUMN = "裂变系数-总裂变UV/首层UV"
 
 MINIAPP_FILE = "小程序T15成熟系数_中文_20260712.csv"
 GZH_FILE = "公众号T15成熟系数_中文_20260712.csv"
@@ -45,26 +48,44 @@ class ParameterRelease:
     observation_end_date: str
     run_suffix: str
     file_sha256: Mapping[str, str]
+    source_version: str
+    includes_total_to_first: bool
 
 
 RELEASES: Mapping[str, ParameterRelease] = {
+    LEGACY_FISSION_PARAMETER_VERSION: ParameterRelease(
+        version=LEGACY_FISSION_PARAMETER_VERSION,
+        cohort_date="20260712",
+        observation_end_date="20260727",
+        run_suffix="f0712v1",
+        file_sha256={
+            MINIAPP_FILE: "9358408ad4957bbd22f689f8bbf66b7abe765f210e5af444d11e329815fee6a4",
+            GZH_FILE: "1f6f17b1403aa9b56a33195da3c324fbe06cbb714aa2ebabcdbceb891aa7a951",
+            FALLBACK_FILE: "6e4d517595ec2632dc54a65000abe94000b5313e77420547759b2419314d60ac",
+        },
+        source_version=LEGACY_FISSION_PARAMETER_VERSION,
+        includes_total_to_first=False,
+    ),
     DEFAULT_FISSION_PARAMETER_VERSION: ParameterRelease(
         version=DEFAULT_FISSION_PARAMETER_VERSION,
         cohort_date="20260712",
         observation_end_date="20260727",
-        run_suffix="f0712v1",
+        run_suffix="f0712v2",
         file_sha256={
             MINIAPP_FILE: "9358408ad4957bbd22f689f8bbf66b7abe765f210e5af444d11e329815fee6a4",
             GZH_FILE: "1f6f17b1403aa9b56a33195da3c324fbe06cbb714aa2ebabcdbceb891aa7a951",
             FALLBACK_FILE: "6e4d517595ec2632dc54a65000abe94000b5313e77420547759b2419314d60ac",
         },
-    )
+        source_version=LEGACY_FISSION_PARAMETER_VERSION,
+        includes_total_to_first=True,
+    ),
 }
 
 
 @dataclass(frozen=True)
 class FissionMultiplierMatch:
     multiplier: float
+    multiplier_vs_first: float | None
     match_level: str
     source: str
 
@@ -79,13 +100,14 @@ class FissionMultiplierParameters:
     gzh_exact: Mapping[tuple[str, str], float]
     gzh_by_partner: Mapping[str, float]
     gzh_channel: float
+    total_to_first: Mapping[tuple[str, str, str, str], float]
     miniapp_exact_rows: int
     miniapp_exact_available_rows: int
     gzh_exact_rows: int
     gzh_exact_available_rows: int
 
     def snapshot(self) -> dict[str, object]:
-        return {
+        snapshot = {
             "version": self.release.version,
             "cohort_date": self.release.cohort_date,
             "observation_end_date": self.release.observation_end_date,
@@ -97,6 +119,28 @@ class FissionMultiplierParameters:
             "miniapp_channel_multiplier": self.miniapp_channel,
             "gzh_channel_multiplier": self.gzh_channel,
         }
+        if self.release.includes_total_to_first:
+            snapshot["includes_total_to_first"] = True
+            snapshot["source_version"] = self.release.source_version
+        return snapshot
+
+    def _match_result(
+        self,
+        multiplier: float,
+        entity_type: str,
+        match_level: str,
+        key_primary: str,
+        key_secondary: str,
+        source: str,
+    ) -> FissionMultiplierMatch:
+        return FissionMultiplierMatch(
+            multiplier=multiplier,
+            multiplier_vs_first=self.total_to_first.get(
+                (entity_type, match_level, key_primary, key_secondary)
+            ),
+            match_level=match_level,
+            source=source,
+        )
 
     def match(
         self,
@@ -112,26 +156,38 @@ class FissionMultiplierParameters:
             goal_key = normalize_name(optimize_goal)
             exact_key = (package_key, goal_key)
             if goal_key and exact_key in self.miniapp_exact:
-                return FissionMultiplierMatch(
+                return self._match_result(
                     self.miniapp_exact[exact_key],
+                    "self",
                     MATCH_MINIAPP_EXACT,
+                    package_key,
+                    goal_key,
                     f"人群包+转化目标精确:{package_key}|{goal_key}",
                 )
             if package_key in self.miniapp_by_package:
-                return FissionMultiplierMatch(
+                return self._match_result(
                     self.miniapp_by_package[package_key],
+                    "self",
                     MATCH_MINIAPP_PACKAGE,
+                    package_key,
+                    "",
                     f"人群包回退:{package_key}",
                 )
             if goal_key and goal_key in self.miniapp_by_goal:
-                return FissionMultiplierMatch(
+                return self._match_result(
                     self.miniapp_by_goal[goal_key],
+                    "self",
                     MATCH_MINIAPP_GOAL,
+                    goal_key,
+                    "",
                     f"转化目标回退:{goal_key}",
                 )
-            return FissionMultiplierMatch(
+            return self._match_result(
                 self.miniapp_channel,
+                "self",
                 MATCH_MINIAPP_CHANNEL,
+                "",
+                "",
                 "小程序渠道回退",
             )
 
@@ -140,26 +196,36 @@ class FissionMultiplierParameters:
             account_key = normalize_name(official_account)
             exact_key = (partner_key, account_key)
             if partner_key and account_key and exact_key in self.gzh_exact:
-                return FissionMultiplierMatch(
+                return self._match_result(
                     self.gzh_exact[exact_key],
+                    "gzh",
                     MATCH_GZH_EXACT,
+                    partner_key,
+                    account_key,
                     f"合作方+公众号精确:{partner_key}|{account_key}",
                 )
             if partner_key in self.gzh_by_partner:
-                return FissionMultiplierMatch(
+                return self._match_result(
                     self.gzh_by_partner[partner_key],
+                    "gzh",
                     MATCH_GZH_PARTNER,
+                    partner_key,
+                    "",
                     f"合作方回退:{partner_key}",
                 )
-            return FissionMultiplierMatch(
+            return self._match_result(
                 self.gzh_channel,
+                "gzh",
                 MATCH_GZH_CHANNEL,
+                "",
+                "",
                 "公众号渠道回退",
             )
 
         if entity_type == "qiwei":
             return FissionMultiplierMatch(
                 QIWEI_REFERENCE_MULTIPLIER,
+                None,
                 MATCH_QIWEI_REFERENCE,
                 f"企微渠道临时参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待补算:"
                 f"{normalize_name(partner)}",
@@ -182,15 +248,20 @@ def parameter_values(
         key_secondary: str,
         multiplier: float,
     ) -> None:
-        rows.append(
-            {
+        row = {
                 "entity_type": entity_type,
                 "match_level": match_level,
                 "key_primary": key_primary,
                 "key_secondary": key_secondary,
                 "multiplier": float(multiplier),
             }
-        )
+        if parameters.release.includes_total_to_first:
+            key = (entity_type, match_level, key_primary, key_secondary)
+            multiplier_vs_first = parameters.total_to_first.get(key)
+            if multiplier_vs_first is None:
+                raise ValueError(f"传播裂变系数缺少总裂变UV/首层UV参数: {key}")
+            row["multiplier_vs_first"] = float(multiplier_vs_first)
+        rows.append(row)
 
     for (package, goal), multiplier in parameters.miniapp_exact.items():
         append("self", MATCH_MINIAPP_EXACT, package, goal, multiplier)
@@ -241,6 +312,7 @@ def parameters_from_database(
 
     metadata = json.loads(str(release_row.get("metadata_json") or "{}"))
     file_sha256 = metadata.get("file_sha256") or {}
+    registered_release = RELEASES.get(str(release_row["version"]))
     release = ParameterRelease(
         version=str(release_row["version"]),
         cohort_date=str(release_row["cohort_date"]).replace("-", ""),
@@ -249,8 +321,17 @@ def parameters_from_database(
         ),
         run_suffix=str(release_row["run_suffix"]),
         file_sha256=file_sha256,
+        source_version=str(
+            metadata.get("source_version")
+            or (registered_release.source_version if registered_release else release_row["version"])
+        ),
+        includes_total_to_first=bool(
+            metadata.get("includes_total_to_first")
+            or (registered_release.includes_total_to_first if registered_release else False)
+        ),
     )
     indexed: dict[tuple[str, str, str, str], float] = {}
+    total_to_first: dict[tuple[str, str, str, str], float] = {}
     for row in value_rows:
         key = (
             str(row["entity_type"]),
@@ -264,6 +345,16 @@ def parameters_from_database(
         if not math.isfinite(multiplier) or multiplier < 1:
             raise ValueError(f"数据库传播裂变系数无效: {key}={multiplier}")
         indexed[key] = multiplier
+        raw_vs_first = row.get("multiplier_vs_first")
+        if raw_vs_first is not None:
+            multiplier_vs_first = float(raw_vs_first)
+            if not math.isfinite(multiplier_vs_first) or multiplier_vs_first < 0:
+                raise ValueError(
+                    f"数据库总裂变UV/首层UV系数无效: {key}={multiplier_vs_first}"
+                )
+            total_to_first[key] = multiplier_vs_first
+    if release.includes_total_to_first and set(total_to_first) != set(indexed):
+        raise ValueError("数据库总裂变UV/首层UV参数与T0参数匹配键不一致")
 
     def values(entity_type: str, level: str) -> dict[tuple[str, str], float]:
         return {
@@ -301,6 +392,7 @@ def parameters_from_database(
         gzh_exact=gzh_exact,
         gzh_by_partner=gzh_partner,
         gzh_channel=single("gzh", MATCH_GZH_CHANNEL),
+        total_to_first=total_to_first,
         miniapp_exact_rows=int(metadata["miniapp_exact_rows"]),
         miniapp_exact_available_rows=int(metadata["miniapp_exact_available_rows"]),
         gzh_exact_rows=int(metadata["gzh_exact_rows"]),
@@ -366,6 +458,7 @@ def _validated_multipliers(
         "A0-A15累计裂变活跃人次",
         "T1-T15相对T0尾部系数",
         SOURCE_MULTIPLIER_COLUMN,
+        SOURCE_TOTAL_TO_FIRST_COLUMN,
     ]
     for column in numeric_columns:
         if column in result:
@@ -391,6 +484,21 @@ def _validated_multipliers(
     if bool(tail_errors.gt(1e-9).any()):
         raise ValueError(f"{label}参数不满足尾部系数=传播裂变系数-1")
 
+    expected_vs_first = (
+        result["A0-A15累计裂变活跃人次"] / result["首层UV"]
+    ).where(result["首层UV"].gt(0))
+    if SOURCE_TOTAL_TO_FIRST_COLUMN in frame:
+        checkable_vs_first = result["首层UV"].gt(0) & result[
+            SOURCE_TOTAL_TO_FIRST_COLUMN
+        ].notna()
+        errors_vs_first = (
+            result.loc[checkable_vs_first, SOURCE_TOTAL_TO_FIRST_COLUMN]
+            - expected_vs_first[checkable_vs_first]
+        ).abs()
+        if bool(errors_vs_first.gt(1e-9).any()):
+            raise ValueError(f"{label}参数不满足累计活跃人次/首层UV恒等式")
+    result[SOURCE_TOTAL_TO_FIRST_COLUMN] = expected_vs_first
+
     available = result[result["样本状态"].eq("可用")].copy()
     valid = available[SOURCE_MULTIPLIER_COLUMN].map(
         lambda value: bool(pd.notna(value) and math.isfinite(float(value)) and value >= 1)
@@ -406,6 +514,7 @@ def _mapping(
     *,
     normalizers: list,
     label: str,
+    value_column: str = SOURCE_MULTIPLIER_COLUMN,
 ) -> dict:
     available = frame[frame["样本状态"].eq("可用")].copy()
     keys = [
@@ -416,19 +525,23 @@ def _mapping(
         raise ValueError(f"{label}存在空匹配键")
     if len(keys) != len(set(keys)):
         raise ValueError(f"{label}存在重复匹配键")
-    values = available[SOURCE_MULTIPLIER_COLUMN].astype(float).tolist()
+    values = available[value_column].astype(float).tolist()
     if len(key_columns) == 1:
         return {key[0]: value for key, value in zip(keys, values)}
     return dict(zip(keys, values))
 
 
-def _single_channel_multiplier(frame: pd.DataFrame, level: str) -> float:
+def _single_channel_multiplier(
+    frame: pd.DataFrame,
+    level: str,
+    value_column: str = SOURCE_MULTIPLIER_COLUMN,
+) -> float:
     rows = frame[
         frame["参数层级"].eq(level) & frame["样本状态"].eq("可用")
     ]
     if len(rows) != 1:
         raise ValueError(f"{level}必须且只能有一条可用参数，实际为{len(rows)}")
-    return float(rows.iloc[0][SOURCE_MULTIPLIER_COLUMN])
+    return float(rows.iloc[0][value_column])
 
 
 def load_fission_multiplier_parameters(
@@ -440,7 +553,7 @@ def load_fission_multiplier_parameters(
         release = RELEASES[version]
     except KeyError as exc:
         raise ValueError(f"未知传播裂变系数参数版本: {version}") from exc
-    directory = (parameter_root or PARAMETER_ROOT) / version
+    directory = (parameter_root or PARAMETER_ROOT) / release.source_version
 
     exact_required = {
         "首层投放日期",
@@ -496,44 +609,133 @@ def load_fission_multiplier_parameters(
     mini_goal_rows = fallback[fallback["参数层级"].eq("小程序转化目标回退")]
     gzh_partner_rows = fallback[fallback["参数层级"].eq("公众号合作方回退")]
 
+    miniapp_exact = _mapping(
+        miniapp,
+        ["人群包", "转化目标"],
+        normalizers=[normalize_package, normalize_name],
+        label="小程序精确参数",
+    )
+    miniapp_by_package = _mapping(
+        mini_package_rows,
+        ["人群包"],
+        normalizers=[normalize_package],
+        label="小程序人群包回退参数",
+    )
+    miniapp_by_goal = _mapping(
+        mini_goal_rows,
+        ["转化目标"],
+        normalizers=[normalize_name],
+        label="小程序转化目标回退参数",
+    )
+    gzh_exact = _mapping(
+        gzh,
+        ["合作方", "公众号"],
+        normalizers=[normalize_name, normalize_name],
+        label="公众号精确参数",
+    )
+    gzh_by_partner = _mapping(
+        gzh_partner_rows,
+        ["合作方"],
+        normalizers=[normalize_name],
+        label="公众号合作方回退参数",
+    )
+
+    total_to_first: dict[tuple[str, str, str, str], float] = {}
+
+    def add_total_to_first(
+        entity_type: str,
+        match_level: str,
+        mapping: Mapping,
+    ) -> None:
+        for raw_key, value in mapping.items():
+            keys = raw_key if isinstance(raw_key, tuple) else (raw_key, "")
+            total_to_first[(entity_type, match_level, keys[0], keys[1])] = value
+
+    if release.includes_total_to_first:
+        add_total_to_first(
+            "self",
+            MATCH_MINIAPP_EXACT,
+            _mapping(
+                miniapp,
+                ["人群包", "转化目标"],
+                normalizers=[normalize_package, normalize_name],
+                label="小程序精确参数",
+                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
+            ),
+        )
+        add_total_to_first(
+            "self",
+            MATCH_MINIAPP_PACKAGE,
+            _mapping(
+                mini_package_rows,
+                ["人群包"],
+                normalizers=[normalize_package],
+                label="小程序人群包回退参数",
+                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
+            ),
+        )
+        add_total_to_first(
+            "self",
+            MATCH_MINIAPP_GOAL,
+            _mapping(
+                mini_goal_rows,
+                ["转化目标"],
+                normalizers=[normalize_name],
+                label="小程序转化目标回退参数",
+                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
+            ),
+        )
+        total_to_first[("self", MATCH_MINIAPP_CHANNEL, "", "")] = (
+            _single_channel_multiplier(
+                fallback,
+                "小程序渠道回退",
+                SOURCE_TOTAL_TO_FIRST_COLUMN,
+            )
+        )
+        add_total_to_first(
+            "gzh",
+            MATCH_GZH_EXACT,
+            _mapping(
+                gzh,
+                ["合作方", "公众号"],
+                normalizers=[normalize_name, normalize_name],
+                label="公众号精确参数",
+                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
+            ),
+        )
+        add_total_to_first(
+            "gzh",
+            MATCH_GZH_PARTNER,
+            _mapping(
+                gzh_partner_rows,
+                ["合作方"],
+                normalizers=[normalize_name],
+                label="公众号合作方回退参数",
+                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
+            ),
+        )
+        total_to_first[("gzh", MATCH_GZH_CHANNEL, "", "")] = (
+            _single_channel_multiplier(
+                fallback,
+                "公众号渠道回退",
+                SOURCE_TOTAL_TO_FIRST_COLUMN,
+            )
+        )
+
     return FissionMultiplierParameters(
         release=release,
-        miniapp_exact=_mapping(
-            miniapp,
-            ["人群包", "转化目标"],
-            normalizers=[normalize_package, normalize_name],
-            label="小程序精确参数",
-        ),
-        miniapp_by_package=_mapping(
-            mini_package_rows,
-            ["人群包"],
-            normalizers=[normalize_package],
-            label="小程序人群包回退参数",
-        ),
-        miniapp_by_goal=_mapping(
-            mini_goal_rows,
-            ["转化目标"],
-            normalizers=[normalize_name],
-            label="小程序转化目标回退参数",
-        ),
+        miniapp_exact=miniapp_exact,
+        miniapp_by_package=miniapp_by_package,
+        miniapp_by_goal=miniapp_by_goal,
         miniapp_channel=_single_channel_multiplier(
             fallback, "小程序渠道回退"
         ),
-        gzh_exact=_mapping(
-            gzh,
-            ["合作方", "公众号"],
-            normalizers=[normalize_name, normalize_name],
-            label="公众号精确参数",
-        ),
-        gzh_by_partner=_mapping(
-            gzh_partner_rows,
-            ["合作方"],
-            normalizers=[normalize_name],
-            label="公众号合作方回退参数",
-        ),
+        gzh_exact=gzh_exact,
+        gzh_by_partner=gzh_by_partner,
         gzh_channel=_single_channel_multiplier(
             fallback, "公众号渠道回退"
         ),
+        total_to_first=total_to_first,
         miniapp_exact_rows=len(miniapp),
         miniapp_exact_available_rows=int(miniapp["样本状态"].eq("可用").sum()),
         gzh_exact_rows=len(gzh),
@@ -552,6 +754,7 @@ def apply_fission_multiplier(
 
     result = daily.copy()
     multipliers: list[float] = []
+    multipliers_vs_first: list[float | None] = []
     levels: list[str] = []
     sources: list[str] = []
     for row in result.to_dict("records"):
@@ -563,10 +766,12 @@ def apply_fission_multiplier(
             official_account=row["公众号名"],
         )
         multipliers.append(match.multiplier)
+        multipliers_vs_first.append(match.multiplier_vs_first)
         levels.append(match.match_level)
         sources.append(match.source)
 
     result[DISPLAY_MULTIPLIER_COLUMN] = multipliers
+    result[DISPLAY_TOTAL_TO_FIRST_COLUMN] = multipliers_vs_first
     result["传播裂变系数匹配层级"] = levels
     result["传播裂变系数来源"] = sources
     result["传播裂变参数版本"] = parameters.release.version

examples/auto_put_ad_mini/roi_control/metrics.py

@@ -13,6 +13,7 @@ import pandas as pd
 
 from .fission_multiplier import (
     DISPLAY_MULTIPLIER_COLUMN,
+    DISPLAY_TOTAL_TO_FIRST_COLUMN,
     FissionMultiplierParameters,
     apply_fission_multiplier,
 )
@@ -161,6 +162,7 @@ def _summarize_entity(
             key_values = (key_values,)
         parameter_columns = [
             DISPLAY_MULTIPLIER_COLUMN,
+            DISPLAY_TOTAL_TO_FIRST_COLUMN,
             "传播裂变系数匹配层级",
             "传播裂变系数来源",
             "传播裂变参数版本",
@@ -174,7 +176,10 @@ def _summarize_entity(
             for column in parameter_columns
         }
         inconsistent = [
-            column for column, values in parameter_values.items() if len(values) != 1
+            column
+            for column, values in parameter_values.items()
+            if len(values) > 1
+            or (len(values) == 0 and column != DISPLAY_TOTAL_TO_FIRST_COLUMN)
         ]
         if inconsistent:
             raise ValueError(
@@ -236,10 +241,14 @@ def _summarize_entity(
             by_date["T0实际裂变收入"].sum()
         )
         for column, values in parameter_values.items():
-            record[column] = values[0]
+            record[column] = values[0] if len(values) else np.nan
         record[DISPLAY_MULTIPLIER_COLUMN] = float(
             record[DISPLAY_MULTIPLIER_COLUMN]
         )
+        if pd.notna(record[DISPLAY_TOTAL_TO_FIRST_COLUMN]):
+            record[DISPLAY_TOTAL_TO_FIRST_COLUMN] = float(
+                record[DISPLAY_TOTAL_TO_FIRST_COLUMN]
+            )
         record["实际全链路效率收入"] = float(
             by_date["实际全链路效率收入"].sum()
         )

examples/auto_put_ad_mini/roi_control/policy.py

@@ -8,7 +8,10 @@ from typing import Any
 
 import pandas as pd
 
-from .fission_multiplier import DISPLAY_MULTIPLIER_COLUMN
+from .fission_multiplier import (
+    DISPLAY_MULTIPLIER_COLUMN,
+    DISPLAY_TOTAL_TO_FIRST_COLUMN,
+)
 from .metrics import ENTITY_KEYS, ENTITY_SELF
 
 
@@ -178,6 +181,9 @@ def annotate_execution(
                 "fission_multiplier_vs_t0": _finite(
                     row.get(DISPLAY_MULTIPLIER_COLUMN)
                 ),
+                "fission_multiplier_vs_first": _finite(
+                    row.get(DISPLAY_TOTAL_TO_FIRST_COLUMN)
+                ),
                 "fission_match_level": str(
                     row.get("传播裂变系数匹配层级") or ""
                 ),

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -12,7 +12,10 @@ from openpyxl.worksheet.datavalidation import DataValidation
 from openpyxl.formatting.rule import ColorScaleRule
 from openpyxl.styles import Alignment, Font, PatternFill
 from openpyxl.utils import get_column_letter
-from .fission_multiplier import DISPLAY_MULTIPLIER_COLUMN
+from .fission_multiplier import (
+    DISPLAY_MULTIPLIER_COLUMN,
+    DISPLAY_TOTAL_TO_FIRST_COLUMN,
+)
 from .metrics import ENTITY_GZH, ENTITY_QIWEI, ENTITY_SELF
 
 
@@ -23,8 +26,12 @@ UP_FILL = PatternFill("solid", fgColor="D9EAD3")
 ADJUST_FILL = PatternFill("solid", fgColor="FFF2CC")
 APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
 APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
-REPORT_VERSION = "roi_report_v5"
-REPORT_RUN_SUFFIX = "r5"
+REPORT_VERSION = "roi_report_v8"
+REPORT_RUN_SUFFIX = "r8"
+
+T0_FISSION_MULTIPLIER_COLUMN = "裂变系数-总裂变UV/T0裂变UV"
+TOTAL_FISSION_TO_FIRST_UV_COLUMN = DISPLAY_TOTAL_TO_FIRST_COLUMN
+FINAL_ROI_COLUMN = "最终效率ROI"
 
 
 BASE_COLUMNS: Dict[str, Sequence[str]] = {
@@ -43,8 +50,9 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "T0裂变效率收入",
         "LTV预测效率收入",
         "日均成本",
-        DISPLAY_MULTIPLIER_COLUMN,
-        "预测ROI",
+        T0_FISSION_MULTIPLIER_COLUMN,
+        TOTAL_FISSION_TO_FIRST_UV_COLUMN,
+        FINAL_ROI_COLUMN,
         "动作",
         "关停线",
         "审批选择",
@@ -60,8 +68,9 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "T0裂变效率收入",
         "LTV预测效率收入",
         "日均成本",
-        DISPLAY_MULTIPLIER_COLUMN,
-        "预测ROI",
+        T0_FISSION_MULTIPLIER_COLUMN,
+        TOTAL_FISSION_TO_FIRST_UV_COLUMN,
+        FINAL_ROI_COLUMN,
         "动作",
         "关停线",
     ),
@@ -73,10 +82,11 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "T0裂变效率收入",
         "LTV预测效率收入",
         "日均成本",
-        DISPLAY_MULTIPLIER_COLUMN,
+        T0_FISSION_MULTIPLIER_COLUMN,
+        TOTAL_FISSION_TO_FIRST_UV_COLUMN,
         "传播裂变参数状态",
         "调控参与状态",
-        "预测ROI",
+        FINAL_ROI_COLUMN,
         "动作",
         "关停线",
     ),
@@ -129,7 +139,13 @@ def _sheet_frame(candidates: pd.DataFrame, sheet_name: str) -> pd.DataFrame:
     subset["日均成本"] = (
         pd.to_numeric(subset.get("成本"), errors="coerce") / coverage_days
     )
-    subset["预测ROI"] = subset["ROI"]
+    subset[T0_FISSION_MULTIPLIER_COLUMN] = pd.to_numeric(
+        subset.get(DISPLAY_MULTIPLIER_COLUMN), errors="coerce"
+    )
+    subset[TOTAL_FISSION_TO_FIRST_UV_COLUMN] = pd.to_numeric(
+        subset.get(DISPLAY_TOTAL_TO_FIRST_COLUMN), errors="coerce"
+    )
+    subset[FINAL_ROI_COLUMN] = subset["ROI"]
     subset["关停线"] = subset["t_stop"]
     subset["扩量线"] = subset["t_up"]
     visible_columns = list(BASE_COLUMNS[sheet_name])
@@ -142,7 +158,7 @@ def _sheet_frame(candidates: pd.DataFrame, sheet_name: str) -> pd.DataFrame:
     ordered = visible_columns + hidden_columns
     subset = subset[ordered]
     if not subset.empty:
-        subset = subset.sort_values("预测ROI", ascending=True)
+        subset = subset.sort_values(FINAL_ROI_COLUMN, ascending=True)
     return subset
 
 
@@ -196,7 +212,12 @@ def _format_sheet(ws, sheet_name: str) -> None:
         width = min(max(max(map(len, values)), len(str(header))) + 2, 42)
         ws.column_dimensions[get_column_letter(column_index)].width = max(width, 11)
 
-        if "ROI" in str(header) or "裂变率" in str(header) or "百分位" in str(header):
+        if (
+            "ROI" in str(header)
+            or "裂变率" in str(header)
+            or "百分位" in str(header)
+            or str(header).startswith("裂变系数-")
+        ):
             for row in range(2, max_row + 1):
                 ws.cell(row, column_index).number_format = "0.000"
         elif "成本" in str(header) or header in {
@@ -263,7 +284,7 @@ def _write_summary(
         ("UV口径", "COUNT(DISTINCT mid)"),
         (
             "ROI口径",
-            "T0裂变收入为实际值；仅预测T1-T15增量。预测总收入=首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-对T0裂变-1)",
+            "T0裂变收入为实际值；仅预测T1-T15增量。预测总收入=首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(裂变系数-总裂变UV/T0裂变UV-1)",
         ),
         (
             "企微口径",
@@ -274,7 +295,7 @@ def _write_summary(
             "阈值样本条件",
             f"连续3天数据完整；小程序日均首层UV>{self_min_uv:g}；"
             f"公众号日均首层UV>{partner_min_uv:g}；连续3天每天成本>"
-            f"{min_daily_cost:g}；三日成本>0且预测ROI有效；企微仅展示、不进入样本池",
+            f"{min_daily_cost:g}；三日成本>0且最终效率ROI有效；企微仅展示、不进入样本池",
         ),
         ("动作优先级", "关停 > 扩量 > 调整封面&落地页视频"),
         ("审批方式", "在小程序投流表黄色【审批选择】列逐行选择批准或拒绝；批准即为最终确认并自动执行"),

examples/auto_put_ad_mini/roi_control/repository.py

@@ -93,8 +93,8 @@ def publish_fission_parameter_release(
                 """
                 INSERT INTO roi_fission_parameter_value
                     (version, entity_type, match_level, key_primary,
-                     key_secondary, multiplier)
-                VALUES (%s,%s,%s,%s,%s,%s)
+                     key_secondary, multiplier, multiplier_vs_first)
+                VALUES (%s,%s,%s,%s,%s,%s,%s)
                 """,
                 [
                     (
@@ -104,6 +104,7 @@ def publish_fission_parameter_release(
                         row["key_primary"],
                         row["key_secondary"],
                         row["multiplier"],
+                        row.get("multiplier_vs_first"),
                     )
                     for row in value_rows
                 ],
@@ -133,7 +134,7 @@ def load_fission_parameter_release(
             cursor.execute(
                 """
                 SELECT entity_type, match_level, key_primary,
-                       key_secondary, multiplier
+                       key_secondary, multiplier, multiplier_vs_first
                 FROM roi_fission_parameter_value
                 WHERE version=%s
                 ORDER BY entity_type, match_level, key_primary, key_secondary
@@ -221,7 +222,8 @@ def replace_run_results(
                 "dynamic_creative_id", "audience_name", "conversion_goal",
                 "partner_name", "official_account_name",
                 "fission_parameter_version", "fission_cohort_date",
-                "fission_multiplier_vs_t0", "fission_match_level",
+                "fission_multiplier_vs_t0", "fission_multiplier_vs_first",
+                "fission_match_level",
                 "fission_source", "ad_age",
                 "avg_first_uv", "first_uv",
                 "t0_fission_count", "t0_fission_rate", "cost",

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -10,7 +10,17 @@ from roi_control.metrics import (
     QIWEI_CHANNEL,
     SELF_CHANNEL,
 )
-from roi_control.fission_multiplier import load_fission_multiplier_parameters
+from roi_control.fission_multiplier import (
+    DISPLAY_TOTAL_TO_FIRST_COLUMN,
+    load_fission_multiplier_parameters,
+)
+from roi_control.reporting import (
+    BASE_COLUMNS,
+    FINAL_ROI_COLUMN,
+    T0_FISSION_MULTIPLIER_COLUMN,
+    TOTAL_FISSION_TO_FIRST_UV_COLUMN,
+    _sheet_frame,
+)
 from roi_control.rules import evaluate_rules as _evaluate_rules
 
 
@@ -252,6 +262,23 @@ class RoiRulesTest(unittest.TestCase):
         self.assertFalse(candidates["广告id"].eq("ad-9").any())
         self.assertEqual(thresholds.iloc[0]["小程序样本数"], 9)
 
+    def test_report_exposes_final_roi_and_both_fission_coefficients(self):
+        candidates, _, _ = evaluate_rules(self.build_daily(), DATES)
+        frame = _sheet_frame(candidates, "小程序投流")
+
+        visible_columns = list(BASE_COLUMNS["小程序投流"])
+        self.assertIn(FINAL_ROI_COLUMN, visible_columns)
+        self.assertIn(T0_FISSION_MULTIPLIER_COLUMN, visible_columns)
+        self.assertIn(TOTAL_FISSION_TO_FIRST_UV_COLUMN, visible_columns)
+        self.assertTrue(
+            frame[FINAL_ROI_COLUMN].equals(frame["ROI"])
+        )
+        pd.testing.assert_series_equal(
+            frame[TOTAL_FISSION_TO_FIRST_UV_COLUMN],
+            frame[DISPLAY_TOTAL_TO_FIRST_COLUMN],
+            check_names=False,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

examples/auto_put_ad_mini/test_roi_fission_multiplier.py

@@ -9,6 +9,7 @@ from roi_control.data_source import build_daily_sql
 from roi_control.fission_multiplier import (
     DEFAULT_FISSION_PARAMETER_VERSION,
     DISPLAY_MULTIPLIER_COLUMN,
+    DISPLAY_TOTAL_TO_FIRST_COLUMN,
     MATCH_GZH_EXACT,
     MATCH_GZH_PARTNER,
     MATCH_MINIAPP_EXACT,
@@ -73,6 +74,16 @@ class RoiFissionMultiplierTest(unittest.TestCase):
         self.assertEqual(insufficient_exact.match_level, MATCH_MINIAPP_GOAL)
         self.assertAlmostEqual(insufficient_exact.multiplier, 2.610471, places=6)
 
+    def test_r500_uses_fixed_cohort_coefficients(self):
+        match = self.parameters.match(
+            "self",
+            package="R500*全品类",
+            optimize_goal="关键页面访问次数",
+        )
+        self.assertEqual(match.match_level, MATCH_MINIAPP_EXACT)
+        self.assertAlmostEqual(match.multiplier, 49567 / 18467)
+        self.assertAlmostEqual(match.multiplier_vs_first, 49567 / 7068)
+
     def test_gzh_insufficient_exact_uses_partner_fallback(self):
         exact = self.parameters.match(
             "gzh", partner="szhx", official_account="尊闻"
@@ -178,6 +189,14 @@ class RoiFissionMultiplierTest(unittest.TestCase):
         self.assertAlmostEqual(result["预测全链路效率收入"], 50 + expected_fission)
         self.assertAlmostEqual(result["ROI"], (50 + expected_fission) / 100)
         self.assertAlmostEqual(result["实际ROI"], 0.8)
+        self.assertAlmostEqual(
+            result[DISPLAY_TOTAL_TO_FIRST_COLUMN],
+            self.parameters.match(
+                "self",
+                package="泛人群",
+                optimize_goal="关键页面访问次数",
+            ).multiplier_vs_first,
+        )
 
     def test_daily_sql_preserves_explicit_optimize_goal(self):
         sql = build_daily_sql("20260725", "20260727")
@@ -191,7 +210,7 @@ class RoiFissionMultiplierTest(unittest.TestCase):
             / "roi_control"
             / "sql"
             / "fission_multiplier"
-            / DEFAULT_FISSION_PARAMETER_VERSION
+            / self.parameters.release.source_version
         )
         expected_miniapp = (
             audit_dir / "小程序人群包转化目标_A0-A15_20260712.sql"

examples/tencent_realtime_control/schema.sql

@@ -92,6 +92,7 @@ CREATE TABLE IF NOT EXISTS roi_fission_parameter_value (
     key_primary VARCHAR(255) NOT NULL DEFAULT '',
     key_secondary VARCHAR(255) NOT NULL DEFAULT '',
     multiplier DOUBLE NOT NULL,
+    multiplier_vs_first DOUBLE DEFAULT NULL,
     created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     UNIQUE KEY uk_roi_fission_parameter_value
         (version, entity_type, match_level, key_primary, key_secondary),
@@ -116,6 +117,7 @@ CREATE TABLE IF NOT EXISTS roi_entity_snapshot (
     fission_parameter_version VARCHAR(64) DEFAULT NULL,
     fission_cohort_date VARCHAR(8) DEFAULT NULL,
     fission_multiplier_vs_t0 DECIMAL(18,8) DEFAULT NULL,
+    fission_multiplier_vs_first DECIMAL(18,8) DEFAULT NULL,
     fission_match_level VARCHAR(64) DEFAULT NULL,
     fission_source VARCHAR(512) DEFAULT NULL,
     ad_age INT DEFAULT NULL,

examples/tencent_realtime_control/storage.py

@@ -108,6 +108,7 @@ def initialize_schema() -> None:
                     "fission_parameter_version": "VARCHAR(64) DEFAULT NULL",
                     "fission_cohort_date": "VARCHAR(8) DEFAULT NULL",
                     "fission_multiplier_vs_t0": "DECIMAL(18,8) DEFAULT NULL",
+                    "fission_multiplier_vs_first": "DECIMAL(18,8) DEFAULT NULL",
                     "fission_match_level": "VARCHAR(64) DEFAULT NULL",
                     "fission_source": "VARCHAR(512) DEFAULT NULL",
                     "actual_total_revenue": "DECIMAL(20,4) DEFAULT NULL",
@@ -115,6 +116,9 @@ def initialize_schema() -> None:
                     "predicted_tail_revenue": "DECIMAL(20,4) DEFAULT NULL",
                     "predicted_fission_revenue": "DECIMAL(20,4) DEFAULT NULL",
                 },
+                "roi_fission_parameter_value": {
+                    "multiplier_vs_first": "DOUBLE DEFAULT NULL",
+                },
                 "roi_action_item": {
                     "approval_status": "VARCHAR(32) NOT NULL DEFAULT 'PENDING'",
                     "approval_source": "VARCHAR(32) DEFAULT NULL",