rm_traffic/availability_algos.py

"""
9 Availability Algorithms for Maximum Traffic

Each algo controls when/how the profile goes available/offline to maximize
visibility, search rank, and traffic attraction.

Algos:
  1. JitterBurst      — brief offline pulses → "recently available" signal
  2. PeakHourSync     — match availability to historical peak traffic hours
  3. CompetitorGap    — go available when competitors drop off
  4. RefreshCascade   — staggered refresh before expiry, never gap
  5. SearchRankBoost  — toggle to appear in "available now" filters more
  6. DemandPulse      — short availability windows during low-traffic to capture pent-up demand
  7. GeoRotation      — rotate timing to catch multiple timezone traffic waves
  8. EngagementTrigger— refresh immediately after receiving a visit/message
  9. BackoffRecovery  — exponential backoff on API failure, never lose presence
"""

import time
import random
import logging
import sqlite3
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field

log = logging.getLogger("rm.avail_algos")


@dataclass
class AlgoState:
    """Shared state across all algos."""
    available: bool = False
    last_refresh: float = 0
    last_visit: float = 0
    last_message: float = 0
    competitor_count: int = 0
    competitor_available: int = 0
    profile_views_1h: int = 0
    profile_views_24h: int = 0
    current_hour: int = 0
    day_of_week: int = 0
    api_failures: int = 0
    last_api_error: str = ""
    availability_option: int = 0
    availability_expires: float = 0
    bursts_executed: int = 0
    refreshes_executed: int = 0
    toggles_executed: int = 0
    algo_history: List[Dict] = field(default_factory=list)
    # ── Traffic Attribution ──
    attribution: Dict[str, Dict] = field(default_factory=lambda: {})
    baseline_views: int = 0
    last_attribution_ts: float = 0
    view_samples: List[Dict] = field(default_factory=list)
    algo_fire_log: List[Dict] = field(default_factory=list)
    baseline_rate: float = 0.0  # views per minute when no algo recently fired


def _record(state: AlgoState, algo: str, action: str, detail: str = ""):
    entry = {"ts": time.time(), "algo": algo, "action": action, "detail": detail}
    state.algo_history.append(entry)
    if len(state.algo_history) > 200:
        state.algo_history = state.algo_history[-200:]
    log.info(f"[{algo}] {action} — {detail}")


def _capture_attribution(state: AlgoState, algo_name: str, action: str):
    """Record that an algo fired — attribution computed later in get_attribution_summary."""
    now = time.time()
    state.algo_fire_log.append({
        "ts": now,
        "algo": algo_name,
        "action": action,
        "views_at_fire": state.profile_views_24h,
    })
    if len(state.algo_fire_log) > 500:
        state.algo_fire_log = state.algo_fire_log[-500:]


def sample_views(state: AlgoState):
    """Call this every cycle to sample current view count.
    Builds a continuous timeline for baseline + lift computation.
    """
    now = time.time()
    state.view_samples.append({
        "ts": now,
        "views_24h": state.profile_views_24h,
        "available": state.available,
    })
    if len(state.view_samples) > 1000:
        state.view_samples = state.view_samples[-1000:]


def _compute_baseline_rate(state: AlgoState) -> float:
    """Compute baseline view rate (views/min) during periods with no algo fire in the prior 30 min."""
    if len(state.view_samples) < 2:
        return 0.0
    fire_ts_set = [f["ts"] for f in state.algo_fire_log]
    quiet_samples = []
    for s in state.view_samples:
        recent_fire = any(abs(s["ts"] - ft) < 1800 for ft in fire_ts_set)
        if not recent_fire:
            quiet_samples.append(s)
    if len(quiet_samples) < 2:
        quiet_samples = state.view_samples
    deltas = []
    for i in range(1, len(quiet_samples)):
        dt = quiet_samples[i]["ts"] - quiet_samples[i-1]["ts"]
        dv = quiet_samples[i]["views_24h"] - quiet_samples[i-1]["views_24h"]
        if dt > 0:
            deltas.append(dv / dt * 60)  # views per minute
    if not deltas:
        return 0.0
    deltas.sort()
    mid = len(deltas) // 2
    return deltas[mid]  # median rate


def get_attribution_summary(state: AlgoState) -> Dict:
    """Compute per-algo traffic attribution using lift over baseline.

    For each algo fire, we look at views gained in the 30-minute window
    after the fire, compare to expected baseline views in that same window.
    Lift = actual - expected. Attribution = lift credited to that algo.
    """
    baseline_rate = _compute_baseline_rate(state)
    state.baseline_rate = baseline_rate
    window = 1800  # 30 min attribution window
    now = time.time()

    per_algo = {}
    for fire in state.algo_fire_log:
        algo = fire["algo"]
        fire_ts = fire["ts"]
        views_at_fire = fire["views_at_fire"]
        # Find view sample closest to fire_ts + window
        end_ts = fire_ts + window
        closest = None
        for s in state.view_samples:
            if abs(s["ts"] - end_ts) < 300:  # within 5 min of window end
                if closest is None or abs(s["ts"] - end_ts) < abs(closest["ts"] - end_ts):
                    closest = s
        if closest is None:
            continue
        actual_views = closest["views_24h"] - views_at_fire
        if actual_views < 0:
            actual_views = 0
        expected_views = baseline_rate * (window / 60)
        lift = actual_views - expected_views
        if algo not in per_algo:
            per_algo[algo] = {
                "fires": 0,
                "total_actual_views": 0,
                "total_expected_views": 0,
                "total_lift": 0,
                "last_action": "",
                "last_ts": 0,
            }
        a = per_algo[algo]
        a["fires"] += 1
        a["total_actual_views"] += actual_views
        a["total_expected_views"] += expected_views
        a["total_lift"] += lift
        a["last_action"] = fire["action"]
        a["last_ts"] = fire_ts

    summary = []
    for name, a in per_algo.items():
        avg_lift = a["total_lift"] / max(a["fires"], 1)
        summary.append({
            "algo": name,
            "fires": a["fires"],
            "total_actual_views": a["total_actual_views"],
            "total_expected_views": round(a["total_expected_views"], 1),
            "total_lift": round(a["total_lift"], 1),
            "avg_lift_per_fire": round(avg_lift, 2),
            "last_action": a["last_action"],
            "last_ts": a["last_ts"],
        })
    summary.sort(key=lambda x: x["total_lift"], reverse=True)
    total_lift = sum(s["total_lift"] for s in summary)
    for s in summary:
        s["share_pct"] = round(s["total_lift"] / max(total_lift, 1) * 100, 1) if total_lift > 0 else 0
        s["verdict"] = (
            "POSITIVE" if s["total_lift"] > 0 and s["fires"] >= 2 else
            "NEGATIVE" if s["total_lift"] < 0 and s["fires"] >= 2 else
            "INSUFFICIENT_DATA"
        )
    return {
        "total_lift": round(total_lift, 1),
        "baseline_rate_views_per_min": round(baseline_rate, 3),
        "algos": summary,
        "current_views_24h": state.profile_views_24h,
        "samples_collected": len(state.view_samples),
        "algo_fires_logged": len(state.algo_fire_log),
        "attribution_window_min": window // 60,
    }


# ═══════════════════════════════════════════════════════════════════
# Algo 1: JitterBurst — brief offline pulse then back online
# ═══════════════════════════════════════════════════════════════════

class JitterBurst:
    """2-5s offline pulse every 30 min to trigger 'recently available' signal."""

    name = "jitter_burst"
    interval = 1800  # 30 min
    min_offline = 2.0
    max_offline = 5.0

    def __init__(self):
        self.last_burst = 0

    def should_fire(self, state: AlgoState) -> bool:
        if not state.available:
            return False
        return (time.time() - self.last_burst) > self.interval

    def execute(self, api, state: AlgoState) -> Dict:
        duration = random.uniform(self.min_offline, self.max_offline)
        ts = time.time()
        try:
            api.set_availability(option=2, duration=0)
            state.available = False
            time.sleep(duration)
            api.set_availability(option=1, duration=5)
            state.available = True
            state.bursts_executed += 1
            self.last_burst = ts
            _record(state, self.name, "burst", f"offline {duration:.1f}s")
            return {"algo": self.name, "action": "burst", "duration": round(duration, 1), "verified": True}
        except Exception as e:
            try:
                api.set_availability(option=1, duration=5)
                state.available = True
            except Exception:
                pass
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 2: PeakHourSync — availability matched to traffic peaks
# ═══════════════════════════════════════════════════════════════════

class PeakHourSync:
    """Ensure availability during historically high-traffic hours."""

    name = "peak_hour_sync"
    # Peak hours: 8-11am, 1-3pm, 7-11pm (local time, assumed ET)
    PEAK_HOURS = {8, 9, 10, 11, 13, 14, 15, 19, 20, 21, 22, 23}
    SHOULDER_HOURS = {7, 12, 16, 17, 18, 0, 1}

    def should_fire(self, state: AlgoState) -> bool:
        hour = state.current_hour
        if hour in self.PEAK_HOURS and not state.available:
            return True
        if hour in self.SHOULDER_HOURS and not state.available:
            return random.random() < 0.5
        return False

    def execute(self, api, state: AlgoState) -> Dict:
        hour = state.current_hour
        is_peak = hour in self.PEAK_HOURS
        duration = 6 if is_peak else 3
        try:
            api.set_availability(option=1, duration=duration)
            state.available = True
            state.refreshes_executed += 1
            _record(state, self.name, "sync", f"hour={hour} peak={is_peak} dur={duration}h")
            return {"algo": self.name, "action": "sync", "hour": hour, "peak": is_peak, "duration": duration}
        except Exception as e:
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 3: CompetitorGap — go available when competitors drop
# ═══════════════════════════════════════════════════════════════════

class CompetitorGap:
    """Monitor competitor availability — go available when they drop."""

    name = "competitor_gap"
    threshold_ratio = 0.3  # if <30% of competitors available, fire

    def should_fire(self, state: AlgoState) -> bool:
        if state.competitor_count == 0:
            return False
        ratio = state.competitor_available / state.competitor_count
        return ratio < self.threshold_ratio and not state.available

    def execute(self, api, state: AlgoState) -> Dict:
        ratio = state.competitor_available / max(state.competitor_count, 1)
        try:
            api.set_availability(option=1, duration=5)
            state.available = True
            state.refreshes_executed += 1
            _record(state, self.name, "gap_fill", f"competitor_ratio={ratio:.2f}")
            return {"algo": self.name, "action": "gap_fill", "competitor_ratio": round(ratio, 2)}
        except Exception as e:
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 4: RefreshCascade — staggered refresh before expiry
# ═══════════════════════════════════════════════════════════════════

class RefreshCascade:
    """Refresh availability in a cascade: 1h before, 30min before, 5min before."""

    name = "refresh_cascade"
    CASCADE_POINTS = [3600, 1800, 300]  # 1h, 30min, 5min before expiry

    def should_fire(self, state: AlgoState) -> bool:
        if state.availability_expires <= 0:
            return False
        remaining = state.availability_expires - time.time()
        return remaining < self.CASCADE_POINTS[0] and remaining > 0

    def execute(self, api, state: AlgoState) -> Dict:
        remaining = state.availability_expires - time.time()
        cascade_level = 0
        for i, point in enumerate(self.CASCADE_POINTS):
            if remaining < point:
                cascade_level = i + 1
        try:
            api.set_availability(option=1, duration=5)
            state.available = True
            state.refreshes_executed += 1
            state.availability_expires = time.time() + 5 * 3600
            _record(state, self.name, "cascade", f"level={cascade_level} remaining={remaining:.0f}s")
            return {"algo": self.name, "action": "cascade", "level": cascade_level, "remaining_s": int(remaining)}
        except Exception as e:
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 5: SearchRankBoost — toggle to appear in "available now" more
# ═══════════════════════════════════════════════════════════════════

class SearchRankBoost:
    """Periodic toggle: 10s offline → back available → fresh in 'available now' sort."""

    name = "search_rank_boost"
    interval = 7200  # every 2 hours
    offline_duration = 10.0

    def __init__(self):
        self.last_toggle = 0

    def should_fire(self, state: AlgoState) -> bool:
        return state.available and (time.time() - self.last_toggle) > self.interval

    def execute(self, api, state: AlgoState) -> Dict:
        try:
            api.set_availability(option=2, duration=0)
            state.available = False
            time.sleep(self.offline_duration)
            api.set_availability(option=1, duration=5)
            state.available = True
            state.toggles_executed += 1
            self.last_toggle = time.time()
            _record(state, self.name, "rank_boost", f"offline {self.offline_duration}s")
            return {"algo": self.name, "action": "rank_boost", "offline_duration": self.offline_duration}
        except Exception as e:
            try:
                api.set_availability(option=1, duration=5)
                state.available = True
            except Exception:
                pass
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 6: DemandPulse — short availability windows in low-traffic periods
# ═══════════════════════════════════════════════════════════════════

class DemandPulse:
    """During low-traffic hours, pulse availability 15min on / 5min off."""

    name = "demand_pulse"
    LOW_HOURS = {2, 3, 4, 5, 6}
    PULSE_ON = 900   # 15 min available
    PULSE_OFF = 300  # 5 min unavailable

    def __init__(self):
        self.pulse_start = 0
        self.pulsing = False

    def should_fire(self, state: AlgoState) -> bool:
        if state.current_hour not in self.LOW_HOURS:
            self.pulsing = False
            return False
        if not self.pulsing:
            return not state.available
        elapsed = time.time() - self.pulse_start
        if state.available and elapsed > self.PULSE_ON:
            return True  # time to go offline briefly
        if not state.available and elapsed > self.PULSE_OFF:
            return True  # time to go back available
        return False

    def execute(self, api, state: AlgoState) -> Dict:
        hour = state.current_hour
        if not self.pulsing:
            self.pulsing = True
            self.pulse_start = time.time()
            try:
                api.set_availability(option=1, duration=1)
                state.available = True
                _record(state, self.name, "pulse_on", f"hour={hour}")
                return {"algo": self.name, "action": "pulse_on", "hour": hour}
            except Exception as e:
                return {"algo": self.name, "action": "error", "error": str(e)}
        elapsed = time.time() - self.pulse_start
        if state.available and elapsed > self.PULSE_ON:
            try:
                api.set_availability(option=2, duration=0)
                state.available = False
                self.pulse_start = time.time()
                _record(state, self.name, "pulse_off", f"hour={hour} after {self.PULSE_ON}s")
                return {"algo": self.name, "action": "pulse_off", "hour": hour}
            except Exception as e:
                return {"algo": self.name, "action": "error", "error": str(e)}
        if not state.available and elapsed > self.PULSE_OFF:
            try:
                api.set_availability(option=1, duration=1)
                state.available = True
                self.pulse_start = time.time()
                state.refreshes_executed += 1
                _record(state, self.name, "pulse_on", f"hour={hour} after {self.PULSE_OFF}s")
                return {"algo": self.name, "action": "pulse_on", "hour": hour}
            except Exception as e:
                return {"algo": self.name, "action": "error", "error": str(e)}
        return {"algo": self.name, "action": "noop"}


# ═══════════════════════════════════════════════════════════════════
# Algo 7: GeoRotation — rotate timing for multiple timezone waves
# ═══════════════════════════════════════════════════════════════════

class GeoRotation:
    """Stagger availability to catch ET, CT, PT traffic waves."""

    name = "geo_rotation"
    # ET peak: 8-11, 19-23 | CT peak: 9-12, 20-00 | PT peak: 11-14, 22-02
    # Combined: ensure availability 8-14 and 19-02
    GEO_PEAKS = [(8, 14), (19, 24), (0, 2)]

    def should_fire(self, state: AlgoState) -> bool:
        hour = state.current_hour
        in_peak = any(start <= hour < end for start, end in self.GEO_PEAKS)
        return in_peak and not state.available

    def execute(self, api, state: AlgoState) -> Dict:
        hour = state.current_hour
        # Calculate duration until peak window ends
        duration = 2
        for start, end in self.GEO_PEAKS:
            if start <= hour < end:
                duration = min(6, end - hour)
                break
        try:
            api.set_availability(option=1, duration=duration)
            state.available = True
            state.refreshes_executed += 1
            _record(state, self.name, "geo_sync", f"hour={hour} dur={duration}h")
            return {"algo": self.name, "action": "geo_sync", "hour": hour, "duration": duration}
        except Exception as e:
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 8: EngagementTrigger — refresh after visit/message
# ═══════════════════════════════════════════════════════════════════

class EngagementTrigger:
    """Immediately refresh availability after receiving a visit or message."""

    name = "engagement_trigger"
    COOLDOWN = 300  # 5 min between triggers

    def __init__(self):
        self.last_trigger = 0

    def should_fire(self, state: AlgoState) -> bool:
        now = time.time()
        if (now - self.last_trigger) < self.COOLDOWN:
            return False
        # Fire if recent visit or message within last 2 min
        recent_visit = (now - state.last_visit) < 120
        recent_msg = (now - state.last_message) < 120
        return recent_visit or recent_msg

    def execute(self, api, state: AlgoState) -> Dict:
        try:
            api.set_availability(option=1, duration=5)
            state.available = True
            state.refreshes_executed += 1
            self.last_trigger = time.time()
            trigger = "visit" if (time.time() - state.last_visit) < 120 else "message"
            _record(state, self.name, "engagement_refresh", f"trigger={trigger}")
            return {"algo": self.name, "action": "engagement_refresh", "trigger": trigger}
        except Exception as e:
            _record(state, self.name, "error", str(e))
            return {"algo": self.name, "action": "error", "error": str(e)}


# ═══════════════════════════════════════════════════════════════════
# Algo 9: BackoffRecovery — exponential backoff on API failure
# ═══════════════════════════════════════════════════════════════════

class BackoffRecovery:
    """Never lose presence: retry with exponential backoff on API failures."""

    name = "backoff_recovery"
    MAX_RETRIES = 5
    BASE_DELAY = 2.0
    MAX_DELAY = 120.0

    def should_fire(self, state: AlgoState) -> bool:
        return state.api_failures > 0 and not state.available

    def execute(self, api, state: AlgoState) -> Dict:
        for attempt in range(self.MAX_RETRIES):
            delay = min(self.MAX_DELAY, self.BASE_DELAY * (2 ** attempt))
            try:
                time.sleep(delay)
                api.set_availability(option=1, duration=5)
                state.available = True
                state.api_failures = 0
                _record(state, self.name, "recovered", f"attempt={attempt+1} delay={delay:.0f}s")
                return {"algo": self.name, "action": "recovered", "attempts": attempt + 1, "delay": delay}
            except Exception as e:
                state.api_failures += 1
                state.last_api_error = str(e)
                _record(state, self.name, "retry", f"attempt={attempt+1} delay={delay:.0f}s err={str(e)[:80]}")
        _record(state, self.name, "exhausted", f"failures={state.api_failures}")
        return {"algo": self.name, "action": "exhausted", "failures": state.api_failures}


# ═══════════════════════════════════════════════════════════════════
# Orchestrator — runs all 9 algos in priority order
# ═══════════════════════════════════════════════════════════════════

ALL_ALGOS = [
    BackoffRecovery(),    # 9: recover first if broken
    RefreshCascade(),     # 4: don't let availability lapse
    EngagementTrigger(),  # 8: capitalize on engagement
    CompetitorGap(),      # 3: fill competitor gaps
    PeakHourSync(),       # 2: sync to peaks
    GeoRotation(),       # 7: timezone coverage
    DemandPulse(),       # 6: low-traffic pulsing
    JitterBurst(),       # 1: traffic attraction bursts
    SearchRankBoost(),   # 5: search rank freshness
]

ALGO_NAMES = [a.name for a in ALL_ALGOS]


def run_algos(api, state: AlgoState) -> List[Dict]:
    """Run all algos that should fire. Returns list of results."""
    results = []
    for algo in ALL_ALGOS:
        try:
            if algo.should_fire(state):
                result = algo.execute(api, state)
                _capture_attribution(state, algo.name, result.get("action", "unknown"))
                results.append(result)
        except Exception as e:
            log.error(f"algo {algo.name} crashed: {e}")
            _capture_attribution(state, algo.name, "crash")
            results.append({"algo": algo.name, "action": "crash", "error": str(e)})
    return results


def algo_status(state: AlgoState) -> Dict[str, Any]:
    """Get status of all algos for API response."""
    return {
        "available": state.available,
        "bursts": state.bursts_executed,
        "refreshes": state.refreshes_executed,
        "toggles": state.toggles_executed,
        "api_failures": state.api_failures,
        "last_error": state.last_api_error,
        "algos": ALGO_NAMES,
        "recent": state.algo_history[-10:],
        "attribution": get_attribution_summary(state),
    }