"""Pry — Competitive Intelligence Engine. Historical snapshots, anomaly detection, natural-language alerts, weekly reports.""" # SPDX-License-Identifier: MIT # Copyright (c) 2026 Rug Munch Media LLC # # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper # Licensed under MIT. See LICENSE. import logging import statistics import time from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import desc from sqlalchemy.exc import SQLAlchemyError from db import IntelSnapshot, session_scope logger = logging.getLogger(__name__) def record_snapshot( competitor_id: str, competitor_name: str, url: str, fields: dict[str, Any], ) -> dict[str, Any]: """Record a data snapshot for a competitor.""" try: with session_scope() as session: snap = IntelSnapshot( competitor_id=competitor_id, competitor_name=competitor_name, url=url, fields=fields, ) session.add(snap) session.flush() ts = snap.ts result = { "ts": ts.isoformat(), "unix_ts": ts.timestamp(), "competitor_id": competitor_id, "competitor_name": competitor_name, "url": url, "fields": fields, } logger.info("snapshot_recorded", extra={"competitor": competitor_name}) return result except SQLAlchemyError as e: logger.error( "snapshot_record_failed", extra={"competitor": competitor_name, "error": str(e)} ) return {"error": str(e)} def get_snapshots( competitor_id: str, limit: int = 50, since_hours: int | None = None, ) -> list[dict[str, Any]]: """Get snapshots for a competitor, most recent first.""" try: with session_scope() as session: query = session.query(IntelSnapshot).filter( IntelSnapshot.competitor_id == competitor_id, ) if since_hours is not None: cutoff = datetime.now(UTC) - timedelta(hours=since_hours) query = query.filter(IntelSnapshot.ts >= cutoff) query = query.order_by(desc(IntelSnapshot.ts)).limit(limit) rows = query.all() return [ { "ts": row.ts.isoformat() if row.ts else "", "unix_ts": row.ts.timestamp() if row.ts else 0, "competitor_id": row.competitor_id, "competitor_name": row.competitor_name, "url": row.url, "fields": row.fields or {}, } for row in rows ] except SQLAlchemyError as e: logger.error("snapshot_query_failed", extra={"competitor": competitor_id, "error": str(e)}) return [] def compute_field_statistics( snapshots: list[dict[str, Any]], field: str, ) -> dict[str, Any]: """Compute statistics for a field across snapshots.""" values = [s.get("fields", {}).get(field) for s in snapshots] values = [v for v in values if v is not None] if not values or len(values) < 2: return {"count": len(values), "has_history": False} numeric_values = [v for v in values if isinstance(v, (int, float))] string_values = [str(v) for v in values if isinstance(v, str)] result: dict[str, Any] = { "count": len(values), "has_history": True, "field": field, } if numeric_values: result["mean"] = round(statistics.mean(numeric_values), 2) result["median"] = round(statistics.median(numeric_values), 2) result["min"] = min(numeric_values) result["max"] = max(numeric_values) if len(numeric_values) > 2: result["stdev"] = round(statistics.stdev(numeric_values), 2) result["latest"] = numeric_values[-1] result["previous"] = numeric_values[-2] if len(numeric_values) >= 2 else None if string_values: result["unique_values"] = len(set(string_values)) result["latest"] = string_values[-1] result["previous"] = string_values[-2] if len(string_values) >= 2 else None return result def detect_anomalies_numeric( current_value: float, history: list[float], z_score_threshold: float = 2.0, ) -> dict[str, Any]: """Detect anomalies in a numeric field using z-score.""" if len(history) < 3: return {"anomaly": False, "reason": "Insufficient history"} mean = statistics.mean(history) stdev = statistics.stdev(history) if len(history) > 1 else 1.0 if stdev == 0: return {"anomaly": current_value != mean, "reason": "Value changed from constant history"} z_score = abs((current_value - mean) / stdev) pct_change = ((current_value - mean) / mean) * 100 if mean != 0 else 0 return { "anomaly": z_score >= z_score_threshold, "z_score": round(z_score, 2), "pct_change": round(pct_change, 1), "mean": round(mean, 2), "stdev": round(stdev, 2), "severity": "high" if z_score >= 3.0 else "medium" if z_score >= 2.0 else "low", } def generate_alert( competitor_name: str, field: str, old_value: Any, new_value: Any, anomaly_info: dict[str, Any] | None = None, ) -> str: """Generate a natural-language alert for a detected change.""" intro = f"*{competitor_name}*" if isinstance(new_value, (int, float)) and isinstance(old_value, (int, float)): pct = ((new_value - old_value) / old_value) * 100 if old_value != 0 else 0 direction = "increased" if pct > 0 else "decreased" change_part = f"{direction} {field} from {old_value} to {new_value} ({abs(pct):.1f}%)" elif isinstance(new_value, str) and isinstance(old_value, str): if len(new_value) > 50 or len(old_value) > 50: change_part = ( f"changed {field} (length: {len(old_value)} \u2192 {len(new_value)} chars)" ) else: change_part = f'changed {field}: "{old_value}" \u2192 "{new_value}"' else: change_part = f"updated {field}" severity = "" if anomaly_info and anomaly_info.get("anomaly"): severity = " \u26a0\ufe0f *ANOMALY DETECTED*" alert = f"{intro} {change_part}{severity}" if anomaly_info and anomaly_info.get("z_score"): alert += f" (z-score: {anomaly_info['z_score']})" if anomaly_info and anomaly_info.get("pct_change"): alert += f" \u2014 unusual change of {anomaly_info['pct_change']}% vs historical average" return alert def generate_weekly_report( competitors: list[dict[str, Any]], days_back: int = 7, ) -> dict[str, Any]: """Generate a weekly competitive intelligence report.""" cutoff_ts = time.time() - (days_back * 86400) report_sections: list[dict[str, Any]] = [] for comp in competitors: comp_id = comp.get("id", comp.get("name", "").lower().replace(" ", "_")) comp_name = comp.get("name", "Unknown") snapshots = get_snapshots(comp_id) weekly = [s for s in snapshots if s.get("unix_ts", 0) >= cutoff_ts] if not weekly: continue if len(weekly) >= 2: latest = weekly[0].get("fields", {}) oldest = weekly[-1].get("fields", {}) changes = [] all_fields = set(latest.keys()) | set(oldest.keys()) for field in all_fields: old_val = oldest.get(field) new_val = latest.get(field) if old_val != new_val: changes.append( { "field": field, "from": old_val, "to": new_val, "alert": generate_alert(comp_name, field, old_val, new_val), } ) if changes: report_sections.append( { "competitor": comp_name, "changes_count": len(changes), "snapshots_this_week": len(weekly), "changes": changes, } ) total_changes = sum(s["changes_count"] for s in report_sections) most_active = ( max(report_sections, key=lambda x: x["changes_count"]) if report_sections else None ) return { "report_period": f"Last {days_back} days", "generated_at": datetime.now(UTC).isoformat(), "competitors_tracked": len(competitors), "competitors_with_changes": len(report_sections), "total_changes": total_changes, "most_active_competitor": most_active["competitor"] if most_active else None, "sections": report_sections, "summary": _generate_report_summary(report_sections, total_changes, len(competitors)), } def _generate_report_summary( sections: list[dict[str, Any]], total_changes: int, total_competitors: int, ) -> str: """Generate a text summary of the weekly report.""" if not sections: return f"No significant changes detected across {total_competitors} tracked competitors." most_active = max(sections, key=lambda x: x["changes_count"]) lines = [ "Weekly Competitive Intelligence Summary", "", f"Tracked {total_competitors} competitors over the past 7 days.", f"Detected {total_changes} changes across {len(sections)} competitors.", "", f"Most active: {most_active['competitor']} with {most_active['changes_count']} changes.", ] for section in sections[:5]: lines.append("") lines.append(f"\u2500\u2500 {section['competitor']} \u2500\u2500") for change in section["changes"][:3]: lines.append(f" \u2022 {change['alert']}") if len(section["changes"]) > 3: lines.append(f" ... and {len(section['changes']) - 3} more changes") return "\n".join(lines)