Bot detection across 14 languages: language-invariant behavioral features and cross-platform sockpuppet fingerprinting

from dataclasses import dataclass
from typing import Optional

@dataclass
class BotFeatureVector:
    account_id: str
    language_code: str       # ISO 639-1; used only for threshold calibration

    # Feature 1: Posting interval entropy
    # Low entropy = machine-timed posts; high entropy = human irregular timing
    # Computed over the last 500 posts, binned into 1-minute intervals
    posting_interval_entropy: float      # bits; range 0-8.97 (log2(500))

    # Feature 2: Reply graph outdegree ratio
    # Bots initiate interactions to visible accounts; humans reply to conversations
    # outdegree = mentions_sent / (mentions_sent + replies_received)
    reply_outdegree_ratio: float         # 0.0-1.0

    # Feature 3: Content hash cluster density
    # Fraction of posts within Hamming distance 3 of another post by a different account
    # High density = copied/templated content across coordinated accounts
    content_cluster_density: float      # 0.0-1.0

    # Feature 4: Account age-activity velocity
    # (total_posts / account_age_days); normalized by platform median
    # Very high velocity in early account life = bot signup burst
    age_velocity_zscore: float           # z-score relative to platform median

    # Feature 5: Quote-to-original ratio
    # Amplifier bots have high quote/retweet ratio; content farms have low
    # Combined with feature 4 to distinguish amplifiers from content generators
    quote_to_original_ratio: float      # 0.0-inf; capped at 20.0

    # Feature 6: URL recycling rate
    # Fraction of posts containing a URL that was posted by another account
    # within the same 15-minute window; coordinated link drops show high rate
    url_recycling_rate: float           # 0.0-1.0

    # Feature 7: Cross-platform timing correlation
    # Pearson r between this account's posting times and the same content
    # appearing on a different platform (0.0 if no cross-platform match found)
    cross_platform_correlation: float   # -1.0 to 1.0; 0.0 = not found

    # Feature 8: Bio change cadence
    # Number of profile bio changes in the last 90 days
    # Coordinated networks often update bios in synchronized bursts
    bio_change_count_90d: int           # raw count; not normalized

import numpy as np
from typing import Sequence

def compute_posting_interval_entropy(
    timestamps: Sequence[float],
    n_bins: int = 60,
) -> float:
    """
    Compute Shannon entropy of the posting interval distribution.

    timestamps: Unix timestamps of posts, sorted ascending.
    n_bins: number of histogram bins (default: 60 = 1-minute bins over 1 hour).

    Returns entropy in bits. Range: 0 (all posts at identical interval)
    to log2(n_bins) (uniform distribution across all bins).
    """
    if len(timestamps) < 10:
        return float('nan')

    intervals = np.diff(timestamps)           # seconds between consecutive posts
    # Clip to [0, 3600] to exclude account dormancy periods > 1 hour
    intervals = intervals[intervals <= 3600]
    if len(intervals) < 5:
        return float('nan')

    counts, _ = np.histogram(intervals, bins=n_bins, range=(0, 3600))

    # Add-1 smoothing to avoid log(0)
    counts = counts + 1.0
    probs = counts / counts.sum()
    entropy = -np.sum(probs * np.log2(probs))

    # Subtract baseline entropy from the smoothing term
    baseline = -np.log2(1.0 / n_bins)  # uniform distribution entropy
    return float(min(entropy, baseline))

# Empirical thresholds (calibrated on 120K labeled accounts):
# entropy < 2.1 bits: HIGH bot probability (automated scheduler)
# entropy 2.1-4.8 bits: AMBIGUOUS (jitter-based bot or low-activity human)
# entropy > 4.8 bits: LOW bot probability (human irregular pattern)

import imagehash
from PIL import Image
import io
import redis

REDIS_PREFIX = 'phash:'
HAMMING_THRESHOLD = 8

def compute_profile_phash(image_bytes: bytes) -> str:
    """Compute 64-bit perceptual hash of a profile photo."""
    img = Image.open(io.BytesIO(image_bytes)).convert('RGB')
    # Resize to 32x32 for consistent hash regardless of source resolution
    img = img.resize((32, 32), Image.LANCZOS)
    return str(imagehash.phash(img, hash_size=8))  # 64-bit pHash

def store_phash(r: redis.Redis, account_id: str, phash_str: str) -> None:
    """Store pHash indexed by its first 16 bits (bucket key)."""
    bucket = phash_str[:4]  # first 4 hex chars = 16 bits
    r.hset(f'{REDIS_PREFIX}{bucket}', account_id, phash_str)

def find_phash_matches(
    r: redis.Redis,
    query_phash: str,
    source_account_id: str,
) -> list[tuple[str, int]]:
    """
    Find accounts with similar profile photos across all platforms.
    Returns [(account_id, hamming_distance)] sorted ascending.
    """
    query_hash = imagehash.hex_to_hash(query_phash)
    matches = []

    # Check the query bucket and adjacent buckets (covers 1-bit bucket boundary errors)
    query_bucket_int = int(query_phash[:4], 16)
    buckets_to_check = set()
    for delta in range(-2, 3):
        bucket = hex(max(0, min(0xFFFF, query_bucket_int + delta)))[2:].zfill(4)
        buckets_to_check.add(bucket)

    for bucket in buckets_to_check:
        bucket_entries = r.hgetall(f'{REDIS_PREFIX}{bucket}')
        for acct_id, stored_phash in bucket_entries.items():
            if acct_id == source_account_id:
                continue
            stored_hash = imagehash.hex_to_hash(stored_phash.decode())
            distance = query_hash - stored_hash
            if distance <= HAMMING_THRESHOLD:
                matches.append((acct_id.decode(), distance))

    return sorted(matches, key=lambda x: x[1])

from xgboost import XGBClassifier
from sklearn.model_selection import StratifiedGroupKFold
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import label_binarize
import numpy as np
import pandas as pd

XGB_PARAMS = {
    'n_estimators': 600,
    'max_depth': 5,
    'learning_rate': 0.04,
    'subsample': 0.8,
    'colsample_bytree': 0.75,
    'min_child_weight': 3,
    'scale_pos_weight': 4.2,  # bot:human ratio ~1:4.2 in training set
    'eval_metric': 'aucpr',
    'early_stopping_rounds': 40,
    'random_state': 42,
}

FEATURE_COLS = [
    'posting_interval_entropy',
    'reply_outdegree_ratio',
    'content_cluster_density',
    'age_velocity_zscore',
    'quote_to_original_ratio',
    'url_recycling_rate',
    'cross_platform_correlation',
    'bio_change_count_90d',
]

def train_multilingual_bot_classifier(
    df: pd.DataFrame,  # must have FEATURE_COLS + 'is_bot' + 'language_code'
) -> tuple[XGBClassifier, dict[str, LogisticRegression]]:
    """
    Train XGBoost on pooled corpus; calibrate per language.
    Returns (base_classifier, per_language_calibrators).
    """
    X = df[FEATURE_COLS].values
    y = df['is_bot'].values.astype(int)
    groups = df['language_code'].values

    cv = StratifiedGroupKFold(n_splits=5)
    oof_scores = np.zeros(len(df))

    clf = XGBClassifier(**XGB_PARAMS)

    for fold, (train_idx, val_idx) in enumerate(cv.split(X, y, groups)):
        X_tr, X_val = X[train_idx], X[val_idx]
        y_tr, y_val = y[train_idx], y[val_idx]
        clf.fit(X_tr, y_tr, eval_set=[(X_val, y_val)], verbose=False)
        oof_scores[val_idx] = clf.predict_proba(X_val)[:, 1]

    # Fit final model on full training set
    clf.fit(X, y, verbose=False)

    # Per-language Platt scaling calibration on OOF predictions
    per_lang_calibrators: dict[str, LogisticRegression] = {}
    for lang in np.unique(groups):
        mask = groups == lang
        if mask.sum() < 50:
            continue
        lr = LogisticRegression(C=1.0, solver='lbfgs')
        lr.fit(oof_scores[mask].reshape(-1, 1), y[mask])
        per_lang_calibrators[lang] = lr

    return clf, per_lang_calibrators