Blame for services/pattern_mining.py

"""
Association-rule mining over watch history.

Every rule is scored the same way, so they can be ranked against each other:

    support    how many times it happened (guards against small-sample noise)
    confidence P(outcome | channel) -- how reliable the habit is
    lift       confidence / P(outcome) -- how much stronger than chance

Lift is what makes a rule interesting. "You watch this channel on Fridays 20% of the
time" sounds like a pattern until you notice 1/7 is 14%, so the lift is only 1.4.

On channel eligibility: the previous gate was max(10, len(watch_events) // 100), which
scaled with total watches. A user with 20,626 watches needed 206 on one channel to
qualify, leaving 4 eligible channels and zero patterns, while a lighter user with the
same habits got plenty. The gate is now an absolute floor -- it exists for statistical
validity, not to ration results by how much someone watches.
"""

from collections import Counter, defaultdict
from datetime import datetime
from itertools import combinations
from typing import Dict, List, Optional, Set

# Minimum observations before a rule is trustworthy at all.
MIN_SUPPORT = 15
# How many of a user's channels are eligible. Top-N already restricts to their mainstays.
TOP_CHANNELS = 40
# Rules must be this much stronger than chance to be worth showing.
MIN_LIFT = 1.5
# A gap this long ends a viewing session.
SESSION_GAP_MINUTES = 60
# Sessions longer than this are almost always autoplay drift, not deliberate pairing.
MAX_SESSION_CHANNELS = 60
# A channel must span this many months before "seasonal" means anything. One watched
# only in April was never available to appear elsewhere, so its concentration carries
# no information -- that is a discovery, not a season.
MIN_SEASONAL_MONTHS = 3
# Cap per rule type so one kind cannot fill the card and hide the others.
MAX_PER_TYPE = 2

DAY_NAMES = [
    "Monday", "Tuesday", "Wednesday", "Thursday",
    "Friday", "Saturday", "Sunday",
]
MONTH_NAMES = [
    "January", "February", "March", "April", "May", "June",
    "July", "August", "September", "October", "November", "December",
]
# name -> number of hours in the slot, used as the chance baseline
SLOTS = {
    "late night": (0, 5),
    "morning": (5, 12),
    "afternoon": (12, 17),
    "evening": (17, 21),
    "night": (21, 24),
}
SLOT_PHRASE = {
    "late night": "after midnight",
    "morning": "in the mornings",
    "afternoon": "in the afternoons",
    "evening": "in the evenings",
    "night": "late in the evening",
}


def top_channels(
    watch_events: List[Dict],
    limit: int = TOP_CHANNELS,
    min_support: int = MIN_SUPPORT,
) -> Set[str]:
    """The user's mainstay channels: most-watched, above an absolute support floor."""
    counts = Counter(
        e.get("channel_clean") for e in watch_events if e.get("channel_clean")
    )
    return {ch for ch, n in counts.most_common(limit) if n >= min_support}


def _slot_for(hour: int) -> Optional[str]:
    for name, (start, end) in SLOTS.items():
        if start <= hour < end:
            return name
    return None


def _rule(kind: str, count: int, confidence: float, lift: float, insight: str, **extra) -> Dict:
    return {
        "type": kind,
        "count": count,
        "confidence": confidence,
        "lift": lift,
        "insight": insight,
        **extra,
    }


def _association_rules(
    watch_events: List[Dict],
    channels: Set[str],
    key,
    label,
    phrase,
    kind: str,
    min_confidence: float,
    extra_field: str,
    min_distinct: int = 1,
) -> List[Dict]:
    """Shared channel x attribute mining, used for day, slot and month rules."""
    per_channel: Dict[str, Counter] = defaultdict(Counter)
    channel_total: Counter = Counter()
    attribute_total: Counter = Counter()

    for event in watch_events:
        value = key(event)
        if value is None:
            continue

        # The chance baseline is the share of ALL viewing with this attribute, not just
        # viewing on eligible channels. Restricting it to eligible channels compares a
        # channel partly against itself, which drives lift toward 1 and hides real
        # patterns whenever few channels qualify.
        attribute_total[value] += 1

        channel = event.get("channel_clean")
        if channel in channels:
            per_channel[channel][value] += 1
            channel_total[channel] += 1

    total = sum(attribute_total.values())
    if not total:
        return []

    rules = []
    for channel, values in per_channel.items():
        # Concentration only means something if the channel could have appeared
        # elsewhere; see MIN_SEASONAL_MONTHS.
        if len(values) < min_distinct:
            continue
        for value, count in values.items():
            if count < MIN_SUPPORT:
                continue
            confidence = count / channel_total[channel]
            baseline = attribute_total[value] / total
            lift = confidence / baseline if baseline else 0.0
            if confidence >= min_confidence and lift >= MIN_LIFT:
                rules.append(
                    _rule(
                        kind, count, confidence, lift,
                        phrase(channel, label(value), confidence),
                        channel=channel, **{extra_field: label(value)},
                    )
                )
    return rules


def channel_day_rules(watch_events: List[Dict], channels: Set[str]) -> List[Dict]:
    """Channels tied to a day of the week. Chance baseline is 1/7."""
    return _association_rules(
        watch_events, channels,
        key=lambda e: e.get("day_of_week"),
        label=lambda d: DAY_NAMES[d] if 0 <= d < 7 else "Unknown",
        phrase=lambda ch, day, conf: (
            f"You watch **{ch}** on **{day}s** ({int(conf * 100)}% of the time)"
        ),
        kind="channel_day", min_confidence=0.25, extra_field="day",
    )


def channel_slot_rules(watch_events: List[Dict], channels: Set[str]) -> List[Dict]:
    """Channels tied to a time of day. Baseline is the slot's real share of hours."""
    return _association_rules(
        watch_events, channels,
        key=lambda e: _slot_for(e["hour_local"]) if e.get("hour_local") is not None else None,
        label=lambda s: s,
        phrase=lambda ch, slot, conf: (
            f"**{ch}** is your go-to {SLOT_PHRASE.get(slot, slot)} "
            f"({int(conf * 100)}% of the time)"
        ),
        kind="channel_time", min_confidence=0.35, extra_field="slot",
    )


def seasonal_rules(watch_events: List[Dict], channels: Set[str]) -> List[Dict]:
    """Channels concentrated in one month -- sports seasons, exam periods, holidays."""
    return _association_rules(
        watch_events, channels,
        key=lambda e: e.get("month_local"),
        label=lambda m: MONTH_NAMES[m - 1] if 1 <= m <= 12 else "Unknown",
        phrase=lambda ch, month, conf: (
            f"**{ch}** is your **{month}** channel ({int(conf * 100)}% of it landed there)"
        ),
        kind="seasonal", min_confidence=0.35, extra_field="month",
        min_distinct=MIN_SEASONAL_MONTHS,
    )


def weekend_rule(watch_events: List[Dict], channels: Set[str]) -> Optional[Dict]:
    """Whether a mainstay channel skews to weekends. Chance baseline is 2/7."""
    weekend = Counter()
    totals = Counter()

    for event in watch_events:
        channel = event.get("channel_clean")
        day = event.get("day_of_week")
        if channel not in channels or day is None:
            continue
        totals[channel] += 1
        if day in (5, 6):
            weekend[channel] += 1

    best, best_lift, best_count = None, 0.0, 0
    for channel, total in totals.items():
        if total < MIN_SUPPORT:
            continue
        confidence = weekend[channel] / total
        lift = confidence / (2 / 7)
        if lift > best_lift:
            best, best_lift, best_count = channel, lift, weekend[channel]

    if best is None or best_lift < MIN_LIFT:
        return None

    confidence = best_count / totals[best]
    return _rule(
        "weekend", best_count, confidence, best_lift,
        f"**{best}** is your weekend channel ({int(confidence * 100)}% on Sat/Sun)",
        channel=best,
    )


def _sessions(watch_events: List[Dict]) -> List[List[str]]:
    """Channel sequences, split wherever viewing pauses for SESSION_GAP_MINUTES."""
    timed = []
    for event in watch_events:
        stamp = event.get("timestamp_local") or event.get("timestamp_utc")
        channel = event.get("channel_clean")
        if not stamp or not channel:
            continue
        try:
            timed.append((datetime.fromisoformat(stamp.replace("Z", "+00:00")), channel))
        except ValueError:
            continue

    if not timed:
        return []

    timed.sort(key=lambda pair: pair[0])
    sessions, current = [], [timed[0][1]]

    for index in range(1, len(timed)):
        gap = (timed[index][0] - timed[index - 1][0]).total_seconds() / 60
        if gap > SESSION_GAP_MINUTES:
            sessions.append(current)
            current = []
        current.append(timed[index][1])

    sessions.append(current)
    return sessions


def cooccurrence_rules(watch_events: List[Dict]) -> List[Dict]:
    """Channel pairs that show up in the same sitting far more often than chance.

    This is the classic market-basket rule -- sessions are baskets, channels are items.
    It surfaces groupings nobody configured: artists of a genre, a sport and its
    highlights reel.
    """
    sessions = _sessions(watch_events)
    if not sessions:
        return []

    total = len(sessions)
    appears = Counter()
    together = Counter()

    for session in sessions:
        unique = sorted(set(session))
        for channel in unique:
            appears[channel] += 1
        # Very long sessions are autoplay drift; every pair in them would look related.
        if len(unique) > MAX_SESSION_CHANNELS:
            continue
        for pair in combinations(unique, 2):
            together[pair] += 1

    rules = []
    for (first, second), count in together.items():
        if count < MIN_SUPPORT:
            continue
        expected = (appears[first] / total) * (appears[second] / total)
        lift = (count / total) / expected if expected else 0.0
        if lift < MIN_LIFT * 2:  # pairs are noisier, so demand a stronger signal
            continue
        confidence = max(count / appears[first], count / appears[second])
        rules.append(
            _rule(
                "cooccurrence", count, confidence, lift,
                f"**{first}** and **{second}** go together "
                f"({count} sessions, {lift:.1f}x more than chance)",
                channel_a=first, channel_b=second,
            )
        )
    return rules


def opener_rules(watch_events: List[Dict]) -> List[Dict]:
    """Channels that tend to start a viewing session -- the gateway into a run."""
    sessions = _sessions(watch_events)
    if not sessions:
        return []

    total = len(sessions)
    opens = Counter(session[0] for session in sessions if session)
    appears = Counter()
    for session in sessions:
        for channel in set(session):
            appears[channel] += 1

    # Chance baseline: across every channel-appearance, what share are openers? Each
    # session has exactly one opener out of however many channels it contains, so this
    # is 1/(average distinct channels per session). Dividing openers by sessions instead
    # would give 1.0 always, making every lift exactly 1.
    appearances = sum(appears.values())
    baseline = total / appearances if appearances else 0

    rules = []
    for channel, count in opens.items():
        if count < MIN_SUPPORT:
            continue
        confidence = count / appears[channel]
        lift = confidence / baseline if baseline else 0.0
        if lift < MIN_LIFT:
            continue
        rules.append(
            _rule(
                "opener", count, confidence, lift,
                f"**{channel}** is how you start a session "
                f"({count} times, {int(confidence * 100)}% of its sessions)",
                channel=channel,
            )
        )
    rules.sort(key=lambda r: r["count"], reverse=True)
    return rules


def find_patterns(watch_events: List[Dict], limit: int = 5) -> List[Dict]:
    """All rule types, ranked by confidence x lift, strongest first."""
    if not watch_events:
        return []

    channels = top_channels(watch_events)
    if not channels:
        return []

    rules: List[Dict] = []
    rules += channel_day_rules(watch_events, channels)
    rules += channel_slot_rules(watch_events, channels)
    rules += seasonal_rules(watch_events, channels)
    rules += cooccurrence_rules(watch_events)
    rules += opener_rules(watch_events)

    weekend = weekend_rule(watch_events, channels)
    if weekend:
        rules.append(weekend)

    rules.sort(key=lambda r: r["lift"] * r["confidence"], reverse=True)

    # Take the strongest rules, but never more than MAX_PER_TYPE of one kind: five
    # variations on "X is your April channel" reads as a malfunction, not a discovery.
    selected: List[Dict] = []
    per_type: Counter = Counter()
    for rule in rules:
        if per_type[rule["type"]] >= MAX_PER_TYPE:
            continue
        selected.append(rule)
        per_type[rule["type"]] += 1
        if len(selected) == limit:
            break

    return selected