pipeline/insights.py

"""CS#1 "what worked" creative analysis (SPEC_NEW §16.4).

`analyze_winners(days=14)` is a Claude skill over the last N days of
`metrics_daily` joined to the creative attributes that drive learning — the ad
ANGLE (`videos.tags->>'angle'`), the startframe AUDIENCE (`avatars.audience`),
and per-ad performance (`publishes` → `metrics_daily`). It determines WHAT
WORKED: which audience performed, which angles won, and which startframe traits
correlate with purchases / low CPA. It writes ONE `creative_insights` row
`{period_days, winning_audience, winning_angles[], winning_startframe_traits,
evidence}` (the owner's original `cs1_analyses` intent) and returns it with
uuids stringified.

The output feeds the strategist (§7.5) AND the audience→startframe selector
(§16.5). It reuses the §7.5 EXPLOIT data guard: a creative needs ≥ 2,000
impressions to count as PROVEN. Claude (`INSIGHTS_SYSTEM`) decides the winners
when `ANTHROPIC_API_KEY` is set — parsed defensively, retried once with the
error appended; otherwise a deterministic fallback ranks the proven creatives by
purchases then CPA (so dry-run / tests work with no key). An empty DB still
writes a cold-start row.
"""

from __future__ import annotations

import json

from psycopg.types.json import Json

from .config import get_settings

# The §7.5 EXPLOIT data guard, restated here so the SQL pre-filter, the prompt,
# and the fallback all agree on what "proven" means (≥ 2,000 impressions).
MIN_PROVEN_IMPRESSIONS = 2000


INSIGHTS_SYSTEM = f"""You analyse what is WORKING in an automated Swedish UGC \
video-ad account, so the pipeline can make more of it. You see per-creative \
performance for the last N days, aggregated by ad ANGLE and by the startframe's \
target AUDIENCE (gender, age, appearance, setting, vibe — free text, no fixed \
buckets). Each row carries impressions, spend, purchases, revenue, CPA, the 3s \
hook rate, and whether it is PROVEN.

Your job is to name what worked:
- DATA GUARD. A creative only counts as PROVEN once it has at least \
{MIN_PROVEN_IMPRESSIONS} impressions. Never call something a winner on thinner \
data — the rows already flag this as `proven`; reason only from proven rows when \
deciding winners (mention unproven ones only as "too early to tell").
- WINNING AUDIENCE. From the proven rows, decide which AUDIENCE performed best \
(most purchases at the lowest CPA). Describe it free-form as an object with the \
same shape as the audience metadata you were given (gender, age, appearance, \
setting, vibe) — the single audience the data says to lean into.
- WINNING ANGLES. List the ad angles that won, best first.
- WINNING STARTFRAME TRAITS. The startframe/audience traits (look, setting, \
vibe, age, gender) that correlate with purchases and low CPA — the levers to \
reuse when generating the next startframes.
- EVIDENCE. A short snapshot of the ranked proven creatives your call rests on.

Output ONLY this JSON object — no prose, no markdown fences:
{{"winning_audience": {{...}} | null, "winning_angles": ["string", ...], \
"winning_startframe_traits": {{...}} | null, "evidence": {{...}} | [...]}}"""


def _aggregate_winner_stats(days: int) -> list[dict]:
    """Aggregate metrics_daily (last `days`) per creative attribute — angle
    (videos.tags->>'angle') + the startframe AUDIENCE — joining publishes →
    videos and LEFT JOIN avatars for the startframe's audience (SPEC_NEW §16.4).

    The feature space is creative levers ONLY: this SQL reads videos.tags and
    avatars.audience, never videos.seed (§12.4). Each row carries the §7.5 guard
    verdict (`proven`, ≥ 2000 impressions) so the prompt and the fallback agree
    on what counts.
    """
    from . import db

    rows = db.fetch_all(
        """
        select
            v.tags->>'angle'            as angle,
            a.id                        as avatar_id,
            a.name                      as avatar_name,
            a.audience                  as audience,
            a.selection_text            as selection_text,
            coalesce(sum(m.impressions), 0) as impressions,
            coalesce(sum(m.spend), 0)       as spend,
            coalesce(sum(m.video_3s), 0)    as video_3s,
            coalesce(sum(m.purchases), 0)   as purchases,
            coalesce(sum(m.revenue), 0)     as revenue
          from metrics_daily m
          join publishes p on p.ad_id = m.ad_id
          join videos    v on v.id = p.video_id
          left join avatars a on a.id = v.avatar_id
         where m.date >= (current_date - %s::int)
         group by v.tags->>'angle', a.id, a.name, a.audience, a.selection_text
         order by purchases desc, revenue desc
        """,
        (days,),
    )
    stats: list[dict] = []
    for r in rows:
        impressions = int(r["impressions"] or 0)
        spend = float(r["spend"] or 0.0)
        video_3s = int(r["video_3s"] or 0)
        purchases = int(r["purchases"] or 0)
        revenue = float(r["revenue"] or 0.0)
        cpa = round(spend / purchases, 2) if purchases else None
        three_s_rate = round(video_3s / impressions, 4) if impressions else None
        # avatars.audience reads back as a python dict (jsonb); {} for legacy/no avatar.
        audience = r["audience"] if isinstance(r["audience"], dict) else {}
        stats.append({
            "angle": r["angle"],
            "avatar_id": str(r["avatar_id"]) if r["avatar_id"] else None,
            "avatar_name": r["avatar_name"],
            "audience": audience,
            "selection_text": r["selection_text"],
            "impressions": impressions,
            "spend": round(spend, 2),
            "video_3s": video_3s,
            "purchases": purchases,
            "revenue": round(revenue, 2),
            "cpa": cpa,
            "three_s_rate": three_s_rate,
            "proven": impressions >= MIN_PROVEN_IMPRESSIONS,
        })
    return stats


def _audience_summary(audience: dict | None) -> str:
    """One-line human summary of an audience object (for traits text / evidence)."""
    if not audience:
        return ""
    parts = [str(audience[k]) for k in ("gender", "age", "appearance", "setting", "vibe")
             if audience.get(k)]
    return ", ".join(parts)


def _fallback_insight(stats: list[dict]) -> dict:
    """Deterministic CS#1 for no-key / dry-run (SPEC_NEW §16.4).

    Rank the PROVEN creatives (≥ 2000 impressions) by purchases then lowest CPA;
    the best one's audience is the winning audience, its angle leads the winning
    angles, and its audience traits become the winning startframe traits. The
    evidence is the ranked proven snapshot. Cold start (no proven creatives)
    yields empty winners with a cold_start evidence note — never a crash.
    """
    proven = [s for s in stats if s.get("proven")]
    # Most purchases first, then lowest CPA (None CPA sorts last).
    proven.sort(
        key=lambda s: (
            -int(s.get("purchases") or 0),
            s.get("cpa") if s.get("cpa") is not None else float("inf"),
        )
    )

    if not proven:
        return {
            "winning_audience": None,
            "winning_angles": [],
            "winning_startframe_traits": None,
            "evidence": {
                "cold_start": True,
                "proven_creatives": 0,
                "note": (
                    f"No creative has cleared the {MIN_PROVEN_IMPRESSIONS}-impression "
                    "guard yet, so there is nothing proven to learn from."
                ),
                "ranked": [],
            },
        }

    best = proven[0]
    # Winning angles: distinct proven angles ordered by total purchases (best first).
    by_angle: dict[str, int] = {}
    for s in proven:
        angle = (s.get("angle") or "").strip()
        if angle:
            by_angle[angle] = by_angle.get(angle, 0) + int(s.get("purchases") or 0)
    winning_angles = [a for a, _ in sorted(by_angle.items(), key=lambda kv: kv[1], reverse=True)]

    best_audience = best.get("audience") or {}
    traits = dict(best_audience) if best_audience else {}
    if traits:
        traits["summary"] = _audience_summary(best_audience)

    return {
        "winning_audience": best_audience or {
            "summary": f"Best creative: {best.get('avatar_name') or 'unknown avatar'}"
        },
        "winning_angles": winning_angles,
        "winning_startframe_traits": traits or None,
        "evidence": {
            "cold_start": False,
            "proven_creatives": len(proven),
            "best": {
                "angle": best.get("angle"),
                "avatar_name": best.get("avatar_name"),
                "audience": best_audience,
                "purchases": best.get("purchases"),
                "cpa": best.get("cpa"),
                "impressions": best.get("impressions"),
            },
            "ranked": proven,
        },
    }


def _parse_insight_json(raw: str) -> dict:
    """Defensive parse of the insights JSON (first '{' to last '}'). Returns the
    normalised insight dict; raises ValueError on anything malformed (SPEC_NEW §16.4)."""
    s = raw.strip()
    a, b = s.find("{"), s.rfind("}")
    if a == -1 or b == -1:
        raise ValueError("no JSON object in insights output")
    obj = json.loads(s[a:b + 1])
    if not isinstance(obj, dict):
        raise ValueError("insights JSON is not an object")

    angles_raw = obj.get("winning_angles")
    if angles_raw is None:
        angles_raw = []
    if not isinstance(angles_raw, list):
        raise ValueError("winning_angles must be a list")
    winning_angles = [str(x).strip() for x in angles_raw if str(x).strip()]

    audience = obj.get("winning_audience")
    if audience is not None and not isinstance(audience, (dict, str)):
        raise ValueError("winning_audience must be an object, string, or null")
    traits = obj.get("winning_startframe_traits")
    if traits is not None and not isinstance(traits, (dict, str, list)):
        raise ValueError("winning_startframe_traits must be an object, string, list, or null")
    evidence = obj.get("evidence")
    if evidence is not None and not isinstance(evidence, (dict, list)):
        raise ValueError("evidence must be an object, list, or null")

    return {
        "winning_audience": audience,
        "winning_angles": winning_angles,
        "winning_startframe_traits": traits,
        "evidence": evidence,
    }


def _build_user_prompt(stats: list[dict], days: int) -> str:
    """The user message for the Claude call: the period and the per-creative stats
    (proven flag included). JSON so the model reads exact numbers."""
    return (
        f"Period: last {days} days. Per-creative performance, aggregated by angle "
        f"and startframe audience (a creative is `proven` once it clears "
        f"{MIN_PROVEN_IMPRESSIONS} impressions):\n\n"
        f"{json.dumps(stats, ensure_ascii=False, default=str)}\n\n"
        "Decide the winning audience, the winning angles, the winning startframe "
        "traits, and the evidence. Reply with ONLY the JSON object."
    )


def analyze_winners(days: int = 14) -> dict:
    """Run CS#1 "what worked" analysis and insert ONE creative_insights row (SPEC_NEW §16.4).

    Aggregates the last `days` of metrics_daily per creative attribute (angle +
    startframe audience), reusing the §7.5 EXPLOIT guard (≥ 2000 impressions =
    PROVEN). Claude (INSIGHTS_SYSTEM) names the winning_audience, winning_angles,
    winning_startframe_traits, and evidence when ANTHROPIC_API_KEY is set —
    parsed defensively, retried ONCE with the error appended; otherwise a
    deterministic fallback ranks the proven creatives by purchases then CPA (so
    dry-run / tests work with no key). An empty DB still inserts a cold-start row.

    Returns the inserted creative_insights row (id stringified).
    """
    from . import db

    days = int(days)
    if days < 1:
        raise ValueError("analyze_winners: days must be >= 1")

    s = get_settings()
    stats = _aggregate_winner_stats(days)

    insight: dict | None = None
    if s.anthropic_api_key:
        import anthropic

        client = anthropic.Anthropic(api_key=s.anthropic_api_key)
        user = _build_user_prompt(stats, days)
        last_err: str | None = None
        for attempt in range(2):
            content = user if attempt == 0 else (
                f"{user}\n\nYour previous reply could not be used ({last_err}). "
                f"Reply with ONLY valid JSON in the required format."
            )
            resp = client.messages.create(
                model=s.anthropic_model, max_tokens=2048,
                system=INSIGHTS_SYSTEM, messages=[{"role": "user", "content": content}],
            )
            text = "".join(b.text for b in resp.content if getattr(b, "type", None) == "text")
            try:
                insight = _parse_insight_json(text)
            except Exception as exc:  # noqa: BLE001 — retry once, then fall back
                last_err = str(exc)
                continue
            break
        if insight is None:  # Claude failed twice → deterministic fallback (still works)
            insight = _fallback_insight(stats)
    else:
        insight = _fallback_insight(stats)

    # Persist exactly ONE creative_insights row; the four learning surfaces are
    # jsonb (via Json(...)). evidence defaults to the raw stats snapshot.
    with db.connect() as conn:
        row = conn.execute(
            """
            insert into creative_insights
                (period_days, winning_audience, winning_angles,
                 winning_startframe_traits, evidence)
            values (%s, %s, %s, %s, %s)
            returning id, period_days, winning_audience, winning_angles,
                      winning_startframe_traits, evidence, created_at
            """,
            (
                days,
                Json(insight.get("winning_audience")),
                Json(insight.get("winning_angles") or []),
                Json(insight.get("winning_startframe_traits")),
                Json(insight.get("evidence") if insight.get("evidence") is not None else stats),
            ),
        ).fetchone()
    d = dict(row)
    d["id"] = str(d["id"])
    return d