企微系数计算

AGENTS.md

@@ -35,7 +35,7 @@
 - ROI 指标计算必须保持纯数据输入/输出,不能依赖飞书审批、腾讯写操作或具体动作执行器。动作策略只能消费一个明确的 `metric_version`。
 - 当前日级 ROI 指标版本为 `north_star_roi_t15_v4`,报表版本为 `roi_report_v5`,使用已发布参数 `20260712_A0-A15_v1`。T0 裂变收入是实际值,只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数-对T0裂变 - 1)`;预测总收入为 `首层实际效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止重复计算 T0。
 - 小程序传播裂变系数按 `人群包+转化目标精确 -> 人群包回退 -> 转化目标回退 -> 渠道回退` 匹配;公众号按 `合作方+公众号精确 -> 合作方回退 -> 渠道回退` 匹配。样本不足的精确实体不能使用自身系数。
-- 小程序日级 ODPS 数据必须保留 `广告优化目标`;缺失目标只能进入人群包回退,不得默认成关键页面访问。企微 ROI 口径未复核,当前按渠道/合作方保留快照和报表,临时参考系数为 1.0 并明确标注待补算;企微不进入阈值样本池且不得生成调控动作。
+- 小程序日级 ODPS 数据必须保留 `广告优化目标`;缺失目标只能进入人群包回退,不得默认成关键页面访问。企微 ROI 口径未复核,优先使用已发布的合作方 SQL 计算结果,合作方未匹配、样本不足或旧版本无企微参数时默认参考系数为 2.5 并明确标注;企微不进入阈值样本池且不得生成调控动作。
 - ROI 审批表金额和 UV 默认展示三日窗口的单日均值,UV 显示为整数;预测 ROI、实际 ROI、裂变率和阈值必须继续使用三日汇总后的加权口径,不能对每日比例做算术平均。原始三日汇总和每日明细字段保留为隐藏审计列。
 - ROI 审批表应展示所有进入阈值样本池的小程序和公众号实体,以及全部企微参考实体;不能只展示有动作的数据。每个渠道 Sheet 默认按预测 ROI 升序排列。
 - 成熟参数是独立版本化只读资产。审计后的参数必须通过更新脚本发布到 MySQL 的 `roi_fission_parameter_release` 和 `roi_fission_parameter_value`;日级 ROI 只从数据库消费 `ROI_FISSION_PARAMETER_VERSION` 指定且通过内容哈希校验的版本,不能在日常任务中现场重算或静默回退到镜像文件。离线复算只能生成草稿,经恒等式、行数、匹配率和新旧结果对账后才能发布。

examples/auto_put_ad_mini/docs/roi/T15成熟系数计算说明.md

@@ -110,9 +110,13 @@ python3 calculate_final_t15_coefficients.py --cohort-date 20260712
 
 - `outputs/小程序T15成熟系数_中文_20260712.csv`
 - `outputs/公众号T15成熟系数_中文_20260712.csv`
+- `outputs/企微T15成熟系数_中文_20260712.csv`
 - `outputs/T15分级回退系数_中文_20260712.csv`
 - `outputs/最终T15成熟系数_中文_20260712.xlsx`
+- `outputs/企微T15成熟系数_中文_20260712.xlsx`
 - `outputs/小程序人群包转化目标_A0-A15_20260712.sql`
 - `outputs/公众号合作即转_A0-A15_20260712.sql`
+- `outputs/企微合作方_A0-A15_20260712.sql`
 
 所有交付CSV和Excel使用中文表头；SQL保留源字段名称以确保可直接执行。
+企微 CSV 只包含合作方参数，不写入小程序和公众号共用的分级回退文件。

examples/auto_put_ad_mini/docs/roi/T15成熟系数迁移指南.md

@@ -85,6 +85,11 @@ outputs/小程序人群包转化目标_A0-A15_20260712.sql
 outputs/公众号合作即转_A0-A15_20260712.sql
 ```
 
+```text
+outputs/企微合作方_A0-A15_20260712.sql
+outputs/企微T15成熟系数_中文_20260712.csv
+```
+
 Python依赖至少包括：
 
 ```text
@@ -113,7 +118,7 @@ pyodps
 | `subsessionid` | 作为`rootSessionId`追踪后续裂变 |
 | `包名` | 小程序人群包 |
 | `广告优化目标` | 小程序转化目标 |
-| `合作方名` | 公众号合作方 |
+| `合作方名` | 公众号合作方或企微合作方 |
 | `公众号名` | 公众号实体 |
 
 渠道值：
@@ -121,6 +126,7 @@ pyodps
 ```text
 小程序投流-稳定
 公众号合作-即转-稳定
+群/企微合作-稳定
 ```
 
 ### 4.2 `loghubods.user_share_log`
@@ -135,7 +141,15 @@ pyodps
 | `machinecode` | 裂变活跃用户去重 |
 | `usersharedepth` | 只保留`>0`裂变层 |
 
-迁移前必须确认新环境能读取这两张表，并且`subsessionid`与`rootsessionid`仍是同一归因键。
+迁移前必须确认新环境能读取上述核心表，并验证企微首层归因覆盖率。
+
+企微首层归因需要先用 `opengid_base_data.subsessionid + mid` 关联
+`user_share_log.subsessionid + machinecode`，以取得 `user_share_log.rootsessionid`。
+
+### 4.3 `loghubods.ods_user_active_log_info_day`
+
+企微裂变活跃还需左关联 `loghubods.ods_user_active_log_info_day`，并排除其中
+同一 `machinecode + subsessionid` 且 `user_share_depth=0` 的首层活跃记录。
 
 ## 5. 算法口径
 
@@ -352,7 +366,7 @@ HORIZON = 15
 5. **转化目标接入**：当前日常ROI旧流程尚未加载“人群包+转化目标”新参数，迁移工程必须确认日常数据是否包含转化目标。
 6. **收入边界**：接入前必须再次确认“首层实际效率收入”不包含准备单独放大的T0裂变收入，避免收入重复。
 7. **公众号新账号**：首层UV不超过500的公众号不会进入精确参数查询；应使用合作方或渠道回退。
-8. **企微渠道**：企微T0裂变口径仍待复核，本次迁移不包含企微正式传播裂变系数；日级ROI临时统一使用参考系数2.5，且不进入阈值和调控。
+8. **企微渠道**：企微T0裂变口径仍待复核；当前已发布版本不包含企微正式传播裂变系数，日级ROI默认使用参考系数2.5。新版本完成企微文件校验和合作方参数发布后优先使用SQL计算结果；企微不设渠道聚合回退，合作方未匹配或样本不足时继续使用默认2.5，且口径审计完成前仍不进入阈值和调控。
 9. **源表稳定性**：`opengid_base_data`与`user_share_log`生产逻辑变更时，需要重新做字段血缘和归因覆盖验证。
 
 ## 12. 回滚方案

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -158,10 +158,12 @@ docker compose --env-file /dev/null run --rm \
 日级 ROI 单独分两阶段启用：
 
 当前 `north_star_roi_t15_v4` 使用版本化传播裂变参数，报表版本为 `roi_report_v5`。小程序按人群包和转化目标、
-公众号按合作方和公众号匹配传播裂变系数；企微口径未确认，当前按渠道/合作方统一使用临时参考系数 2.5 保留展示，但不进入阈值和调控。T0 收入是实际值，
+公众号按合作方和公众号匹配传播裂变系数；企微优先按合作方精确匹配，合作方未匹配或样本不足时使用默认参考系数 2.5，不做渠道聚合回退。企微仅保留展示，不进入阈值和调控。T0 收入是实际值，
 只预测 T1-T15 增量。日级任务不会现场重算 cohort 参数。部署前应保持
 `ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1`。参数必须先发布到 MySQL，
 启动日志、数据库发布版本和 Excel 摘要中的参数版本、cohort 日期必须一致。
+默认企微参考值使用独立版本 `qiwei_reference_2p5_v1`，并进入 ROI 批次幂等键，
+避免与其他临时系数结果复用同一个运行批次。
 ROI 在线表只上传一次，并发送到 `ROI_FEISHU_CHAT_ID` 和
 `FEISHU_OPERATOR_CHAT_ID`；两个变量相同时只发送一次。
 ROI 表格使用获得链接者可编辑权限。黄色【审批选择】列只接受“批准”或“拒绝”，
@@ -201,8 +203,10 @@ docker compose run --rm ad_daily_service \
   --sql-only
 ```
 
-去掉 `--sql-only` 才会执行两条只读 cohort SQL 并生成草稿 CSV/Excel。草稿不会自动
+去掉 `--sql-only` 才会固定执行小程序、公众号和企微三条只读 cohort SQL，并生成草稿 CSV/Excel。草稿不会自动
 进入生产；必须审计并在 `roi_control/fission_multiplier.py` 登记版本和文件 SHA256。
+小程序和公众号写入联合发布草稿，企微另行生成独立草稿。企微正式版本只需
+登记企微 CSV 的 SHA256，数据库参数行只允许合作方精确匹配；不存在企微文件或参数行的旧版本继续使用默认参考系数 2.5。
 首次上线或切换版本前，显式发布到 MySQL：
 
 ```bash

examples/auto_put_ad_mini/refresh_roi_fission_multiplier.py

@@ -32,7 +32,9 @@ from roi_control.fission_multiplier_refresh import (
     HORIZON,
     build_gzh_activity_curve_sql,
     build_miniapp_activity_curve_sql,
+    build_qiwei_activity_curve_sql,
     write_draft_release,
+    write_qiwei_draft,
 )
 from roi_control.odps_client import ODPSClient
 from roi_control.repository import (
@@ -120,29 +122,41 @@ def main() -> int:
 
     miniapp_sql = build_miniapp_activity_curve_sql(args.cohort_date, HORIZON)
     gzh_sql = build_gzh_activity_curve_sql(args.cohort_date, HORIZON)
+    qiwei_sql = build_qiwei_activity_curve_sql(args.cohort_date, HORIZON)
     miniapp_sql_path = output_dir / f"小程序人群包转化目标_A0-A15_{args.cohort_date}.sql"
     gzh_sql_path = output_dir / f"公众号合作即转_A0-A15_{args.cohort_date}.sql"
+    qiwei_sql_path = output_dir / f"企微合作方_A0-A15_{args.cohort_date}.sql"
     miniapp_sql_path.write_text(miniapp_sql + "\n", encoding="utf-8")
     gzh_sql_path.write_text(gzh_sql + "\n", encoding="utf-8")
+    qiwei_sql_path.write_text(qiwei_sql + "\n", encoding="utf-8")
     print(f"SQL已生成: {miniapp_sql_path}")
     print(f"SQL已生成: {gzh_sql_path}")
+    print(f"SQL已生成: {qiwei_sql_path}")
     if args.sql_only:
         return 0
 
     client = ODPSClient()
-    print("[1/2] 查询小程序 cohort", flush=True)
+    print("[1/3] 查询小程序 cohort", flush=True)
     miniapp_raw = client.execute_sql(miniapp_sql)
-    print("[2/2] 查询公众号 cohort", flush=True)
+    print("[2/3] 查询公众号 cohort", flush=True)
     gzh_raw = client.execute_sql(gzh_sql)
+    print("[3/3] 查询企微 cohort", flush=True)
+    qiwei_raw = client.execute_sql(qiwei_sql)
     paths = write_draft_release(
         output_dir,
         args.cohort_date,
         miniapp_raw,
         gzh_raw,
     )
+    qiwei_paths = write_qiwei_draft(
+        output_dir,
+        args.cohort_date,
+        qiwei_raw,
+    )
     print(
         f"草稿完成: 小程序={len(miniapp_raw)}行 公众号={len(gzh_raw)}行 "
-        f"workbook={paths['workbook']}",
+        f"企微={len(qiwei_raw)}行 联合workbook={paths['workbook']} "
+        f"企微workbook={qiwei_paths['workbook']}",
         flush=True,
     )
     print("提示: 草稿不会自动发布；审计后需登记版本和SHA256。", flush=True)

examples/auto_put_ad_mini/roi_control/fission_multiplier.py

@@ -25,6 +25,7 @@ DISPLAY_MULTIPLIER_COLUMN = "传播裂变系数-对T0裂变"
 
 MINIAPP_FILE = "小程序T15成熟系数_中文_20260712.csv"
 GZH_FILE = "公众号T15成熟系数_中文_20260712.csv"
+QIWEI_FILE = "企微T15成熟系数_中文_20260712.csv"
 FALLBACK_FILE = "T15分级回退系数_中文_20260712.csv"
 
 MATCH_MINIAPP_EXACT = "miniapp_package_goal_exact"
@@ -34,8 +35,11 @@ MATCH_MINIAPP_CHANNEL = "miniapp_channel_fallback"
 MATCH_GZH_EXACT = "gzh_partner_account_exact"
 MATCH_GZH_PARTNER = "gzh_partner_fallback"
 MATCH_GZH_CHANNEL = "gzh_channel_fallback"
-MATCH_QIWEI_REFERENCE = "qiwei_channel_reference_pending"
+MATCH_QIWEI_PARTNER = "qiwei_partner_exact"
+MATCH_QIWEI_REFERENCE = "qiwei_reference_pending"
 QIWEI_REFERENCE_MULTIPLIER = 2.5
+QIWEI_REFERENCE_VERSION = "qiwei_reference_2p5_v1"
+QIWEI_REFERENCE_RUN_SUFFIX = "q2p5v1"
 
 
 @dataclass(frozen=True)
@@ -79,13 +83,16 @@ class FissionMultiplierParameters:
     gzh_exact: Mapping[tuple[str, str], float]
     gzh_by_partner: Mapping[str, float]
     gzh_channel: float
+    qiwei_by_partner: Mapping[str, float]
     miniapp_exact_rows: int
     miniapp_exact_available_rows: int
     gzh_exact_rows: int
     gzh_exact_available_rows: int
+    qiwei_exact_rows: int
+    qiwei_exact_available_rows: int
 
     def snapshot(self) -> dict[str, object]:
-        return {
+        result: dict[str, object] = {
             "version": self.release.version,
             "cohort_date": self.release.cohort_date,
             "observation_end_date": self.release.observation_end_date,
@@ -97,6 +104,16 @@ class FissionMultiplierParameters:
             "miniapp_channel_multiplier": self.miniapp_channel,
             "gzh_channel_multiplier": self.gzh_channel,
         }
+        # Keep legacy release hashes stable until a release explicitly carries
+        # audited WeCom parameters.
+        if self.qiwei_by_partner:
+            result.update(
+                {
+                    "qiwei_exact_rows": self.qiwei_exact_rows,
+                    "qiwei_exact_available_rows": self.qiwei_exact_available_rows,
+                }
+            )
+        return result
 
     def match(
         self,
@@ -158,11 +175,18 @@ class FissionMultiplierParameters:
             )
 
         if entity_type == "qiwei":
+            partner_key = normalize_name(partner)
+            if partner_key and partner_key in self.qiwei_by_partner:
+                return FissionMultiplierMatch(
+                    self.qiwei_by_partner[partner_key],
+                    MATCH_QIWEI_PARTNER,
+                    f"企微合作方精确:{partner_key}",
+                )
             return FissionMultiplierMatch(
                 QIWEI_REFERENCE_MULTIPLIER,
                 MATCH_QIWEI_REFERENCE,
-                f"企微渠道临时参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待补算:"
-                f"{normalize_name(partner)}",
+                f"企微默认参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待发布合作方参数:"
+                f"{partner_key}",
             )
 
         raise ValueError(f"不支持的ROI实体类型: {entity_type}")
@@ -205,6 +229,9 @@ def parameter_values(
     for partner, multiplier in parameters.gzh_by_partner.items():
         append("gzh", MATCH_GZH_PARTNER, partner, "", multiplier)
     append("gzh", MATCH_GZH_CHANNEL, "", "", parameters.gzh_channel)
+
+    for partner, multiplier in parameters.qiwei_by_partner.items():
+        append("qiwei", MATCH_QIWEI_PARTNER, partner, "", multiplier)
     return sorted(
         rows,
         key=lambda row: (
@@ -265,6 +292,19 @@ def parameters_from_database(
             raise ValueError(f"数据库传播裂变系数无效: {key}={multiplier}")
         indexed[key] = multiplier
 
+    unsupported_qiwei_levels = sorted(
+        {
+            match_level
+            for entity_type, match_level, _, _ in indexed
+            if entity_type == "qiwei" and match_level != MATCH_QIWEI_PARTNER
+        }
+    )
+    if unsupported_qiwei_levels:
+        raise ValueError(
+            "数据库企微传播裂变参数只支持合作方精确匹配: "
+            + ", ".join(unsupported_qiwei_levels)
+        )
+
     def values(entity_type: str, level: str) -> dict[tuple[str, str], float]:
         return {
             (primary, secondary): multiplier
@@ -292,6 +332,18 @@ def parameters_from_database(
         key[0]: multiplier
         for key, multiplier in values("gzh", MATCH_GZH_PARTNER).items()
     }
+    qiwei_partner_values = values("qiwei", MATCH_QIWEI_PARTNER)
+    if any(secondary for _, secondary in qiwei_partner_values):
+        raise ValueError("数据库企微合作方参数不能包含次级匹配键")
+    if any(not primary for primary, _ in qiwei_partner_values):
+        raise ValueError("数据库企微合作方参数不能包含空合作方键")
+    qiwei_partner = {
+        key[0]: multiplier for key, multiplier in qiwei_partner_values.items()
+    }
+    qiwei_metadata = {"qiwei_exact_rows", "qiwei_exact_available_rows"}
+    if qiwei_partner and not qiwei_metadata.issubset(metadata):
+        missing = sorted(qiwei_metadata - set(metadata))
+        raise ValueError(f"数据库企微传播裂变参数元数据缺失: {', '.join(missing)}")
     result = FissionMultiplierParameters(
         release=release,
         miniapp_exact=mini_exact,
@@ -301,10 +353,17 @@ def parameters_from_database(
         gzh_exact=gzh_exact,
         gzh_by_partner=gzh_partner,
         gzh_channel=single("gzh", MATCH_GZH_CHANNEL),
+        qiwei_by_partner=qiwei_partner,
         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"]),
         gzh_exact_available_rows=int(metadata["gzh_exact_available_rows"]),
+        qiwei_exact_rows=(
+            int(metadata["qiwei_exact_rows"]) if qiwei_partner else 0
+        ),
+        qiwei_exact_available_rows=(
+            int(metadata["qiwei_exact_available_rows"]) if qiwei_partner else 0
+        ),
     )
     expected_hash = str(release_row["content_sha256"])
     if parameter_content_sha256(result) != expected_hash:
@@ -463,6 +522,15 @@ def load_fission_multiplier_parameters(
         label="公众号",
         expected_sha256=release.file_sha256[GZH_FILE],
     )
+    has_qiwei_file = QIWEI_FILE in release.file_sha256
+    qiwei = None
+    if has_qiwei_file:
+        qiwei = _read_parameter_file(
+            directory / QIWEI_FILE,
+            required=exact_required | {"合作方"},
+            label="企微",
+            expected_sha256=release.file_sha256[QIWEI_FILE],
+        )
     fallback = _read_parameter_file(
         directory / FALLBACK_FILE,
         required={
@@ -483,9 +551,14 @@ def load_fission_multiplier_parameters(
     )
     miniapp = _validated_multipliers(miniapp, label="小程序")
     gzh = _validated_multipliers(gzh, label="公众号")
+    if qiwei is not None:
+        qiwei = _validated_multipliers(qiwei, label="企微")
     fallback = _validated_multipliers(fallback, label="分级回退")
 
-    for label, frame in (("小程序", miniapp), ("公众号", gzh)):
+    exact_frames = [("小程序", miniapp), ("公众号", gzh)]
+    if qiwei is not None:
+        exact_frames.append(("企微", qiwei))
+    for label, frame in exact_frames:
         dates = {str(value) for value in frame["首层投放日期"].dropna()}
         if dates != {release.cohort_date}:
             raise ValueError(
@@ -495,6 +568,16 @@ def load_fission_multiplier_parameters(
     mini_package_rows = fallback[fallback["参数层级"].eq("小程序人群包回退")]
     mini_goal_rows = fallback[fallback["参数层级"].eq("小程序转化目标回退")]
     gzh_partner_rows = fallback[fallback["参数层级"].eq("公众号合作方回退")]
+    qiwei_by_partner = (
+        _mapping(
+            qiwei,
+            ["合作方"],
+            normalizers=[normalize_name],
+            label="企微合作方参数",
+        )
+        if qiwei is not None
+        else {}
+    )
 
     return FissionMultiplierParameters(
         release=release,
@@ -534,10 +617,17 @@ def load_fission_multiplier_parameters(
         gzh_channel=_single_channel_multiplier(
             fallback, "公众号渠道回退"
         ),
+        qiwei_by_partner=qiwei_by_partner,
         miniapp_exact_rows=len(miniapp),
         miniapp_exact_available_rows=int(miniapp["样本状态"].eq("可用").sum()),
         gzh_exact_rows=len(gzh),
         gzh_exact_available_rows=int(gzh["样本状态"].eq("可用").sum()),
+        qiwei_exact_rows=len(qiwei) if qiwei is not None else 0,
+        qiwei_exact_available_rows=(
+            int(qiwei["样本状态"].eq("可用").sum())
+            if qiwei is not None
+            else 0
+        ),
     )
 
 
@@ -572,17 +662,24 @@ def apply_fission_multiplier(
     result["传播裂变参数版本"] = parameters.release.version
     result["传播裂变参数cohort日期"] = parameters.release.cohort_date
     qiwei = result["entity_type"].eq("qiwei")
-    result["传播裂变参数状态"] = "正式参数"
-    result.loc[qiwei, "传播裂变参数状态"] = (
-        f"临时参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待分渠道补算"
+    qiwei_reference = qiwei & result["传播裂变系数匹配层级"].eq(
+        MATCH_QIWEI_REFERENCE
     )
+    qiwei_formal = qiwei & ~qiwei_reference
+    result["传播裂变参数状态"] = "正式参数"
     result["调控参与状态"] = "参与阈值和调控"
     result.loc[qiwei, "调控参与状态"] = "仅展示_不进入阈值和调控"
     result["ROI计算口径"] = (
         "首层实际效率收入+T0实际裂变收入+预测T1-T15裂变收入"
     )
-    result.loc[qiwei, "ROI计算口径"] = (
+    result.loc[qiwei_formal, "传播裂变参数状态"] = "企微正式参数_仅展示"
+    result.loc[qiwei_reference, "传播裂变参数版本"] = QIWEI_REFERENCE_VERSION
+    result.loc[qiwei_reference, "传播裂变参数cohort日期"] = ""
+    result.loc[qiwei_reference, "传播裂变参数状态"] = (
+        f"默认参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待发布合作方参数"
+    )
+    result.loc[qiwei_reference, "ROI计算口径"] = (
         "首层实际效率收入+T0实际裂变收入+预测T1-T15裂变收入_"
-        f"传播裂变系数临时取{QIWEI_REFERENCE_MULTIPLIER:g}"
+        f"传播裂变系数默认取{QIWEI_REFERENCE_MULTIPLIER:g}"
     )
     return result

examples/auto_put_ad_mini/roi_control/fission_multiplier_refresh.py

@@ -247,6 +247,142 @@ ORDER BY b.first_uv DESC
 """.strip()
 
 
+def build_qiwei_activity_curve_sql(
+    cohort_date: str,
+    horizon: int = HORIZON,
+    min_first_uv: int = 500,
+) -> str:
+    """Build the audited WeCom partner-level cohort SQL."""
+
+    start = _parse_date(cohort_date)
+    if horizon < 0:
+        raise ValueError("horizon不能小于0")
+    if min_first_uv < 0:
+        raise ValueError("首层UV门槛不能小于0")
+    dates = [
+        (start + timedelta(days=lag)).strftime("%Y%m%d")
+        for lag in range(horizon + 1)
+    ]
+    end_date = dates[-1]
+    active_counts = ",\n".join(
+        f"    SUM(CASE WHEN active_dt='{date}' THEN 1 ELSE 0 END) AS a{lag}_active_uv"
+        for lag, date in enumerate(dates)
+    )
+    active_select = ",\n".join(
+        f"  NVL(a.a{lag}_active_uv,0) AS a{lag}_active_uv"
+        for lag in range(horizon + 1)
+    )
+    rate_select = ",\n".join(
+        f"  CASE WHEN b.first_uv>0 THEN NVL(a.a{lag}_active_uv,0)/b.first_uv END AS a{lag}_active_rate"
+        for lag in range(horizon + 1)
+    )
+    active_sum = "+".join(
+        f"NVL(a.a{lag}_active_uv,0)" for lag in range(horizon + 1)
+    )
+    active_tail_sum = "+".join(
+        f"NVL(a.a{lag}_active_uv,0)" for lag in range(1, horizon + 1)
+    ) or "0"
+    return f"""
+WITH first_cohort AS (
+  SELECT DISTINCT
+    CASE WHEN d.合作方名 IS NULL OR d.合作方名='' THEN 'null' ELSE d.合作方名 END AS partner_name,
+    d.mid AS first_mid,
+    l.rootsessionid AS rootsessionid
+  FROM loghubods.opengid_base_data d
+  JOIN loghubods.user_share_log l
+    ON d.subsessionid=l.subsessionid AND d.mid=l.machinecode
+  WHERE d.dt='{cohort_date}' AND l.dt='{cohort_date}'
+    AND d.channel='群/企微合作-稳定'
+    AND d.usersharedepth=0
+    AND d.hotsencetype != 1167
+    AND d.videoid IS NOT NULL
+    AND d.mid IS NOT NULL
+    AND d.subsessionid IS NOT NULL
+),
+base_stats AS (
+  SELECT partner_name,
+         COUNT(DISTINCT first_mid) AS first_uv,
+         COUNT(DISTINCT rootsessionid) AS root_session_cnt
+  FROM first_cohort
+  GROUP BY partner_name
+  HAVING COUNT(DISTINCT first_mid)>{min_first_uv}
+),
+root_entity AS (
+  SELECT DISTINCT f.partner_name, f.rootsessionid
+  FROM first_cohort f
+  JOIN base_stats b ON f.partner_name=b.partner_name
+),
+first_users AS (
+  SELECT DISTINCT f.partner_name, f.first_mid
+  FROM first_cohort f
+  JOIN base_stats b ON f.partner_name=b.partner_name
+),
+raw_active_fission AS (
+  SELECT DISTINCT r.partner_name, s.machinecode AS active_mid, s.dt AS active_dt
+  FROM root_entity r
+  JOIN (
+    SELECT DISTINCT a.dt, a.machinecode, a.rootsessionid
+    FROM loghubods.user_share_log a
+    LEFT JOIN (
+      SELECT DISTINCT
+        apptype,
+        machinecode,
+        rootsessionid,
+        channel_type,
+        user_share_depth,
+        real_user_share_depth,
+        subsessionid
+      FROM loghubods.ods_user_active_log_info_day
+      WHERE dt BETWEEN '{cohort_date}' AND '{end_date}'
+        AND user_share_depth=0
+    ) b
+      ON a.machinecode=b.machinecode AND a.subsessionid=b.subsessionid
+    WHERE a.dt BETWEEN '{cohort_date}' AND '{end_date}'
+      AND a.topic='click'
+      AND b.machinecode IS NULL
+      AND CAST(a.usersharedepth AS BIGINT)>0
+  ) s
+    ON s.rootsessionid=r.rootsessionid
+  WHERE s.dt BETWEEN '{cohort_date}' AND '{end_date}'
+    AND s.machinecode IS NOT NULL
+),
+active_fission AS (
+  SELECT c.*
+  FROM raw_active_fission c
+  LEFT JOIN first_users f
+    ON c.partner_name=f.partner_name AND c.active_mid=f.first_mid
+  WHERE f.first_mid IS NULL
+),
+active_counts AS (
+  SELECT partner_name,
+{active_counts},
+    COUNT(*) AS active_person_days
+  FROM active_fission
+  GROUP BY partner_name
+)
+SELECT
+  '{cohort_date}' AS cohort_dt,
+  '群/企微合作-稳定' AS channel,
+  b.partner_name,
+  b.first_uv,
+  b.root_session_cnt,
+{active_select},
+{rate_select},
+  NVL(a.active_person_days,0) AS active_person_days,
+  CASE WHEN b.first_uv>0 THEN ({active_sum})/b.first_uv END AS activity_rate_sum,
+  CASE WHEN NVL(a.a0_active_uv,0)>0
+       THEN ({active_tail_sum})/a.a0_active_uv
+  END AS active_tail_vs_a0,
+  CASE WHEN NVL(a.a0_active_uv,0)>0
+       THEN ({active_sum})/a.a0_active_uv
+  END AS activity_maturity_vs_a0,
+  CASE WHEN b.first_uv>0 THEN 1+({active_sum})/b.first_uv END AS steady_dau_multiplier
+FROM base_stats b
+LEFT JOIN active_counts a ON b.partner_name=a.partner_name
+ORDER BY b.first_uv DESC
+""".strip()
+
+
 def _sample_status(first_uv: pd.Series, a0_uv: pd.Series) -> pd.Series:
     status = pd.Series("可用", index=first_uv.index, dtype="object")
     status.loc[first_uv < 500] = "首层UV不足500"
@@ -352,6 +488,10 @@ def _rollup(raw: pd.DataFrame, groups: list[str], level: str) -> pd.DataFrame:
     ]
 
 
+def build_qiwei_release_frame(qiwei_raw: pd.DataFrame) -> pd.DataFrame:
+    return localize_exact(qiwei_raw, "企微合作方")
+
+
 def build_release_frames(
     miniapp_raw: pd.DataFrame,
     gzh_raw: pd.DataFrame,
@@ -386,6 +526,20 @@ def _format_workbook(path: Path) -> None:
     workbook.save(path)
 
 
+def _draft_notes(cohort_date: str) -> pd.DataFrame:
+    return pd.DataFrame(
+        {
+            "项目": ["Cohort日期", "观察周期", "参数状态", "收入公式"],
+            "定义": [
+                cohort_date,
+                "A0至A15，同一用户同日去重、隔日重新计入",
+                "草稿；完成审计并登记版本与SHA256后才能被日级ROI消费",
+                "首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-对T0裂变-1)",
+            ],
+        }
+    )
+
+
 def write_draft_release(
     output_dir: Path,
     cohort_date: str,
@@ -403,21 +557,29 @@ def write_draft_release(
     miniapp.to_csv(paths["miniapp"], index=False, encoding="utf-8-sig")
     gzh.to_csv(paths["gzh"], index=False, encoding="utf-8-sig")
     fallback.to_csv(paths["fallback"], index=False, encoding="utf-8-sig")
-    notes = pd.DataFrame(
-        {
-            "项目": ["Cohort日期", "观察周期", "参数状态", "收入公式"],
-            "定义": [
-                cohort_date,
-                "A0至A15，同一用户同日去重、隔日重新计入",
-                "草稿；完成审计并登记版本与SHA256后才能被日级ROI消费",
-                "首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-对T0裂变-1)",
-            ],
-        }
-    )
     with pd.ExcelWriter(paths["workbook"], engine="openpyxl") as writer:
         miniapp.to_excel(writer, sheet_name="小程序精确系数", index=False)
         gzh.to_excel(writer, sheet_name="公众号精确系数", index=False)
         fallback.to_excel(writer, sheet_name="分级回退系数", index=False)
-        notes.to_excel(writer, sheet_name="计算说明", index=False)
+        _draft_notes(cohort_date).to_excel(writer, sheet_name="计算说明", index=False)
+    _format_workbook(paths["workbook"])
+    return paths
+
+
+def write_qiwei_draft(
+    output_dir: Path,
+    cohort_date: str,
+    qiwei_raw: pd.DataFrame,
+) -> dict[str, Path]:
+    exact = build_qiwei_release_frame(qiwei_raw)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    paths = {
+        "exact": output_dir / f"企微T15成熟系数_中文_{cohort_date}.csv",
+        "workbook": output_dir / f"企微T15成熟系数_中文_{cohort_date}.xlsx",
+    }
+    exact.to_csv(paths["exact"], index=False, encoding="utf-8-sig")
+    with pd.ExcelWriter(paths["workbook"], engine="openpyxl") as writer:
+        exact.to_excel(writer, sheet_name="企微合作方系数", index=False)
+        _draft_notes(cohort_date).to_excel(writer, sheet_name="计算说明", index=False)
     _format_workbook(paths["workbook"])
     return paths

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -267,7 +267,8 @@ def _write_summary(
         ),
         (
             "企微口径",
-            "企微按渠道/合作方保留展示，临时参考系数为2.5，待补算正式参数；不进入阈值样本池和调控。",
+            "企微按合作方匹配已发布参数；未匹配或样本不足时默认取2.5。"
+            "企微仅展示，不进入阈值样本池和调控。",
         ),
         ("阈值口径", "小程序和公众号合格实体进入同一全局池：三日聚合ROI的P20=关停线、P80=扩量线；企微不参与"),
         (
@@ -311,8 +312,10 @@ def _write_summary(
                 "gzh_exact_available_rows",
                 "miniapp_channel_multiplier",
                 "gzh_channel_multiplier",
+                "qiwei_exact_available_rows",
             ):
-                ws.append([key, fission_multiplier.get(key)])
+                if key in fission_multiplier:
+                    ws.append([key, fission_multiplier[key]])
 
     ws.append([])
     display_thresholds = thresholds.rename(

examples/auto_put_ad_mini/roi_control/service.py

@@ -25,6 +25,8 @@ from .fission_multiplier import (
     DEFAULT_FISSION_PARAMETER_VERSION,
     FissionMultiplierParameters,
     QIWEI_REFERENCE_MULTIPLIER,
+    QIWEI_REFERENCE_RUN_SUFFIX,
+    QIWEI_REFERENCE_VERSION,
     parameters_from_database,
 )
 from .policy import annotate_execution
@@ -76,12 +78,14 @@ def _run_identity(
     fission_parameters: FissionMultiplierParameters,
 ) -> tuple[str, str]:
     release = fission_parameters.release
+    run_suffixes = [release.run_suffix, QIWEI_REFERENCE_RUN_SUFFIX]
+    run_key_versions = [release.version, QIWEI_REFERENCE_VERSION]
     run_id = (
         f"roi_{end_date}_{METRIC_RUN_SUFFIX}_{POLICY_RUN_SUFFIX}_"
-        f"{release.run_suffix}_{REPORT_RUN_SUFFIX}"
+        f"{'_'.join(run_suffixes)}_{REPORT_RUN_SUFFIX}"
     )
     run_key = (
-        f"{METRIC_VERSION}:{POLICY_VERSION}:{release.version}:"
+        f"{METRIC_VERSION}:{POLICY_VERSION}:{':'.join(run_key_versions)}:"
         f"{REPORT_VERSION}:{end_date}"
     )
     return run_id, run_key
@@ -108,6 +112,7 @@ def run_daily_roi(
     run_config = config.snapshot()
     run_config["fission_multiplier"] = fission_parameters.snapshot()
     run_config["qiwei_reference_multiplier"] = QIWEI_REFERENCE_MULTIPLIER
+    run_config["qiwei_reference_version"] = QIWEI_REFERENCE_VERSION
     run_config["report_version"] = REPORT_VERSION
     client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
     end_date = resolve_end_date(client, requested_end_date)

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -1,4 +1,5 @@
 import unittest
+from dataclasses import replace
 
 import pandas as pd
 
@@ -10,8 +11,13 @@ 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 (
+    QIWEI_REFERENCE_RUN_SUFFIX,
+    QIWEI_REFERENCE_VERSION,
+    load_fission_multiplier_parameters,
+)
 from roi_control.rules import evaluate_rules as _evaluate_rules
+from roi_control.service import _run_identity
 
 
 DATES = ["20260720", "20260721", "20260722"]
@@ -65,6 +71,21 @@ def row(entity_type, channel, entity_id, dt, roi, uv=600, age=10, cost=200.0):
 
 
 class RoiRulesTest(unittest.TestCase):
+    def test_run_identity_tracks_qiwei_parameter_source(self):
+        temporary_id, temporary_key = _run_identity("20260727", FISSION_PARAMETERS)
+        self.assertIn(QIWEI_REFERENCE_RUN_SUFFIX, temporary_id)
+        self.assertIn(QIWEI_REFERENCE_VERSION, temporary_key)
+
+        formal = replace(
+            FISSION_PARAMETERS,
+            qiwei_by_partner={"合作方": 2.0},
+            qiwei_exact_rows=1,
+            qiwei_exact_available_rows=1,
+        )
+        formal_id, formal_key = _run_identity("20260727", formal)
+        self.assertIn(QIWEI_REFERENCE_RUN_SUFFIX, formal_id)
+        self.assertIn(QIWEI_REFERENCE_VERSION, formal_key)
+
     def build_daily(self):
         rows = []
         # 每日形成稳定分布，低值实体连续低于后20%，高值实体连续高于前20%。

examples/auto_put_ad_mini/test_roi_fission_multiplier.py

@@ -1,7 +1,12 @@
+import hashlib
 import json
+import shutil
+import tempfile
 import unittest
+from dataclasses import replace
 from datetime import date
 from pathlib import Path
+from unittest.mock import patch
 
 import pandas as pd
 
@@ -9,13 +14,22 @@ from roi_control.data_source import build_daily_sql
 from roi_control.fission_multiplier import (
     DEFAULT_FISSION_PARAMETER_VERSION,
     DISPLAY_MULTIPLIER_COLUMN,
+    FALLBACK_FILE,
+    GZH_FILE,
     MATCH_GZH_EXACT,
     MATCH_GZH_PARTNER,
     MATCH_MINIAPP_EXACT,
     MATCH_MINIAPP_GOAL,
     MATCH_MINIAPP_PACKAGE,
     MATCH_QIWEI_REFERENCE,
+    MATCH_QIWEI_PARTNER,
+    MINIAPP_FILE,
+    ParameterRelease,
+    QIWEI_FILE,
     QIWEI_REFERENCE_MULTIPLIER,
+    QIWEI_REFERENCE_VERSION,
+    RELEASES,
+    apply_fission_multiplier,
     load_fission_multiplier_parameters,
     parameter_content_sha256,
     parameter_values,
@@ -25,6 +39,8 @@ from roi_control.metrics import prepare_daily_metrics
 from roi_control.fission_multiplier_refresh import (
     build_gzh_activity_curve_sql,
     build_miniapp_activity_curve_sql,
+    build_qiwei_activity_curve_sql,
+    build_qiwei_release_frame,
 )
 
 
@@ -47,6 +63,16 @@ class RoiFissionMultiplierTest(unittest.TestCase):
         self.assertEqual(snapshot["gzh_exact_available_rows"], 93)
         self.assertAlmostEqual(snapshot["miniapp_channel_multiplier"], 2.608491, places=6)
         self.assertAlmostEqual(snapshot["gzh_channel_multiplier"], 2.476153, places=6)
+        self.assertFalse(self.parameters.qiwei_by_partner)
+        self.assertNotIn("qiwei_formal", snapshot)
+        self.assertNotIn("qiwei_channel_multiplier", snapshot)
+        self.assertFalse(
+            any(row["entity_type"] == "qiwei" for row in parameter_values(self.parameters))
+        )
+        self.assertEqual(
+            parameter_content_sha256(self.parameters),
+            "c447dc53fb560a9ca7dd954b27966ec46bb4e25e1dfd6916d607431c106242c6",
+        )
 
     def test_miniapp_uses_exact_then_published_fallbacks(self):
         exact = self.parameters.match(
@@ -89,7 +115,8 @@ class RoiFissionMultiplierTest(unittest.TestCase):
         match = self.parameters.match("qiwei", partner="企微合作方")
         self.assertEqual(match.match_level, MATCH_QIWEI_REFERENCE)
         self.assertEqual(match.multiplier, QIWEI_REFERENCE_MULTIPLIER)
-        self.assertIn("待补算", match.source)
+        self.assertEqual(QIWEI_REFERENCE_MULTIPLIER, 2.5)
+        self.assertIn("待发布合作方参数", match.source)
 
     def test_qiwei_roi_uses_reference_multiplier(self):
         frame = pd.DataFrame(
@@ -121,9 +148,154 @@ class RoiFissionMultiplierTest(unittest.TestCase):
             result["预测全链路效率收入"],
             50 + 30 * QIWEI_REFERENCE_MULTIPLIER,
         )
-        self.assertAlmostEqual(result["ROI"], 1.25)
+        self.assertAlmostEqual(
+            result["ROI"],
+            (50 + 30 * QIWEI_REFERENCE_MULTIPLIER) / 100,
+        )
+        self.assertEqual(result["传播裂变参数版本"], QIWEI_REFERENCE_VERSION)
+        self.assertEqual(result["传播裂变参数cohort日期"], "")
         self.assertEqual(result["调控参与状态"], "仅展示_不进入阈值和调控")
 
+    def test_qiwei_parameters_match_partner_then_default(self):
+        formal = replace(
+            self.parameters,
+            qiwei_by_partner={"合作方A": 2.2},
+            qiwei_exact_rows=2,
+            qiwei_exact_available_rows=1,
+        )
+
+        exact = formal.match("qiwei", partner="合作方A")
+        fallback = formal.match("qiwei", partner="未知合作方")
+        self.assertEqual(exact.match_level, MATCH_QIWEI_PARTNER)
+        self.assertEqual(exact.multiplier, 2.2)
+        self.assertEqual(fallback.match_level, MATCH_QIWEI_REFERENCE)
+        self.assertEqual(fallback.multiplier, QIWEI_REFERENCE_MULTIPLIER)
+
+    def test_qiwei_formal_and_default_rows_keep_separate_audit_fields(self):
+        formal = replace(
+            self.parameters,
+            qiwei_by_partner={"合作方A": 2.2},
+            qiwei_exact_rows=2,
+            qiwei_exact_available_rows=1,
+        )
+        daily = pd.DataFrame(
+            [
+                {
+                    "entity_type": "qiwei",
+                    "包名": "",
+                    "广告优化目标": "",
+                    "合作方名": "合作方A",
+                    "公众号名": "",
+                },
+                {
+                    "entity_type": "qiwei",
+                    "包名": "",
+                    "广告优化目标": "",
+                    "合作方名": "未知合作方",
+                    "公众号名": "",
+                },
+            ]
+        )
+
+        result = apply_fission_multiplier(daily, formal)
+        exact, fallback = result.iloc[0], result.iloc[1]
+        self.assertEqual(exact["传播裂变参数版本"], formal.release.version)
+        self.assertEqual(exact["传播裂变参数状态"], "企微正式参数_仅展示")
+        self.assertEqual(fallback["传播裂变参数版本"], QIWEI_REFERENCE_VERSION)
+        self.assertEqual(
+            fallback[DISPLAY_MULTIPLIER_COLUMN],
+            QIWEI_REFERENCE_MULTIPLIER,
+        )
+        self.assertIn("默认参考系数", fallback["传播裂变参数状态"])
+
+    def test_formal_qiwei_database_round_trip(self):
+        formal = replace(
+            self.parameters,
+            qiwei_by_partner={"合作方A": 2.2},
+            qiwei_exact_rows=2,
+            qiwei_exact_available_rows=1,
+        )
+        metadata = formal.snapshot()
+        metadata["file_sha256"] = dict(formal.release.file_sha256)
+        restored = parameters_from_database(
+            {
+                "version": formal.release.version,
+                "cohort_date": date(2026, 7, 12),
+                "observation_end_date": date(2026, 7, 27),
+                "run_suffix": formal.release.run_suffix,
+                "content_sha256": parameter_content_sha256(formal),
+                "metadata_json": json.dumps(metadata, ensure_ascii=False),
+            },
+            parameter_values(formal),
+        )
+
+        self.assertEqual(restored.qiwei_by_partner, {"合作方A": 2.2})
+        self.assertNotIn("qiwei_formal", restored.snapshot())
+        self.assertNotIn("qiwei_channel_multiplier", restored.snapshot())
+        self.assertEqual(
+            parameter_content_sha256(restored),
+            parameter_content_sha256(formal),
+        )
+
+    def test_formal_qiwei_parameter_files_are_loaded_and_serialized(self):
+        version = "20260712_A0-A15_qiwei_test"
+        source = (
+            PROJECT_DIR
+            / "roi_control"
+            / "data"
+            / "fission_multiplier"
+            / DEFAULT_FISSION_PARAMETER_VERSION
+        )
+        with tempfile.TemporaryDirectory() as temporary:
+            parameter_root = Path(temporary)
+            target = parameter_root / version
+            target.mkdir()
+            shutil.copy2(source / MINIAPP_FILE, target / MINIAPP_FILE)
+            shutil.copy2(source / GZH_FILE, target / GZH_FILE)
+            shutil.copy2(source / FALLBACK_FILE, target / FALLBACK_FILE)
+
+            qiwei = pd.DataFrame(
+                [
+                    {
+                        "首层投放日期": "20260712",
+                        "合作方": "合作方A",
+                        "首层UV": 600,
+                        "A0活跃裂变UV": 200,
+                        "A0-A15累计裂变活跃人次": 400,
+                        "T1-T15相对T0尾部系数": 1.0,
+                        "T15成熟系数（相对T0）": 2.0,
+                        "样本状态": "可用",
+                    }
+                ]
+            )
+            qiwei.to_csv(target / QIWEI_FILE, index=False, encoding="utf-8-sig")
+
+            def sha256(path: Path) -> str:
+                return hashlib.sha256(path.read_bytes()).hexdigest()
+
+            release = ParameterRelease(
+                version=version,
+                cohort_date="20260712",
+                observation_end_date="20260727",
+                run_suffix="qtest",
+                file_sha256={
+                    name: sha256(target / name)
+                    for name in (MINIAPP_FILE, GZH_FILE, QIWEI_FILE, FALLBACK_FILE)
+                },
+            )
+            with patch.dict(RELEASES, {version: release}):
+                parameters = load_fission_multiplier_parameters(
+                    version,
+                    parameter_root=parameter_root,
+                )
+
+        self.assertEqual(parameters.qiwei_by_partner, {"合作方A": 2.0})
+        qiwei_values = [
+            row for row in parameter_values(parameters) if row["entity_type"] == "qiwei"
+        ]
+        self.assertEqual(len(qiwei_values), 1)
+        self.assertEqual(qiwei_values[0]["match_level"], MATCH_QIWEI_PARTNER)
+
     def test_database_parameter_round_trip_preserves_release(self):
         metadata = self.parameters.snapshot()
         metadata["file_sha256"] = dict(self.parameters.release.file_sha256)
@@ -208,6 +380,57 @@ class RoiFissionMultiplierTest(unittest.TestCase):
             expected_gzh,
         )
 
+    def test_qiwei_recompute_sql_groups_by_partner(self):
+        sql = build_qiwei_activity_curve_sql("20260712")
+
+        self.assertIn("d.channel='群/企微合作-稳定'", sql)
+        self.assertIn("JOIN loghubods.user_share_log l", sql)
+        self.assertIn(
+            "ON d.subsessionid=l.subsessionid AND d.mid=l.machinecode",
+            sql,
+        )
+        self.assertIn("d.dt='20260712' AND l.dt='20260712'", sql)
+        self.assertIn("FROM loghubods.ods_user_active_log_info_day", sql)
+        self.assertIn("user_share_depth=0", sql)
+        self.assertIn("a.machinecode=b.machinecode", sql)
+        self.assertIn("a.subsessionid=b.subsessionid", sql)
+        self.assertIn("b.machinecode IS NULL", sql)
+        self.assertIn("a.topic='click'", sql)
+        self.assertIn("CAST(a.usersharedepth AS BIGINT)>0", sql)
+        self.assertIn("s.dt BETWEEN '20260712' AND '20260727'", sql)
+        self.assertIn("GROUP BY partner_name", sql)
+        self.assertIn("HAVING COUNT(DISTINCT first_mid)>500", sql)
+        self.assertIn("AS a0_active_uv", sql)
+        self.assertIn("AS a15_active_uv", sql)
+        self.assertIn("END AS activity_maturity_vs_a0", sql)
+        self.assertNotIn("-NVL(a.a0_active_uv,0)", sql)
+        self.assertNotIn("account_name", sql)
+        self.assertNotIn("公众号名", sql)
+
+    def test_qiwei_recompute_sql_validates_arguments(self):
+        with self.assertRaisesRegex(ValueError, "horizon不能小于0"):
+            build_qiwei_activity_curve_sql("20260712", horizon=-1)
+        with self.assertRaisesRegex(ValueError, "首层UV门槛不能小于0"):
+            build_qiwei_activity_curve_sql("20260712", min_first_uv=-1)
+
+    def test_qiwei_release_frame_has_no_channel_fallback(self):
+        row = {
+            "cohort_dt": "20260712",
+            "partner_name": "合作方A",
+            "first_uv": 600,
+            "root_session_cnt": 600,
+            "active_person_days": 400,
+            "activity_rate_sum": 2 / 3,
+            "active_tail_vs_a0": 1.0,
+            "activity_maturity_vs_a0": 2.0,
+            "steady_dau_multiplier": 5 / 3,
+        }
+        for lag in range(16):
+            row[f"a{lag}_active_uv"] = 200 if lag == 0 else 0
+            row[f"a{lag}_active_rate"] = 1 / 3 if lag == 0 else 0
+        frame = build_qiwei_release_frame(pd.DataFrame([row]))
+        self.assertEqual(frame["参数层级"].tolist(), ["企微合作方"])
+        self.assertNotIn("企微渠道回退", frame.to_string())
 
 if __name__ == "__main__":
     unittest.main()