pipeline/rules.py

"""Kill/winner rules (SPEC_NEW §7.8, Phase 5) — read-only reporting.

v1 NEVER auto-pauses on Meta; these helpers only *surface* candidates for the
`report` command (and the UI). Thresholds derive from `TARGET_CPA` (env).

- **Kill:** spend ≥ 1.5× target CPA with 0 purchases, OR hook rate (3s/impr)
  < 15% after 2,000 impressions → surface (would mark video `killed`).
- **Winner:** ≥ 3 purchases and CPA ≤ 0.8× target CPA → surface for manual scaling.

Both aggregate `metrics_daily` per `ad_id` (sum the counters), join
`publishes → videos` to attach `video_id`, `ad_name`, and `angle`
(`videos.tags->>'angle'`). All divisions are null-safe via `nullif`, all sums
via `coalesce`.
"""

from __future__ import annotations

from . import db
from .config import get_settings

# Aggregate metrics_daily per ad_id and attach the creative's video/angle. One
# row per published ad_id that has at least one metrics_daily row. nullif guards
# every division (impressions/purchases can be 0 or null); coalesce guards sums.
_BASE_CTE = """
with agg as (
    select
        m.ad_id,
        sum(coalesce(m.impressions, 0)) as impressions,
        sum(coalesce(m.spend, 0))       as spend,
        sum(coalesce(m.purchases, 0))   as purchases,
        sum(coalesce(m.revenue, 0))     as revenue,
        sum(coalesce(m.video_3s, 0))    as video_3s
    from metrics_daily m
    group by m.ad_id
),
joined as (
    select
        p.video_id,
        agg.ad_id,
        p.ad_name,
        v.tags->>'angle' as angle,
        agg.impressions,
        agg.spend,
        agg.purchases,
        agg.revenue,
        agg.video_3s,
        agg.video_3s::numeric / nullif(agg.impressions, 0) as hook_rate,
        agg.spend / nullif(agg.purchases, 0)               as cpa
    from agg
    join publishes p on p.ad_id = agg.ad_id
    left join videos v on v.id = p.video_id
)
"""


def kill_candidates() -> list[dict]:
    """Ads that should be killed (SPEC_NEW §7.8). Read-only — surfaced, not paused.

    KILL when spend ≥ 1.5×TARGET_CPA AND purchases = 0, OR impressions ≥ 2000
    AND hook_rate < 0.15. `reason` is a short human-readable string.
    """
    target_cpa = get_settings().target_cpa
    rows = db.fetch_all(
        _BASE_CTE
        + """
        select
            video_id, ad_id, ad_name, angle,
            case
                when spend >= %(spend_kill)s and purchases = 0
                    then 'spend ' || round(spend) || ' SEK, 0 purchases'
                else 'hook rate ' || round(coalesce(hook_rate, 0) * 100) ||
                     '%% < 15%% over ' || impressions || ' impressions'
            end as reason,
            spend, impressions, purchases, hook_rate
        from joined
        where (spend >= %(spend_kill)s and purchases = 0)
           or (impressions >= 2000 and coalesce(hook_rate, 0) < 0.15)
        order by spend desc
        """,
        {"spend_kill": 1.5 * target_cpa},
    )
    return rows


def winner_candidates() -> list[dict]:
    """Ads worth scaling (SPEC_NEW §7.8). Read-only — surfaced for manual scaling.

    WINNER when purchases ≥ 3 AND CPA ≤ 0.8×TARGET_CPA.
    """
    target_cpa = get_settings().target_cpa
    rows = db.fetch_all(
        _BASE_CTE
        + """
        select
            video_id, ad_id, ad_name, angle,
            purchases, cpa, revenue, spend
        from joined
        where purchases >= 3
          and cpa is not null
          and cpa <= %(cpa_win)s
        order by cpa asc
        """,
        {"cpa_win": 0.8 * target_cpa},
    )
    return rows