sourcecode/scoring/pflip_plus_model.py

"""Train and apply a model to predict which notes will lose CRH status.

This module implements a supervised model designed to be applied throughout the period
when a enters or be in status stabilization. The model predicts whether the note will ultimately
lock to CRH status or flip back to NMR.  The features used by the model include:
* Note author and timing of the note in relation to the associated post.
* Individual user Helpfulness and tag ratings.
* Ratios of how common each tag is in ratings associated with the note.
* Aggregate statistics about rater factors (e.g. standard deviation of rater factors for
  Helpful ratings, number of {Helpful, Somewhat Helpful, Not Helpful} ratings from users
  with {positive, center, negative} factors, etc.).
* Burstiness of ratings (e.g. how many ratings occur soon after note creation, over short
  periods of time or within the recent past).
* Creation, ratings and scoring outcomes of peer notes associated with the same post.

The training data includes notes that were CRH at some point and have either locked to
CRH or NMR, except for any note that either drifted to NMR status after locking.  Notes
included in training must be associated with a post created after August 19, 2024, which
marks the point when usage of the timestampMillisOfFirstNmrDueToMinStableCrhTime changed
such that the value always reflects the timestamp when a note first entered stabilization
or NaN if the note never entered stabilization.

Since the model is designed to be applied throughout the entire stabilization period we include
each note twice, with features reflecting both when the note enters stabilization and when the
note achieves CRH status.  If the note skips stabilization or exits stabilization without achieving
CRH status, then the note is included twice with identical features to maintain consistent
weighting across notes.

The feature extraction and model training are implemented entirely in scikit-learn.
"""

# Standard libraries
from io import BytesIO
import logging
from typing import Any, Dict, List, Optional, Tuple, Union

# Project libraries
from . import constants as c
from .enums import Scorers
from .pandas_utils import get_df_fingerprint

# 3rd party libraries
import joblib
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.feature_selection import SelectPercentile, VarianceThreshold, chi2
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import auc as area_under_curve, confusion_matrix, roc_curve
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import (
  FunctionTransformer,
  KBinsDiscretizer,
  OneHotEncoder,
  PolynomialFeatures,
)


# Configure logger
logger = logging.getLogger("birdwatch.pflip_plus_model")
logger.setLevel(logging.INFO)

# Exposed constants
LABEL = "LABEL"
CRH = "CRH"
FLIP = "FLIP"

# Internal constants
_MAX = "MAX"
_MIN = "MIN"
_MIN_TWEET_ID = 1825679568688054351

# Internal column names
_RATER_FACTOR = "RATER_FACTOR"
_SCORING_CUTOFF_MTS = "SCORING_CUTOFF_MTS"
_SCORING_CUTOFF_MODE = "SCORING_CUTOFF_MODE"
_USER_HELPFULNESS_RATINGS = "USER_HELPFULNESS_RATINGS"
_USER_HELPFUL_TAGS = "USER_HELPFUL_TAGS"
_USER_NOT_HELPFUL_TAGS = "USER_NOT_HELPFUL_TAGS"
_NEGATIVE = "NEGATIVE"
_NEUTRAL = "NEUTRAL"
_POSITIVE = "POSITIVE"
_BUCKET_COUNT_COLS = [
  f"{viewpoint}_{rating}"
  for viewpoint in [_NEGATIVE, _NEUTRAL, _POSITIVE]
  for rating in [c.helpfulValueTsv, c.somewhatHelpfulValueTsv, c.notHelpfulValueTsv]
]
_RATING_TIME_BUCKETS = [1, 5, 15, 60]
_QUICK_RATING_COLS = [f"FIRST_{cutoff}_TOTAL" for cutoff in _RATING_TIME_BUCKETS] + [
  f"FIRST_{cutoff}_RATIO" for cutoff in _RATING_TIME_BUCKETS
]
_BURST_RATING_COLS = [f"BURST_{cutoff}_TOTAL" for cutoff in _RATING_TIME_BUCKETS] + [
  f"BURST_{cutoff}_RATIO" for cutoff in _RATING_TIME_BUCKETS
]
_RECENT_RATING_COLS = [f"RECENT_{cutoff}_TOTAL" for cutoff in _RATING_TIME_BUCKETS] + [
  f"RECENT_{cutoff}_RATIO" for cutoff in _RATING_TIME_BUCKETS
]
_NOTE_WRITING_LATENCY = "NOTE_WRITING_LATENCY"
_MAX_POS_HELPFUL = "MAX_POS_HELPFUL"
_MAX_NEG_HELPFUL = "MAX_NEG_HELPFUL"
_MEAN_POS_HELPFUL = "MEAN_POS_HELPFUL"
_MEAN_NEG_HELPFUL = "MEAN_NEG_HELPFUL"
_STD_HELPFUL = "STD_HELPFUL"
_MAX_DIFF = "MAX_DIFF"
_MEAN_DIFF = "MEAN_DIFF"
_STATS_COLS = [
  _MAX_POS_HELPFUL,
  _MAX_NEG_HELPFUL,
  _MEAN_POS_HELPFUL,
  _MEAN_NEG_HELPFUL,
  _STD_HELPFUL,
  _MAX_DIFF,
  _MEAN_DIFF,
]
_NOTE_CREATION_MILLIS = "noteCreationMillis"
_TWEET_CREATION_MILLIS = "tweetCreationMillis"
_RATED_ON_NOTE_ID = "ratedOnNoteId"
_LOCAL = "LOCAL"
_PEER_MISLEADING = "PEER_MISLEADING"
_PEER_NON_MISLEADING = "PEER_NON_MISLEADING"
_TOTAL_PEER_NOTES = "TOTAL_PEER_NOTES"
_TOTAL_PEER_MISLEADING_NOTES = "TOTAL_PEER_MISLEADING_NOTES"
_TOTAL_PEER_NON_MISLEADING_NOTES = "TOTAL_PEER_NON_MISLEADING_NOTES"
_TOTAL_PEER_STABILIZATION_NOTES = "TOTAL_PEER_STABILIZATION_NOTES"
_TOTAL_PEER_CRH_NOTES = "TOTAL_PEER_CRH_NOTES"


# Define helper functions at module level so feature extraction pipeline doesn't require
# any lambda functions (and consequently can be pickled.)
def _identity(x: Any) -> Any:
  """Used to create modified preprocessing and tokenization for CountVectorizer."""
  return x


def _feature_log(features: pd.Series) -> pd.Series:
  """Helper to log-scale features while allowing for zero valued features."""
  return np.log(1 + features) / np.log(2)


def _get_timestamp_from_snowflake(snowflake: int) -> int:
  """Helper function to recover a timestamp from a snowflake ID."""
  return (snowflake >> 22) + 1288834974657


def _set_to_list(series: pd.Series) -> List:
  """Helper function to convert a Series containing sets to lists."""
  assert isinstance(series, pd.Series)
  return [list(x) for x in series]


def _fill_na(series: pd.Series) -> np.ndarray:
  return series.fillna(-1).values


def _reshape(series: pd.Series) -> np.ndarray:
  assert isinstance(series, pd.Series)
  return series.values.reshape(-1, 1)


class PFlipPlusModel(object):
  def __init__(
    self,
    ratingRecencyCutoffMinutes: int = 15,
    helpfulnessRaterMin: int = 1,
    tagRaterMin: int = 5,
    helpfulTagPercentile: int = 10,
    notHelpfulTagPercentile: int = 50,
    penalty: str = "l1",
    C: float = 0.04,
    maxIter: int = 2500,
    verbose: int = 0,
    classWeight: Dict[int, int] = {0: 1, 1: 1},
    trainSize: float = 0.9,
    solver: str = "liblinear",
    tol: float = 1e-4,
    seed: Optional[int] = None,
    crhFpRate: float = 0.6,
    tag_ratio_bins: int = 10,
    rating_count_bins: int = 7,
    factor_stats_bins: int = 10,
    burst_bins: int = 5,
    latency_bins: int = 10,
    peer_note_count_bins: int = 4,
  ):
    """Configure PFlipModel.

    Args:
      ratingRecencyCutoffMinutes: Avoiding skew in training requires identifying the time at which
        the note would have been scored in production and which notes / ratings would have been
        available at that time.  ratingRecencyCutoffMinutes specifies the assumed latency from
        rating creation through to when a rating becomes available in final scoring (e.g. if we know
        a note was set CRH at a particular point in time then we require ratings be created at least
        ratingRecencyCutoffMinutes minutes earlier).
    """
    self._pipeline: Optional[Pipeline] = None
    self._predictionThreshold: Optional[float] = None
    self._ratingRecencyCutoff = 1000 * 60 * ratingRecencyCutoffMinutes
    self._helpfulnessRaterMin = helpfulnessRaterMin
    self._tagRaterMin = tagRaterMin
    self._helpfulTagPercentile = helpfulTagPercentile
    self._notHelpfulTagPercentile = notHelpfulTagPercentile
    self._penalty = penalty
    self._C = C
    self._maxIter = maxIter
    self._verbose = verbose
    self._classWeight = classWeight
    self._solver = solver
    self._tol = tol
    self._trainSize = trainSize
    self._seed = seed
    self._crhFpRate = crhFpRate
    self._tag_ratio_bins = tag_ratio_bins
    self._rating_count_bins = rating_count_bins
    self._factor_stats_bins = factor_stats_bins
    self._burst_bins = burst_bins
    self._latency_bins = latency_bins
    self._peer_note_count_bins = peer_note_count_bins
    self._column_thresholds: Dict[str, float] = dict()

  def _get_notes(
    self,
    notes: pd.DataFrame,
    noteStatusHistory: pd.DataFrame,
  ):
    """Determine notes to include in scoring and associated metadata.

    To be included, notes must have an associated tweet and classification.  Returned columns
    include: noteId, tweetId, classification, noteCreationMillis, tweetCreationMillis

    Args:
      notes: Input DataFrame containing creation times.
      noteStatusHistory: Input DataFrame containing creation times.
    """
    # Validate that "notes" contains a subset of noteStatusHistory.
    assert notes[c.noteIdKey].nunique() == len(notes), "notes contains duplicate noteIds"
    assert noteStatusHistory[c.noteIdKey].nunique() == len(
      noteStatusHistory
    ), "noteStatusHistory contains duplicate noteIds"
    assert len(notes) == len(
      noteStatusHistory[[c.noteIdKey, c.createdAtMillisKey]].merge(
        notes[[c.noteIdKey, c.createdAtMillisKey]],
        on=[c.noteIdKey, c.createdAtMillisKey],
        how="inner",
      )
    ), "notes is not a subset of noteStatusHistory"
    # Develop set of candidate notes
    candidateNotes = noteStatusHistory[[c.noteIdKey, c.createdAtMillisKey]].rename(
      columns={c.createdAtMillisKey: _NOTE_CREATION_MILLIS}
    )
    candidateNotes = candidateNotes.merge(notes[[c.noteIdKey, c.tweetIdKey, c.classificationKey]])
    candidateNotes[_TWEET_CREATION_MILLIS] = [
      _get_timestamp_from_snowflake(tweetId) for tweetId in candidateNotes[c.tweetIdKey]
    ]
    # Prune candidates to require that associated tweet and classification are available, and possibly
    # that associated tweet is recent.
    candidateNotes = candidateNotes[
      (candidateNotes[c.tweetIdKey] > 0) & (candidateNotes[c.classificationKey].notna())
    ]
    return candidateNotes

  def _compute_scoring_cutoff(
    self, noteStatusHistory: pd.DataFrame, ratings: pd.DataFrame, mode: str, minRatings: int = 5
  ) -> pd.DataFrame:
    """Compute time limits on which ratings to include for each note.

    Recall that some notes may enter stabilization and then later exit to CRH, some may enter stabilization
    and exit to NMR, and some may skip stabilizaiton and go directly to CRH.  We support taking the min
    and max of the two timestamps so that we have a datapoint for notes both at the beginning and end of
    any stabilization period.  For notes that skip stabilization or never go CRH, this means the note is
    included twice with the same features, which is OK since it ensures weight parity with notes that
    are included twice with different features.  We do want to capture notes at both the beginning and
    end of the stabilization period so that the model performs well throughout the entire stabilization
    period.

    Notes that never enter stabilization or go CRH will be dropped because they should not be included
    in training.

    Args:
      noteStatusHistory: pd.DataFrame used to determine time of the first CRH status,
        if applicable.
      mode: whether to return the max or min of timestampMillisOfFirstNmrDueToMinStableCrhTime
        and timestampMillisOfNoteFirstNonNMRLabel

    Returns:
      pd.DataFrame with noteId and STATUS_MTS columns
    """
    # Select columns and prune to notes that either entered stabilization or went CRH.  Note that
    # the goal here is to not set a rating cutoff for notes that went CRNH since the model won't
    # be applied to these notes in prod.
    scoringCutoff = noteStatusHistory[
      [
        c.noteIdKey,
        c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey,
        c.timestampMillisOfNoteFirstNonNMRLabelKey,
        c.firstNonNMRLabelKey,
      ]
    ].copy()
    scoringCutoff = scoringCutoff[
      (scoringCutoff[c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey].notna())
      | (scoringCutoff[c.firstNonNMRLabelKey] == c.currentlyRatedHelpful)
    ].drop(columns=c.firstNonNMRLabelKey)
    # Set cutoff timestamps
    if mode == _MAX:
      cutoffs = scoringCutoff[
        [
          c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey,
          c.timestampMillisOfNoteFirstNonNMRLabelKey,
        ]
      ].max(axis=1)
    else:
      assert mode == _MIN, f"unexpected mode: {mode}"
      cutoffs = scoringCutoff[
        [
          c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey,
          c.timestampMillisOfNoteFirstNonNMRLabelKey,
        ]
      ].min(axis=1)
    # cutoffs doesn't contain any NaN values and has float type, but Pandas generates an
    # error when casting directly to Int64Dtype.  There is no error if cast to int64 first.
    assert cutoffs.dtype == np.float64
    assert cutoffs.isna().sum() == 0
    scoringCutoff[_SCORING_CUTOFF_MTS] = cutoffs.astype(np.int64).astype(pd.Int64Dtype())
    scoringCutoff[_SCORING_CUTOFF_MTS] = (
      scoringCutoff[_SCORING_CUTOFF_MTS] - self._ratingRecencyCutoff
    )
    # Compute cutoff that is guaranteed to include at least 5 ratings
    ratings = ratings[[c.noteIdKey, c.createdAtMillisKey]]
    cutoffByRatings = (
      ratings.groupby(c.noteIdKey)
      .max()
      .reset_index(drop=False)
      .rename(columns={c.createdAtMillisKey: "maxRatingMts"})
      .astype(pd.Int64Dtype())
    )
    cutoffByRatings = cutoffByRatings.merge(
      ratings.sort_values(c.createdAtMillisKey, ascending=True)
      .groupby(c.noteIdKey)
      .nth(minRatings - 1)
      .rename(columns={c.createdAtMillisKey: "nthRatingMts"})
      .astype(pd.Int64Dtype()),
      how="left",
    )
    cutoffByRatings["ratingMin"] = cutoffByRatings[["maxRatingMts", "nthRatingMts"]].min(axis=1)
    # Merge cutoffs by time and by ratings
    beforeMerge = len(scoringCutoff)
    scoringCutoff = scoringCutoff.merge(cutoffByRatings[[c.noteIdKey, "ratingMin"]])
    assert len(scoringCutoff) == beforeMerge
    scoringCutoff[_SCORING_CUTOFF_MTS] = scoringCutoff[[_SCORING_CUTOFF_MTS, "ratingMin"]].max(
      axis=1
    )
    scoringCutoff[_SCORING_CUTOFF_MODE] = mode
    return scoringCutoff[[c.noteIdKey, _SCORING_CUTOFF_MTS, _SCORING_CUTOFF_MODE]]

  def _label_notes(
    self,
    noteStatusHistory: pd.DataFrame,
  ) -> pd.DataFrame:
    """Generate a DataFrame mapping noteIds to labels.

    We define a CRH note as any note that is locked to CRH, and a FLIP note as any note
    that was scored as CRH at some point but has since locked to NMR.  Note that we exclude
    notes that are locked to CRH but decided by ScoringDriftGuard, since that indicates the
    model wanted to score the note as NMR (and therefore it is unclear whether the note is
    best labeled CRH or FLIP).

    Args:
      noteStatusHistory: pd.DataFrame used to determine locked status and whether there was
        a prior CRH status.

    Returns:
      pd.DataFrame with noteId and LABEL columns
    """
    # Assemble relevant data for labeling
    labels = noteStatusHistory[
      [
        c.noteIdKey,
        # If set, implies note was on track to be CRH at some point
        c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey,
        # If set to CRH, implies note was actually CRH at some point
        c.firstNonNMRLabelKey,
        # Use to determine final status and whether note is locked
        c.lockedStatusKey,
        # If set to ScoringDriftGuard, indicates note may be prone to flipping
        c.currentDecidedByKey,
      ]
    ].copy()
    labels[LABEL] = pd.NA
    labels.loc[(labels[c.lockedStatusKey] == c.currentlyRatedHelpful), LABEL] = CRH
    labels.loc[
      (labels[c.firstNonNMRLabelKey] == c.currentlyRatedHelpful)
      & (labels[c.lockedStatusKey].isin({c.needsMoreRatings, c.currentlyRatedNotHelpful})),
      LABEL,
    ] = FLIP
    labels.loc[
      (~labels[c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey].isna())
      & (labels[c.firstNonNMRLabelKey].isna())
      & (labels[c.lockedStatusKey].isin({c.needsMoreRatings, c.currentlyRatedNotHelpful})),
      LABEL,
    ] = FLIP
    labels = labels.dropna(subset=LABEL)
    logger.info(f"labels before ScoringDriftGuard:\n{labels[LABEL].value_counts(dropna=False)}")
    # Note that we don't exclude notes decided by ScoringDriftGuard when a note locks to NMR
    # after being CRH and is now decided by ScoringDriftGuard (implying the note was once again
    # scored as CRH) because in that case the note is labeled as FLIP and the involvement of
    # ScoringDriftGuard only provides further evidence that the note flips status.
    dropRows = (labels[LABEL] == CRH) & (
      np.array(
        [decider.startswith("ScoringDriftGuard") for decider in labels[c.currentDecidedByKey]]
      )
    )
    labels = labels[~dropRows][[c.noteIdKey, LABEL]]
    logger.info(f"labels after ScoringDriftGuard:\n{labels[LABEL].value_counts(dropna=False)}")
    return labels

  def _compute_rater_factors(self, prescoringRaterModelOutput: pd.DataFrame) -> pd.DataFrame:
    """Generate a DataFrame mapping raterParticipantIds to factors.

    Each rater is assigned their factor from either Core, Expansion or ExpansionPlus,
    prioritizing Core, then Expansion, then ExpansionPlus.

    Args:
      prescoringRaterModelOutput: pd.DataFrame used to determine rater factors.

    Returns:
      pd.DataFrame with raterParticipantId and RATER_FACTOR columns
    """
    # Obtain prescoring rater factors
    coreFactors = prescoringRaterModelOutput[
      prescoringRaterModelOutput[c.scorerNameKey] == Scorers.MFCoreScorer.name
    ][[c.raterParticipantIdKey, c.internalRaterFactor1Key]].rename(
      columns={c.internalRaterFactor1Key: c.coreRaterFactor1Key}
    )
    expansionFactors = prescoringRaterModelOutput[
      prescoringRaterModelOutput[c.scorerNameKey] == Scorers.MFExpansionScorer.name
    ][[c.raterParticipantIdKey, c.internalRaterFactor1Key]].rename(
      columns={c.internalRaterFactor1Key: c.expansionRaterFactor1Key}
    )
    expansionPlusFactors = prescoringRaterModelOutput[
      prescoringRaterModelOutput[c.scorerNameKey] == Scorers.MFExpansionPlusScorer.name
    ][[c.raterParticipantIdKey, c.internalRaterFactor1Key]].rename(
      columns={c.internalRaterFactor1Key: c.expansionPlusRaterFactor1Key}
    )
    # Combine and prioritize factors
    raterFactors = coreFactors.merge(expansionFactors, how="outer").merge(
      expansionPlusFactors, how="outer"
    )
    raterFactors[_RATER_FACTOR] = raterFactors[c.expansionPlusRaterFactor1Key]
    raterFactors.loc[
      ~raterFactors[c.expansionRaterFactor1Key].isna(), _RATER_FACTOR
    ] = raterFactors[c.expansionRaterFactor1Key]
    raterFactors.loc[~raterFactors[c.coreRaterFactor1Key].isna(), _RATER_FACTOR] = raterFactors[
      c.coreRaterFactor1Key
    ]
    return raterFactors[[c.raterParticipantIdKey, _RATER_FACTOR]]

  def _prepare_local_ratings(
    self,
    ratings: pd.DataFrame,
    scoredNotes: pd.DataFrame,
  ) -> pd.DataFrame:
    """Filter ratings DF to only include ratings on the notes being scored.

    Args:
      ratings: pd.DataFrame containing all ratings used in scoring
      scoredNotes: pd.DataFrame specifying which notes are actually being scored
    """
    return ratings.merge(scoredNotes[[c.noteIdKey]])

  def _prepare_peer_ratings(
    self, ratings: pd.DataFrame, notes: pd.DataFrame, scoredNotes: pd.DataFrame
  ) -> pd.DataFrame:
    """Construct a ratings DF that captures ratings on peer notes.

    Peer notes are defined as different notes that are on the same post.  We consider the
    ratings for the peer note to provide signal for the note being scored (e.g. if another
    note received many "note not needed" ratings, then it is more likely that the post
    itself does not need a note).

    Since each unique {note, peer} pair is a row in the output dataframe, the column name
    for the note being scored remains "noteId" and the note which originally received the
    rating is the "ratedOnNoteId".

    Args:
      ratings: pd.DataFrame containing all ratings used in scoring
      notes: pd.DataFrame relating notes to posts
      scoredNotes: pd.DataFrame specifying which notes are actually being scored
    """
    # Augment ratings with tweetId and classification
    beforeMerge = len(ratings)
    ratings = ratings.merge(notes[[c.noteIdKey, c.tweetIdKey, c.classificationKey]])
    assert len(ratings) == beforeMerge
    ratings = ratings.rename(columns={c.noteIdKey: _RATED_ON_NOTE_ID})
    # Create one copy of each rating for each scored note on the post, excluding
    # ratings that occur on that post itself
    assert len(scoredNotes) == len(scoredNotes[[c.noteIdKey, c.tweetIdKey]].drop_duplicates())
    ratings = ratings.merge(scoredNotes[[c.noteIdKey, c.tweetIdKey]], on=c.tweetIdKey)
    ratings = ratings[
      ratings[c.noteIdKey] != ratings[_RATED_ON_NOTE_ID]
    ]  # throw out ratings that occur on the note itself
    ratings = ratings.drop(columns={c.tweetIdKey})
    return ratings

  def _apply_cutoff(self, ratings: pd.DataFrame, scoredNotes: pd.DataFrame) -> pd.DataFrame:
    """Filter a ratings DF to only contain ratings before an applicable cutoff.

    Args:
      ratings: pd.DataFrame containing peer and/or local ratings.
      scoredNotes: pd.DataFrame specifying scoring cutoff timestamps
    """
    assert scoredNotes[_SCORING_CUTOFF_MTS].isna().sum() == 0
    beforeMerge = len(ratings)
    ratings = ratings.merge(scoredNotes[[c.noteIdKey, _SCORING_CUTOFF_MTS]])
    assert len(ratings) == beforeMerge
    return ratings[ratings[c.createdAtMillisKey] <= ratings[_SCORING_CUTOFF_MTS]].drop(
      columns=_SCORING_CUTOFF_MTS
    )

  def _get_note_writing_latency(self, notes: pd.DataFrame) -> pd.DataFrame:
    """Identify the time in milliseconds from post to note creation."""
    notes = notes[[c.noteIdKey, _NOTE_CREATION_MILLIS, _TWEET_CREATION_MILLIS]].copy()
    notes[_NOTE_WRITING_LATENCY] = notes[_NOTE_CREATION_MILLIS] - notes[_TWEET_CREATION_MILLIS]
    return notes[[c.noteIdKey, _NOTE_WRITING_LATENCY]]

  def _get_quick_rating_stats(self, notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
    """Return counts and ratios of how many ratings occurred in the first 1/5/15/60 minutes.

    Args:
      notes: DF specifying note creation timestamps.
      ratings: DF specifying local rating timestamps.
    """
    ratingTotals = (
      ratings[[c.noteIdKey]]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": "total"})
    )
    ratingTotals = notes[[c.noteIdKey]].merge(ratingTotals, how="left")
    ratingTotals = ratingTotals.fillna({"total": 0}).astype(pd.Int64Dtype())
    for cutoff in _RATING_TIME_BUCKETS:
      beforeCutoff = ratings[[c.noteIdKey, c.createdAtMillisKey]].rename(
        columns={c.createdAtMillisKey: "ratingCreationMts"}
      )
      beforeCutoff = beforeCutoff.merge(notes[[c.noteIdKey, _NOTE_CREATION_MILLIS]])
      beforeCutoff = beforeCutoff[
        beforeCutoff["ratingCreationMts"]
        < (beforeCutoff[_NOTE_CREATION_MILLIS] + (1000 * 60 * cutoff))
      ]
      cutoffCount = (
        beforeCutoff[[c.noteIdKey]]
        .value_counts()
        .to_frame()
        .reset_index(drop=False)
        .rename(columns={"count": f"FIRST_{cutoff}_TOTAL"})
      )
      ratingTotals = ratingTotals.merge(cutoffCount, how="left").fillna(0)
    ratingTotals = ratingTotals.astype(pd.Int64Dtype())
    for cutoff in _RATING_TIME_BUCKETS:
      ratingTotals[f"FIRST_{cutoff}_RATIO"] = ratingTotals[f"FIRST_{cutoff}_TOTAL"] / (
        ratingTotals["total"].clip(lower=1)
      )
    return ratingTotals[[c.noteIdKey] + _QUICK_RATING_COLS]

  def _get_burst_rating_stats(self, notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
    """Return counts and ratios of the max ratings in 1/5/15/60 minute windows.

    Args:
      ratings: DF specifying local rating timestamps.
    """
    ratingTotals = (
      ratings[[c.noteIdKey]]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": "total"})
    )
    initialNotes = len(notes)
    ratingTotals = notes[[c.noteIdKey]].merge(ratingTotals, how="left")
    ratingTotals = ratingTotals.fillna({"total": 0}).astype(pd.Int64Dtype())
    for cutoff in _RATING_TIME_BUCKETS:
      ratingCounts = []
      for offset in range(cutoff):
        offsetRatings = ratings[[c.noteIdKey, c.createdAtMillisKey]].copy()
        offsetRatings[c.createdAtMillisKey] = (
          offsetRatings[c.createdAtMillisKey] + (1000 * 60 * offset)
        ) // (1000 * 60 * cutoff)
        offsetRatings = (
          offsetRatings.value_counts()
          .to_frame()
          .reset_index(drop=False)[[c.noteIdKey, "count"]]
          .groupby(c.noteIdKey)
          .max()
          .reset_index(drop=False)
        )
        ratingCounts.append(offsetRatings)
      ratingCounts = (
        pd.concat(ratingCounts)
        .groupby(c.noteIdKey)
        .max()
        .reset_index(drop=False)
        .rename(columns={"count": f"BURST_{cutoff}_TOTAL"})
      ).astype(pd.Int64Dtype())
      ratingTotals = ratingTotals.merge(ratingCounts, how="left").fillna(
        {f"BURST_{cutoff}_TOTAL": 0}
      )
      ratingTotals[f"BURST_{cutoff}_RATIO"] = ratingTotals[f"BURST_{cutoff}_TOTAL"] / (
        ratingTotals["total"].clip(lower=1)
      )
    assert (
      len(ratingTotals) == initialNotes
    ), f"unexpected length mismatch: {len(ratingTotals)} vs. {initialNotes}"
    return ratingTotals[[c.noteIdKey] + _BURST_RATING_COLS]

  def _get_recent_rating_stats(
    self, scoredNotes: pd.DataFrame, ratings: pd.DataFrame, prepareForTraining: bool
  ):
    """Generate counts of ratings within the last 1/5/15/20 minutes.

    Note that this process must work differently during training and in production, since
    during training future ratings will be available.  We solve this by treating the
    _SCORING_CUTOFF_MTS as the effective time at which scoring occurred, and include ratings
    based on that timestamp.

    Args:
      scoredNotes: pd.DataFrame specifying scoring cutoff timestamps
      ratings: pd.DataFrame containing local ratings.
      prepareForTraining: bool specifying whether to prune ratings.
    """
    # Define notion of effective present for each rating
    initialRatings = len(ratings)
    ratings = ratings[[c.noteIdKey, c.createdAtMillisKey]].copy()
    if prepareForTraining:
      ratings = ratings.merge(
        scoredNotes[[c.noteIdKey, _SCORING_CUTOFF_MTS]].rename(
          columns={_SCORING_CUTOFF_MTS: "effectivePresent"}
        )
      )
    else:
      ratings["effectivePresent"] = ratings[c.createdAtMillisKey].max()
    assert len(ratings) == initialRatings
    assert (ratings[c.createdAtMillisKey] > ratings["effectivePresent"]).sum() == 0
    assert ratings[c.createdAtMillisKey].isna().sum() == 0
    assert ratings["effectivePresent"].isna().sum() == 0
    # Develop counts and ratios of recent ratings in specific time ranges
    ratingTotals = (
      ratings[[c.noteIdKey]]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": "total"})
    )
    ratingTotals = scoredNotes[[c.noteIdKey]].merge(ratingTotals, how="left")
    ratingTotals = ratingTotals.fillna({"total": 0}).astype(pd.Int64Dtype())
    for cutoff in _RATING_TIME_BUCKETS:
      afterCutoff = ratings[
        ratings[c.createdAtMillisKey] > (ratings["effectivePresent"] - (1000 * 60 * cutoff))
      ]
      cutoffCount = (
        afterCutoff[[c.noteIdKey]]
        .value_counts()
        .to_frame()
        .reset_index(drop=False)
        .rename(columns={"count": f"RECENT_{cutoff}_TOTAL"})
      )
      ratingTotals = ratingTotals.merge(cutoffCount, how="left").fillna(0)
    ratingTotals = ratingTotals.astype(pd.Int64Dtype())
    for cutoff in _RATING_TIME_BUCKETS:
      ratingTotals[f"RECENT_{cutoff}_RATIO"] = ratingTotals[f"RECENT_{cutoff}_TOTAL"] / (
        ratingTotals["total"].clip(lower=1)
      )
    return ratingTotals[[c.noteIdKey] + _RECENT_RATING_COLS]

  def _get_helpfulness_ratings(self, notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
    """Return a DataFrame with one row per note and a column with a nested list of helpfulness ratings.

    Args:
      notes: pd.DataFrame used to determine full set of notes
      ratings: pd.DataFrame containing ratings that should contribute to model features.

    Returns:
      pd.DataFrame with two columns: noteIds and all user helpfulness ratings on the note.
    """
    raters = ratings[c.raterParticipantIdKey].astype(str)
    helpfulnessRatings = ratings[[c.noteIdKey]].copy()
    helpfulnessRatings.loc[:, _USER_HELPFULNESS_RATINGS] = (
      raters + ":" + ratings[c.helpfulnessLevelKey].astype(str)
    )
    helpfulnessRatings = (
      helpfulnessRatings[[c.noteIdKey, _USER_HELPFULNESS_RATINGS]]
      .groupby(c.noteIdKey)
      .agg(set)
      .reset_index(drop=False)
    )
    helpfulnessRatings = notes.merge(helpfulnessRatings, how="left")
    helpfulnessRatings[_USER_HELPFULNESS_RATINGS] = helpfulnessRatings[
      _USER_HELPFULNESS_RATINGS
    ].apply(lambda d: d if isinstance(d, set) else {})
    return helpfulnessRatings

  def _get_tag_ratios(self, notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
    """Produce a DataFrame specifying the ratio of ratings for each note that included each tag.

    Args:
      notes: pd.DataFrame used to specify the universe of all notes to include.
      ratings: pd.DataFrame containing all ratings for feature extraction.

    Returns:
      pd.DataFrame containing one row per note and one column per rating tag.
    """
    tags = ratings[[c.noteIdKey] + c.helpfulTagsTSVOrder + c.notHelpfulTagsTSVOrder].copy()
    total_ratings = "total_ratings"
    tags[total_ratings] = 1
    tags = tags.groupby(c.noteIdKey).sum().reset_index(drop=False)
    tags[c.helpfulTagsTSVOrder + c.notHelpfulTagsTSVOrder] = tags[
      c.helpfulTagsTSVOrder + c.notHelpfulTagsTSVOrder
    ].divide(tags[total_ratings], axis=0)
    tags = notes[[c.noteIdKey]].merge(tags.drop(columns=total_ratings), how="left")
    return tags[[c.noteIdKey] + c.helpfulTagsTSVOrder + c.notHelpfulTagsTSVOrder]

  def _get_user_tag_ratings(
    self,
    notes: pd.DataFrame,
    ratings: pd.DataFrame,
    outCol: str,
    tagCols: List[str],
  ) -> pd.DataFrame:
    """Return a DataFrame with one row per note and a column with a nested list of user rating tags.

    Args:
      notes: pd.DataFrame used to specify the universe of all notes to include.
      ratings: pd.DataFrame containing all ratings for feature extraction.
      outCol: str identifying output column to contain list of user tag ratings
      tagCols: List of tag columns to include in outCol

    Returns:
      pd.DataFrame containing one row per note and one column containing all user rating tags.
    """
    ratingTags = ratings[[c.noteIdKey, c.raterParticipantIdKey] + tagCols].copy()
    tagStrs = np.array(tagCols)
    ratingTags[outCol] = [
      {f"{rater}:{tag}" for tag in tagStrs[np.where(row)[0]]}
      for rater, row in zip(ratingTags[c.raterParticipantIdKey], ratingTags[tagCols].values)
    ]
    ratingTags = (
      ratingTags[[c.noteIdKey, outCol]]
      .groupby(c.noteIdKey)
      .agg(lambda x: set().union(*x))
      .reset_index(drop=False)
    )
    ratingTags = notes[[c.noteIdKey]].merge(ratingTags, how="left")
    ratingTags[outCol] = ratingTags[outCol].apply(
      lambda userTags: userTags if isinstance(userTags, set) else {}
    )
    return ratingTags

  def _get_bucket_count_totals(self, notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
    """Returns a DataFrame with one row per note and 9 columns containing buckets of rating counts.

    Args:
      notes: pd.DataFrame used to specify the universe of all notes to include.
      ratings: pd.DataFrame containing all ratings for feature extraction.

    Returns:
      pd.DataFrame containing one row per note and one column containing all user rating tags.
    """
    summary = ratings[[c.noteIdKey, _RATER_FACTOR, c.helpfulnessLevelKey]].copy()
    summary = summary[~summary[_RATER_FACTOR].isna()]
    summary[_NEGATIVE] = summary[_RATER_FACTOR] < -0.3
    summary[_NEUTRAL] = (summary[_RATER_FACTOR] >= -0.3) & (summary[_RATER_FACTOR] <= 0.3)
    summary[_POSITIVE] = summary[_RATER_FACTOR] > 0.3
    summary[c.helpfulValueTsv] = summary[c.helpfulnessLevelKey] == c.helpfulValueTsv
    summary[c.somewhatHelpfulValueTsv] = summary[c.helpfulnessLevelKey] == c.somewhatHelpfulValueTsv
    summary[c.notHelpfulValueTsv] = summary[c.helpfulnessLevelKey] == c.notHelpfulValueTsv
    for viewpoint in [_NEGATIVE, _NEUTRAL, _POSITIVE]:
      for rating in [c.helpfulValueTsv, c.somewhatHelpfulValueTsv, c.notHelpfulValueTsv]:
        summary[f"{viewpoint}_{rating}"] = summary[viewpoint].multiply(summary[rating])
    summary = summary[[c.noteIdKey] + _BUCKET_COUNT_COLS]
    summary = summary.groupby(c.noteIdKey).sum().reset_index(drop=False)
    summary[_BUCKET_COUNT_COLS] = summary[_BUCKET_COUNT_COLS].astype(np.float64)
    summary = (
      notes[[c.noteIdKey]]
      .merge(summary, on=c.noteIdKey, how="left")
      .fillna(0.0)
      .astype(pd.Int64Dtype())
    )
    return summary

  def _get_helpful_rating_stats(self, notes: pd.DataFrame, ratings: pd.DataFrame) -> pd.DataFrame:
    """Compute aggregate statistics about the Helpful ratings on a note.

    Args:
      notes: pd.DataFrame used to specify the universe of all notes to include.
      ratings: pd.DataFrame containing all ratings for feature extraction.

    Returns:
      pd.DataFrame with one row per note and 7 columns for aggregate statistics about rater
        factors for Helpful ratings.
    """
    # Prune ratings to only include Helpful ratings from users with a factor
    ratings = ratings[[c.noteIdKey, _RATER_FACTOR, c.helpfulnessLevelKey]].copy()
    ratings = ratings[~ratings[_RATER_FACTOR].isna()]
    ratings = ratings[ratings[c.helpfulnessLevelKey] == c.helpfulValueTsv]
    ratings = ratings.drop(columns=c.helpfulnessLevelKey)
    # Compute rating stats
    maxPosHelpful = (
      ratings[ratings[_RATER_FACTOR] > 0]
      .groupby(c.noteIdKey)
      .max()
      .reset_index()
      .rename(columns={_RATER_FACTOR: _MAX_POS_HELPFUL})
    )
    maxNegHelpful = (
      ratings[ratings[_RATER_FACTOR] <= 0]
      .groupby(c.noteIdKey)
      .min()
      .abs()
      .reset_index()
      .rename(columns={_RATER_FACTOR: _MAX_NEG_HELPFUL})
    )
    meanPosHelpful = (
      ratings[ratings[_RATER_FACTOR] > 0]
      .groupby(c.noteIdKey)
      .mean()
      .reset_index()
      .rename(columns={_RATER_FACTOR: _MEAN_POS_HELPFUL})
    )
    meanNegHelpful = (
      ratings[ratings[_RATER_FACTOR] <= 0]
      .groupby(c.noteIdKey)
      .mean()
      .abs()
      .reset_index()
      .rename(columns={_RATER_FACTOR: _MEAN_NEG_HELPFUL})
    )
    stdHelpful = (
      ratings.groupby(c.noteIdKey).std().reset_index().rename(columns={_RATER_FACTOR: _STD_HELPFUL})
    )
    # Compile into features per-note
    notes = notes[[c.noteIdKey]].merge(maxPosHelpful, on=c.noteIdKey, how="left")
    notes = notes.merge(maxNegHelpful, on=c.noteIdKey, how="left")
    notes = notes.merge(meanPosHelpful, on=c.noteIdKey, how="left")
    notes = notes.merge(meanNegHelpful, on=c.noteIdKey, how="left")
    notes = notes.merge(stdHelpful, on=c.noteIdKey, how="left")
    notes[_MAX_DIFF] = notes[_MAX_POS_HELPFUL] + notes[_MAX_NEG_HELPFUL]
    notes[_MEAN_DIFF] = notes[_MEAN_POS_HELPFUL] + notes[_MEAN_NEG_HELPFUL]
    return notes

  def _make_note_info_from_ratings(
    self,
    notes: pd.DataFrame,
    ratings: pd.DataFrame,
  ) -> pd.DataFrame:
    """Generate features derived from peer or local ratings associated with a note.

    Generated features include user helpfulness and tag ratings, buckets of rating counts,
    stats about rater factor distributions and ratios of tags across ratings.

    Args:
      notes: DF specifying which notes should be included in the output
      ratings: DF containing peer or local ratings to base the features on
    """
    # Augment notes with features.  Note that attributes of the note (e.g. author,
    # creation time) should always be available because we filter to notes with the creation time
    # in the last year, inherently removing any deleted notes where the creation time is unavailable.
    helpfulnessRatings = self._get_helpfulness_ratings(notes[[c.noteIdKey]], ratings)
    notes = notes.merge(helpfulnessRatings, how="inner")
    userHelpfulTags = self._get_user_tag_ratings(
      notes[[c.noteIdKey]], ratings, _USER_HELPFUL_TAGS, c.helpfulTagsTSVOrder
    )
    notes = notes.merge(userHelpfulTags, how="inner")
    userNotHelpfulTags = self._get_user_tag_ratings(
      notes[[c.noteIdKey]], ratings, _USER_NOT_HELPFUL_TAGS, c.notHelpfulTagsTSVOrder
    )
    notes = notes.merge(userNotHelpfulTags, how="inner")
    bucketCounts = self._get_bucket_count_totals(notes[[c.noteIdKey]], ratings)
    notes = notes.merge(bucketCounts, how="inner")
    helpfulStats = self._get_helpful_rating_stats(notes[[c.noteIdKey]], ratings)
    notes = notes.merge(helpfulStats, how="inner")
    tagRatios = self._get_tag_ratios(notes[[c.noteIdKey]], ratings)
    notes = notes.merge(tagRatios, how="inner")
    return notes

  def _get_note_counts(
    self,
    scoredNotes: pd.DataFrame,
    notes: pd.DataFrame,
    noteStatusHistory: pd.DataFrame,
    prepareForTraining: bool,
  ) -> pd.DataFrame:
    """Return counts of relevant events on peer notes.

    Counts should include: number of misleading and non-misleading notes created on the same post,
    number of misleading notes on the same post that went CRH and ratios of the above values.

    Note that aggregation works differently during training and in production, since training
    effectively has access to data from the future.


    Args:
      scoredNotes: DF containing {note, tweet} associations and effective scoring times for training.
      notes: DF containing set of all notes, not just those that are being scored.
      noteStatusHistory: DF specifying key scoring timestamps for peer notes.
      prepareForTraining: whether notes should be pruned to discard future data.
    """
    # Merge inputs into a single dataframe with relevant info
    assert scoredNotes[c.tweetIdKey].min() > 0  # tweet should be set for all notes being scored
    peerNotes = scoredNotes[[c.noteIdKey, c.tweetIdKey]].merge(
      notes[[c.tweetIdKey, c.noteIdKey, c.classificationKey]].rename(
        columns={c.noteIdKey: "peerNoteId"}
      )
    )
    assert len(scoredNotes) == peerNotes[c.noteIdKey].nunique()
    assert peerNotes[c.classificationKey].isna().sum() == 0
    peerNotes = peerNotes.merge(noteStatusHistory[[c.noteIdKey, c.createdAtMillisKey]])
    peerNotes = peerNotes.merge(
      noteStatusHistory[
        [
          c.noteIdKey,
          c.createdAtMillisKey,
          c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey,
          c.timestampMillisOfNoteFirstNonNMRLabelKey,
          c.firstNonNMRLabelKey,
        ]
      ].rename(columns={c.noteIdKey: "peerNoteId", c.createdAtMillisKey: "peerCreatedAtMillis"})
    )
    assert len(scoredNotes) == peerNotes[c.noteIdKey].nunique()

    # If we are in training, prune scope of what info should be available
    assert (_SCORING_CUTOFF_MTS in scoredNotes) == prepareForTraining
    if _SCORING_CUTOFF_MTS in scoredNotes:
      peerNotes = peerNotes.merge(scoredNotes[[c.noteIdKey, _SCORING_CUTOFF_MTS]])
      # Prune any notes created after the cutoff
      peerNotes = peerNotes[peerNotes["peerCreatedAtMillis"] <= peerNotes[_SCORING_CUTOFF_MTS]]
      # Null any scoring events that happened after the cutoff
      peerNotes.loc[
        peerNotes[c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey]
        > peerNotes[_SCORING_CUTOFF_MTS],
        c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey,
      ] = np.nan
      peerNotes.loc[
        peerNotes[c.timestampMillisOfNoteFirstNonNMRLabelKey] > peerNotes[_SCORING_CUTOFF_MTS],
        [c.timestampMillisOfNoteFirstNonNMRLabelKey, c.firstNonNMRLabelKey],
      ] = np.nan

    # Compute totals
    peerNotes = peerNotes[peerNotes[c.noteIdKey] != peerNotes["peerNoteId"]]
    totalPeerNotes = (
      peerNotes[c.noteIdKey]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _TOTAL_PEER_NOTES})
    )
    totalPeerMisleadingNotes = (
      peerNotes[peerNotes[c.classificationKey] == c.notesSaysTweetIsMisleadingKey][c.noteIdKey]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _TOTAL_PEER_MISLEADING_NOTES})
    )
    totalPeerNotMisleadingNotes = (
      peerNotes[peerNotes[c.classificationKey] == c.noteSaysTweetIsNotMisleadingKey][c.noteIdKey]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _TOTAL_PEER_NON_MISLEADING_NOTES})
    )
    totalPeerStabilizationNotes = (
      peerNotes[
        (peerNotes[c.classificationKey] == c.notesSaysTweetIsMisleadingKey)
        & (peerNotes[c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey].notna())
      ][c.noteIdKey]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _TOTAL_PEER_STABILIZATION_NOTES})
    )
    totalPeerCrhNotes = (
      peerNotes[
        (peerNotes[c.classificationKey] == c.notesSaysTweetIsMisleadingKey)
        & (peerNotes[c.firstNonNMRLabelKey] == c.currentlyRatedHelpful)
      ][c.noteIdKey]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _TOTAL_PEER_CRH_NOTES})
    )
    return (
      scoredNotes[[c.noteIdKey]]
      .merge(totalPeerNotes, how="left")
      .merge(totalPeerMisleadingNotes, how="left")
      .merge(totalPeerNotMisleadingNotes, how="left")
      .merge(totalPeerStabilizationNotes, how="left")
      .merge(totalPeerCrhNotes, how="left")
      .fillna(0)
      .astype(pd.Int64Dtype())
    )

  def _prepare_note_info(
    self,
    notes: pd.DataFrame,
    ratings: pd.DataFrame,
    noteStatusHistory: pd.DataFrame,
    prescoringRaterModelOutput: pd.DataFrame,
    prepareForTraining: bool,
    cutoff: Optional[str],
  ) -> pd.DataFrame:
    """Generate a DataFrame with one row per note containing all feature information.

    Note that some columns contain list values (e.g. a single column contains all
    Helpfulness ratings, where each rating is a unique string containing the rater ID
    and their Helpfulness rating).

    Args:
      notes: pd.DataFrame
      ratings: pd.DataFrame
      noteStatusHistory: pd.DataFrame
      prescoringRaterModelOutput: pd.DataFrame
      prepareForTraining: True if ratings should be filtered to discard data after the
        point of scoring to avoid skew.
      cutoff: Whether to prune ratings to those available when a note enters stabilization
        or gains CRH status.  None if prepareForTraining=False.

    Returns:
      pd.DataFrame containing all feature information with one row per note.
    """
    # Validate and normalize types
    assert ((prepareForTraining == False) & (cutoff is None)) | (
      prepareForTraining & (cutoff in {_MIN, _MAX})
    )
    notes[c.tweetIdKey] = notes[c.tweetIdKey].astype(pd.Int64Dtype())
    noteStatusHistory[c.createdAtMillisKey] = noteStatusHistory[c.createdAtMillisKey].astype(
      pd.Int64Dtype()
    )

    # Prep notes
    scoredNotes = self._get_notes(notes, noteStatusHistory)
    if prepareForTraining:
      assert cutoff is not None
      # Prune to recent notes
      scoredNotes = scoredNotes[scoredNotes[c.tweetIdKey] > _MIN_TWEET_ID]
      # Validate that stabilization timestamps are valid
      assert (
        noteStatusHistory[[c.noteIdKey, c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey]].merge(
          scoredNotes[[c.noteIdKey]]
        )[c.timestampMillisOfFirstNmrDueToMinStableCrhTimeKey]
        < 0
      ).sum() == 0
      # Compute flip labels
      labels = self._label_notes(noteStatusHistory.merge(scoredNotes[[c.noteIdKey]]))
      # Compute scoring cutoffs based on when the note entered and left stabilization
      scoringCutoff = self._compute_scoring_cutoff(
        noteStatusHistory.merge(scoredNotes[[c.noteIdKey]]),
        ratings.merge(scoredNotes[[c.noteIdKey]]),
        cutoff,
      )
      # Validate and merge data, effectively pruning to notes that have a label
      scoredNotes = scoredNotes.merge(scoringCutoff, on=c.noteIdKey)
      assert len(scoredNotes) == len(scoringCutoff)
      assert len(labels) == len(
        scoringCutoff.merge(labels)
      )  # labels should be a subset of scoringCutoff
      scoredNotes = scoredNotes.merge(labels)
      assert len(scoredNotes) == len(labels)
    totalScoredNotes = len(scoredNotes)

    # Prep ratings
    # Prune ratings to only include scored notes and other notes on the same post
    assert scoredNotes[c.tweetIdKey].min() > 0  # tweet should be set for all notes being scored
    adjacentNotes = notes[[c.noteIdKey, c.tweetIdKey]].merge(
      scoredNotes[[c.tweetIdKey]].drop_duplicates()
    )[[c.noteIdKey]]
    assert len(adjacentNotes) == adjacentNotes[c.noteIdKey].nunique()
    ratings = ratings.merge(adjacentNotes)
    assert len(ratings) == len(ratings[[c.noteIdKey, c.raterParticipantIdKey]].drop_duplicates())
    # Associate rater factors
    raterFactors = self._compute_rater_factors(prescoringRaterModelOutput)
    assert len(raterFactors) == raterFactors[c.raterParticipantIdKey].nunique()
    ratings = ratings.merge(raterFactors, how="left")
    # Generate rating datasets for self, peer misleading and peer non-misleading notes
    localRatings = self._prepare_local_ratings(ratings, scoredNotes[[c.noteIdKey]])
    peerRatings = self._prepare_peer_ratings(
      ratings,
      notes[[c.noteIdKey, c.tweetIdKey, c.classificationKey]],
      scoredNotes[[c.noteIdKey, c.tweetIdKey]],
    )
    peerMisleadingRatings = peerRatings[
      peerRatings[c.classificationKey] == c.notesSaysTweetIsMisleadingKey
    ]
    peerNonMisleadingRatings = peerRatings[
      peerRatings[c.classificationKey] == c.noteSaysTweetIsNotMisleadingKey
    ]
    if prepareForTraining:
      localRatings = self._apply_cutoff(localRatings, scoredNotes)
      peerMisleadingRatings = self._apply_cutoff(peerMisleadingRatings, scoredNotes)
      peerNonMisleadingRatings = self._apply_cutoff(peerNonMisleadingRatings, scoredNotes)

    # Extract featuers
    # Generate features that depend on self ratings only
    writingLatency = self._get_note_writing_latency(
      scoredNotes[[c.noteIdKey, _TWEET_CREATION_MILLIS, _NOTE_CREATION_MILLIS]]
    )
    scoredNotes = scoredNotes.merge(writingLatency, how="inner")
    noteAuthors = noteStatusHistory[[c.noteIdKey, c.noteAuthorParticipantIdKey]]
    scoredNotes = scoredNotes.merge(noteAuthors, how="inner")
    quickRatings = self._get_quick_rating_stats(
      scoredNotes[[c.noteIdKey, _NOTE_CREATION_MILLIS]], localRatings
    )
    scoredNotes = scoredNotes.merge(quickRatings, how="inner")
    burstRatings = self._get_burst_rating_stats(
      scoredNotes[[c.noteIdKey]], localRatings[[c.noteIdKey, c.createdAtMillisKey]]
    )
    scoredNotes = scoredNotes.merge(burstRatings, how="inner")
    recentRatings = self._get_recent_rating_stats(
      scoredNotes, localRatings[[c.noteIdKey, c.createdAtMillisKey]], prepareForTraining
    )
    scoredNotes = scoredNotes.merge(recentRatings, how="inner")
    peerNoteCounts = self._get_note_counts(
      scoredNotes, notes, noteStatusHistory, prepareForTraining
    )
    scoredNotes = scoredNotes.merge(peerNoteCounts, how="inner")
    # Generate features based on self and peer ratings
    for df, prefix in [
      (localRatings, _LOCAL),
      (peerMisleadingRatings, _PEER_MISLEADING),
      (peerNonMisleadingRatings, _PEER_NON_MISLEADING),
    ]:
      features = self._make_note_info_from_ratings(scoredNotes, df)
      overlapCols = (set(scoredNotes.columns) & set(features.columns)) - {c.noteIdKey}
      features = features[[col for col in features.columns if col not in overlapCols]]
      features = features.rename(
        columns={col: f"{prefix}_{col}" for col in features if col != c.noteIdKey}
      )
      scoredNotes = scoredNotes.merge(features, how="left")

    # Merge rating totals for debugging / info
    scoredNotes = scoredNotes.merge(
      localRatings[[c.noteIdKey]]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _LOCAL}),
      how="left",
    )
    scoredNotes = scoredNotes.merge(
      peerMisleadingRatings[[c.noteIdKey]]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _PEER_MISLEADING}),
      how="left",
    )
    scoredNotes = scoredNotes.merge(
      peerNonMisleadingRatings[[c.noteIdKey]]
      .value_counts()
      .to_frame()
      .reset_index(drop=False)
      .rename(columns={"count": _PEER_NON_MISLEADING}),
      how="left",
    )
    scoredNotes = scoredNotes.fillna(
      {_LOCAL: 0, _PEER_MISLEADING: 0, _PEER_NON_MISLEADING: 0}
    ).astype(
      {
        _LOCAL: pd.Int64Dtype(),
        _PEER_MISLEADING: pd.Int64Dtype(),
        _PEER_NON_MISLEADING: pd.Int64Dtype(),
      }
    )
    assert len(scoredNotes) == totalScoredNotes, f"{len(scoredNotes)} vs {totalScoredNotes}"
    return scoredNotes

  def _get_feature_pipeline(self, noteInfo: pd.DataFrame) -> Pipeline:
    # Begin with author pipeline
    columnPipes: List[Tuple[str, Any, Union[str, List[str]]]] = [
      (
        c.noteAuthorParticipantIdKey,
        Pipeline([("onehot", OneHotEncoder(handle_unknown="ignore"))]),
        [c.noteAuthorParticipantIdKey],
      )
    ]
    # Add pipelines for individual user helpfulness ratings
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      column = f"{prefix}_{_USER_HELPFULNESS_RATINGS}"
      pipe = Pipeline(
        [
          ("_set_to_list", FunctionTransformer(_set_to_list)),
          (
            "onehot",
            CountVectorizer(
              tokenizer=_identity, preprocessor=_identity, min_df=self._helpfulnessRaterMin
            ),
          ),
        ]
      )
      columnPipes.append((column, pipe, column))
    # Add pipelines for individual helpful tag ratings
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      column = f"{prefix}_{_USER_HELPFUL_TAGS}"
      pipe = Pipeline(
        [
          ("_set_to_list", FunctionTransformer(_set_to_list)),
          (
            "onehot",
            CountVectorizer(tokenizer=_identity, preprocessor=_identity, min_df=self._tagRaterMin),
          ),
          ("selection", SelectPercentile(chi2, percentile=self._helpfulTagPercentile)),
        ]
      )
      columnPipes.append((column, pipe, column))
    # Add pipelines for individual not-helpful tag ratings
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      column = f"{prefix}_{_USER_NOT_HELPFUL_TAGS}"
      pipe = Pipeline(
        [
          ("_set_to_list", FunctionTransformer(_set_to_list)),
          (
            "onehot",
            CountVectorizer(tokenizer=_identity, preprocessor=_identity, min_df=self._tagRaterMin),
          ),
          ("selection", SelectPercentile(chi2, percentile=self._notHelpfulTagPercentile)),
        ]
      )
      columnPipes.append((column, pipe, column))
    # Add pipelines for tag ratio columns
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      for tagset in [c.notHelpfulTagsTSVOrder, c.helpfulTagsTSVOrder]:
        for tag in tagset:
          column = f"{prefix}_{tag}"
          self._column_thresholds[column] = noteInfo[column].quantile(0.99)
          if noteInfo[column].min() == noteInfo[column].max():
            continue
          pipe = Pipeline(
            [
              ("reshape", FunctionTransformer(_reshape)),
              ("drop_constants", VarianceThreshold()),
              (
                "binize",
                KBinsDiscretizer(n_bins=self._tag_ratio_bins, encode="onehot", strategy="kmeans"),
              ),
            ]
          )
          columnPipes.append((column, pipe, column))
    # Add pipelines for rating counts across notes
    columns = []
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      for col in _BUCKET_COUNT_COLS:
        columns.append(f"{prefix}_{col}")
    assert noteInfo[columns].isna().sum().sum() == 0
    for col in columns:
      pipe = Pipeline(
        [
          ("log", FunctionTransformer(_feature_log)),
          (
            "binize",
            KBinsDiscretizer(n_bins=self._rating_count_bins, encode="onehot", strategy="kmeans"),
          ),
        ]
      )
      columnPipes.append((col, pipe, [col]))
    for degree in [2, 3]:
      pipe = Pipeline(
        [
          ("log", FunctionTransformer(_feature_log)),
          (
            "binize",
            KBinsDiscretizer(n_bins=self._rating_count_bins, encode="onehot", strategy="kmeans"),
          ),
          ("drop_rare", VarianceThreshold(threshold=0.001)),
          (
            "cross",
            PolynomialFeatures(degree=(degree, degree), interaction_only=True, include_bias=False),
          ),
          ("drop_rare_again", VarianceThreshold(threshold=0.001)),
        ]
      )
      columnPipes.append((f"cross_note_counts_degree_{degree}", pipe, columns))
    # Add pipelines for rating counts within notes
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      columns = []
      for col in _BUCKET_COUNT_COLS:
        columns.append(f"{prefix}_{col}")
      assert noteInfo[columns].isna().sum().sum() == 0
      pipe = Pipeline(
        [
          ("log", FunctionTransformer(_feature_log)),
          (
            "binize",
            KBinsDiscretizer(n_bins=self._rating_count_bins, encode="onehot", strategy="kmeans"),
          ),
          ("drop_rare", VarianceThreshold(threshold=0.001)),
          ("cross_0", PolynomialFeatures(degree=(2, 2), interaction_only=True, include_bias=False)),
          ("cross_1", PolynomialFeatures(degree=(2, 2), interaction_only=True, include_bias=False)),
          ("drop_rare_again", VarianceThreshold(threshold=0.001)),
        ]
      )
      columnPipes.append((f"{prefix}_cross_note_counts_degree_4", pipe, columns))
    # Add pipelines for rater factor stats
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      columns = []
      for col in _STATS_COLS:
        columns.append(f"{prefix}_{col}")
      pipe = Pipeline(
        [
          ("fill_nans_df", FunctionTransformer(_fill_na)),
          (
            "binize",
            KBinsDiscretizer(n_bins=self._factor_stats_bins, encode="onehot", strategy="kmeans"),
          ),
          ("cross", PolynomialFeatures(degree=(1, 2), interaction_only=True, include_bias=False)),
          ("drop_rare", VarianceThreshold(threshold=0.001)),
        ]
      )
      columnPipes.append((f"{prefix}_stat_cols", pipe, columns))
    # Add pipelines for rating bursts
    for colset in [_QUICK_RATING_COLS, _BURST_RATING_COLS, _RECENT_RATING_COLS]:
      for col in colset:
        assert noteInfo[col].isna().sum() == 0
        if "RATIO" in col:
          self._column_thresholds[col] = noteInfo[col].quantile(0.999)
          pipe = Pipeline(
            [
              (
                "binize",
                KBinsDiscretizer(n_bins=self._burst_bins, encode="onehot", strategy="kmeans"),
              ),
            ]
          )
        else:
          assert "TOTAL" in col
          self._column_thresholds[col] = noteInfo[col].quantile(0.999)
          pipe = Pipeline(
            [
              (
                "binize",
                KBinsDiscretizer(n_bins=self._burst_bins, encode="onehot", strategy="kmeans"),
              ),
            ]
          )
        columnPipes.append((col, pipe, [col]))
    # Add pipeline for note writing latency
    assert noteInfo[_NOTE_WRITING_LATENCY].isna().sum() == 0
    self._column_thresholds[_NOTE_WRITING_LATENCY] = noteInfo[_NOTE_WRITING_LATENCY].quantile(0.999)
    pipe = Pipeline(
      [
        ("binize", KBinsDiscretizer(n_bins=self._latency_bins, encode="onehot", strategy="kmeans")),
      ]
    )
    columnPipes.append((_NOTE_WRITING_LATENCY, pipe, [_NOTE_WRITING_LATENCY]))
    # Add columns for peer notes
    peerNoteCols = [
      _TOTAL_PEER_NOTES,
      _TOTAL_PEER_MISLEADING_NOTES,
      _TOTAL_PEER_NON_MISLEADING_NOTES,
      _TOTAL_PEER_CRH_NOTES,
      _TOTAL_PEER_STABILIZATION_NOTES,
    ]
    assert noteInfo[peerNoteCols].isna().sum().sum() == 0
    for col in peerNoteCols:
      self._column_thresholds[col] = noteInfo[col].quantile(0.9999)
      pipe = Pipeline(
        [
          (
            "binize",
            KBinsDiscretizer(n_bins=self._peer_note_count_bins, encode="onehot", strategy="kmeans"),
          ),
        ]
      )
      columnPipes.append((col, pipe, [col]))
    pipe = Pipeline(
      [
        ("log", FunctionTransformer(_feature_log)),
        (
          "binize",
          KBinsDiscretizer(n_bins=self._peer_note_count_bins, encode="onehot", strategy="kmeans"),
        ),
        ("cross", PolynomialFeatures(degree=(2, 2), interaction_only=True, include_bias=False)),
        ("drop_rare", VarianceThreshold(threshold=0.001)),
      ]
    )
    columnPipes.append(("peer_note_cross_degree_2", pipe, peerNoteCols))

    # Build and return column transformer
    return ColumnTransformer(columnPipes, verbose=True)

  def _get_model_pipeline(self, noteInfo: pd.DataFrame) -> Pipeline:
    """Return a full model pipeline including feature extraction and model.

    Note that the pipeline includes a VarianceThreshold filter between feature extraction
    and model training to drop any features that have zero variance (i.e. always have the
    same value).

    Returns:
      Pipeline containing feature extraction, variance threshold and model.
    """
    return Pipeline(
      [
        ("feature_extraction", self._get_feature_pipeline(noteInfo)),
        ("drop_constants", VarianceThreshold()),
        (
          "lr",
          LogisticRegression(
            penalty=self._penalty,
            C=self._C,
            max_iter=self._maxIter,
            verbose=self._verbose,
            class_weight=self._classWeight,
            solver=self._solver,
            tol=self._tol,
          ),
        ),
      ]
    )

  def _transform_note_info(self, noteInfo: pd.DataFrame) -> pd.DataFrame:
    noteInfo = noteInfo.copy()
    # Transform tag ratio columns
    for prefix in [_LOCAL, _PEER_MISLEADING, _PEER_NON_MISLEADING]:
      for tagset in [c.notHelpfulTagsTSVOrder, c.helpfulTagsTSVOrder]:
        for tag in tagset:
          column = f"{prefix}_{tag}"
          threshold = self._column_thresholds[column]
          noteInfo[column] = noteInfo[column].fillna(-0.25).clip(-0.25, threshold)
    # Transform for rating bursts
    for colset in [_QUICK_RATING_COLS, _BURST_RATING_COLS, _RECENT_RATING_COLS]:
      for column in colset:
        threshold = self._column_thresholds[column]
        if "RATIO" in column:
          noteInfo[column] = noteInfo[column].clip(0, threshold)
        else:
          assert "TOTAL" in column
          noteInfo[column] = np.log(1 + noteInfo[column].clip(0, threshold)) / np.log(2)
    # Transform for writing latency
    threshold = self._column_thresholds[_NOTE_WRITING_LATENCY]
    noteInfo[_NOTE_WRITING_LATENCY] = np.log(
      noteInfo[_NOTE_WRITING_LATENCY].clip(0, threshold)
    ) / np.log(2)
    # Transform for peer notes
    peerNoteCols = [
      _TOTAL_PEER_NOTES,
      _TOTAL_PEER_MISLEADING_NOTES,
      _TOTAL_PEER_NON_MISLEADING_NOTES,
      _TOTAL_PEER_CRH_NOTES,
      _TOTAL_PEER_STABILIZATION_NOTES,
    ]
    for column in peerNoteCols:
      threshold = self._column_thresholds[column]
      noteInfo[column] = np.log(1 + noteInfo[column].clip(0, threshold)) / np.log(2)
    return noteInfo

  def _get_label_vector(self, noteInfo: pd.DataFrame) -> np.array:
    """Extract a binary label vector from a noteInfo DataFrame."""
    return (noteInfo[LABEL] == FLIP).values.astype(np.int8)

  def _evaluate_model(
    self, noteInfo: pd.DataFrame, threshold: Optional[float] = None
  ) -> Tuple[float, float, float, float]:
    """Apply a pipeline to noteInfo and return the AUC, TPR, FPR and associated threshold.

    Assumes that the pipeline has already been fit.  If the threshold is specified as a
    command line argument then uses the provided threshold.  Otherwise, select the
    threshold to yield a 25% FPR.

    Args:
      noteInfo: pd.DataFrame containing raw feature information and labels.

    Returns:
      Tuple containing AUC, TPR and FPR.
    """
    assert self._pipeline is not None, "pipeline must be fit prior to evaluation"
    labels = self._get_label_vector(noteInfo)
    predictions = self._pipeline.decision_function(noteInfo)
    fpRates, tpRates, thresholds = roc_curve(labels, predictions)
    auc = area_under_curve(fpRates, tpRates)
    if threshold is None:
      idx = np.argmin(np.abs(fpRates - self._crhFpRate))
      threshold = thresholds[idx]
      fpr = fpRates[idx]
      tpr = tpRates[idx]
    else:
      tn, fp, fn, tp = confusion_matrix(
        labels, (predictions > threshold).astype(np.int8), labels=np.arange(2)
      ).ravel()
      fpr = fp / (fp + tn)
      tpr = tp / (tp + fn)
    logger.info(f"threshold={threshold}  tpr={tpr}  fpr={fpr}  auc={auc}")
    return (threshold, tpr, fpr, auc)

  def _convert_col_types(self, noteInfo: pd.DataFrame) -> pd.DataFrame:
    """Convert pandas types to numpy types.

    This conversion is necessary for scikit-learn compatibility.
    """
    for col, dtype in noteInfo.dtypes.to_dict().items():
      if isinstance(dtype, pd.Int64Dtype):
        assert noteInfo[col].isna().sum() == 0
        noteInfo[col] = noteInfo[col].astype(np.int64)
      if isinstance(dtype, pd.Float64Dtype):
        noteInfo[col] = noteInfo[col].astype(np.float64)
    return noteInfo

  def _profile_pipeline(self, pipe: Pipeline, noteInfo: pd.DataFrame) -> str:
    """Generate a numerical profile of each extracted feature.

    For each feature, we examine the dimensionality and sparsity of the feature.  For low
    dimensional features representing discretized continuous values, we also profile the
    size and boundaries of each bin.
    """
    # Generate feature matrix
    matrix = pipe.transform(noteInfo)
    # Profile matrix
    start = 0
    lines = []
    for name, transformer, _ in pipe.transformers_:
      if name == "remainder":
        continue
      end = start + len(transformer[-1].get_feature_names_out())
      total = int(matrix[:, start:end].sum())
      colMin = int(matrix[:, start:end].sum(axis=0).min())
      colMean = total / (end - start)
      colMax = int(matrix[:, start:end].sum(axis=0).max())
      rowMin = int(matrix[:, start:end].sum(axis=1).min())
      rowMean = total / (matrix.shape[0])
      rowMax = int(matrix[:, start:end].sum(axis=1).max())
      columns = [
        f"{name:<60}pos=[{start:8} {end:8} {end-start:8}]",
        f"total={total:9}",
        f"col=[{colMin:8} {colMean:8.1f} {colMax:8}]",
        f"row=[{rowMin:8} {rowMean:8.1f} {rowMax:8}]",
      ]
      if (end - start) <= 10:
        columns.append(f"{str(matrix[:, start:end].sum(axis=0).astype(np.int64)):<80}")
        columns.append(str(transformer[-1].bin_edges_[0].round(3).tolist()))
      lines.append("    ".join(columns))
      start = end
    return "\n".join(lines)

  def fit(
    self,
    notes: pd.DataFrame,
    ratings: pd.DataFrame,
    noteStatusHistory: pd.DataFrame,
    prescoringRaterModelOutput: pd.DataFrame,
  ) -> None:
    """Fit and evaluate a sklearn pipeline for predicting note status.

    Args:
      notes: pd.DataFrame
      ratings: pd.DataFrame
      noteStatusHistory: pd.DataFrame
      prescoringRaterModelOutput: pd.DataFrame

    Returns:
      sklearn pipeline covering containing full process of feature extraction, feature
      selection and prediction.
    """
    # Apply seed if necessary
    if self._seed is not None:
      logger.info(f"seeding pflip: {self._seed}")
      np.random.seed(self._seed)
    # Prepare datasets
    noteInfo = pd.concat(
      [
        self._prepare_note_info(
          notes,
          ratings,
          noteStatusHistory,
          prescoringRaterModelOutput,
          prepareForTraining=True,
          cutoff=_MIN,
        ),
        self._prepare_note_info(
          notes,
          ratings,
          noteStatusHistory,
          prescoringRaterModelOutput,
          prepareForTraining=True,
          cutoff=_MAX,
        ),
      ]
    )
    noteInfo = self._convert_col_types(noteInfo)
    noteInfo = noteInfo.sort_values(c.noteIdKey)
    if len(noteInfo) == 0:
      return
    logger.info(f"noteInfo summary: {get_df_fingerprint(noteInfo, [c.noteIdKey])}")
    # Dividing training data temporally provides a more accurate measurement, but would also
    # require excluding the newest data from training.
    trainDataFrame, validationDataFrame = train_test_split(noteInfo, train_size=self._trainSize)
    logger.info(f"pflip training data size: {len(trainDataFrame)}")
    logger.info(f"trainDataFrame summary: {get_df_fingerprint(trainDataFrame, [c.noteIdKey])}")
    logger.info(f"pflip validation data size: {len(validationDataFrame)}")
    logger.info(
      f"validationDataFrame summary: {get_df_fingerprint(validationDataFrame, [c.noteIdKey])}"
    )
    # Fit model
    self._pipeline = self._get_model_pipeline(trainDataFrame)
    trainDataFrame = self._transform_note_info(trainDataFrame)
    validationDataFrame = self._transform_note_info(validationDataFrame)
    self._pipeline.fit(trainDataFrame, self._get_label_vector(trainDataFrame))
    featureProfile = self._profile_pipeline(self._pipeline[0], trainDataFrame)
    logger.info(f"\ntraining feature matrix profile:\n{featureProfile}")
    # Evaluate model
    logger.info("Training Results:")
    threshold, _, _, _ = self._evaluate_model(trainDataFrame)
    self._predictionThreshold = threshold
    logger.info("Validation Results:")
    self._evaluate_model(validationDataFrame, threshold=threshold)

  def serialize(self) -> bytes:
    """Return a serialized version of the PFlipModel object.

    Note that since the pflip pipeline includes CountVectorizer instances that have
    functions as parameters, joblib must be called from within this same module to be
    able to serialize the functions.

    Returns:
      bytes containing a serialized PFlipModel object
    """
    buffer = BytesIO()
    joblib.dump(self, buffer)
    return buffer.getvalue()

  def predict(
    self,
    notes: pd.DataFrame,
    ratings: pd.DataFrame,
    noteStatusHistory: pd.DataFrame,
    prescoringRaterModelOutput: pd.DataFrame,
    maxBatchSize: int = 10000,
  ) -> pd.DataFrame:
    """Given input DataFrames, predict which notes will flip and lose CRH status.

    Args:
      notes: pd.DataFrame
      ratings: pd.DataFrame
      noteStatusHistory: pd.DataFrame
      prescoringRaterModelOutput: pd.DataFrame

    Returns:
      pd.DataFrame containing noteIds and predicted labels
    """
    assert self._pipeline is not None, "pipeline must be initialized prior to prediction"
    assert (
      self._predictionThreshold is not None
    ), "threshold must be initialized prior to prediction"
    # Build list of unique tweetIds
    tweetIds = notes[[c.tweetIdKey]].drop_duplicates()
    tweetIds = tweetIds[tweetIds[c.tweetIdKey] != "-1"]
    # Iterate through batches
    results = [pd.DataFrame({c.noteIdKey: [], LABEL: []})]
    start = 0
    while start < len(tweetIds):
      logger.info(f"processing prediction batch: {start}")
      # Note that all peer tweets must be included in the same batch for correct feature extraction
      tweetBatch = tweetIds.iloc[start : (start + maxBatchSize)]
      noteBatch = (
        notes[[c.noteIdKey, c.tweetIdKey]].merge(tweetBatch)[[c.noteIdKey]].drop_duplicates()
      )
      noteInfo = self._prepare_note_info(
        notes.merge(noteBatch),
        ratings.merge(noteBatch),
        noteStatusHistory.merge(noteBatch),
        prescoringRaterModelOutput,
        prepareForTraining=False,
        cutoff=None,
      )
      noteInfo = self._convert_col_types(noteInfo)
      noteInfo = self._transform_note_info(noteInfo)
      predictions = self._pipeline.decision_function(noteInfo)
      results.append(
        pd.DataFrame(
          {
            c.noteIdKey: noteInfo[c.noteIdKey],
            LABEL: [FLIP if p > self._predictionThreshold else CRH for p in predictions],
          }
        )
      )
      start += maxBatchSize
    return pd.concat(results)