sourcecode/scoring/post_selection_similarity_old.py

import gc
import logging
from typing import Dict

from . import constants as c

import numpy as np
import pandas as pd


logger = logging.getLogger("birdwatch.post_selection_similarity")
logger.setLevel(logging.INFO)


class PostSelectionSimilarity:
  def __init__(self):
    pass

  def initialize(
    self,
    notes: pd.DataFrame,
    ratings: pd.DataFrame,
    pmiRegularization: int = 500,
    smoothedNpmiThreshold: float = 0.45,
    minimumRatingProportionThreshold: float = 0.4,
    minUniquePosts: int = 10,
    minSimPseudocounts: int = 10,
    windowMillis: int = 1000 * 60 * 20,
  ):
    self.ratings = _preprocess_ratings(notes, ratings)
    self.pairCounts = _get_pair_tuples(self.ratings, windowMillis=windowMillis)
    self.pairStatsDf = _tuples_to_df(self.pairCounts)
    self.uniqueRatingsOnTweets = self.ratings[
      [c.tweetIdKey, c.raterParticipantIdKey]
    ].drop_duplicates()
    self.pairStatsDf = _join_rater_totals(self.pairStatsDf, self.uniqueRatingsOnTweets)
    self.pmiDf = _compute_pmi(
      self.pairStatsDf, len(self.uniqueRatingsOnTweets), pmiRegularization, minSimPseudocounts
    )

    self.filter_edges_below_threshold(
      smoothedNpmiThreshold, minimumRatingProportionThreshold, minUniquePosts
    )

  def filter_edges_below_threshold(
    self, smoothedNpmiThreshold, minimumRatingProportionThreshold, minUniquePosts
  ):
    self.graphDf = self.pmiDf[
      (self.pmiDf["smoothedNpmi"] >= smoothedNpmiThreshold)
      | (
        (self.pmiDf["minSimRatingProp"] >= minimumRatingProportionThreshold)
        & (self.pmiDf["minTotal"] >= minUniquePosts)
      )
    ]

  def get_high_post_selection_similarity_raters(self):
    highPostSelectionSimilarityRaters = pd.concat(
      [
        self.graphDf[["leftRaterId"]].rename(columns={"leftRaterId": c.raterParticipantIdKey}),
        self.graphDf[["rightRaterId"]].rename(columns={"rightRaterId": c.raterParticipantIdKey}),
      ]
    ).drop_duplicates()
    highPostSelectionSimilarityRaters[c.postSelectionValueKey] = 1
    return highPostSelectionSimilarityRaters

  def get_post_selection_similarity_values(self):
    """
    Returns dataframe with [raterParticipantId, postSelectionSimilarityValue] columns.
    postSelectionSimilarityValue is None by default.
    """
    cliqueToUserMap, userToCliqueMap = aggregate_into_cliques(self.graphDf)

    # Convert dict to pandas dataframe
    cliquesDfList = []
    for cliqueId in cliqueToUserMap.keys():
      for userId in cliqueToUserMap[cliqueId]:
        cliquesDfList.append({c.raterParticipantIdKey: userId, c.postSelectionValueKey: cliqueId})
    cliquesDf = pd.DataFrame(
      cliquesDfList, columns=[c.raterParticipantIdKey, c.postSelectionValueKey]
    )
    return cliquesDf


def filter_ratings_by_post_selection_similarity(notes, ratings, postSelectionSimilarityValues):
  """
  Filters out ratings after the first on each note from raters who have high post selection similarity,
  or filters all if the note is authored by a user with the same post selection similarity value.
  """
  ratingsWithPostSelectionSimilarity = (
    ratings.merge(
      postSelectionSimilarityValues,
      on=c.raterParticipantIdKey,
      how="left",
      unsafeAllowed=c.postSelectionValueKey,
    )
    .merge(notes[[c.noteIdKey, c.noteAuthorParticipantIdKey]], on=c.noteIdKey, how="left")
    .merge(
      postSelectionSimilarityValues,
      left_on=c.noteAuthorParticipantIdKey,
      right_on=c.raterParticipantIdKey,
      how="left",
      suffixes=("", "_note_author"),
      unsafeAllowed={c.postSelectionValueKey, c.postSelectionValueKey + "_note_author"},
    )
  )
  ratingsWithNoPostSelectionSimilarityValue = ratingsWithPostSelectionSimilarity[
    pd.isna(ratingsWithPostSelectionSimilarity[c.postSelectionValueKey])
  ]
  ratingsWithPostSelectionSimilarityValue = ratingsWithPostSelectionSimilarity[
    (~pd.isna(ratingsWithPostSelectionSimilarity[c.postSelectionValueKey]))
    & (
      ratingsWithPostSelectionSimilarity[c.postSelectionValueKey]
      != ratingsWithPostSelectionSimilarity[c.postSelectionValueKey + "_note_author"]
    )
  ]
  ratingsWithPostSelectionSimilarityValue.sort_values(
    by=[c.noteIdKey, c.createdAtMillisKey], ascending=True, inplace=True
  )
  ratingsWithPostSelectionSimilarityValue.drop_duplicates(
    subset=[c.noteIdKey, c.postSelectionValueKey], keep="first", inplace=True
  )

  ratings = pd.concat(
    [ratingsWithPostSelectionSimilarityValue, ratingsWithNoPostSelectionSimilarityValue], axis=0
  )
  ratings.drop(
    columns={c.noteAuthorParticipantIdKey, c.raterParticipantIdKey + "_note_author"},
    errors="ignore",
    inplace=True,
  )
  return ratings


def filter_all_ratings_by_post_selection_similarity(ratings, highPostSelectionSimilarityRaters):
  """
  Deprecated.
  Filters out all ratings from raters who have high post selection similarity.
  """
  ratings = ratings.merge(
    highPostSelectionSimilarityRaters, on=c.raterParticipantIdKey, how="left", indicator=True
  )
  ratings = ratings[ratings["_merge"] == "left_only"]
  ratings = ratings.drop(columns=["_merge"])
  return ratings


def _compute_pmi(
  pairStatsDf: pd.DataFrame, N: int, pmiPseudocounts: int = 500, minSimPseudocounts: int = 10
) -> pd.DataFrame:
  """
  Compute PMI between raters.
  """
  numerator = pairStatsDf["pairRatings"] * N
  denominator = (pairStatsDf["leftTotal"] + pmiPseudocounts) * (
    pairStatsDf["rightTotal"] + pmiPseudocounts
  )
  pairStatsDf["smoothedPmi"] = np.log(numerator / denominator)
  pairStatsDf["smoothedNpmi"] = pairStatsDf["smoothedPmi"] / -np.log(pairStatsDf["pairRatings"] / N)
  pairStatsDf["minTotal"] = np.minimum(pairStatsDf["leftTotal"], pairStatsDf["rightTotal"])
  pairStatsDf["minSimRatingProp"] = pairStatsDf["pairRatings"] / (
    pairStatsDf["minTotal"] + minSimPseudocounts
  )
  return pairStatsDf


def _preprocess_ratings(notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
  """
  Preprocess ratings dataframe.
  """
  ratings = notes[[c.noteIdKey, c.tweetIdKey]].merge(
    ratings[[c.raterParticipantIdKey, c.noteIdKey, c.createdAtMillisKey]],
    on=c.noteIdKey,
    how="inner",
  )
  ratings = ratings[(ratings[c.tweetIdKey] != -1) & (ratings[c.tweetIdKey] != "-1")]
  return ratings


def _join_rater_totals(
  pairStatsDf: pd.DataFrame, uniqueRatingsOnTweets: pd.DataFrame, minRatings: int = 10
):
  raterTotals = uniqueRatingsOnTweets[c.raterParticipantIdKey].value_counts().reset_index()
  raterTotals.columns = [c.raterParticipantIdKey, "count"]
  raterTotals = raterTotals[raterTotals["count"] >= minRatings]
  pairStatsDf = pairStatsDf.merge(
    raterTotals.rename(columns={c.raterParticipantIdKey: "leftRaterId", "count": "leftTotal"})
  )
  pairStatsDf = pairStatsDf.merge(
    raterTotals.rename(columns={c.raterParticipantIdKey: "rightRaterId", "count": "rightTotal"})
  )
  return pairStatsDf


def aggregate_into_cliques(graphDf):
  with c.time_block("Aggregate into cliques by post selection similarity"):
    userToCliqueMap = dict()
    cliqueToUserMap = dict()

    nextNewCliqueId = 1  # start cliqueIdxs from 1
    for i, row in graphDf.iterrows():
      sid = row["leftRaterId"]
      tid = row["rightRaterId"]
      if sid in userToCliqueMap:
        if tid in userToCliqueMap:
          # both in map. merge if not same clique
          if userToCliqueMap[sid] != userToCliqueMap[tid]:
            # merge. assign all member's of target clique to source clique.
            # slow way: iterate over all values here.
            # fast way: maintain a reverse map of cliqueToUserMap.
            sourceDestClique = userToCliqueMap[sid]
            oldTargetCliqueToDel = userToCliqueMap[tid]

            for userId in cliqueToUserMap[oldTargetCliqueToDel]:
              cliqueToUserMap[sourceDestClique].append(userId)
              userToCliqueMap[userId] = sourceDestClique
            del cliqueToUserMap[oldTargetCliqueToDel]
            gc.collect()

        else:
          # source in map; target not. add target to source's clique
          sourceClique = userToCliqueMap[sid]
          userToCliqueMap[tid] = sourceClique
          cliqueToUserMap[sourceClique].append(tid)
      elif tid in userToCliqueMap:
        # target in map; source not. add source to target's clique
        targetClique = userToCliqueMap[tid]
        userToCliqueMap[sid] = targetClique
        cliqueToUserMap[targetClique].append(sid)
      else:
        # new clique
        userToCliqueMap[sid] = nextNewCliqueId
        userToCliqueMap[tid] = nextNewCliqueId
        cliqueToUserMap[nextNewCliqueId] = [sid, tid]
        nextNewCliqueId += 1
  return cliqueToUserMap, userToCliqueMap


def _make_rater_stats_df(pairCounts):
  with c.time_block("Making rater stats dataframe from pair counts dict"):
    leftRater, rightRater, pairRatings = [], [], []
    for i, ((left, right), count) in enumerate(pairCounts.items()):
      leftRater.append(left)
      rightRater.append(right)
      pairRatings.append(count)
  return pd.DataFrame(
    {
      "leftRaterId": np.array(leftRater),
      "rightRaterId": np.array(rightRater),
      "pairRatings": np.array(pairRatings),
    }
  )

def _get_pair_counts_df_dict(ratings, windowMillis):
    import numpy as np
    import pandas as pd
    from collections import defaultdict

    # Assign column keys to local variables for faster access
    noteIdKey = c.noteIdKey
    createdAtMillisKey = c.createdAtMillisKey
    raterParticipantIdKey = c.raterParticipantIdKey

    # Sort ratings by noteIdKey and createdAtMillisKey
    ratings_sorted = ratings.sort_values([noteIdKey, createdAtMillisKey])

    # Initialize a defaultdict to store counts of pairs
    pair_counts = defaultdict(int)

    # Group by noteIdKey to process each note individually
    grouped = ratings_sorted.groupby(noteIdKey, sort=False)

    for noteId, group in grouped:
        # Extract relevant columns as numpy arrays for efficient computation
        times = group[createdAtMillisKey].values
        raters = group[raterParticipantIdKey].values

        n = len(group)
        window_start = 0

        for i in range(n):
            # Move the window start forward if the time difference exceeds windowMillis
            while times[i] - times[window_start] > windowMillis:
                window_start += 1

            # For all indices within the sliding window (excluding the current index)
            for j in range(window_start, i):
                if raters[i] != raters[j]:
                    left_rater, right_rater = tuple(sorted((raters[i], raters[j])))
                    # Update the count for this pair
                    pair_counts[(left_rater, right_rater)] += 1

    # Convert the pair_counts dictionary to a DataFrame
    if pair_counts:
        pairs = np.array(list(pair_counts.keys()))
        counts = np.array(list(pair_counts.values()))
        df = pd.DataFrame({
            'leftRaterId': pairs[:, 0],
            'rightRaterId': pairs[:, 1],
            'pairRatings': counts
        })
    else:
        # Return an empty DataFrame with appropriate columns
        df = pd.DataFrame(columns=['leftRaterId', 'rightRaterId', 'pairRatings'])

    return df


def _get_pair_ratings_df_optimized(ratings, windowMillis):

    # Assign column keys to local variables for faster access
    noteIdKey = c.noteIdKey
    createdAtMillisKey = c.createdAtMillisKey
    raterParticipantIdKey = c.raterParticipantIdKey
    tweetIdKey = c.tweetIdKey

    # Sort ratings by noteIdKey and createdAtMillisKey
    ratings_sorted = ratings.sort_values([noteIdKey, createdAtMillisKey])

    # Initialize lists to collect data
    left_raters = []
    right_raters = []
    tweet_ids = []

    # Group by noteIdKey to process each note individually
    grouped = ratings_sorted.groupby(noteIdKey, sort=False)

    for noteId, group in grouped:
        # Extract relevant columns as numpy arrays for efficient computation
        times = group[createdAtMillisKey].values
        raters = group[raterParticipantIdKey].values
        tweetId = group[tweetIdKey].iloc[0]  # Assuming tweetIdKey is constant within a note

        n = len(group)
        window_start = 0

        for i in range(n):
            # Move the window start forward if the time difference exceeds windowMillis
            while times[i] - times[window_start] > windowMillis:
                window_start += 1

            # For all indices within the sliding window (excluding the current index)
            for j in range(window_start, i):
                if raters[i] != raters[j]:
                    left_rater, right_rater = tuple(sorted((raters[i], raters[j])))
                    left_raters.append(left_rater)
                    right_raters.append(right_rater)
                    tweet_ids.append(tweetId)

    # Convert lists to numpy arrays for efficient DataFrame creation
    left_raters = np.array(left_raters)
    right_raters = np.array(right_raters)
    tweet_ids = np.array(tweet_ids)

    # Create the DataFrame from the collected data
    df = pd.DataFrame({
        'leftRaterId': left_raters,
        'rightRaterId': right_raters,
        'tweetId': tweet_ids,
    })

    # Drop duplicates
    df = df.drop_duplicates()

    # Group by leftRaterId and rightRaterId and count the number of occurrences
    df = (
        df.groupby(['leftRaterId', 'rightRaterId'], as_index=False)
        .agg(pairRatings=('tweetId', 'count'))
    )
    return df


# get number of ratings per pair in same time window
def _get_pair_tuples(ratings, windowMillis):
  tuples = []
  ratings = ratings.sort_values([c.noteIdKey, c.createdAtMillisKey])
  values = ratings[
    [c.noteIdKey, c.createdAtMillisKey, c.raterParticipantIdKey, c.tweetIdKey]
  ].values
  print(len(values))
  for i in range(len(values)):
    priorNote, priorTs, priorRater, priorTweet = values[i]
    if i == 0 or i == 1000 or i == 100000 or i % 5000000 == 0:
      print(f"i={i}  len(tuples)={len(tuples)}")
    j = i + 1
    while j < len(values):
      nextNote, nextTs, nextRater, nextTweet = values[j]
      assert priorNote <= nextNote, (priorNote, nextNote)
      if nextNote != priorNote:
        break  # break if we're onto a new note
      assert priorTweet == nextTweet, (priorTweet, nextTweet)  # tweet should be same
      assert priorRater != nextRater, (priorRater, nextRater)  # rater should be different
      assert priorTs <= nextTs, (priorTs, nextTs)
      if nextTs > (priorTs + windowMillis):
        break  # break if we're beyond the overlap window
      leftRater, rigthRater = tuple(sorted((priorRater, nextRater)))
      tuples.append((leftRater, rigthRater, priorTweet))
      j += 1
  return tuples

def _get_pair_tuples_optimized(ratings, windowMillis):

    # Sort ratings by noteIdKey and createdAtMillisKey
    ratings_sorted = ratings.sort_values([c.noteIdKey, c.createdAtMillisKey])

    # Initialize an empty list to store the result
    tuples = []

    # Group by noteIdKey to process each note individually
    grouped = ratings_sorted.groupby(c.noteIdKey, sort=False)

    for noteId, group in grouped:
        # Extract relevant columns as numpy arrays for efficient computation
        times = group[c.createdAtMillisKey].values
        raters = group[c.raterParticipantIdKey].values
        priorTweet = group[c.tweetIdKey].iloc[0]

        n = len(group)
        window_start = 0  # Start index of the sliding window

        for i in range(n):
            # Move the window start forward if the time difference exceeds windowMillis
            while times[i] - times[window_start] > windowMillis:
                window_start += 1

            # For all indices within the sliding window (excluding the current index)
            for j in range(window_start, i):
                # Check if raters are different
                if raters[i] != raters[j]:
                    # Sort raters to maintain consistency
                    leftRater, rightRater = tuple(sorted((raters[i], raters[j])))
                    tuples.append((leftRater, rightRater, priorTweet))

    return tuples


import multiprocessing as mp

def _get_pair_tuples_parallel(ratings, windowMillis):
    # Sort and group ratings
    ratings_sorted = ratings.sort_values([c.noteIdKey, c.createdAtMillisKey])
    grouped = ratings_sorted.groupby(c.noteIdKey, sort=False)

    # Prepare arguments for parallel processing
    args = [(group, windowMillis) for _, group in grouped]

    # Use multiprocessing Pool
    with mp.Pool(mp.cpu_count()) as pool:
        results = pool.starmap(_get_pair_tuples_process_group, args)

    # Flatten the list of results
    tuples = [tup for sublist in results for tup in sublist]
    return tuples

def _get_pair_tuples_process_group(group, windowMillis):
    # Same logic as before, applied to a single group
    times = group[c.createdAtMillisKey].values
    raters = group[c.raterParticipantIdKey].values
    priorTweet = group[c.tweetIdKey].iloc[0]

    n = len(group)
    window_start = 0
    tuples = []

    for i in range(n):
        while times[i] - times[window_start] > windowMillis:
            window_start += 1
        for j in range(window_start, i):
            if raters[i] != raters[j]:
                leftRater, rightRater = tuple(sorted((raters[i], raters[j])))
                tuples.append((leftRater, rightRater, priorTweet))
    return tuples



def _tuples_to_df(tuples, name="pairRatings"):
  leftRater, rightRater, tweetId = zip(*tuples)
  df = pd.DataFrame(
    {
      "leftRaterId": np.array(leftRater),
      "rightRaterId": np.array(rightRater),
      "tweetId": np.array(tweetId),
    }
  )
  print(len(df))
  df = df.drop_duplicates()
  print(len(df))
  df = (
    df.groupby(["leftRaterId", "rightRaterId"])
    .count()
    .reset_index(drop=False)
    .rename(columns={"tweetId": name})
  )
  print(len(df))
  return df


def _get_pair_counts(ratings: pd.DataFrame, windowMillis: int = 1000 * 60 * 30) -> Dict:
  """
  Compute counts of unique posts that were co-rated within windowMillis millis of each other
  by different users.

  Returns dict: (raterId1, raterId2) => count.
  """
  with c.time_block("Computing rating pair counts"):
    counts = dict()
    seen = set()
    ratings = ratings.sort_values([c.noteIdKey, c.createdAtMillisKey])
    values = ratings[
      [c.noteIdKey, c.createdAtMillisKey, c.raterParticipantIdKey, c.tweetIdKey]
    ].values
    logger.info(len(values))
    for i in range(len(values)):
      priorNote, priorTs, priorRater, priorTweet = values[i]
      if i == 0 or i == 1000 or i == 100000 or i % 5000000 == 0:
        logger.info(f"get_pair_counts i={i}")
      j = i + 1
      while j < len(values):
        nextNote, nextTs, nextRater, nextTweet = values[j]
        assert priorNote <= nextNote, (priorNote, nextNote)
        if nextNote != priorNote:
          break  # break if we're onto a new note
        assert priorTweet == nextTweet, (priorTweet, nextTweet)  # tweet should be same
        assert priorRater != nextRater, (priorRater, nextRater)  # rater should be different
        assert priorTs <= nextTs, (priorTs, nextTs)
        if nextTs > (priorTs + windowMillis):
          break  # break if we're beyond windowMillis
        raterPairKey = tuple(sorted((priorRater, nextRater)))
        raterTweetPairKey = (raterPairKey, priorTweet)
        if raterTweetPairKey in seen:
          break  # break if we already counted a match on this tweet
        seen.add(raterTweetPairKey)
        if raterPairKey not in counts:
          counts[raterPairKey] = 0
        counts[raterPairKey] += 1
        j += 1
    return counts