Merge pull request #247 from twitter/jbaxter/2024_07_26

Allow scorers to firm_reject, finish split scorer refactoring & QOL++

Author: jbaxter

sourcecode/scoring/constants.py

@@ -36,7 +36,8 @@
 # Max flip rates
 prescoringAllUnlockedNotesMaxCrhChurn = 0.04
 finalUnlockedNotesWithNoNewRatingsMaxCrhChurn = 0.03
-finalNotesWithNewRatingsMaxCrhChurn = 0.40
+finalNotesWithNewRatingsMaxNewCrhChurn = 0.80
+finalNotesWithNewRatingsMaxOldCrhChurn = 0.25
 finalNotesThatJustFlippedStatusMaxCrhChurn = 1e8
 finalNotesThatFlippedRecentlyMaxCrhChurn = 1e8
 
@@ -236,6 +237,9 @@ def rater_factor_key(i):
 currentlyRatedHelpful = "CURRENTLY_RATED_HELPFUL"
 currentlyRatedNotHelpful = "CURRENTLY_RATED_NOT_HELPFUL"
 needsMoreRatings = "NEEDS_MORE_RATINGS"
+# FIRM_REJECT is set by individual scorers to indicate downstream scorers should not CRH
+# a note, but is never set as the finalRatingStatus of a note.
+firmReject = "FIRM_REJECT"
 
 # Boolean Note Status Labels
 currentlyRatedHelpfulBoolKey = "crhBool"
@@ -511,7 +515,7 @@ def rater_factor_key(i):
   (successfulRatingNeededToEarnIn, np.int64),
   (timestampOfLastStateChange, np.int64),
   (timestampOfLastEarnOut, np.double),  # double because nullable.
-  (modelingPopulationKey, str),
+  (modelingPopulationKey, "category"),
   (modelingGroupKey, np.float64),
   (numberOfTimesEarnedOutKey, np.int64),
 ]
@@ -801,6 +805,9 @@ class PrescoringMetaScorerOutput:
   globalIntercept: Optional[float]
   lowDiligenceGlobalIntercept: Optional[ReputationGlobalIntercept]
   tagFilteringThresholds: Optional[Dict[str, float]]  # tag => threshold
+  finalRoundNumRatings: Optional[int]
+  finalRoundNumNotes: Optional[int]
+  finalRoundNumUsers: Optional[int]
 
 
 @dataclass
@@ -886,5 +893,6 @@ class RescoringRuleID(Enum):
 @dataclass
 class NoteSubset:
   noteSet: Optional[set]
-  maxCrhChurnRate: float
+  maxNewCrhChurnRate: float
+  maxOldCrhChurnRate: float
   description: RescoringRuleID

sourcecode/scoring/matrix_factorization/matrix_factorization.py

@@ -65,7 +65,7 @@ def __init__(
         if logging:
           print(f"Using pos weight: {self._posWeight} with BCEWithLogitsLoss")
         self.criterion = torch.nn.BCEWithLogitsLoss(
-          pos_weight=torch.Tensor(np.array(self._posWeight)), reduction="none"
+          pos_weight=torch.FloatTensor(np.array(self._posWeight)), reduction="none"
         )
       else:
         if logging:
@@ -84,6 +84,9 @@ def __init__(
     self.trainModelData: Optional[ModelData] = None
     self.validateModelData: Optional[ModelData] = None
 
+    self._ratingPerNoteLossRatio: Optional[float] = None
+    self._ratingPerUserLossRatio: Optional[float] = None
+
   def get_final_train_error(self) -> Optional[float]:
     return self.train_errors[-1] if self.train_errors else None
 
@@ -214,7 +217,7 @@ def _initialize_parameters(
       if self._logging:
         print("initialized global intercept")
       self.mf_model.global_intercept = torch.nn.parameter.Parameter(
-        torch.ones(1, 1) * globalInterceptInit
+        torch.ones(1, 1, dtype=torch.float32) * globalInterceptInit
       )
 
   def _get_parameters_from_trained_model(self) -> Tuple[pd.DataFrame, pd.DataFrame]:
@@ -359,16 +362,55 @@ def _get_loss(self, epoch: Optional[int] = None):
     return loss
 
   def _get_reg_loss(self):
-    l2_reg_loss = torch.tensor(0.0).to(self.mf_model.device)
-    l2_reg_loss += self._userFactorLambda * (self.mf_model.user_factors.weight**2).mean()
-    l2_reg_loss += self._noteFactorLambda * (self.mf_model.note_factors.weight**2).mean()
-    l2_reg_loss += self._userInterceptLambda * (self.mf_model.user_intercepts.weight**2).mean()
-    l2_reg_loss += self._noteInterceptLambda * (self.mf_model.note_intercepts.weight**2).mean()
+    l2_reg_loss = torch.tensor(0.0, dtype=torch.float32).to(self.mf_model.device)
+
+    if self._ratingPerUserLossRatio is None:
+      l2_reg_loss += self._userFactorLambda * (self.mf_model.user_factors.weight**2).mean()
+      l2_reg_loss += self._userInterceptLambda * (self.mf_model.user_intercepts.weight**2).mean()
+    else:
+      simulatedNumberOfRatersForLoss = (
+        len(self.trainModelData.rating_labels) / self._ratingPerUserLossRatio
+      )
+      l2_reg_loss += (
+        self._userFactorLambda
+        * (self.mf_model.user_factors.weight**2).sum()
+        / simulatedNumberOfRatersForLoss
+      )
+      l2_reg_loss += (
+        self._userInterceptLambda
+        * (self.mf_model.user_intercepts.weight**2).sum()
+        / simulatedNumberOfRatersForLoss
+      )
+
+    if self._ratingPerNoteLossRatio is None:
+      l2_reg_loss += self._noteFactorLambda * (self.mf_model.note_factors.weight**2).mean()
+      l2_reg_loss += self._noteInterceptLambda * (self.mf_model.note_intercepts.weight**2).mean()
+      l2_reg_loss += (
+        self._diamondLambda
+        * (self.mf_model.note_factors.weight * self.mf_model.note_intercepts.weight).abs().mean()
+      )
+    else:
+      simulatedNumberOfNotesForLoss = (
+        len(self.trainModelData.rating_labels) / self._ratingPerNoteLossRatio
+      )
+      l2_reg_loss += (
+        self._noteFactorLambda
+        * (self.mf_model.note_factors.weight**2).sum()
+        / simulatedNumberOfNotesForLoss
+      )
+      l2_reg_loss += (
+        self._noteInterceptLambda
+        * (self.mf_model.note_intercepts.weight**2).sum()
+        / simulatedNumberOfNotesForLoss
+      )
+      l2_reg_loss += (
+        self._diamondLambda
+        * (self.mf_model.note_factors.weight * self.mf_model.note_intercepts.weight).abs().sum()
+        / simulatedNumberOfNotesForLoss
+      )
+
     l2_reg_loss += self._globalInterceptLambda * (self.mf_model.global_intercept**2).mean()
-    l2_reg_loss += (
-      self._diamondLambda
-      * (self.mf_model.note_factors.weight * self.mf_model.note_intercepts.weight).abs().mean()
-    )
+
     return l2_reg_loss
 
   def _fit_model(
@@ -434,10 +476,10 @@ def prepare_features_and_labels(
     rating_labels = torch.FloatTensor(ratingFeaturesAndLabels[self._labelCol].values).to(
       self.mf_model.device
     )
-    user_indexes = torch.LongTensor(ratingFeaturesAndLabels[Constants.raterIndexKey].values).to(
+    user_indexes = torch.IntTensor(ratingFeaturesAndLabels[Constants.raterIndexKey].values).to(
       self.mf_model.device
     )
-    note_indexes = torch.LongTensor(ratingFeaturesAndLabels[Constants.noteIndexKey].values).to(
+    note_indexes = torch.IntTensor(ratingFeaturesAndLabels[Constants.noteIndexKey].values).to(
       self.mf_model.device
     )
     self.modelData = ModelData(rating_labels, user_indexes, note_indexes)
@@ -451,6 +493,9 @@ def run_mf(
     specificNoteId: Optional[int] = None,
     validatePercent: Optional[float] = None,
     freezeRaterParameters: bool = False,
+    freezeGlobalParameters: bool = False,
+    ratingPerNoteLossRatio: Optional[float] = None,
+    ratingPerUserLossRatio: Optional[float] = None,
   ):
     """Train matrix factorization model.
 
@@ -469,13 +514,33 @@ def run_mf(
           raterParams: contains one row per rating, including raterId and learned rater parameters
           globalIntercept: learned global intercept parameter
     """
+    self._ratingPerNoteLossRatio = ratingPerNoteLossRatio
+    self._ratingPerUserLossRatio = ratingPerUserLossRatio
+
     self._initialize_note_and_rater_id_maps(ratings)
 
     self._create_mf_model(noteInit, userInit, globalInterceptInit)
     assert self.mf_model is not None
 
+    print(
+      f"Ratings per note in dataset: {len(ratings)/self.mf_model.note_factors.weight.data.shape[0]}"
+    )
+    print(
+      f"Ratings per user in dataset: {len(ratings)/self.mf_model.user_factors.weight.data.shape[0]}"
+    )
+    if ratingPerNoteLossRatio is not None:
+      print(
+        f"Correcting loss function to simulate rating per note loss ratio = {ratingPerNoteLossRatio}"
+      )
+    if ratingPerUserLossRatio is not None:
+      print(
+        f"Correcting loss function to simulate rating per user loss ratio = {ratingPerUserLossRatio}"
+      )
+
     if freezeRaterParameters:
       self.mf_model._freeze_parameters(set({"user"}))
+    if freezeGlobalParameters:
+      self.mf_model._freeze_parameters(set({"global"}))
     if specificNoteId is not None:
       self.mf_model.freeze_rater_and_global_parameters()
     self.prepare_features_and_labels(specificNoteId)

sourcecode/scoring/matrix_factorization/model.py

@@ -7,8 +7,8 @@
 @dataclass
 class ModelData:
   rating_labels: Optional[torch.FloatTensor]
-  user_indexes: Optional[torch.LongTensor]
-  note_indexes: Optional[torch.LongTensor]
+  user_indexes: Optional[torch.IntTensor]
+  note_indexes: Optional[torch.IntTensor]
 
 
 class BiasedMatrixFactorization(torch.nn.Module):
@@ -35,14 +35,14 @@ def __init__(
 
     self._logging = logging
 
-    self.user_factors = torch.nn.Embedding(n_users, n_factors, sparse=False)
-    self.note_factors = torch.nn.Embedding(n_notes, n_factors, sparse=False)
+    self.user_factors = torch.nn.Embedding(n_users, n_factors, sparse=False, dtype=torch.float32)
+    self.note_factors = torch.nn.Embedding(n_notes, n_factors, sparse=False, dtype=torch.float32)
 
-    self.user_intercepts = torch.nn.Embedding(n_users, 1, sparse=False)
-    self.note_intercepts = torch.nn.Embedding(n_notes, 1, sparse=False)
+    self.user_intercepts = torch.nn.Embedding(n_users, 1, sparse=False, dtype=torch.float32)
+    self.note_intercepts = torch.nn.Embedding(n_notes, 1, sparse=False, dtype=torch.float32)
 
     self.use_global_intercept = use_global_intercept
-    self.global_intercept = torch.nn.parameter.Parameter(torch.zeros(1, 1))
+    self.global_intercept = torch.nn.parameter.Parameter(torch.zeros(1, 1, dtype=torch.float32))
     torch.nn.init.xavier_uniform_(self.user_factors.weight)
     torch.nn.init.xavier_uniform_(self.note_factors.weight)
     self.user_intercepts.weight.data.fill_(0.0)

sourcecode/scoring/matrix_factorization/normalized_loss.py

@@ -125,9 +125,16 @@ def __init__(
     # Finalize weights
     weightMap = dict(
       ((rater, note), weight)
-      for (rater, note, weight) in ratings[[c.raterParticipantIdKey, c.noteIdKey, "weights"]].values
+      for (rater, note, weight) in zip(
+        ratings[c.raterParticipantIdKey], ratings[c.noteIdKey], ratings["weights"]
+      )
+    )
+    self.weights = torch.FloatTensor(
+      [
+        weightMap[(rater, note)]
+        for (rater, note) in zip(ratingOrder[c.raterParticipantIdKey], ratingOrder[c.noteIdKey])
+      ]
     )
-    self.weights = torch.tensor([weightMap[(rater, note)] for (rater, note) in ratingOrder.values])
     assert len(self.weights) == len(self.targets)
 
   def forward(self, pred):

sourcecode/scoring/matrix_factorization/pseudo_raters.py

@@ -3,6 +3,7 @@
 from .. import constants as c
 from .matrix_factorization import Constants as mf_c, MatrixFactorization
 
+import numpy as np
 import pandas as pd
 import torch
 
@@ -236,6 +237,13 @@ def _create_dataset_with_extreme_rating_on_each_note(self, ratingToAddWithoutNot
       extremeRatingsToAdd = pd.DataFrame(ratingsWithNoteIds).drop(
         [c.internalRaterInterceptKey, c.internalRaterFactor1Key], axis=1
       )
+      extremeRatingsToAdd[c.noteIdKey] = extremeRatingsToAdd[c.noteIdKey].astype(np.int64)
+      if isinstance(self.ratingFeaturesAndLabels[c.raterParticipantIdKey].dtype, pd.Int64Dtype):
+        # Only convert ID type from string to Int64 if is necessary to match existing IDs (which is
+        # expected when running in prod, but not always in unit tests or public data.)
+        extremeRatingsToAdd[c.raterParticipantIdKey] = extremeRatingsToAdd[
+          c.raterParticipantIdKey
+        ].astype(pd.Int64Dtype())
       ratingFeaturesAndLabelsWithExtremeRatings = pd.concat(
         [self.ratingFeaturesAndLabels, extremeRatingsToAdd]
       )

sourcecode/scoring/mf_base_scorer.py

@@ -184,6 +184,7 @@ def __init__(
     useReputation: bool = True,
     tagFilterPercentile: int = 95,
     incorrectFilterThreshold: float = 2.5,
+    firmRejectThreshold: Optional[float] = None,
   ):
     """Configure MatrixFactorizationScorer object.
 
@@ -256,6 +257,7 @@ def __init__(
     self._useReputation = useReputation
     self._tagFilterPercentile = tagFilterPercentile
     self._incorrectFilterThreshold = incorrectFilterThreshold
+    self._firmRejectThreshold = firmRejectThreshold
     mfArgs = dict(
       [
         pair
@@ -611,6 +613,7 @@ def _prescore_notes_and_users(
           incorrectFilterThreshold=self._incorrectFilterThreshold,
           tagFilterThresholds=None,
           finalRound=False,
+          firmRejectThreshold=self._firmRejectThreshold,
         )
       if self._saveIntermediateState:
         self.prescoringScoredNotes = scoredNotes
@@ -826,6 +829,7 @@ def _prescore_notes_and_users(
       incorrectFilterThreshold=self._incorrectFilterThreshold,
       finalRound=False,
       factorThreshold=self._factorThreshold,
+      firmRejectThreshold=self._firmRejectThreshold,
     )
 
     # Compute meta output
@@ -847,6 +851,9 @@ def _prescore_notes_and_users(
           + c.notHelpfulTagsTSVOrder
         ],
       ),
+      finalRoundNumRatings=len(finalRoundRatings),
+      finalRoundNumNotes=finalRoundRatings[c.noteIdKey].nunique(),
+      finalRoundNumUsers=finalRoundRatings[c.raterParticipantIdKey].nunique(),
     )
 
     # Compute user incorrect tag aggregates
@@ -950,6 +957,16 @@ def _score_notes_and_users(
       if self._saveIntermediateState:
         self.finalRoundRatings = finalRoundRatings
 
+    assert (
+      prescoringMetaScorerOutput.finalRoundNumNotes is not None
+    ), "Missing final round num notes"
+    assert (
+      prescoringMetaScorerOutput.finalRoundNumRatings is not None
+    ), "Missing final round num ratings"
+    assert (
+      prescoringMetaScorerOutput.finalRoundNumUsers is not None
+    ), "Missing final round num users"
+
     # Re-runs matrix factorization using only ratings given by helpful raters.
     with self.time_block("Final helpfulness-filtered MF"):
       noteParams, raterParams, globalBias = self._mfRanker.run_mf(
@@ -958,6 +975,9 @@ def _score_notes_and_users(
         userInit=prescoringRaterModelOutput,
         globalInterceptInit=prescoringMetaScorerOutput.globalIntercept,
         freezeRaterParameters=True,
+        freezeGlobalParameters=True,
+        ratingPerNoteLossRatio=prescoringMetaScorerOutput.finalRoundNumRatings
+        / prescoringMetaScorerOutput.finalRoundNumNotes,
       )
 
     if self._saveIntermediateState:
@@ -994,6 +1014,10 @@ def _score_notes_and_users(
         noteInitStateDiligence=prescoringNoteModelOutput,
         raterInitStateDiligence=prescoringRaterModelOutput,
         globalInterceptDiligence=prescoringMetaScorerOutput.lowDiligenceGlobalIntercept,
+        ratingsPerNoteLossRatio=prescoringMetaScorerOutput.finalRoundNumRatings
+        / prescoringMetaScorerOutput.finalRoundNumNotes,
+        ratingsPerUserLossRatio=prescoringMetaScorerOutput.finalRoundNumRatings
+        / prescoringMetaScorerOutput.finalRoundNumUsers,
       )
       print(f"diligenceNP cols: {diligenceNoteParams.columns}")
       noteParams = noteParams.merge(diligenceNoteParams, on=c.noteIdKey)
@@ -1033,6 +1057,7 @@ def _score_notes_and_users(
         lowDiligenceThreshold=self._lowDiligenceThreshold,
         finalRound=True,
         factorThreshold=self._factorThreshold,
+        firmRejectThreshold=self._firmRejectThreshold,
       )
       print(f"sn cols: {scoredNotes.columns}")
 

sourcecode/scoring/mf_core_scorer.py

@@ -12,6 +12,7 @@ def __init__(
     useStableInitialization: bool = True,
     saveIntermediateState: bool = False,
     threads: int = c.defaultNumThreads,
+    firmRejectThreshold: Optional[float] = 0.3,
   ) -> None:
     """Configure MFCoreScorer object.
 
@@ -29,6 +30,7 @@ def __init__(
       useStableInitialization=useStableInitialization,
       saveIntermediateState=saveIntermediateState,
       threads=threads,
+      firmRejectThreshold=firmRejectThreshold,
     )
 
   def get_name(self):

sourcecode/scoring/mf_expansion_scorer.py

@@ -14,6 +14,7 @@ def __init__(
     useStableInitialization: bool = True,
     saveIntermediateState: bool = False,
     threads: int = c.defaultNumThreads,
+    firmRejectThreshold: Optional[float] = 0.3,
   ) -> None:
     """Configure MFExpansionScorer object.
 
@@ -30,6 +31,7 @@ def __init__(
       useStableInitialization=useStableInitialization,
       saveIntermediateState=saveIntermediateState,
       threads=threads,
+      firmRejectThreshold=firmRejectThreshold,
     )
 
   def get_name(self):