Merge branch 'agent_auto_ad_put_ad_adjust_0509_creative' into task/ad-creative-build-20260805

AGENTS.md

@@ -58,6 +58,19 @@
 - `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0` 只允许计算、快照、报表和审批预览;只有 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=1`、`ROI_SHEET_APPROVAL_ENABLED=1` 时,表格批准后才允许自动执行腾讯写操作。
 - 禁止配置 `DAILY_ROI_ENABLED=0`、`ROI_APPLY_ENABLED=1`;服务启动时必须拒绝这种不完整配置。
 
+## 实时业务收入预测
+
+- 实时收入预测的累计收入和最近两个窗口统一读取 `loghubods.ads_ad_own_package_detail_15min` 当天 `package_cost_times_today`;不依赖昨日曲线。日表只补充实时业务指标和次日最终对账。
+- `ads_ad_own_package_detail_15min.report_date` 表示 15 分钟窗口的开始时间;预测时点和报表展示必须使用窗口结束时间,即 `report_date + 15 分钟`,不能提前使用尚未结束的窗口。
+- 每个新 `report_date` 快照及对应预测必须幂等写入 MySQL 用于审计;MySQL 不作为近期趋势信号来源。
+- 第一阶段预测是独立只读能力,只输出预测全天收入、置信区间和建议总成本;不得调用腾讯写接口、飞书审批或 ROI 动作执行器。
+- 默认从每天 `06:15` 开始,按自然 15 分钟刻度预测一次。主模型使用当前时点最近两个连续15分钟窗口;窗口不足、不连续、数据过期或累计值异常时必须停止预测并记录原因,不能补造趋势数据。
+- 5 分钟与 15 分钟区间收入可能无法与日表最终收入对齐;预测内部的今日累计和速度窗口必须保持同一15分钟 `today` 口径,日表不得混入预测公式。历史训练只接受96个连续窗口的完整日,异常日期不能进入参数发布或算法评价。
+- `REVENUE_FORECAST_ENABLED` 默认关闭。算法、成本系数或窗口口径变化时必须升级 `REVENUE_FORECAST_VERSION`,保留历史预测结果。
+- 当前主预测使用最近两个15分钟窗口的加权速度乘以历史同时间点剩余倍数。最近窗口默认权重为2/3,历史参数必须按版本离线发布到 MySQL;实时任务只能读取 `REVENUE_SPEED_PARAMETER_VERSION` 指定版本,不能现场扫描历史或使用预测日及未来数据。
+- 剩余倍数参数必须保存训练日期范围、样本数、P10/P50/P90 和 MAD。P50 用于正式预测,P10/P90 只表示历史经验区间;参数缺失、样本不足或窗口异常时停止本次预测并记录原因,不得使用其他算法兜底。
+- 收入预测模型上线前必须做严格走步回测,至少比较 MAPE、带符号偏差、P90 绝对误差、超过20%误差比例和相邻预测跳变。第一阶段结果不得直接触发腾讯或飞书调控。
+
 ## 模块 B 创意创建规则
 
 - 目标是单广告最终合格创意数,不是单次生成的 pending 行数。

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -67,6 +67,19 @@ RTC_COMMAND_LLM_TIMEOUT_SECONDS=20
 RTC_COMMAND_LLM_CONFIDENCE_THRESHOLD=0.85
 OPEN_ROUTER_API_KEY=sk-or-v1-xxx
 
+REVENUE_FORECAST_ENABLED=0
+REVENUE_FORECAST_POLL_SECONDS=900
+REVENUE_FORECAST_START_TIME=06:15
+REVENUE_FORECAST_STOP_TIME=22:30
+REVENUE_FORECAST_TREND_WINDOW_MINUTES=15
+REVENUE_FORECAST_TREND_WINDOW_COUNT=2
+REVENUE_FORECAST_TARGET_RATIO=3.5
+REVENUE_FORECAST_MAX_LAG_MINUTES=20
+REVENUE_FORECAST_VERSION=revenue_forecast_v6_weighted_30m_speed
+REVENUE_SPEED_PARAMETER_VERSION=revenue_speed_params_v1
+REVENUE_SPEED_LATEST_WEIGHT=0.66666667
+REVENUE_SPEED_MINIMUM_SAMPLES=3
+
 DAILY_CREATION_ENABLED=0
 DAILY_CREATION_HOUR=10
 DAILY_CREATION_MINUTE=30
@@ -160,6 +173,30 @@ docker compose --env-file /dev/null run --rm \
   python /app/examples/tencent_realtime_control/init_db.py
 ```
 
+收入预测首次上线保持 `REVENUE_FORECAST_ENABLED=0`，先执行一次只读预测。命令会
+读取 ODPS 并写入收入快照和预测结果表，不会调用腾讯或飞书：
+
+```bash
+docker compose --env-file /dev/null run --rm \
+  ad_control_service \
+  python /app/examples/tencent_realtime_control/build_revenue_speed_parameters.py \
+  --start-date 20260727 --end-date 20260805 \
+  --parameter-version revenue_speed_params_v1
+
+docker compose --env-file /dev/null run --rm \
+  ad_control_service \
+  python /app/examples/tencent_realtime_control/run_revenue_forecast.py \
+  --ignore-runtime-window
+```
+
+确认 `forecast_method=weighted_30m_speed`、参数版本、样本数、P50预测、P10/P90
+经验区间、建议总成本、昨日非小程序预留成本和小程序目标成本后，
+再设置 `REVENUE_FORECAST_ENABLED=1` 并只重建 `ad_control_service`。预留成本临时
+读取 `opengid_base_data` 昨日完整日数据；缺失时返回 `WAIT_COST_DATA`，不生成
+小程序预算。模型参数或连续窗口不可用时会输出 `WAIT_SPEED_MODEL` 和明确原因，
+本轮不生成预测。
+已发布参数版本不可覆盖；训练日期变化时发布新版本并同步修改环境变量。
+
 执行无腾讯写入的 CPM dry-run：
 
 ```bash

examples/tencent_realtime_control/.env.example

@@ -26,6 +26,20 @@ RTC_COMMAND_LLM_MODEL=google/gemini-3-flash-preview
 RTC_COMMAND_LLM_TIMEOUT_SECONDS=20
 RTC_COMMAND_LLM_CONFIDENCE_THRESHOLD=0.85
 
+REVENUE_FORECAST_ENABLED=0
+REVENUE_FORECAST_POLL_SECONDS=900
+REVENUE_FORECAST_START_TIME=06:15
+REVENUE_FORECAST_STOP_TIME=22:30
+REVENUE_FORECAST_TREND_WINDOW_MINUTES=15
+REVENUE_FORECAST_TREND_WINDOW_COUNT=2
+REVENUE_FORECAST_TARGET_RATIO=3.5
+REVENUE_FORECAST_MAX_LAG_MINUTES=20
+REVENUE_FORECAST_VERSION=revenue_forecast_v6_weighted_30m_speed
+REVENUE_SPEED_PARAMETER_VERSION=revenue_speed_params_v1
+REVENUE_SPEED_LATEST_WEIGHT=0.66666667
+REVENUE_SPEED_MINIMUM_SAMPLES=3
+REVENUE_FORECAST_DB_LOCK_NAME=revenue_forecast
+
 DAILY_ROI_ENABLED=0
 DAILY_ROI_HOUR=11
 DAILY_ROI_MINUTE=0

examples/tencent_realtime_control/README.md

@@ -2,6 +2,69 @@
 
 独立管理实时投放指标采集、状态判断和腾讯广告调控，不依赖广告创建或历史调价分析流程。
 
+## 实时业务收入预测
+
+第一阶段从 `loghubods.ads_ad_own_package_detail_15min` 读取当天
+`package_cost_times_today`，预测当天最终收入并计算建议总成本。日表仅补充 CPM、
+曝光、DAU 等业务指标和次日最终对账，不进入预测收入公式。该能力只采集、计算和
+落库，不调用腾讯写接口，也不发送飞书审批。
+
+当前主模型使用“加权30分钟速度 × 历史同时间点剩余倍数”：最近15分钟收入权重
+默认为 2/3，前一个15分钟权重为 1/3；每个历史完整日在同一预测时点计算
+`(全天收入-当时累计收入)/加权速度`，发布 P10/P50/P90 和 MAD。实时任务只读取
+`REVENUE_SPEED_PARAMETER_VERSION` 指定的已发布参数，P50 是正式预测，P10/P90
+形成经验区间。参数缺失、样本不足、窗口不连续或速度非正时会停止本轮预测并记录
+明确原因，不能现场重算参数或改用其他算法兜底。
+
+表中的 `report_date` 是窗口开始时间，预测与报表统一按窗口结束时间
+`report_date + 15 分钟`记录。默认从 06:15 开始，按自然15分钟刻度预测一次；
+建议总成本默认等于预测收入除以3.5。临时预算口径从
+`loghubods.opengid_base_data` 读取昨日完整日渠道成本，
+将小程序投流以外的渠道成本全部预留；小程序目标成本等于建议总成本减去该预留，
+最低为 0。该口径只用于预测验证，其他渠道尚不具备实时成本。
+原始快照与版本化预测分别保存在 `revenue_forecast_observation` 和
+`revenue_forecast_result`；历史速度样本和已发布参数分别保存在
+`revenue_forecast_speed_sample`、`revenue_forecast_speed_parameter`。
+参数版本发布后不可覆盖；增加训练日期或调整权重时必须使用新的
+`REVENUE_SPEED_PARAMETER_VERSION`。
+
+发布参数版本：
+
+```bash
+.venv/bin/python \
+  examples/tencent_realtime_control/build_revenue_speed_parameters.py \
+  --start-date 20260727 --end-date 20260805 \
+  --parameter-version revenue_speed_params_v1
+```
+
+单次执行：
+
+```bash
+.venv/bin/python \
+  examples/tencent_realtime_control/run_revenue_forecast.py
+```
+
+仅为排查数据而在配置时段外运行：
+
+```bash
+.venv/bin/python \
+  examples/tencent_realtime_control/run_revenue_forecast.py \
+  --ignore-runtime-window
+```
+
+常驻任务由 `ad-control-service` 承载，默认关闭。完整环境变量见 `.env.example`。
+
+严格走步回测，每个预测日只使用此前日期：
+
+```bash
+.venv/bin/python \
+  examples/tencent_realtime_control/backtest_revenue_speed_forecast.py \
+  --history-start 20260727 \
+  --start-date 20260728 --end-date 20260805
+```
+
+回测只读取 ODPS，不写 MySQL，也不会调用腾讯或飞书。
+
 ## 数据和范围
 
 - 从 `loghubods.advertiser_data_da_hour` 读取当天整体小时数据。

examples/tencent_realtime_control/backtest_revenue_speed_forecast.py

@@ -0,0 +1,284 @@
+#!/usr/bin/env python
+"""Strict walk-forward backtest of the weighted-speed forecast."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+from datetime import date, datetime, time, timedelta
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+from odps_source import build_odps_client
+from revenue_forecast import (
+    RevenueObservation,
+    RevenueTrendWindow,
+)
+from revenue_forecast_config import RevenueForecastConfig
+from revenue_forecast_source import fetch_interval_revenue_series
+from revenue_speed_forecast import (
+    aggregate_speed_parameters,
+    build_speed_samples,
+    calculate_speed_forecast,
+    decimal_quantile,
+)
+from run_once import load_environment
+
+
+OUTPUT_DIR = Path(__file__).resolve().parent / "outputs"
+
+
+def _parse_date(value: str) -> date:
+    return datetime.strptime(value, "%Y%m%d").date()
+
+
+def _date_range(start: date, end: date) -> tuple[date, ...]:
+    return tuple(
+        start + timedelta(days=offset)
+        for offset in range((end - start).days + 1)
+    )
+
+
+def _pct_error(predicted: Decimal, actual: Decimal) -> Decimal:
+    return (predicted - actual) / actual * Decimal("100")
+
+
+def _mean(values: list[Decimal]) -> Decimal | None:
+    return sum(values, Decimal("0")) / len(values) if values else None
+
+
+def _model_summary(rows: list[dict[str, Any]]) -> dict[str, Any]:
+    errors = [abs(row["error_pct"]) for row in rows]
+    signed = [row["error_pct"] for row in rows]
+    jumps = [
+        row["jump_pct"]
+        for row in rows
+        if row["jump_pct"] is not None
+    ]
+    return {
+        "rows": len(rows),
+        "mape_pct": _mean(errors),
+        "signed_bias_pct": _mean(signed),
+        "p90_abs_error_pct": (
+            decimal_quantile(errors, Decimal("0.9")) if errors else None
+        ),
+        "over_20pct_rate_pct": (
+            Decimal(sum(error > 20 for error in errors))
+            / Decimal(len(errors))
+            * Decimal("100")
+            if errors
+            else None
+        ),
+        "mean_prediction_jump_pct": _mean(jumps),
+    }
+
+
+def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {"rows": len(rows), "windows": {}}
+    for label, start_at in (
+        ("from_07", "07:00"),
+        ("from_09", "09:00"),
+        ("from_12", "12:00"),
+        ("from_15", "15:00"),
+    ):
+        selected = [row for row in rows if row["time"] >= start_at]
+        result["windows"][label] = {
+            "weighted_speed": _model_summary(selected),
+            "interval_hit_rate_pct": (
+                Decimal(sum(row["interval_hit"] for row in selected))
+                / Decimal(len(selected))
+                * Decimal("100")
+                if selected
+                else None
+            ),
+        }
+    result["by_date"] = {}
+    for data_date in sorted({row["date"] for row in rows}):
+        selected = [row for row in rows if row["date"] == data_date]
+        result["by_date"][data_date] = {
+            "weighted_speed": _model_summary(selected),
+        }
+    return result
+
+
+def _json_default(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        return str(value)
+    raise TypeError(f"Unsupported JSON value: {type(value)!r}")
+
+
+def run_backtest(
+    *,
+    series: dict[date, list[tuple[datetime, Decimal]]],
+    prediction_dates: tuple[date, ...],
+    config: RevenueForecastConfig,
+    start_at: time,
+    stop_at: time,
+) -> list[dict[str, Any]]:
+    all_samples = []
+    samples_by_date_time = {}
+    for data_date, intervals in series.items():
+        samples = build_speed_samples(
+            data_date=data_date,
+            intervals=intervals,
+            parameter_version=config.speed_parameter_version,
+            latest_weight=config.speed_latest_weight,
+            start_time=start_at,
+            stop_time=stop_at,
+        )
+        all_samples.extend(samples)
+        for sample in samples:
+            samples_by_date_time[(data_date, sample.report_time.time())] = sample
+
+    rows: list[dict[str, Any]] = []
+    previous_predictions: dict[date, Decimal] = {}
+    for prediction_date in prediction_dates:
+        current_intervals = series.get(prediction_date, [])
+        if len(current_intervals) != 96:
+            continue
+        actual = sum((value for _, value in current_intervals), Decimal("0"))
+        cumulative = Decimal("0")
+        trend_windows: list[RevenueTrendWindow] = []
+        for window_start, revenue in current_intervals:
+            report_time = window_start + timedelta(minutes=15)
+            cumulative += revenue
+            trend_windows.append(
+                RevenueTrendWindow(
+                    report_time=report_time,
+                    today_revenue=revenue,
+                )
+            )
+            if not start_at <= report_time.time() <= stop_at:
+                continue
+            current_sample = samples_by_date_time.get(
+                (prediction_date, report_time.time())
+            )
+            if current_sample is None:
+                continue
+            training_samples = [
+                item
+                for item in all_samples
+                if item.data_date < prediction_date
+                and item.report_time.time() == report_time.time()
+            ]
+            parameters = aggregate_speed_parameters(training_samples)
+            if not parameters:
+                continue
+            parameter = parameters[0]
+            observation = RevenueObservation(
+                partition=prediction_date.strftime("%Y%m%d"),
+                report_time=report_time,
+                today_revenue=cumulative,
+            )
+            speed = calculate_speed_forecast(
+                observation,
+                trend_windows[-3:],
+                parameter,
+                forecast_version=config.forecast_version,
+                parameter_version=config.speed_parameter_version,
+                target_cost_ratio=config.target_cost_ratio,
+                minimum_samples=1,
+            )
+            time_text = report_time.strftime("%H:%M")
+            row = {
+                "date": prediction_date.strftime("%Y%m%d"),
+                "time": time_text,
+                "training_start": parameter.training_start.strftime("%Y%m%d"),
+                "training_end": parameter.training_end.strftime("%Y%m%d"),
+                "training_samples": parameter.sample_count,
+                "actual_revenue": actual,
+                "current_cumulative": cumulative,
+                "latest_15m_revenue": revenue,
+                "previous_15m_revenue": trend_windows[-2].today_revenue,
+                "weighted_speed": speed.weighted_speed,
+                "remaining_multiplier_p10": parameter.multiplier_p10,
+                "remaining_multiplier_p50": parameter.multiplier_p50,
+                "remaining_multiplier_p90": parameter.multiplier_p90,
+                "forecast_p10": speed.forecast_lower,
+                "forecast_p50": speed.forecast_revenue,
+                "forecast_p90": speed.forecast_upper,
+                "error_pct": _pct_error(speed.forecast_revenue, actual),
+                "interval_hit": (
+                    speed.forecast_lower <= actual <= speed.forecast_upper
+                ),
+            }
+            previous = previous_predictions.get(prediction_date)
+            row["jump_pct"] = (
+                abs(speed.forecast_revenue - previous)
+                / previous
+                * Decimal("100")
+                if previous is not None and previous > 0
+                else None
+            )
+            previous_predictions[prediction_date] = speed.forecast_revenue
+            rows.append(row)
+    return rows
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--start-date", required=True)
+    parser.add_argument("--end-date", required=True)
+    parser.add_argument("--history-start", required=True)
+    parser.add_argument("--start-time", default="07:00")
+    parser.add_argument("--stop-time", default="22:30")
+    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
+    return parser.parse_args()
+
+
+def main() -> int:
+    load_environment()
+    args = parse_args()
+    start_date = _parse_date(args.start_date)
+    end_date = _parse_date(args.end_date)
+    history_start = _parse_date(args.history_start)
+    if not history_start < start_date <= end_date:
+        raise ValueError("history-start must be before the prediction range")
+    dates = _date_range(history_start, end_date)
+    series = fetch_interval_revenue_series(build_odps_client(), dates)
+    rows = run_backtest(
+        series=series,
+        prediction_dates=_date_range(start_date, end_date),
+        config=RevenueForecastConfig.from_env(),
+        start_at=datetime.strptime(args.start_time, "%H:%M").time(),
+        stop_at=datetime.strptime(args.stop_time, "%H:%M").time(),
+    )
+    if not rows:
+        raise RuntimeError("No strict walk-forward rows produced")
+
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    suffix = f"{args.start_date}_{args.end_date}"
+    csv_path = args.output_dir / f"revenue_speed_backtest_{suffix}.csv"
+    json_path = args.output_dir / f"revenue_speed_backtest_{suffix}.json"
+    xlsx_path = args.output_dir / f"revenue_speed_backtest_{suffix}.xlsx"
+    with csv_path.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
+        writer.writeheader()
+        writer.writerows(rows)
+    summary = summarize(rows)
+    json_path.write_text(
+        json.dumps(summary, ensure_ascii=False, indent=2, default=_json_default),
+        encoding="utf-8",
+    )
+    pd.DataFrame(rows).to_excel(xlsx_path, index=False)
+    print(
+        json.dumps(
+            {
+                "summary": summary,
+                "csv": str(csv_path),
+                "json": str(json_path),
+                "xlsx": str(xlsx_path),
+            },
+            ensure_ascii=False,
+            default=_json_default,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/tencent_realtime_control/build_revenue_speed_parameters.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python
+"""Build and publish weighted-speed parameters from completed revenue days."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import date, datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from odps_source import build_odps_client
+from revenue_forecast_config import RevenueForecastConfig
+from revenue_forecast_repository import publish_speed_parameter_release
+from revenue_forecast_source import fetch_interval_revenue_series
+from revenue_speed_forecast import (
+    aggregate_speed_parameters,
+    build_speed_samples,
+)
+from run_once import load_environment
+from storage import initialize_schema
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+
+
+def _parse_date(value: str) -> date:
+    return datetime.strptime(value, "%Y%m%d").date()
+
+
+def _date_range(start: date, end: date) -> tuple[date, ...]:
+    if start > end:
+        raise ValueError("start date must not be after end date")
+    return tuple(
+        start + timedelta(days=offset)
+        for offset in range((end - start).days + 1)
+    )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--start-date", help="Training start date in YYYYMMDD")
+    parser.add_argument("--end-date", help="Training end date in YYYYMMDD")
+    parser.add_argument("--parameter-version", help="Published parameter version")
+    return parser.parse_args()
+
+
+def main() -> int:
+    load_environment()
+    args = parse_args()
+    config = RevenueForecastConfig.from_env()
+    today = datetime.now(SHANGHAI).date()
+    yesterday = today - timedelta(days=1)
+    end_date = _parse_date(args.end_date) if args.end_date else yesterday
+    start_date = (
+        _parse_date(args.start_date)
+        if args.start_date
+        else end_date - timedelta(days=29)
+    )
+    if end_date >= today:
+        raise ValueError("Training end date must be a completed historical day")
+    parameter_version = (
+        args.parameter_version or config.speed_parameter_version
+    )
+    dates = _date_range(start_date, end_date)
+    series = fetch_interval_revenue_series(build_odps_client(), dates)
+
+    samples = []
+    complete_dates = []
+    skipped_dates = []
+    for data_date in dates:
+        day_samples = build_speed_samples(
+            data_date=data_date,
+            intervals=series.get(data_date, []),
+            parameter_version=parameter_version,
+            latest_weight=config.speed_latest_weight,
+            start_time=config.start_time,
+            stop_time=config.stop_time,
+        )
+        if not day_samples:
+            skipped_dates.append(data_date.strftime("%Y%m%d"))
+            continue
+        complete_dates.append(data_date.strftime("%Y%m%d"))
+        samples.extend(day_samples)
+
+    parameters = aggregate_speed_parameters(samples)
+    if not parameters:
+        raise RuntimeError("No complete historical day available for publication")
+    initialize_schema()
+    publish_speed_parameter_release(
+        parameter_version=parameter_version,
+        samples=samples,
+        parameters=parameters,
+    )
+    print(
+        json.dumps(
+            {
+                "status": "PUBLISHED",
+                "parameter_version": parameter_version,
+                "latest_weight": str(config.speed_latest_weight),
+                "complete_dates": complete_dates,
+                "skipped_dates": skipped_dates,
+                "sample_rows": len(samples),
+                "time_slots": len(parameters),
+                "minimum_samples_per_slot": min(
+                    item.sample_count for item in parameters
+                ),
+                "maximum_samples_per_slot": max(
+                    item.sample_count for item in parameters
+                ),
+            },
+            ensure_ascii=False,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/tencent_realtime_control/revenue_forecast.py

@@ -0,0 +1,57 @@
+"""Revenue forecast domain values."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime
+from decimal import Decimal
+
+
+@dataclass(frozen=True)
+class RevenueObservation:
+    partition: str
+    report_time: datetime
+    today_revenue: Decimal
+    overall_cpm: Decimal | None = None
+    impressions: int | None = None
+    dau: int | None = None
+    fill_rate: Decimal | None = None
+    revenue_change_pct: Decimal | None = None
+    cpm_change_pct: Decimal | None = None
+    exposure_change_pct: Decimal | None = None
+    dau_change_pct: Decimal | None = None
+    fill_rate_change_pct: Decimal | None = None
+    source_modified_at: datetime | None = None
+
+
+@dataclass(frozen=True)
+class RevenueTrendWindow:
+    report_time: datetime
+    today_revenue: Decimal
+
+
+@dataclass(frozen=True)
+class RevenueForecast:
+    report_time: datetime
+    forecast_version: str
+    parameter_version: str
+    current_cumulative_revenue: Decimal
+    speed_latest_weight: Decimal
+    latest_interval_revenue: Decimal
+    previous_interval_revenue: Decimal
+    weighted_speed: Decimal
+    remaining_multiplier_p10: Decimal
+    remaining_multiplier_p50: Decimal
+    remaining_multiplier_p90: Decimal
+    parameter_sample_count: int
+    forecast_revenue: Decimal
+    forecast_lower: Decimal
+    forecast_upper: Decimal
+    target_cost_ratio: Decimal
+    target_daily_cost: Decimal
+    cost_reserve_date: date | None
+    channel_costs: tuple[tuple[str, Decimal], ...]
+    non_miniapp_reserved_cost: Decimal
+    miniapp_target_daily_cost: Decimal
+    status: str
+    forecast_method: str = "weighted_30m_speed"

examples/tencent_realtime_control/revenue_forecast_config.py

@@ -0,0 +1,115 @@
+"""Runtime configuration for real-time revenue forecasting."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from datetime import time
+from decimal import Decimal
+
+
+def _env_flag(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _parse_time(value: str) -> time:
+    try:
+        hour, minute = (int(part) for part in value.split(":"))
+        return time(hour=hour, minute=minute)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"Invalid HH:MM time: {value!r}") from exc
+
+
+@dataclass(frozen=True)
+class RevenueForecastConfig:
+    enabled: bool = False
+    poll_seconds: int = 900
+    start_time: time = time(6, 15)
+    stop_time: time = time(22, 30)
+    trend_window_minutes: int = 15
+    trend_window_count: int = 2
+    target_cost_ratio: Decimal = Decimal("3.5")
+    max_lag_minutes: int = 20
+    forecast_version: str = "revenue_forecast_v6_weighted_30m_speed"
+    speed_parameter_version: str = "revenue_speed_params_v1"
+    speed_latest_weight: Decimal = Decimal("0.66666667")
+    speed_minimum_samples: int = 3
+    lock_name: str = "revenue_forecast"
+
+    @classmethod
+    def from_env(cls) -> "RevenueForecastConfig":
+        config = cls(
+            enabled=_env_flag("REVENUE_FORECAST_ENABLED", False),
+            poll_seconds=int(os.getenv("REVENUE_FORECAST_POLL_SECONDS", "900")),
+            start_time=_parse_time(
+                os.getenv("REVENUE_FORECAST_START_TIME", "06:15")
+            ),
+            stop_time=_parse_time(
+                os.getenv("REVENUE_FORECAST_STOP_TIME", "22:30")
+            ),
+            trend_window_minutes=int(
+                os.getenv("REVENUE_FORECAST_TREND_WINDOW_MINUTES", "15")
+            ),
+            trend_window_count=int(
+                os.getenv("REVENUE_FORECAST_TREND_WINDOW_COUNT", "2")
+            ),
+            target_cost_ratio=Decimal(
+                os.getenv("REVENUE_FORECAST_TARGET_RATIO", "3.5")
+            ),
+            max_lag_minutes=int(
+                os.getenv("REVENUE_FORECAST_MAX_LAG_MINUTES", "20")
+            ),
+            forecast_version=os.getenv(
+                "REVENUE_FORECAST_VERSION",
+                "revenue_forecast_v6_weighted_30m_speed",
+            ).strip(),
+            speed_parameter_version=os.getenv(
+                "REVENUE_SPEED_PARAMETER_VERSION",
+                "revenue_speed_params_v1",
+            ).strip(),
+            speed_latest_weight=Decimal(
+                os.getenv("REVENUE_SPEED_LATEST_WEIGHT", "0.66666667")
+            ),
+            speed_minimum_samples=int(
+                os.getenv("REVENUE_SPEED_MINIMUM_SAMPLES", "3")
+            ),
+            lock_name=os.getenv(
+                "REVENUE_FORECAST_DB_LOCK_NAME", "revenue_forecast"
+            ).strip(),
+        )
+        if config.start_time >= config.stop_time:
+            raise ValueError("Revenue forecast start time must be before stop time")
+        if config.poll_seconds < 60:
+            raise ValueError("REVENUE_FORECAST_POLL_SECONDS must be at least 60")
+        if config.trend_window_minutes < 5:
+            raise ValueError("Trend window must be at least 5 minutes")
+        if config.trend_window_count < 1:
+            raise ValueError("Trend window count must be positive")
+        if config.target_cost_ratio <= 0:
+            raise ValueError("REVENUE_FORECAST_TARGET_RATIO must be positive")
+        if config.max_lag_minutes < 5:
+            raise ValueError("Maximum source lag must be at least 5 minutes")
+        if config.trend_window_minutes != 15:
+            raise ValueError("Weighted speed forecast requires 15-minute windows")
+        if config.trend_window_count < 2:
+            raise ValueError("Weighted speed forecast requires at least two windows")
+        if "weighted_30m_speed" not in config.forecast_version:
+            raise ValueError(
+                "Revenue forecast version must identify weighted_30m_speed"
+            )
+        if not Decimal("0") <= config.speed_latest_weight <= Decimal("1"):
+            raise ValueError("REVENUE_SPEED_LATEST_WEIGHT must be between 0 and 1")
+        if config.speed_minimum_samples < 1:
+            raise ValueError("REVENUE_SPEED_MINIMUM_SAMPLES must be positive")
+        if not all(
+            (
+                config.forecast_version,
+                config.speed_parameter_version,
+                config.lock_name,
+            )
+        ):
+            raise ValueError("Forecast, parameter version and lock name are required")
+        return config

examples/tencent_realtime_control/revenue_forecast_job.py

@@ -0,0 +1,212 @@
+"""One revenue forecast collection and calculation cycle."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from decimal import Decimal
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from odps_source import build_odps_client
+from revenue_forecast import (
+    RevenueForecast,
+    RevenueObservation,
+)
+from revenue_forecast_config import RevenueForecastConfig
+from revenue_forecast_repository import (
+    load_speed_parameter,
+    load_previous_observation,
+    save_forecast,
+    upsert_observation,
+)
+from revenue_speed_forecast import (
+    SpeedForecastUnavailable,
+    calculate_speed_forecast,
+)
+from revenue_forecast_source import (
+    MINIAPP_CHANNEL,
+    fetch_daily_channel_costs,
+    fetch_revenue_source_snapshot,
+)
+from storage import advisory_lock
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+
+
+def _in_runtime_window(now: datetime, config: RevenueForecastConfig) -> bool:
+    local_time = now.timetz().replace(tzinfo=None)
+    return config.start_time <= local_time <= config.stop_time
+
+
+def _quality_status(
+    observation: RevenueObservation,
+    previous: RevenueObservation | None,
+    now: datetime,
+    config: RevenueForecastConfig,
+) -> tuple[str, str | None, int]:
+    lag_seconds = int((now - observation.report_time).total_seconds())
+    if observation.report_time.date() != now.date():
+        return "INVALID", "report_time is not from the current day", lag_seconds
+    if lag_seconds < -60:
+        return "INVALID", "report_time is in the future", lag_seconds
+    if lag_seconds > config.max_lag_minutes * 60:
+        return "STALE", "source data exceeds maximum lag", lag_seconds
+    if observation.today_revenue < 0:
+        return "INVALID", "today cumulative revenue is negative", lag_seconds
+    if previous and observation.today_revenue < previous.today_revenue:
+        return "CORRECTED", "cumulative revenue moved backwards", lag_seconds
+    return "VALID", None, lag_seconds
+
+
+def _decimal_json(value: Decimal | None) -> str | None:
+    return str(value) if value is not None else None
+
+
+def _forecast_payload(forecast: RevenueForecast) -> dict[str, Any]:
+    return {
+        "forecast_status": forecast.status,
+        "forecast_version": forecast.forecast_version,
+        "forecast_method": forecast.forecast_method,
+        "parameter_version": forecast.parameter_version,
+        "current_cumulative_revenue": str(
+            forecast.current_cumulative_revenue
+        ),
+        "forecast_revenue": str(forecast.forecast_revenue),
+        "forecast_lower": str(forecast.forecast_lower),
+        "forecast_upper": str(forecast.forecast_upper),
+        "target_cost_ratio": str(forecast.target_cost_ratio),
+        "target_daily_cost": str(forecast.target_daily_cost),
+        "cost_reserve_date": (
+            forecast.cost_reserve_date.isoformat()
+            if forecast.cost_reserve_date is not None
+            else None
+        ),
+        "channel_costs": {
+            channel: str(cost) for channel, cost in forecast.channel_costs
+        },
+        "non_miniapp_reserved_cost": str(
+            forecast.non_miniapp_reserved_cost
+        ),
+        "miniapp_target_daily_cost": str(
+            forecast.miniapp_target_daily_cost
+        ),
+        "speed_latest_weight": _decimal_json(forecast.speed_latest_weight),
+        "latest_interval_revenue": _decimal_json(
+            forecast.latest_interval_revenue
+        ),
+        "previous_interval_revenue": _decimal_json(
+            forecast.previous_interval_revenue
+        ),
+        "weighted_speed": _decimal_json(forecast.weighted_speed),
+        "remaining_multiplier_p10": _decimal_json(
+            forecast.remaining_multiplier_p10
+        ),
+        "remaining_multiplier_p50": _decimal_json(
+            forecast.remaining_multiplier_p50
+        ),
+        "remaining_multiplier_p90": _decimal_json(
+            forecast.remaining_multiplier_p90
+        ),
+        "parameter_sample_count": forecast.parameter_sample_count,
+    }
+
+
+def run_revenue_forecast_cycle(
+    *,
+    now: datetime | None = None,
+    config: RevenueForecastConfig | None = None,
+    ignore_runtime_window: bool = False,
+) -> dict[str, Any]:
+    now = (now or datetime.now(SHANGHAI)).astimezone(SHANGHAI)
+    config = config or RevenueForecastConfig.from_env()
+    if not ignore_runtime_window and not _in_runtime_window(now, config):
+        return {"status": "OFF_HOURS", "evaluated_at": now.isoformat()}
+
+    with advisory_lock(config.lock_name) as acquired:
+        if not acquired:
+            return {"status": "LOCKED", "evaluated_at": now.isoformat()}
+
+        client = build_odps_client()
+        source_snapshot = fetch_revenue_source_snapshot(
+            client,
+            now.date(),
+            window_minutes=config.trend_window_minutes,
+            window_count=config.trend_window_count,
+            as_of=now,
+        )
+        if source_snapshot is None:
+            return {"status": "WAIT_DATA", "evaluated_at": now.isoformat()}
+        observation = source_snapshot.observation
+
+        previous = load_previous_observation(
+            observation.report_time,
+            config.forecast_version,
+        )
+        quality_status, quality_message, lag_seconds = _quality_status(
+            observation,
+            previous,
+            now,
+            config,
+        )
+        upsert_observation(
+            observation,
+            source_version=config.forecast_version,
+            lag_seconds=lag_seconds,
+            quality_status=quality_status,
+            quality_message=quality_message,
+        )
+        payload: dict[str, Any] = {
+            "status": quality_status,
+            "quality_status": quality_status,
+            "partition": observation.partition,
+            "report_time": observation.report_time.isoformat(),
+            "lag_seconds": lag_seconds,
+            "today_revenue": str(observation.today_revenue),
+            "quality_message": quality_message,
+            "evaluated_at": now.isoformat(),
+        }
+        if quality_status != "VALID":
+            return payload
+
+        trend_windows = list(source_snapshot.trend_windows)
+        cost_reserve_date = now.date() - timedelta(days=1)
+        channel_costs = fetch_daily_channel_costs(client, cost_reserve_date)
+        if not channel_costs:
+            payload["status"] = "WAIT_COST_DATA"
+            payload["cost_reserve_date"] = cost_reserve_date.isoformat()
+            return payload
+        non_miniapp_reserved_cost = sum(
+            (
+                cost
+                for channel, cost in channel_costs
+                if channel != MINIAPP_CHANNEL
+            ),
+            Decimal("0"),
+        )
+        parameter = load_speed_parameter(
+            observation.report_time,
+            config.speed_parameter_version,
+        )
+        try:
+            forecast = calculate_speed_forecast(
+                observation,
+                trend_windows,
+                parameter,
+                forecast_version=config.forecast_version,
+                parameter_version=config.speed_parameter_version,
+                target_cost_ratio=config.target_cost_ratio,
+                minimum_samples=config.speed_minimum_samples,
+                cost_reserve_date=cost_reserve_date,
+                channel_costs=channel_costs,
+                non_miniapp_reserved_cost=non_miniapp_reserved_cost,
+            )
+        except SpeedForecastUnavailable as exc:
+            payload["status"] = "WAIT_SPEED_MODEL"
+            payload["forecast_status"] = str(exc)
+            payload["parameter_version"] = config.speed_parameter_version
+            return payload
+        save_forecast(forecast)
+        payload["status"] = "FORECASTED"
+        payload.update(_forecast_payload(forecast))
+        return payload

examples/tencent_realtime_control/revenue_forecast_repository.py

@@ -0,0 +1,370 @@
+"""MySQL persistence for revenue observations and forecasts."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from decimal import Decimal
+from typing import Any
+
+from revenue_forecast import RevenueForecast, RevenueObservation
+from revenue_speed_forecast import RevenueSpeedParameter, RevenueSpeedSample
+from storage import connect
+
+
+def _db_datetime(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    return value.replace(tzinfo=None)
+
+
+def _aware_datetime(value: datetime, timezone: Any) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone)
+    return value.astimezone(timezone)
+
+
+def load_previous_observation(
+    report_time: datetime,
+    source_version: str,
+) -> RevenueObservation | None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT *
+                FROM revenue_forecast_observation
+                WHERE report_time < %s
+                  AND report_time >= DATE(%s)
+                  AND source_version = %s
+                  AND quality_status = 'VALID'
+                ORDER BY report_time DESC
+                LIMIT 1
+                """,
+                (
+                    _db_datetime(report_time),
+                    _db_datetime(report_time),
+                    source_version,
+                ),
+            )
+            row = cursor.fetchone()
+            return _row_to_observation(row, report_time.tzinfo) if row else None
+    finally:
+        connection.close()
+
+
+def upsert_observation(
+    observation: RevenueObservation,
+    *,
+    source_version: str,
+    lag_seconds: int,
+    quality_status: str,
+    quality_message: str | None,
+) -> None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO revenue_forecast_observation
+                    (source_partition, source_version, report_time, today_revenue,
+                     overall_cpm, impressions, dau, fill_rate,
+                     revenue_change_pct, cpm_change_pct,
+                     exposure_change_pct, dau_change_pct,
+                     fill_rate_change_pct,
+                     source_modified_at, lag_seconds, quality_status,
+                     quality_message)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                ON DUPLICATE KEY UPDATE
+                    source_partition=VALUES(source_partition),
+                    source_version=VALUES(source_version),
+                    today_revenue=VALUES(today_revenue),
+                    overall_cpm=VALUES(overall_cpm),
+                    impressions=VALUES(impressions),
+                    dau=VALUES(dau),
+                    fill_rate=VALUES(fill_rate),
+                    revenue_change_pct=VALUES(revenue_change_pct),
+                    cpm_change_pct=VALUES(cpm_change_pct),
+                    exposure_change_pct=VALUES(exposure_change_pct),
+                    dau_change_pct=VALUES(dau_change_pct),
+                    fill_rate_change_pct=VALUES(fill_rate_change_pct),
+                    source_modified_at=VALUES(source_modified_at),
+                    lag_seconds=VALUES(lag_seconds),
+                    quality_status=VALUES(quality_status),
+                    quality_message=VALUES(quality_message)
+                """,
+                (
+                    observation.partition,
+                    source_version,
+                    _db_datetime(observation.report_time),
+                    observation.today_revenue,
+                    observation.overall_cpm,
+                    observation.impressions,
+                    observation.dau,
+                    observation.fill_rate,
+                    observation.revenue_change_pct,
+                    observation.cpm_change_pct,
+                    observation.exposure_change_pct,
+                    observation.dau_change_pct,
+                    observation.fill_rate_change_pct,
+                    _db_datetime(observation.source_modified_at),
+                    lag_seconds,
+                    quality_status,
+                    quality_message,
+                ),
+            )
+    finally:
+        connection.close()
+
+
+def save_forecast(forecast: RevenueForecast) -> None:
+    channel_costs_json = json.dumps(
+        {channel: str(cost) for channel, cost in forecast.channel_costs},
+        ensure_ascii=False,
+        sort_keys=True,
+    )
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO revenue_forecast_result
+                    (report_time, forecast_version, forecast_method,
+                     parameter_version, current_cumulative_revenue,
+                     speed_latest_weight, latest_interval_revenue,
+                     previous_interval_revenue, weighted_speed,
+                     remaining_multiplier_p10, remaining_multiplier_p50,
+                     remaining_multiplier_p90, parameter_sample_count,
+                     forecast_revenue, forecast_lower, forecast_upper,
+                     target_cost_ratio, target_daily_cost, cost_reserve_date,
+                     channel_costs_json, non_miniapp_reserved_cost,
+                     miniapp_target_daily_cost, forecast_status)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                ON DUPLICATE KEY UPDATE
+                    forecast_method=VALUES(forecast_method),
+                    parameter_version=VALUES(parameter_version),
+                    current_cumulative_revenue=VALUES(current_cumulative_revenue),
+                    speed_latest_weight=VALUES(speed_latest_weight),
+                    latest_interval_revenue=VALUES(latest_interval_revenue),
+                    previous_interval_revenue=VALUES(previous_interval_revenue),
+                    weighted_speed=VALUES(weighted_speed),
+                    remaining_multiplier_p10=VALUES(remaining_multiplier_p10),
+                    remaining_multiplier_p50=VALUES(remaining_multiplier_p50),
+                    remaining_multiplier_p90=VALUES(remaining_multiplier_p90),
+                    parameter_sample_count=VALUES(parameter_sample_count),
+                    forecast_revenue=VALUES(forecast_revenue),
+                    forecast_lower=VALUES(forecast_lower),
+                    forecast_upper=VALUES(forecast_upper),
+                    target_cost_ratio=VALUES(target_cost_ratio),
+                    target_daily_cost=VALUES(target_daily_cost),
+                    cost_reserve_date=VALUES(cost_reserve_date),
+                    channel_costs_json=VALUES(channel_costs_json),
+                    non_miniapp_reserved_cost=VALUES(non_miniapp_reserved_cost),
+                    miniapp_target_daily_cost=VALUES(miniapp_target_daily_cost),
+                    forecast_status=VALUES(forecast_status)
+                """,
+                (
+                    _db_datetime(forecast.report_time),
+                    forecast.forecast_version,
+                    forecast.forecast_method,
+                    forecast.parameter_version,
+                    forecast.current_cumulative_revenue,
+                    forecast.speed_latest_weight,
+                    forecast.latest_interval_revenue,
+                    forecast.previous_interval_revenue,
+                    forecast.weighted_speed,
+                    forecast.remaining_multiplier_p10,
+                    forecast.remaining_multiplier_p50,
+                    forecast.remaining_multiplier_p90,
+                    forecast.parameter_sample_count,
+                    forecast.forecast_revenue,
+                    forecast.forecast_lower,
+                    forecast.forecast_upper,
+                    forecast.target_cost_ratio,
+                    forecast.target_daily_cost,
+                    forecast.cost_reserve_date,
+                    channel_costs_json,
+                    forecast.non_miniapp_reserved_cost,
+                    forecast.miniapp_target_daily_cost,
+                    forecast.status,
+                ),
+            )
+    finally:
+        connection.close()
+
+
+def load_speed_parameter(
+    report_time: datetime,
+    parameter_version: str,
+) -> RevenueSpeedParameter | None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT *, TIME_FORMAT(time_slot, '%%H:%%i:%%s') AS time_slot_text
+                FROM revenue_forecast_speed_parameter
+                WHERE parameter_version = %s
+                  AND time_slot = TIME(%s)
+                LIMIT 1
+                """,
+                (parameter_version, _db_datetime(report_time)),
+            )
+            row = cursor.fetchone()
+    finally:
+        connection.close()
+    if not row:
+        return None
+    return RevenueSpeedParameter(
+        parameter_version=str(row["parameter_version"]),
+        time_slot=datetime.strptime(row["time_slot_text"], "%H:%M:%S").time(),
+        training_start=row["training_start"],
+        training_end=row["training_end"],
+        sample_count=int(row["sample_count"]),
+        latest_weight=Decimal(str(row["latest_weight"])),
+        multiplier_p10=Decimal(str(row["multiplier_p10"])),
+        multiplier_p50=Decimal(str(row["multiplier_p50"])),
+        multiplier_p90=Decimal(str(row["multiplier_p90"])),
+        multiplier_mad=Decimal(str(row["multiplier_mad"])),
+    )
+
+
+def publish_speed_parameter_release(
+    *,
+    parameter_version: str,
+    samples: list[RevenueSpeedSample],
+    parameters: list[RevenueSpeedParameter],
+) -> None:
+    if not samples or not parameters:
+        raise ValueError("Cannot publish an empty speed parameter release")
+    if any(item.parameter_version != parameter_version for item in samples):
+        raise ValueError("Sample parameter version mismatch")
+    if any(item.parameter_version != parameter_version for item in parameters):
+        raise ValueError("Aggregate parameter version mismatch")
+
+    connection = connect()
+    try:
+        connection.autocommit(False)
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT COUNT(*) AS row_count "
+                "FROM revenue_forecast_speed_parameter "
+                "WHERE parameter_version=%s",
+                (parameter_version,),
+            )
+            if int(cursor.fetchone()["row_count"]) > 0:
+                raise ValueError(
+                    f"Parameter version already published: {parameter_version}"
+                )
+            cursor.executemany(
+                """
+                INSERT INTO revenue_forecast_speed_sample
+                    (data_date, time_slot, parameter_version,
+                     cumulative_revenue, final_revenue,
+                     latest_interval_revenue, previous_interval_revenue,
+                     latest_weight, weighted_speed, remaining_revenue,
+                     remaining_multiplier)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                """,
+                [
+                    (
+                        item.data_date,
+                        item.report_time.time(),
+                        item.parameter_version,
+                        item.cumulative_revenue,
+                        item.final_revenue,
+                        item.latest_interval_revenue,
+                        item.previous_interval_revenue,
+                        item.latest_weight,
+                        item.weighted_speed,
+                        item.remaining_revenue,
+                        item.remaining_multiplier,
+                    )
+                    for item in samples
+                ],
+            )
+            cursor.executemany(
+                """
+                INSERT INTO revenue_forecast_speed_parameter
+                    (parameter_version, time_slot, training_start,
+                     training_end, sample_count, latest_weight,
+                     multiplier_p10, multiplier_p50, multiplier_p90,
+                     multiplier_mad)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                """,
+                [
+                    (
+                        item.parameter_version,
+                        item.time_slot,
+                        item.training_start,
+                        item.training_end,
+                        item.sample_count,
+                        item.latest_weight,
+                        item.multiplier_p10,
+                        item.multiplier_p50,
+                        item.multiplier_p90,
+                        item.multiplier_mad,
+                    )
+                    for item in parameters
+                ],
+            )
+        connection.commit()
+    except Exception:
+        connection.rollback()
+        raise
+    finally:
+        connection.autocommit(True)
+        connection.close()
+
+
+def _row_to_observation(row: dict[str, Any], timezone: Any) -> RevenueObservation:
+    return RevenueObservation(
+        partition=str(row["source_partition"]),
+        report_time=_aware_datetime(row["report_time"], timezone),
+        today_revenue=Decimal(str(row["today_revenue"])),
+        overall_cpm=(
+            Decimal(str(row["overall_cpm"]))
+            if row.get("overall_cpm") is not None
+            else None
+        ),
+        impressions=(
+            int(row["impressions"]) if row.get("impressions") is not None else None
+        ),
+        dau=int(row["dau"]) if row.get("dau") is not None else None,
+        fill_rate=(
+            Decimal(str(row["fill_rate"]))
+            if row.get("fill_rate") is not None
+            else None
+        ),
+        revenue_change_pct=(
+            Decimal(str(row["revenue_change_pct"]))
+            if row.get("revenue_change_pct") is not None
+            else None
+        ),
+        cpm_change_pct=(
+            Decimal(str(row["cpm_change_pct"]))
+            if row.get("cpm_change_pct") is not None
+            else None
+        ),
+        exposure_change_pct=(
+            Decimal(str(row["exposure_change_pct"]))
+            if row.get("exposure_change_pct") is not None
+            else None
+        ),
+        dau_change_pct=(
+            Decimal(str(row["dau_change_pct"]))
+            if row.get("dau_change_pct") is not None
+            else None
+        ),
+        fill_rate_change_pct=(
+            Decimal(str(row["fill_rate_change_pct"]))
+            if row.get("fill_rate_change_pct") is not None
+            else None
+        ),
+        source_modified_at=(
+            _aware_datetime(row["source_modified_at"], timezone)
+            if row.get("source_modified_at") is not None
+            else None
+        ),
+    )

examples/tencent_realtime_control/revenue_forecast_service.py

@@ -0,0 +1,77 @@
+"""Background scheduler for revenue forecast collection."""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from revenue_forecast_config import RevenueForecastConfig
+from revenue_forecast_job import run_revenue_forecast_cycle
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+logger = logging.getLogger("tencent_realtime_control.revenue_forecast")
+
+
+def _next_wake(now: datetime, config: RevenueForecastConfig) -> datetime:
+    local_time = now.timetz().replace(tzinfo=None)
+    if local_time > config.stop_time:
+        return datetime.combine(
+            now.date() + timedelta(days=1),
+            config.start_time,
+            SHANGHAI,
+        )
+    if local_time < config.start_time:
+        return datetime.combine(now.date(), config.start_time, SHANGHAI)
+    seconds = int(now.timestamp())
+    boundary = (
+        seconds // config.poll_seconds + 1
+    ) * config.poll_seconds
+    return datetime.fromtimestamp(boundary, SHANGHAI)
+
+
+def run_revenue_forecast_forever(config: RevenueForecastConfig) -> None:
+    logger.info(
+        "Revenue forecast started hours=%s-%s interval=%ss version=%s",
+        config.start_time.strftime("%H:%M"),
+        config.stop_time.strftime("%H:%M"),
+        config.poll_seconds,
+        config.forecast_version,
+    )
+    while True:
+        try:
+            payload = run_revenue_forecast_cycle(config=config)
+            logger.info(
+                "Revenue forecast cycle status=%s report_time=%s revenue=%s "
+                "target_cost=%s miniapp_target_cost=%s",
+                payload.get("status"),
+                payload.get("report_time"),
+                payload.get("forecast_revenue"),
+                payload.get("target_daily_cost"),
+                payload.get("miniapp_target_daily_cost"),
+            )
+        except Exception:
+            logger.exception("Revenue forecast cycle failed")
+
+        wake_at = _next_wake(datetime.now(SHANGHAI), config)
+        while True:
+            remaining = (wake_at - datetime.now(SHANGHAI)).total_seconds()
+            if remaining <= 0:
+                break
+            time.sleep(min(remaining, 5))
+
+
+def start_revenue_forecast_service(
+    config: RevenueForecastConfig,
+) -> threading.Thread:
+    thread = threading.Thread(
+        target=run_revenue_forecast_forever,
+        args=(config,),
+        name="revenue-forecast",
+        daemon=True,
+    )
+    thread.start()
+    return thread

examples/tencent_realtime_control/revenue_forecast_source.py

@@ -0,0 +1,250 @@
+"""Read same-source 15-minute business revenue series from ODPS."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime, timedelta
+from decimal import Decimal
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from odps import ODPS
+
+from revenue_forecast import RevenueObservation, RevenueTrendWindow
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+SOURCE_TABLE = "ads_ad_own_package_detail_day"
+INTERVAL_TABLE = "ads_ad_own_package_detail_15min"
+COST_TABLE = "opengid_base_data"
+MINIAPP_CHANNEL = "小程序投流-稳定"
+
+
+@dataclass(frozen=True)
+class RevenueSourceSnapshot:
+    observation: RevenueObservation
+    trend_windows: tuple[RevenueTrendWindow, ...]
+
+
+def fetch_daily_channel_costs(
+    client: ODPS,
+    data_date: date,
+) -> tuple[tuple[str, Decimal], ...]:
+    sql = f"""
+SELECT
+  NVL(channel, '') AS channel,
+  SUM(NVL(`成本`, 0)) AS cost_yuan
+FROM loghubods.{COST_TABLE}
+WHERE dt = '{data_date:%Y%m%d}'
+  AND usersharedepth = '0'
+  AND videoid IS NOT NULL
+  AND NVL(hotsencetype, '') <> '1167'
+GROUP BY NVL(channel, '')
+ORDER BY cost_yuan DESC
+""".strip()
+    instance = client.execute_sql(
+        sql,
+        hints={"odps.sql.submit.mode": "script"},
+    )
+    with instance.open_reader(tunnel=True) as reader:
+        rows = reader.to_pandas()
+    if rows.empty:
+        return ()
+
+    costs: list[tuple[str, Decimal]] = []
+    for row in rows.itertuples(index=False):
+        cost = Decimal(str(row.cost_yuan or 0))
+        if cost < 0:
+            raise ValueError(f"Negative channel cost: {row.channel}={cost}")
+        costs.append((str(row.channel), cost))
+    return tuple(costs)
+
+
+def _decimal(value: Any) -> Decimal | None:
+    if value is None or value == "":
+        return None
+    return Decimal(str(value).rstrip("%"))
+
+
+def _latest_today_row(partition: Any) -> Any | None:
+    rows = []
+    with partition.open_reader() as reader:
+        for row in reader:
+            if str(row["data_type"]) == "today" and row["report_date"]:
+                rows.append(row)
+    return max(rows, key=lambda row: str(row["report_date"]), default=None)
+
+
+def _partition(table: Any, data_date: date) -> Any | None:
+    spec = f"dt='{data_date:%Y%m%d}'"
+    if not table.exist_partition(spec):
+        return None
+    return table.get_partition(spec)
+
+
+def _fetch_interval_today_series(
+    client: ODPS,
+    data_dates: tuple[date, ...],
+) -> dict[date, list[tuple[datetime, Decimal]]]:
+    partition_filter = " OR ".join(
+        (
+            f"(dt >= '{item:%Y%m%d}000000' "
+            f"AND dt <= '{item:%Y%m%d}235959')"
+        )
+        for item in data_dates
+    )
+    sql = f"""
+SELECT data_date, report_date, today_revenue
+FROM (
+  SELECT
+    data_date,
+    dt,
+    report_date,
+    today_revenue,
+    ROW_NUMBER() OVER (PARTITION BY dt ORDER BY report_date DESC) AS row_num
+  FROM (
+    SELECT
+      SUBSTR(dt, 1, 8) AS data_date,
+      dt,
+      report_date,
+      MIN(NVL(package_cost_times_today, 0)) AS today_revenue
+    FROM loghubods.{INTERVAL_TABLE}
+    WHERE ({partition_filter})
+      AND data_type = 'today'
+      AND report_date IS NOT NULL
+    GROUP BY SUBSTR(dt, 1, 8), dt, report_date
+  ) deduplicated
+) ranked
+WHERE row_num = 1
+ORDER BY data_date, report_date
+""".strip()
+    instance = client.execute_sql(
+        sql,
+        hints={"odps.sql.submit.mode": "script"},
+    )
+    with instance.open_reader(tunnel=True) as reader:
+        rows = reader.to_pandas()
+
+    result = {item: [] for item in data_dates}
+    for row in rows.itertuples(index=False):
+        data_date = datetime.strptime(str(row.data_date), "%Y%m%d").date()
+        revenue = _decimal(row.today_revenue)
+        if revenue is None:
+            continue
+        report_time = datetime.strptime(
+            str(row.report_date), "%Y%m%d%H%M%S"
+        ).replace(tzinfo=SHANGHAI)
+        result[data_date].append((report_time, revenue))
+    return result
+
+
+def fetch_interval_revenue_series(
+    client: ODPS,
+    data_dates: tuple[date, ...],
+) -> dict[date, list[tuple[datetime, Decimal]]]:
+    """Return each date's deduplicated 15-minute `today` interval series."""
+    if not data_dates:
+        return {}
+    return _fetch_interval_today_series(client, data_dates)
+
+
+def _build_trend_windows(
+    current: list[tuple[datetime, Decimal]],
+    *,
+    window_minutes: int,
+    window_count: int,
+) -> tuple[RevenueTrendWindow, ...]:
+    windows: list[RevenueTrendWindow] = []
+    for report_time, today_revenue in current[-window_count:]:
+        windows.append(
+            RevenueTrendWindow(
+                report_time=report_time + timedelta(minutes=window_minutes),
+                today_revenue=today_revenue,
+            )
+        )
+    return tuple(windows)
+
+
+def fetch_revenue_source_snapshot(
+    client: ODPS,
+    data_date: date,
+    *,
+    window_minutes: int = 15,
+    window_count: int = 3,
+    as_of: datetime | None = None,
+) -> RevenueSourceSnapshot | None:
+    series = _fetch_interval_today_series(client, (data_date,))
+    current_values = series[data_date]
+    if as_of is not None:
+        cutoff = as_of.astimezone(SHANGHAI)
+        current_values = [
+            item
+            for item in current_values
+            if item[0] + timedelta(minutes=window_minutes) <= cutoff
+        ]
+    if not current_values:
+        return None
+
+    latest_window_start = current_values[-1][0]
+    today_revenue = sum((value for _, value in current_values), Decimal("0"))
+
+    metrics_table = client.get_table(SOURCE_TABLE)
+    metrics_partition = _partition(metrics_table, data_date)
+    metrics = (
+        _latest_today_row(metrics_partition)
+        if metrics_partition is not None
+        else None
+    )
+    interval_table = client.get_table(INTERVAL_TABLE)
+    latest_partition = _partition_at_time(interval_table, latest_window_start)
+    modified_at = (
+        latest_partition.last_data_modified_time
+        if latest_partition is not None
+        else None
+    )
+    if modified_at is not None and modified_at.tzinfo is None:
+        modified_at = modified_at.replace(tzinfo=SHANGHAI)
+    elif modified_at is not None:
+        modified_at = modified_at.astimezone(SHANGHAI)
+
+    def metric(name: str) -> Any | None:
+        return metrics[name] if metrics is not None else None
+
+    observation = RevenueObservation(
+        partition=data_date.strftime("%Y%m%d"),
+        report_time=latest_window_start + timedelta(minutes=window_minutes),
+        today_revenue=today_revenue,
+        overall_cpm=_decimal(metric("overall_cpm")),
+        impressions=(
+            int(metric("daily_exposure_cnt"))
+            if metric("daily_exposure_cnt") is not None
+            else None
+        ),
+        dau=(
+            int(metric("dau_today"))
+            if metric("dau_today") is not None
+            else None
+        ),
+        fill_rate=_decimal(metric("fill_rate_today")),
+        revenue_change_pct=_decimal(metric("package_cost_lastday_change_rate")),
+        cpm_change_pct=_decimal(metric("cpm_lastday_change_rate")),
+        exposure_change_pct=_decimal(metric("exposure_lastday_change_rate")),
+        dau_change_pct=_decimal(metric("dau_lastday_change_rate")),
+        fill_rate_change_pct=_decimal(metric("fill_rate_lastday_change_rate")),
+        source_modified_at=modified_at,
+    )
+    return RevenueSourceSnapshot(
+        observation=observation,
+        trend_windows=_build_trend_windows(
+            current_values,
+            window_minutes=window_minutes,
+            window_count=window_count,
+        ),
+    )
+
+
+def _partition_at_time(table: Any, report_time: datetime) -> Any | None:
+    spec = f"dt='{report_time:%Y%m%d%H%M%S}'"
+    if not table.exist_partition(spec):
+        return None
+    return table.get_partition(spec)

examples/tencent_realtime_control/revenue_speed_forecast.py

@@ -0,0 +1,243 @@
+"""Weighted revenue-speed forecast and versioned parameter calculation."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime, time, timedelta
+from decimal import Decimal
+
+from revenue_forecast import (
+    RevenueForecast,
+    RevenueObservation,
+    RevenueTrendWindow,
+)
+
+
+@dataclass(frozen=True)
+class RevenueSpeedSample:
+    data_date: date
+    report_time: datetime
+    parameter_version: str
+    cumulative_revenue: Decimal
+    final_revenue: Decimal
+    latest_interval_revenue: Decimal
+    previous_interval_revenue: Decimal
+    latest_weight: Decimal
+    weighted_speed: Decimal
+    remaining_revenue: Decimal
+    remaining_multiplier: Decimal
+
+
+@dataclass(frozen=True)
+class RevenueSpeedParameter:
+    parameter_version: str
+    time_slot: time
+    training_start: date
+    training_end: date
+    sample_count: int
+    latest_weight: Decimal
+    multiplier_p10: Decimal
+    multiplier_p50: Decimal
+    multiplier_p90: Decimal
+    multiplier_mad: Decimal
+
+
+class SpeedForecastUnavailable(RuntimeError):
+    """The current snapshot cannot produce a weighted-speed forecast."""
+
+
+def decimal_quantile(values: list[Decimal], percentile: Decimal) -> Decimal:
+    if not values:
+        raise ValueError("values must not be empty")
+    if percentile < 0 or percentile > 1:
+        raise ValueError("percentile must be between 0 and 1")
+    ordered = sorted(values)
+    if len(ordered) == 1:
+        return ordered[0]
+    position = Decimal(len(ordered) - 1) * percentile
+    lower_index = int(position)
+    upper_index = min(lower_index + 1, len(ordered) - 1)
+    fraction = position - Decimal(lower_index)
+    return ordered[lower_index] + (
+        ordered[upper_index] - ordered[lower_index]
+    ) * fraction
+
+
+def build_speed_samples(
+    *,
+    data_date: date,
+    intervals: list[tuple[datetime, Decimal]],
+    parameter_version: str,
+    latest_weight: Decimal,
+    start_time: time,
+    stop_time: time,
+) -> list[RevenueSpeedSample]:
+    if latest_weight < 0 or latest_weight > 1:
+        raise ValueError("latest_weight must be between 0 and 1")
+    if len(intervals) != 96:
+        return []
+    intervals = sorted(intervals, key=lambda item: item[0])
+    if any(
+        current[0] - previous[0] != timedelta(minutes=15)
+        for previous, current in zip(intervals, intervals[1:])
+    ):
+        return []
+    if any(window_start.date() != data_date for window_start, _ in intervals):
+        return []
+    final_revenue = sum((value for _, value in intervals), Decimal("0"))
+    cumulative = Decimal("0")
+    samples: list[RevenueSpeedSample] = []
+    for index, (window_start, revenue) in enumerate(intervals):
+        cumulative += revenue
+        report_time = window_start + timedelta(minutes=15)
+        if index == 0 or not start_time <= report_time.time() <= stop_time:
+            continue
+        previous_revenue = intervals[index - 1][1]
+        speed = (
+            latest_weight * revenue
+            + (Decimal("1") - latest_weight) * previous_revenue
+        )
+        remaining = final_revenue - cumulative
+        if speed <= 0 or remaining < 0:
+            continue
+        samples.append(
+            RevenueSpeedSample(
+                data_date=data_date,
+                report_time=report_time,
+                parameter_version=parameter_version,
+                cumulative_revenue=cumulative,
+                final_revenue=final_revenue,
+                latest_interval_revenue=revenue,
+                previous_interval_revenue=previous_revenue,
+                latest_weight=latest_weight,
+                weighted_speed=speed,
+                remaining_revenue=remaining,
+                remaining_multiplier=remaining / speed,
+            )
+        )
+    return samples
+
+
+def aggregate_speed_parameters(
+    samples: list[RevenueSpeedSample],
+) -> list[RevenueSpeedParameter]:
+    by_time: dict[time, list[RevenueSpeedSample]] = {}
+    for sample in samples:
+        by_time.setdefault(sample.report_time.time(), []).append(sample)
+
+    parameters: list[RevenueSpeedParameter] = []
+    for time_slot, items in sorted(by_time.items()):
+        versions = {item.parameter_version for item in items}
+        weights = {item.latest_weight for item in items}
+        if len(versions) != 1 or len(weights) != 1:
+            raise ValueError("Speed samples must use one version and one weight")
+        values = [item.remaining_multiplier for item in items]
+        median = decimal_quantile(values, Decimal("0.5"))
+        mad = decimal_quantile(
+            [abs(value - median) for value in values], Decimal("0.5")
+        )
+        parameters.append(
+            RevenueSpeedParameter(
+                parameter_version=items[0].parameter_version,
+                time_slot=time_slot,
+                training_start=min(item.data_date for item in items),
+                training_end=max(item.data_date for item in items),
+                sample_count=len(items),
+                latest_weight=items[0].latest_weight,
+                multiplier_p10=decimal_quantile(values, Decimal("0.1")),
+                multiplier_p50=median,
+                multiplier_p90=decimal_quantile(values, Decimal("0.9")),
+                multiplier_mad=mad,
+            )
+        )
+    return parameters
+
+
+def calculate_speed_forecast(
+    current: RevenueObservation,
+    trend_windows: list[RevenueTrendWindow],
+    parameter: RevenueSpeedParameter | None,
+    *,
+    forecast_version: str,
+    parameter_version: str,
+    target_cost_ratio: Decimal,
+    minimum_samples: int,
+    cost_reserve_date: date | None = None,
+    channel_costs: tuple[tuple[str, Decimal], ...] = (),
+    non_miniapp_reserved_cost: Decimal = Decimal("0"),
+) -> RevenueForecast:
+    if current.today_revenue < 0:
+        raise ValueError("today_revenue must not be negative")
+    if target_cost_ratio <= 0:
+        raise ValueError("target_cost_ratio must be positive")
+    if non_miniapp_reserved_cost < 0:
+        raise ValueError("non_miniapp_reserved_cost must not be negative")
+    unavailable_reason = None
+    if parameter is None:
+        unavailable_reason = "NO_PARAMETER"
+    elif parameter.parameter_version != parameter_version:
+        unavailable_reason = "PARAMETER_VERSION_MISMATCH"
+    elif parameter.time_slot != current.report_time.time():
+        unavailable_reason = "PARAMETER_TIME_MISMATCH"
+    elif parameter.training_end >= current.report_time.date():
+        unavailable_reason = "PARAMETER_USES_CURRENT_OR_FUTURE_DATA"
+    elif parameter.sample_count < minimum_samples:
+        unavailable_reason = "INSUFFICIENT_PARAMETER_SAMPLES"
+    elif not (
+        Decimal("0")
+        <= parameter.multiplier_p10
+        <= parameter.multiplier_p50
+        <= parameter.multiplier_p90
+    ):
+        unavailable_reason = "INVALID_PARAMETER_QUANTILES"
+    elif len(trend_windows) < 2:
+        unavailable_reason = "INSUFFICIENT_SPEED_WINDOWS"
+    if unavailable_reason is not None:
+        raise SpeedForecastUnavailable(unavailable_reason)
+
+    assert parameter is not None
+    previous_window, latest_window = trend_windows[-2:]
+    if latest_window.report_time - previous_window.report_time != timedelta(
+        minutes=15
+    ):
+        raise SpeedForecastUnavailable("NON_CONTIGUOUS_SPEED_WINDOWS")
+    speed = (
+        parameter.latest_weight * latest_window.today_revenue
+        + (Decimal("1") - parameter.latest_weight)
+        * previous_window.today_revenue
+    )
+    if speed <= 0:
+        raise SpeedForecastUnavailable("NON_POSITIVE_SPEED")
+
+    lower = current.today_revenue + speed * parameter.multiplier_p10
+    forecast = current.today_revenue + speed * parameter.multiplier_p50
+    upper = current.today_revenue + speed * parameter.multiplier_p90
+    target_daily_cost = forecast / target_cost_ratio
+    miniapp_target = max(
+        target_daily_cost - non_miniapp_reserved_cost,
+        Decimal("0"),
+    )
+    return RevenueForecast(
+        report_time=current.report_time,
+        forecast_version=forecast_version,
+        parameter_version=parameter.parameter_version,
+        current_cumulative_revenue=current.today_revenue,
+        speed_latest_weight=parameter.latest_weight,
+        latest_interval_revenue=latest_window.today_revenue,
+        previous_interval_revenue=previous_window.today_revenue,
+        weighted_speed=speed,
+        remaining_multiplier_p10=parameter.multiplier_p10,
+        remaining_multiplier_p50=parameter.multiplier_p50,
+        remaining_multiplier_p90=parameter.multiplier_p90,
+        parameter_sample_count=parameter.sample_count,
+        forecast_revenue=forecast,
+        forecast_lower=lower,
+        forecast_upper=upper,
+        target_cost_ratio=target_cost_ratio,
+        target_daily_cost=target_daily_cost,
+        cost_reserve_date=cost_reserve_date,
+        channel_costs=channel_costs,
+        non_miniapp_reserved_cost=non_miniapp_reserved_cost,
+        miniapp_target_daily_cost=miniapp_target,
+        status="WEIGHTED_SPEED_P50",
+    )

examples/tencent_realtime_control/run_control_service.py

@@ -20,6 +20,7 @@ from run_once import load_environment  # noqa: E402
 from run_scheduler import run_forever  # noqa: E402
 from storage import initialize_schema  # noqa: E402
 from roi_control.config import RoiConfig  # noqa: E402
+from revenue_forecast_config import RevenueForecastConfig  # noqa: E402
 
 
 logger = logging.getLogger("tencent_realtime_control.service")
@@ -47,6 +48,7 @@ def main() -> None:
     args = parse_args()
     apply = args.apply or _env_flag("RTC_APPLY_ENABLED")
     roi_config = RoiConfig.from_env()
+    revenue_config = RevenueForecastConfig.from_env()
     initialize_schema()
     if os.getenv("RTC_COMMAND_ENABLED", "1").strip().lower() in {
         "1",
@@ -66,6 +68,12 @@ def main() -> None:
         logger.info("ROI sheet approval polling started")
     else:
         logger.warning("ROI sheet approval polling is disabled")
+    if revenue_config.enabled:
+        from revenue_forecast_service import start_revenue_forecast_service
+
+        start_revenue_forecast_service(revenue_config)
+    else:
+        logger.info("Revenue forecast is disabled")
     run_forever(apply=apply)
 
 

examples/tencent_realtime_control/run_revenue_forecast.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+"""Collect and calculate one real-time revenue forecast."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+
+from revenue_forecast_job import run_revenue_forecast_cycle
+from run_once import load_environment
+from storage import initialize_schema
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--ignore-runtime-window",
+        action="store_true",
+        help="Run the read-only forecast outside the configured time window.",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    load_environment()
+    args = parse_args()
+    initialize_schema()
+    payload = run_revenue_forecast_cycle(
+        ignore_runtime_window=args.ignore_runtime_window
+    )
+    print(json.dumps(payload, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    raise SystemExit(main())

examples/tencent_realtime_control/schema.sql

@@ -324,3 +324,97 @@ CREATE TABLE IF NOT EXISTS operator_command_draft (
     UNIQUE KEY uk_operator_draft_conversation (chat_id, sender_open_id),
     KEY idx_operator_draft_status_expiry (status, expires_at)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='飞书运营命令多轮草稿';
+
+CREATE TABLE IF NOT EXISTS revenue_forecast_observation (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    source_partition VARCHAR(8) NOT NULL,
+    source_version VARCHAR(64) NOT NULL,
+    report_time DATETIME NOT NULL,
+    today_revenue DECIMAL(20, 4) NOT NULL,
+    overall_cpm DECIMAL(18, 6) DEFAULT NULL,
+    impressions BIGINT DEFAULT NULL,
+    dau BIGINT DEFAULT NULL,
+    fill_rate DECIMAL(12, 6) DEFAULT NULL,
+    revenue_change_pct DECIMAL(12, 6) DEFAULT NULL,
+    cpm_change_pct DECIMAL(12, 6) DEFAULT NULL,
+    exposure_change_pct DECIMAL(12, 6) DEFAULT NULL,
+    dau_change_pct DECIMAL(12, 6) DEFAULT NULL,
+    fill_rate_change_pct DECIMAL(12, 6) DEFAULT NULL,
+    source_modified_at DATETIME DEFAULT NULL,
+    lag_seconds INT NOT NULL,
+    quality_status VARCHAR(32) NOT NULL,
+    quality_message VARCHAR(255) DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_revenue_observation_version_time (report_time, source_version),
+    KEY idx_revenue_observation_quality (quality_status, report_time)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时业务收入原始快照';
+
+CREATE TABLE IF NOT EXISTS revenue_forecast_result (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    report_time DATETIME NOT NULL,
+    forecast_version VARCHAR(64) NOT NULL,
+    forecast_method VARCHAR(64) NOT NULL,
+    parameter_version VARCHAR(64) NOT NULL,
+    current_cumulative_revenue DECIMAL(20, 4) NOT NULL,
+    speed_latest_weight DECIMAL(12, 8) NOT NULL,
+    latest_interval_revenue DECIMAL(20, 4) NOT NULL,
+    previous_interval_revenue DECIMAL(20, 4) NOT NULL,
+    weighted_speed DECIMAL(20, 4) NOT NULL,
+    remaining_multiplier_p10 DECIMAL(20, 8) NOT NULL,
+    remaining_multiplier_p50 DECIMAL(20, 8) NOT NULL,
+    remaining_multiplier_p90 DECIMAL(20, 8) NOT NULL,
+    parameter_sample_count INT NOT NULL,
+    forecast_revenue DECIMAL(20, 4) NOT NULL,
+    forecast_lower DECIMAL(20, 4) NOT NULL,
+    forecast_upper DECIMAL(20, 4) NOT NULL,
+    target_cost_ratio DECIMAL(12, 6) NOT NULL,
+    target_daily_cost DECIMAL(20, 4) NOT NULL,
+    cost_reserve_date DATE DEFAULT NULL,
+    channel_costs_json TEXT DEFAULT NULL,
+    non_miniapp_reserved_cost DECIMAL(20, 4) NOT NULL DEFAULT 0,
+    miniapp_target_daily_cost DECIMAL(20, 4) NOT NULL DEFAULT 0,
+    forecast_status VARCHAR(32) NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_revenue_forecast_result (report_time, forecast_version),
+    KEY idx_revenue_forecast_version (forecast_version, report_time)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时业务收入预测结果';
+
+CREATE TABLE IF NOT EXISTS revenue_forecast_speed_sample (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    data_date DATE NOT NULL,
+    time_slot TIME NOT NULL,
+    parameter_version VARCHAR(64) NOT NULL,
+    cumulative_revenue DECIMAL(20, 4) NOT NULL,
+    final_revenue DECIMAL(20, 4) NOT NULL,
+    latest_interval_revenue DECIMAL(20, 4) NOT NULL,
+    previous_interval_revenue DECIMAL(20, 4) NOT NULL,
+    latest_weight DECIMAL(12, 8) NOT NULL,
+    weighted_speed DECIMAL(20, 4) NOT NULL,
+    remaining_revenue DECIMAL(20, 4) NOT NULL,
+    remaining_multiplier DECIMAL(20, 8) NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_revenue_speed_sample
+        (parameter_version, data_date, time_slot),
+    KEY idx_revenue_speed_sample_date (data_date, time_slot)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='收入速度模型历史日样本';
+
+CREATE TABLE IF NOT EXISTS revenue_forecast_speed_parameter (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    parameter_version VARCHAR(64) NOT NULL,
+    time_slot TIME NOT NULL,
+    training_start DATE NOT NULL,
+    training_end DATE NOT NULL,
+    sample_count INT NOT NULL,
+    latest_weight DECIMAL(12, 8) NOT NULL,
+    multiplier_p10 DECIMAL(20, 8) NOT NULL,
+    multiplier_p50 DECIMAL(20, 8) NOT NULL,
+    multiplier_p90 DECIMAL(20, 8) NOT NULL,
+    multiplier_mad DECIMAL(20, 8) NOT NULL,
+    published_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_revenue_speed_parameter
+        (parameter_version, time_slot),
+    KEY idx_revenue_speed_parameter_training
+        (training_start, training_end)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='已发布收入速度模型参数';

examples/tencent_realtime_control/storage.py

@@ -140,7 +140,7 @@ def initialize_schema() -> None:
                             f"ALTER TABLE {table_name} ADD COLUMN {column} {definition}"
                         )
 
-            roi_migrations = {
+            schema_migrations = {
                 "roi_metric_run": {
                     "fission_parameter_version": "VARCHAR(64) DEFAULT NULL",
                     "fission_cohort_date": "DATE DEFAULT NULL",
@@ -174,8 +174,39 @@ def initialize_schema() -> None:
                     "creative_rows": "INT NOT NULL DEFAULT 0",
                     "ad_rows": "INT NOT NULL DEFAULT 0",
                 },
+                "revenue_forecast_result": {
+                    "cost_reserve_date": "DATE DEFAULT NULL",
+                    "channel_costs_json": "TEXT DEFAULT NULL",
+                    "non_miniapp_reserved_cost": (
+                        "DECIMAL(20,4) NOT NULL DEFAULT 0"
+                    ),
+                    "miniapp_target_daily_cost": (
+                        "DECIMAL(20,4) NOT NULL DEFAULT 0"
+                    ),
+                    "forecast_method": (
+                        "VARCHAR(64) NOT NULL DEFAULT 'weighted_30m_speed'"
+                    ),
+                    "parameter_version": "VARCHAR(64) DEFAULT NULL",
+                    "current_cumulative_revenue": (
+                        "DECIMAL(20,4) DEFAULT NULL"
+                    ),
+                    "speed_latest_weight": "DECIMAL(12,8) DEFAULT NULL",
+                    "latest_interval_revenue": "DECIMAL(20,4) DEFAULT NULL",
+                    "previous_interval_revenue": "DECIMAL(20,4) DEFAULT NULL",
+                    "weighted_speed": "DECIMAL(20,4) DEFAULT NULL",
+                    "remaining_multiplier_p10": "DECIMAL(20,8) DEFAULT NULL",
+                    "remaining_multiplier_p50": "DECIMAL(20,8) DEFAULT NULL",
+                    "remaining_multiplier_p90": "DECIMAL(20,8) DEFAULT NULL",
+                    "parameter_sample_count": "INT DEFAULT NULL",
+                },
+                "revenue_forecast_observation": {
+                    "source_version": (
+                        "VARCHAR(64) NOT NULL DEFAULT "
+                        "'legacy'"
+                    ),
+                },
             }
-            for table_name, columns in roi_migrations.items():
+            for table_name, columns in schema_migrations.items():
                 cursor.execute(
                     """
                     SELECT COLUMN_NAME
@@ -193,6 +224,113 @@ def initialize_schema() -> None:
                             f"ALTER TABLE {table_name} "
                             f"ADD COLUMN {column} {definition}"
                         )
+            cursor.execute(
+                """
+                SELECT COLUMN_NAME, IS_NULLABLE
+                FROM information_schema.COLUMNS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='revenue_forecast_result'
+                  AND COLUMN_NAME IN (
+                    'signal_a', 'trend_growths_json', 'signal_a_weight',
+                    'signal_b_weight', 'confidence_ratio'
+                  )
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            legacy_required_columns = {
+                row["COLUMN_NAME"]
+                for row in cursor.fetchall()
+                if row["IS_NULLABLE"] == "NO"
+            }
+            legacy_definitions = {
+                "signal_a": "DECIMAL(20,4) DEFAULT NULL",
+                "trend_growths_json": "TEXT DEFAULT NULL",
+                "signal_a_weight": "DECIMAL(8,6) DEFAULT NULL",
+                "signal_b_weight": "DECIMAL(8,6) DEFAULT NULL",
+                "confidence_ratio": "DECIMAL(8,6) DEFAULT NULL",
+            }
+            for column in sorted(legacy_required_columns):
+                cursor.execute(
+                    f"ALTER TABLE revenue_forecast_result MODIFY COLUMN "
+                    f"{column} {legacy_definitions[column]}"
+                )
+            cursor.execute(
+                """
+                SELECT COLUMN_NAME, IS_NULLABLE
+                FROM information_schema.COLUMNS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='revenue_forecast_observation'
+                  AND COLUMN_NAME IN (
+                    'yesterday_same_time_revenue',
+                    'yesterday_total_revenue'
+                  )
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            for row in cursor.fetchall():
+                if row["IS_NULLABLE"] == "NO":
+                    cursor.execute(
+                        f"ALTER TABLE revenue_forecast_observation "
+                        f"MODIFY COLUMN {row['COLUMN_NAME']} "
+                        f"DECIMAL(20,4) DEFAULT NULL"
+                    )
+            cursor.execute(
+                """
+                UPDATE revenue_forecast_observation o
+                JOIN (
+                    SELECT report_time, MIN(forecast_version) AS forecast_version
+                    FROM revenue_forecast_result
+                    GROUP BY report_time
+                    HAVING COUNT(DISTINCT forecast_version) = 1
+                ) r ON r.report_time = o.report_time
+                LEFT JOIN revenue_forecast_observation exact_observation
+                  ON exact_observation.report_time = o.report_time
+                 AND exact_observation.source_version = r.forecast_version
+                 AND exact_observation.id <> o.id
+                SET o.source_version = r.forecast_version
+                WHERE o.source_version IN (
+                    'legacy', 'revenue_forecast_v4_45m'
+                )
+                  AND exact_observation.id IS NULL
+                """
+            )
+            cursor.execute(
+                """
+                SELECT INDEX_NAME
+                FROM information_schema.STATISTICS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='revenue_forecast_observation'
+                  AND INDEX_NAME='uk_revenue_observation_report_time'
+                LIMIT 1
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            if cursor.fetchone():
+                cursor.execute(
+                    """
+                    ALTER TABLE revenue_forecast_observation
+                    DROP INDEX uk_revenue_observation_report_time
+                    """
+                )
+            cursor.execute(
+                """
+                SELECT INDEX_NAME
+                FROM information_schema.STATISTICS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='revenue_forecast_observation'
+                  AND INDEX_NAME='uk_revenue_observation_version_time'
+                LIMIT 1
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            if not cursor.fetchone():
+                cursor.execute(
+                    """
+                    CREATE UNIQUE INDEX uk_revenue_observation_version_time
+                    ON revenue_forecast_observation
+                       (report_time, source_version)
+                    """
+                )
             cursor.execute(
                 """
                 SELECT INDEX_NAME

examples/tencent_realtime_control/test_revenue_forecast.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import unittest
+from datetime import date, datetime, time, timedelta
+from decimal import Decimal
+from zoneinfo import ZoneInfo
+
+from revenue_forecast import RevenueObservation, RevenueTrendWindow
+from revenue_speed_forecast import (
+    RevenueSpeedParameter,
+    SpeedForecastUnavailable,
+    aggregate_speed_parameters,
+    build_speed_samples,
+    calculate_speed_forecast,
+)
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+
+
+def observation() -> RevenueObservation:
+    return RevenueObservation(
+        partition="20260803",
+        report_time=datetime(2026, 8, 3, 12, 0, tzinfo=SHANGHAI),
+        today_revenue=Decimal("500"),
+    )
+
+
+def windows() -> list[RevenueTrendWindow]:
+    return [
+        RevenueTrendWindow(
+            report_time=datetime(2026, 8, 3, 11, 45, tzinfo=SHANGHAI),
+            today_revenue=Decimal("30"),
+        ),
+        RevenueTrendWindow(
+            report_time=datetime(2026, 8, 3, 12, 0, tzinfo=SHANGHAI),
+            today_revenue=Decimal("60"),
+        ),
+    ]
+
+
+def parameter(
+    *,
+    samples: int = 10,
+    training_end: date | None = None,
+) -> RevenueSpeedParameter:
+    return RevenueSpeedParameter(
+        parameter_version="params_v1",
+        time_slot=time(12, 0),
+        training_start=date(2026, 7, 1),
+        training_end=training_end or date(2026, 8, 2),
+        sample_count=samples,
+        latest_weight=Decimal("0.66666667"),
+        multiplier_p10=Decimal("5"),
+        multiplier_p50=Decimal("10"),
+        multiplier_p90=Decimal("15"),
+        multiplier_mad=Decimal("2"),
+    )
+
+
+class RevenueForecastTest(unittest.TestCase):
+    def test_forecast_uses_weighted_speed_and_p50(self) -> None:
+        result = calculate_speed_forecast(
+            observation(),
+            windows(),
+            parameter(),
+            forecast_version="forecast_v6",
+            parameter_version="params_v1",
+            target_cost_ratio=Decimal("3.5"),
+            minimum_samples=3,
+        )
+        expected_speed = (
+            Decimal("0.66666667") * Decimal("60")
+            + Decimal("0.33333333") * Decimal("30")
+        )
+        self.assertEqual("WEIGHTED_SPEED_P50", result.status)
+        self.assertEqual(expected_speed, result.weighted_speed)
+        self.assertEqual(Decimal("500") + expected_speed * 10, result.forecast_revenue)
+        self.assertEqual(Decimal("500") + expected_speed * 5, result.forecast_lower)
+        self.assertEqual(Decimal("500") + expected_speed * 15, result.forecast_upper)
+
+    def test_insufficient_parameter_samples_stop_forecast(self) -> None:
+        with self.assertRaisesRegex(
+            SpeedForecastUnavailable,
+            "INSUFFICIENT_PARAMETER_SAMPLES",
+        ):
+            calculate_speed_forecast(
+                observation(),
+                windows(),
+                parameter(samples=1),
+                forecast_version="forecast_v6",
+                parameter_version="params_v1",
+                target_cost_ratio=Decimal("3.5"),
+                minimum_samples=3,
+            )
+
+    def test_current_or_future_training_data_stop_forecast(self) -> None:
+        with self.assertRaisesRegex(
+            SpeedForecastUnavailable,
+            "PARAMETER_USES_CURRENT_OR_FUTURE_DATA",
+        ):
+            calculate_speed_forecast(
+                observation(),
+                windows(),
+                parameter(training_end=date(2026, 8, 3)),
+                forecast_version="forecast_v6",
+                parameter_version="params_v1",
+                target_cost_ratio=Decimal("3.5"),
+                minimum_samples=3,
+            )
+
+    def test_samples_and_parameters_share_one_formula(self) -> None:
+        intervals = [
+            (
+                datetime(2026, 8, 1, 0, 0, tzinfo=SHANGHAI)
+                + timedelta(minutes=15 * index),
+                Decimal(index + 1),
+            )
+            for index in range(96)
+        ]
+        samples = build_speed_samples(
+            data_date=date(2026, 8, 1),
+            intervals=intervals,
+            parameter_version="params_v1",
+            latest_weight=Decimal("0.66666667"),
+            start_time=time(6, 15),
+            stop_time=time(22, 30),
+        )
+        self.assertEqual(66, len(samples))
+        first = samples[0]
+        result = aggregate_speed_parameters([first])[0]
+        self.assertEqual(first.remaining_multiplier, result.multiplier_p50)
+
+    def test_non_contiguous_history_is_not_published(self) -> None:
+        intervals = [
+            (
+                datetime(2026, 8, 1, 0, 0, tzinfo=SHANGHAI)
+                + timedelta(minutes=15 * index),
+                Decimal("1"),
+            )
+            for index in range(96)
+        ]
+        intervals[50] = (intervals[50][0] + timedelta(minutes=1), Decimal("1"))
+        self.assertEqual(
+            [],
+            build_speed_samples(
+                data_date=date(2026, 8, 1),
+                intervals=intervals,
+                parameter_version="params_v1",
+                latest_weight=Decimal("0.66666667"),
+                start_time=time(6, 15),
+                stop_time=time(22, 30),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()