新增导出近7天小时级投放成本、收入、CPM、ROI的脚本 增加小程序投流收入信息

examples/tencent_realtime_control/README.md

@@ -182,6 +182,9 @@ conda run -n agent python \
 输出会包含本地 xlsx 路径和 `sheet_url=...`。默认统计截至当天的 7 个自然日，
 每天包含 `06:00-22:00` 共 17 个小时。每行同时包含当前小时值，以及当天
 `00:00` 至当前小时的累计总收入、累计总投放成本、累计小程序成本和对应收入比。
+小程序投流收入从 `loghubods.ad_pqtid_behavior_info` 按 `dt/hh` 分区汇总
+`真实收入`，并过滤 `增长渠道来源=小程序投流-稳定`；报表同时展示小时收入、
+累计收入，以及它们相对小程序投流成本的比例。
 上传后的飞书链接默认设置为获得链接者可编辑。也可以指定历史区间：
 
 ```bash

examples/tencent_realtime_control/export_historical_hourly_roi.py

@@ -25,6 +25,7 @@ ROOT = Path(__file__).resolve().parent
 MINIAPP_CHANNEL = "小程序投流-稳定"
 REVENUE_TABLE = "ads_ad_own_package_detail_15min"
 COST_TABLE = "opengid_base_data"
+MINIAPP_REVENUE_TABLE = "ad_pqtid_behavior_info"
 DEFAULT_START_HOUR = 6
 DEFAULT_END_HOUR = 22
 
@@ -34,14 +35,18 @@ REPORT_HEADERS = [
     "总投放成本",
     "小程序投流成本",
     "商业化收入",
+    "小程序投流收入",
     "CPM",
     "收入/总投放",
     "收入/小程序成本",
+    "小程序投流收入/小程序成本",
     "当日累计总收入",
+    "当日累计小程序投流收入",
     "当日累计总投放成本",
     "当日累计小程序投流成本",
     "当日累计收入/总投放",
     "当日累计收入/小程序成本",
+    "当日累计小程序投流收入/小程序成本",
     "总去重UV",
     "总单用户成本",
     "小程序去重UV",
@@ -174,6 +179,31 @@ ORDER BY dt, hour_of_day
         return reader.to_pandas()
 
 
+def fetch_miniapp_revenue_rows(
+    client: "ODPS",
+    *,
+    start_date: date,
+    end_date: date,
+    miniapp_channel: str,
+) -> pd.DataFrame:
+    escaped_channel = miniapp_channel.replace("'", "''")
+    sql = f"""
+SELECT
+  dt AS data_date,
+  CAST(hh AS BIGINT) AS hour_of_day,
+  SUM(NVL(`真实收入`, 0)) AS miniapp_revenue
+FROM loghubods.{MINIAPP_REVENUE_TABLE}
+WHERE dt >= '{start_date:%Y%m%d}'
+  AND dt <= '{end_date:%Y%m%d}'
+  AND `增长渠道来源` = '{escaped_channel}'
+GROUP BY dt, CAST(hh AS BIGINT)
+ORDER BY dt, hour_of_day
+""".strip()
+    instance = client.execute_sql(sql, hints={"odps.sql.submit.mode": "script"})
+    with instance.open_reader(tunnel=True) as reader:
+        return reader.to_pandas()
+
+
 def hourly_revenue(revenue_rows: pd.DataFrame) -> pd.DataFrame:
     columns = [
         "data_date",
@@ -255,10 +285,27 @@ def hourly_cost(cost_rows: pd.DataFrame) -> pd.DataFrame:
     return rows[columns].sort_values(["data_date", "hour_of_day"])
 
 
+def hourly_miniapp_revenue(revenue_rows: pd.DataFrame) -> pd.DataFrame:
+    columns = ["data_date", "hour_of_day", "miniapp_revenue"]
+    if revenue_rows.empty:
+        return pd.DataFrame(columns=columns)
+
+    rows = revenue_rows.copy()
+    rows["data_date"] = rows["data_date"].astype(str)
+    rows["hour_of_day"] = pd.to_numeric(rows["hour_of_day"], errors="coerce")
+    rows = rows.dropna(subset=["hour_of_day"])
+    rows["hour_of_day"] = rows["hour_of_day"].astype("int64")
+    rows["miniapp_revenue"] = pd.to_numeric(
+        rows["miniapp_revenue"], errors="coerce"
+    ).fillna(0.0)
+    return rows[columns].sort_values(["data_date", "hour_of_day"])
+
+
 def build_hourly_report(
     *,
     revenue_rows: pd.DataFrame,
     cost_rows: pd.DataFrame,
+    miniapp_revenue_rows: pd.DataFrame | None = None,
     start_date: date,
     end_date: date,
     start_hour: int = DEFAULT_START_HOUR,
@@ -282,12 +329,21 @@ def build_hourly_report(
         hourly_revenue(revenue_rows),
         on=["data_date", "hour_of_day"],
         how="left",
+    ).merge(
+        hourly_miniapp_revenue(
+            miniapp_revenue_rows
+            if miniapp_revenue_rows is not None
+            else pd.DataFrame()
+        ),
+        on=["data_date", "hour_of_day"],
+        how="left",
     )
 
     numeric_defaults = {
         "total_cost": 0.0,
         "miniapp_cost": 0.0,
         "commercial_revenue": 0.0,
+        "miniapp_revenue": 0.0,
         "total_uv": 0.0,
         "total_unit_user_cost": 0.0,
         "miniapp_uv": 0.0,
@@ -313,10 +369,21 @@ def build_hourly_report(
         ),
         axis=1,
     )
+    report["小程序投流收入/小程序成本"] = report.apply(
+        lambda row: (
+            row["miniapp_revenue"] / row["miniapp_cost"]
+            if row["miniapp_cost"] > 0
+            else None
+        ),
+        axis=1,
+    )
     report = report.sort_values(["data_date", "hour_of_day"]).reset_index(drop=True)
     report["当日累计总收入"] = report.groupby("data_date")[
         "commercial_revenue"
     ].cumsum()
+    report["当日累计小程序投流收入"] = report.groupby("data_date")[
+        "miniapp_revenue"
+    ].cumsum()
     report["当日累计总投放成本"] = report.groupby("data_date")[
         "total_cost"
     ].cumsum()
@@ -339,6 +406,14 @@ def build_hourly_report(
         ),
         axis=1,
     )
+    report["当日累计小程序投流收入/小程序成本"] = report.apply(
+        lambda row: (
+            row["当日累计小程序投流收入"] / row["当日累计小程序投流成本"]
+            if row["当日累计小程序投流成本"] > 0
+            else None
+        ),
+        axis=1,
+    )
     report["日期"] = pd.to_datetime(report["data_date"], format="%Y%m%d").dt.strftime(
         "%Y-%m-%d"
     )
@@ -348,6 +423,7 @@ def build_hourly_report(
             "total_cost": "总投放成本",
             "miniapp_cost": "小程序投流成本",
             "commercial_revenue": "商业化收入",
+            "miniapp_revenue": "小程序投流收入",
             "cpm": "CPM",
             "total_uv": "总去重UV",
             "total_unit_user_cost": "总单用户成本",
@@ -395,21 +471,46 @@ def create_workbook(
         cell.alignment = Alignment(horizontal="center", vertical="center")
 
     widths = [
-        13, 9, 14, 16, 14, 10, 14, 17, 16, 18, 22, 20, 24,
-        11, 14, 13, 16, 17, 18,
+        13, 9, 14, 16, 14, 16, 10, 14, 17, 22, 16, 20, 18, 22,
+        20, 24, 28, 11, 14, 13, 16, 17, 18,
     ]
     for index, width in enumerate(widths, start=1):
         sheet.column_dimensions[get_column_letter(index)].width = width
     for row in sheet.iter_rows(min_row=2):
         for cell in row:
             cell.alignment = Alignment(horizontal="center", vertical="center")
-    for column in ("C", "D", "E", "F", "I", "J", "K", "O", "Q"):
+    money_headers = (
+        "总投放成本",
+        "小程序投流成本",
+        "商业化收入",
+        "小程序投流收入",
+        "CPM",
+        "当日累计总收入",
+        "当日累计小程序投流收入",
+        "当日累计总投放成本",
+        "当日累计小程序投流成本",
+        "总单用户成本",
+        "小程序单用户成本",
+    )
+    ratio_headers = (
+        "收入/总投放",
+        "收入/小程序成本",
+        "小程序投流收入/小程序成本",
+        "当日累计收入/总投放",
+        "当日累计收入/小程序成本",
+        "当日累计小程序投流收入/小程序成本",
+    )
+    count_headers = ("总去重UV", "小程序去重UV", "收入15分钟窗口数")
+    for header in money_headers:
+        column = get_column_letter(REPORT_HEADERS.index(header) + 1)
         for cell in sheet[column][1:]:
             cell.number_format = "0.00"
-    for column in ("G", "H", "L", "M"):
+    for header in ratio_headers:
+        column = get_column_letter(REPORT_HEADERS.index(header) + 1)
         for cell in sheet[column][1:]:
             cell.number_format = "0.0000"
-    for column in ("N", "P", "R"):
+    for header in count_headers:
+        column = get_column_letter(REPORT_HEADERS.index(header) + 1)
         for cell in sheet[column][1:]:
             cell.number_format = "0"
     sheet.freeze_panes = "A2"
@@ -423,9 +524,14 @@ def create_workbook(
         ("每日统计时段", f"{start_hour:02d}:00-{end_hour:02d}:59"),
         ("统计小时数", len(report)),
         ("小程序成本渠道", miniapp_channel),
+        (
+            "小程序收入来源",
+            f"loghubods.{MINIAPP_REVENUE_TABLE}.真实收入",
+        ),
         ("总投放成本", float(report["总投放成本"].sum())),
         ("小程序投流成本", float(report["小程序投流成本"].sum())),
         ("商业化收入", float(report["商业化收入"].sum())),
+        ("小程序投流收入", float(report["小程序投流收入"].sum())),
     ]
     summary.append(["字段", "值"])
     for row in summary_rows:
@@ -474,9 +580,16 @@ def main() -> None:
         end_date=end_date,
         miniapp_channel=args.miniapp_channel,
     )
+    miniapp_revenue_rows = fetch_miniapp_revenue_rows(
+        client,
+        start_date=start_date,
+        end_date=end_date,
+        miniapp_channel=args.miniapp_channel,
+    )
     report = build_hourly_report(
         revenue_rows=revenue_rows,
         cost_rows=cost_rows,
+        miniapp_revenue_rows=miniapp_revenue_rows,
         start_date=start_date,
         end_date=end_date,
         start_hour=args.start_hour,

examples/tencent_realtime_control/test_historical_hourly_roi.py

@@ -63,10 +63,25 @@ class HistoricalHourlyRoiTest(unittest.TestCase):
                 }
             ]
         )
+        miniapp_revenue_rows = pd.DataFrame(
+            [
+                {
+                    "data_date": "20260811",
+                    "hour_of_day": 0,
+                    "miniapp_revenue": 3.0,
+                },
+                {
+                    "data_date": "20260811",
+                    "hour_of_day": 6,
+                    "miniapp_revenue": 40.0,
+                },
+            ]
+        )
 
         report = build_hourly_report(
             revenue_rows=revenue_rows,
             cost_rows=cost_rows,
+            miniapp_revenue_rows=miniapp_revenue_rows,
             start_date=date(2026, 8, 11),
             end_date=date(2026, 8, 11),
         )
@@ -79,12 +94,19 @@ class HistoricalHourlyRoiTest(unittest.TestCase):
         self.assertEqual(240.0, hour_6["CPM"])
         self.assertEqual(1.4, hour_6["收入/总投放"])
         self.assertEqual(2.8, hour_6["收入/小程序成本"])
+        self.assertEqual(40.0, hour_6["小程序投流收入"])
+        self.assertEqual(1.6, hour_6["小程序投流收入/小程序成本"])
         self.assertEqual(1, hour_6["收入15分钟窗口数"])
         self.assertEqual(105.0, hour_6["当日累计总收入"])
         self.assertEqual(60.0, hour_6["当日累计总投放成本"])
         self.assertEqual(29.0, hour_6["当日累计小程序投流成本"])
+        self.assertEqual(43.0, hour_6["当日累计小程序投流收入"])
         self.assertAlmostEqual(1.75, hour_6["当日累计收入/总投放"])
         self.assertAlmostEqual(105.0 / 29.0, hour_6["当日累计收入/小程序成本"])
+        self.assertAlmostEqual(
+            43.0 / 29.0,
+            hour_6["当日累计小程序投流收入/小程序成本"],
+        )
         self.assertEqual("06:00", report.iloc[0]["小时"])
         self.assertEqual("22:00", report.iloc[-1]["小时"])
 
@@ -127,10 +149,25 @@ class HistoricalHourlyRoiTest(unittest.TestCase):
                 },
             ]
         )
+        miniapp_revenue_rows = pd.DataFrame(
+            [
+                {
+                    "data_date": "20260811",
+                    "hour_of_day": 6,
+                    "miniapp_revenue": 80.0,
+                },
+                {
+                    "data_date": "20260812",
+                    "hour_of_day": 6,
+                    "miniapp_revenue": 15.0,
+                },
+            ]
+        )
 
         report = build_hourly_report(
             revenue_rows=revenue_rows,
             cost_rows=pd.DataFrame(),
+            miniapp_revenue_rows=miniapp_revenue_rows,
             start_date=date(2026, 8, 11),
             end_date=date(2026, 8, 12),
         )
@@ -139,6 +176,7 @@ class HistoricalHourlyRoiTest(unittest.TestCase):
             (report["日期"] == "2026-08-12") & (report["小时"] == "06:00")
         ].iloc[0]
         self.assertEqual(20.0, day_2_hour_6["当日累计总收入"])
+        self.assertEqual(15.0, day_2_hour_6["当日累计小程序投流收入"])
 
     def test_spreadsheet_import_defaults_to_editable_link(self) -> None:
         notifier = FeishuNotifier()