Agent/examples/tencent_realtime_control/revenue_forecast_source.py

"""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)