Detecting Coordinated Inauthentic Behavior in Social Media at Scale

import numpy as np
import re
from typing import List

_NORMALIZE = re.compile(r"https?://\S+|@\w+|#")
_RNG_SEED = 42

# Pre-generate hash function parameters once at module load
_np_rng = np.random.default_rng(_RNG_SEED)
_HASH_A = _np_rng.integers(1, (1 << 31) - 1, size=128)
_HASH_B = _np_rng.integers(0, (1 << 31) - 1, size=128)
_MERSENNE_PRIME = (1 << 31) - 1

def compute_minhash(text: str, num_hashes: int = 128) -> np.ndarray:
    """
    Compute MinHash signature over character 4-grams.
    Returns array of shape (num_hashes,) — the MinHash signature.
    """
    clean = _NORMALIZE.sub("", text.lower()).strip()
    # Extract character 4-grams
    shingles = set()
    for i in range(max(0, len(clean) - 3)):
        shingles.add(hash(clean[i:i+4]) & 0x7FFFFFFF)

    if not shingles:
        return np.full(num_hashes, _MERSENNE_PRIME, dtype=np.int64)

    shingle_arr = np.array(list(shingles), dtype=np.int64)

    # Universal hash family: h(x) = (a*x + b) mod p
    # Shape: (num_hashes, num_shingles)
    hashed = (
        (_HASH_A[:, None] * shingle_arr[None, :] + _HASH_B[:, None])
        % _MERSENNE_PRIME
    )
    # MinHash: take the minimum hash value per hash function
    return hashed.min(axis=1)   # shape (num_hashes,)


# LSH index: maps band_key -> list of post_ids
# 16 bands × 8 rows; band key = (band_index, tuple of 8 signature values)
_lsh_index: dict[tuple, list[str]] = {}

def lsh_query(signature: np.ndarray, bands: int = 16) -> List[str]:
    """
    Query the LSH index for candidate near-duplicate posts.
    Returns list of candidate post_ids sharing at least one band bucket.
    """
    rows = len(signature) // bands   # 8 rows per band
    candidates: set[str] = set()

    for b in range(bands):
        band_sig = tuple(signature[b * rows:(b + 1) * rows].tolist())
        band_key = (b, band_sig)
        if band_key in _lsh_index:
            candidates.update(_lsh_index[band_key])

    return list(candidates)

def lsh_insert(post_id: str, signature: np.ndarray, bands: int = 16) -> None:
    """Insert a post's signature into all band buckets."""
    rows = len(signature) // bands
    for b in range(bands):
        band_sig = tuple(signature[b * rows:(b + 1) * rows].tolist())
        band_key = (b, band_sig)
        _lsh_index.setdefault(band_key, []).append(post_id)

def jaccard_from_signatures(sig_a: np.ndarray, sig_b: np.ndarray) -> float:
    """Estimate Jaccard similarity from MinHash signatures."""
    return float((sig_a == sig_b).mean())

import redis
import math
import time
from typing import Optional

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def record_cluster_post(
    cluster_id: str,
    account_id: str,
    timestamp: Optional[float] = None,
) -> None:
    """Record that account_id posted content in cluster_id at timestamp."""
    ts = timestamp or time.time()
    # ZADD cluster:{cluster_id} {ts} {account_id}
    # NX: only add, don't update existing member — prevents one account from
    # inflating burst counts by posting repeatedly
    r.zadd(f"cluster:{cluster_id}", {account_id: ts}, nx=True)
    # Expire the key after 2 hours to avoid unbounded growth
    r.expire(f"cluster:{cluster_id}", 7200)

def get_burst_accounts(
    cluster_id: str,
    window_seconds: int = 900,   # 15 minutes
    timestamp: Optional[float] = None,
) -> list[str]:
    """Return distinct account_ids that posted in the cluster within the window."""
    t = timestamp or time.time()
    # ZRANGEBYSCORE cluster:{cluster_id} {t-900} {t}
    return r.zrangebyscore(f"cluster:{cluster_id}", t - window_seconds, t)

def compute_burst_score(
    cluster_id: str,
    account_ages_days: dict[str, float],
    window_seconds: int = 900,
    min_accounts: int = 5,
) -> float:
    """
    Burst score = distinct_account_count × (1 / sqrt(avg_account_age_days + 1))
    Newer accounts contribute more to the burst score.
    Returns 0.0 if fewer than min_accounts in window.
    """
    accounts = get_burst_accounts(cluster_id, window_seconds)
    if len(accounts) < min_accounts:
        return 0.0

    ages = [account_ages_days.get(acct, 0.0) for acct in accounts]
    avg_age = sum(ages) / len(ages)
    # Inverse-sqrt weighting: account_age_days=0 → weight=1.0,
    # age=3 → 0.5, age=99 → 0.1
    age_weight = 1.0 / math.sqrt(avg_age + 1)

    return len(accounts) * age_weight

from dataclasses import dataclass
import numpy as np
import pandas as pd
from scipy.stats import entropy

@dataclass
class AccountFeatures:
    account_age_days: float
    """Days since account creation. Bots often created in recent batches."""

    posts_per_day: float
    """Average daily posting rate. Expected Pareto-distributed: mean ~3.2, median 0.8.
    Values > 50/day are unusual for organic human accounts."""

    reply_ratio: float
    """Fraction of posts that are replies (0.0–1.0).
    Bots typically broadcast; humans converse. Organic mean ~0.42."""

    unique_domains_per_day: float
    """Distinct link domains shared per day. Bot accounts post narrow link sets;
    organic accounts share from varied sources."""

    follower_following_ratio: float
    """follower_count / max(following_count, 1).
    Accounts with 0 followers but many followings are common bot patterns."""

    profile_completeness: float
    """0.0–1.0 composite: profile photo (0.4) + bio text (0.4) + custom header (0.2).
    Most real accounts have at least a profile photo."""

    posting_hour_entropy: float
    """Shannon entropy of the 24-bin hourly posting distribution (nats, max ln(24)≈3.18).
    Organic humans: varies (high entropy). Bots: uniform or scripted peak (low entropy)."""


def compute_account_features(history: pd.DataFrame) -> AccountFeatures:
    """
    Compute coordination features from a DataFrame of the account's post history.

    Required columns:
        created_at      datetime: timestamp of each post
        is_reply        bool: True if the post is a reply
        url_domain      str | None: extracted domain from any link in the post

    Additional context (passed as scalar, not in DataFrame):
        account_created_at  datetime
        follower_count      int
        following_count     int
        profile_photo       bool
        bio_text            str | None
        custom_header       bool
    """
    # Degenerate case — brand new account with no posts
    if len(history) == 0:
        return AccountFeatures(
            account_age_days=0.0,
            posts_per_day=0.0,
            reply_ratio=0.0,
            unique_domains_per_day=0.0,
            follower_following_ratio=0.0,
            profile_completeness=0.0,
            posting_hour_entropy=0.0,
        )

    history = history.copy()
    history["created_at"] = pd.to_datetime(history["created_at"], utc=True)

    span_days = max(
        (history["created_at"].max() - history["created_at"].min()).total_seconds() / 86400,
        1.0,   # floor at 1 day to avoid division by zero
    )

    posts_per_day = len(history) / span_days
    reply_ratio = float(history["is_reply"].mean())

    domain_links = history["url_domain"].dropna()
    unique_domains_per_day = domain_links.nunique() / span_days

    # Hourly posting distribution — 24 bins
    hours = history["created_at"].dt.hour
    hour_counts = np.zeros(24)
    for h, cnt in hours.value_counts().items():
        hour_counts[int(h)] = cnt
    hour_probs = hour_counts / max(hour_counts.sum(), 1)
    posting_hour_entropy = float(entropy(hour_probs + 1e-9))   # nats

    return AccountFeatures(
        account_age_days=span_days,
        posts_per_day=posts_per_day,
        reply_ratio=reply_ratio,
        unique_domains_per_day=unique_domains_per_day,
        follower_following_ratio=0.0,       # populated from profile metadata
        profile_completeness=0.0,           # populated from profile metadata
        posting_hour_entropy=posting_hour_entropy,
    )

import networkx as nx
from typing import List

def build_amplification_graph(retweets: list[dict]) -> nx.DiGraph:
    """
    Build a directed graph from retweet events.
    Each edge (A → B) means: A retweeted B's content.
    Edge weight = number of times A retweeted B.
    """
    G = nx.DiGraph()
    for rt in retweets:
        src, dst = rt["retweeter_id"], rt["original_author_id"]
        if G.has_edge(src, dst):
            G[src][dst]["weight"] += 1
        else:
            G.add_edge(src, dst, weight=1)
    return G

def find_amplification_rings(
    G: nx.DiGraph,
    min_ring_size: int = 3,
    max_ring_size: int = 5,
    min_edge_weight: int = 2,
) -> List[List[str]]:
    """
    Find mutual amplification rings using Johnson's algorithm for cycle detection.
    Filters to cycles of length min_ring_size–max_ring_size where all edges
    meet the minimum mutual-retweet threshold.

    Johnson's algorithm finds all simple cycles in O((V+E)(C+1)) time,
    where C is the number of cycles. On sparse retweet graphs (most accounts
    have few retweet relationships) this is fast enough for hourly batch runs.

    Returns list of cycles, each cycle a list of account_ids in order.
    """
    # Prune weak edges before cycle detection to reduce search space
    pruned = nx.DiGraph()
    for u, v, data in G.edges(data=True):
        if data.get("weight", 1) >= min_edge_weight:
            pruned.add_edge(u, v, **data)

    rings = []
    for cycle in nx.simple_cycles(pruned):
        cycle_len = len(cycle)
        if min_ring_size <= cycle_len <= max_ring_size:
            # Verify the cycle is mutual: every edge in the cycle has a
            # reciprocal edge (i.e., A→B and B→A both exist)
            is_mutual = all(
                pruned.has_edge(cycle[(i + 1) % cycle_len], cycle[i])
                for i in range(cycle_len)
            )
            if is_mutual:
                rings.append(cycle)

    return rings

-- Cross-platform amplification detection query
-- Run hourly against the canonical post table
-- Returns content_hash values with cross-platform coordination signals

WITH recent_posts AS (
    SELECT
        content_hash,
        platform,
        account_id,
        created_at,
        COUNT(DISTINCT platform) OVER (PARTITION BY content_hash) AS platform_count
    FROM canonical_posts
    WHERE created_at >= NOW() - INTERVAL '2 hours'
      AND content_hash IS NOT NULL
),
multi_platform_hashes AS (
    -- Step 1: content appearing on 3+ platforms in the window
    SELECT content_hash
    FROM recent_posts
    GROUP BY content_hash
    HAVING COUNT(DISTINCT platform) >= 3
),
poster_pairs AS (
    -- Step 2: distinct (account_id, platform) pairs per content_hash
    SELECT
        rp.content_hash,
        rp.account_id,
        rp.platform,
        rp.created_at
    FROM recent_posts rp
    INNER JOIN multi_platform_hashes mph ON rp.content_hash = mph.content_hash
),
cross_platform_overlap AS (
    -- Step 3: check whether any poster pair has prior cross-platform relationship
    -- account_cross_links maps (account_id, platform) -> known alt accounts
    SELECT pp.content_hash, COUNT(*) AS linked_accounts
    FROM poster_pairs pp
    INNER JOIN account_cross_links acl
        ON acl.account_id = pp.account_id
        AND acl.linked_platform != pp.platform
    GROUP BY pp.content_hash
)
SELECT
    pp.content_hash,
    ARRAY_AGG(DISTINCT pp.platform ORDER BY pp.platform) AS platforms,
    COUNT(DISTINCT pp.account_id) AS distinct_accounts,
    MIN(pp.created_at) AS first_seen,
    MAX(pp.created_at) AS last_seen,
    EXTRACT(EPOCH FROM (MAX(pp.created_at) - MIN(pp.created_at))) AS window_seconds,
    COALESCE(cpo.linked_accounts, 0) AS known_cross_platform_accounts
FROM poster_pairs pp
LEFT JOIN cross_platform_overlap cpo ON cpo.content_hash = pp.content_hash
GROUP BY pp.content_hash, cpo.linked_accounts
HAVING COALESCE(cpo.linked_accounts, 0) = 0   -- No prior cross-platform relationships
ORDER BY distinct_accounts DESC;

from dataclasses import dataclass
import math

@dataclass
class CoordinationSignals:
    # Content similarity (0–30 points)
    # Driven by cluster size and average within-cluster Jaccard similarity
    cluster_size: int
    avg_jaccard: float               # average pairwise similarity within cluster

    # Temporal burst (0–25 points)
    burst_score: float               # from compute_burst_score() above
    max_burst_score: float = 50.0    # calibration constant for normalization

    # Account features (0–25 points)
    account_feature_logit: float     # raw logit from logistic regression classifier
                                     # trained on 40K labeled accounts

    # Network graph (0–20 points)
    ring_count: int                  # number of amplification rings detected
    synchronized_follow_events: int  # count of 50+accounts/10min follow bursts


def compute_coordination_score(signals: CoordinationSignals) -> int:
    """
    Assemble per-signal scores into a 0–100 coordination score.
    Score > 70  → analyst alert queue
    Score > 90  → automatic flag for election anomaly detection pipeline
    """

    # --- Content similarity score (0–30) ---
    # Larger clusters with higher average similarity score higher
    # log(cluster_size) caps contribution of very large organic clusters
    size_component = min(math.log1p(signals.cluster_size) / math.log1p(100), 1.0)
    similarity_component = max(0.0, (signals.avg_jaccard - 0.45) / (1.0 - 0.45))
    content_score = 30 * (0.5 * size_component + 0.5 * similarity_component)

    # --- Temporal burst score (0–25) ---
    # Normalize burst_score against calibration constant
    burst_normalized = min(signals.burst_score / signals.max_burst_score, 1.0)
    temporal_score = 25 * burst_normalized

    # --- Account feature score (0–25) ---
    # Logistic regression output: sigmoid(logit) gives probability in [0,1]
    account_prob = 1.0 / (1.0 + math.exp(-signals.account_feature_logit))
    account_score = 25 * account_prob

    # --- Network graph score (0–20) ---
    # Rings are a strong signal: each ring contributes up to 8 points (cap at 3)
    ring_contribution = min(signals.ring_count, 3) / 3 * 12
    # Synchronized follow events contribute up to 8 points
    follow_contribution = min(signals.synchronized_follow_events, 2) / 2 * 8
    graph_score = ring_contribution + follow_contribution

    total = content_score + temporal_score + account_score + graph_score
    return min(100, round(total))


def classify_coordination_score(score: int) -> str:
    if score >= 90:
        return "AUTO_FLAG"     # automatic flag for election anomaly pipeline
    elif score >= 70:
        return "ANALYST_ALERT" # enters human review queue
    else:
        return "MONITOR"       # logged, no immediate action

Campaign                     TP Rate    FP Rate    FP Rate    Detection   Detection
                             (confirmed  (@70 thr.) (@90 thr.) latency     latency
                             posts)                             (70+ med.)  (90+ med.)
──────────────────────────────────────────────────────────────────────────────────────────
Campaign A (domestic amp.)   97%         8%         3%         9 min       41 min
Campaign B (cross-platform)  89%         11%        4%         16 min      58 min
Campaign C (ballot measure)  93%         6%         2%         11 min      43 min
──────────────────────────────────────────────────────────────────────────────────────────
Overall (weighted by size)   93%         8%         3%         12 min      47 min