def _prescore_notes_and_users(
    self,
    ratings: pd.DataFrame,
    noteStatusHistory: pd.DataFrame,
    userEnrollmentRaw: pd.DataFrame,
  ) -> Tuple[pd.DataFrame, pd.DataFrame, c.PrescoringMetaScorerOutput]:
    """
    Fit initial matrix factorization model(s) on the ratings data in order to generate
    initial note and rater parameters (and rater helpfulness scores) that are passed to
    the final scoring step. The final scoring step will be able to run faster by
    using the rater helpfulness scores computed here, and also intializing its parameters
    with these parameters.


    Args:
        ratings (pd.DataFrame)
        noteStatusHistory (pd.DataFrame)
        userEnrollmentRaw (pd.DataFrame)

    Returns:
        Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
          noteParamsUnfiltered (pd.DataFrame)
          raterParamsUnfiltered (pd.DataFrame)
          helpfulnessScores (pd.DataFrame)
    """
    if self._seed is not None:
      logger.info(f"seeding with {self._seed}")
      torch.manual_seed(self._seed)

    # Removes ratings where either (1) the note did not receive enough ratings, or
    # (2) the rater did not rate enough notes.
    logger.info(
      f"ratings summary {self.get_name()}: {get_df_fingerprint(ratings, [c.noteIdKey, c.raterParticipantIdKey])}"
    )

    with self.time_block("Prepare ratings"):
      ratingsForTraining = self._prepare_data_for_scoring(
        ratings[
          [
            c.noteIdKey,
            c.raterParticipantIdKey,
            c.helpfulNumKey,
            c.createdAtMillisKey,
            c.helpfulnessLevelKey,
            c.notHelpfulIncorrectTagKey,
            c.notHelpfulIrrelevantSourcesTagKey,
            c.notHelpfulSourcesMissingOrUnreliableTagKey,
            c.notHelpfulSpamHarassmentOrAbuseTagKey,
            c.notHelpfulOtherTagKey,
          ]
        ]
      )
    logger.info(
      f"ratingsForTraining summary {self.get_name()}: {get_df_fingerprint(ratingsForTraining, [c.noteIdKey, c.raterParticipantIdKey])}"
    )
    logger.info(
      f"noteStatusHistory summary {self.get_name()}: {get_df_fingerprint(noteStatusHistory, [c.noteIdKey])}"
    )
    logger.info(
      f"userEnrollmentRaw summary {self.get_name()}: {get_df_fingerprint(userEnrollmentRaw, [c.participantIdKey])}"
    )
    if self._saveIntermediateState:
      self.ratingsForTraining = ratingsForTraining

    # TODO: Save parameters from this first run in note_model_output next time we add extra fields to model output TSV.
    with self.time_block("First MF/stable init"):
      (
        noteParamsUnfiltered,
        raterParamsUnfiltered,
        globalBias,
      ) = self._run_stable_matrix_factorization(
        ratingsForTraining[[c.noteIdKey, c.raterParticipantIdKey, c.helpfulNumKey]],
        userEnrollmentRaw[[c.participantIdKey, c.modelingGroupKey]],
      )
    if self._saveIntermediateState:
      self.noteParamsUnfiltered = noteParamsUnfiltered
      self.raterParamsUnfiltered = raterParamsUnfiltered
      self.globalBias = globalBias
    self.assert_train_error_is_below_threshold(
      ratingsForTraining[[c.noteIdKey]], self._maxFirstMFTrainError
    )

    # If reputation is disabled, generate final intercepts, factors and note status
    # based on the first round scoring results.  Disabling reputation can be desirable
    # in situations where the overall volume of ratings is lower (e.g. topic models).
    if not self._useReputation:
      assert "MFTopicScorer" in self.get_name(), f"Unexpected scorer: {self.get_name()}"
      logger.info(f"Skipping rep-filtering in prescoring for {self.get_name()}")
      helpfulnessScores = raterParamsUnfiltered[[c.raterParticipantIdKey]]
      helpfulnessScores[
        [
          c.crhCrnhRatioDifferenceKey,
          c.meanNoteScoreKey,
          c.raterAgreeRatioKey,
          c.aboveHelpfulnessThresholdKey,
          c.successfulRatingNotHelpfulCount,
          c.successfulRatingHelpfulCount,
          c.unsuccessfulRatingNotHelpfulCount,
          c.unsuccessfulRatingHelpfulCount,
          c.totalHelpfulHarassmentRatingsPenaltyKey,
          c.raterAgreeRatioWithHarassmentAbusePenaltyKey,
        ]
      ] = np.nan
      noteParams = noteParamsUnfiltered
      raterParams = raterParamsUnfiltered
      # TODO: delete after we run prescoring diligence properly
      # diligenceGlobalIntercept = None
      finalRoundRatings = ratingsForTraining
      harassmentAbuseNoteParams = noteParamsUnfiltered[[c.noteIdKey]]
      harassmentAbuseNoteParams[[c.harassmentNoteInterceptKey, c.harassmentNoteFactor1Key]] = np.nan
    else:
      assert "MFTopicScorer" not in self.get_name(), f"Unexpected scorer: {self.get_name()}"
      logger.info(f"Performing rep-filtering for {self.get_name()}")
      # Get a dataframe of scored notes based on the algorithm results above
      with self.time_block("Compute scored notes"):
        scoredNotes = note_ratings.compute_scored_notes(
          ratings[
            [
              c.noteIdKey,
              c.raterParticipantIdKey,
              c.helpfulnessLevelKey,
              c.createdAtMillisKey,
            ]
            + c.notHelpfulTagsTSVOrder
            + c.helpfulTagsTSVOrder
          ],
          keep_columns(
            noteParamsUnfiltered,
            [
              c.noteIdKey,
              c.internalNoteInterceptKey,
              c.internalNoteFactor1Key,
            ]
            + c.noteParameterUncertaintyTSVColumns,
          ),
          raterParamsUnfiltered[
            [
              c.raterParticipantIdKey,
              c.internalRaterFactor1Key,
            ]
          ],
          noteStatusHistory[
            [
              c.noteIdKey,
              c.createdAtMillisKey,
              c.noteAuthorParticipantIdKey,
              c.classificationKey,
              c.currentLabelKey,
              c.lockedStatusKey,
            ]
          ],
          minRatingsNeeded=self._minRatingsNeeded,
          crhThreshold=self._crhThreshold,
          crnhThresholdIntercept=self._crnhThresholdIntercept,
          crnhThresholdNoteFactorMultiplier=self._crnhThresholdNoteFactorMultiplier,
          crnhThresholdNMIntercept=self._crnhThresholdNMIntercept,
          crnhThresholdUCBIntercept=self._crnhThresholdUCBIntercept,
          crhSuperThreshold=self._crhSuperThreshold,
          inertiaDelta=self._inertiaDelta,
          incorrectFilterThreshold=self._incorrectFilterThreshold,
          tagFilterThresholds=None,
          finalRound=False,
          firmRejectThreshold=self._firmRejectThreshold,
        )
      if self._saveIntermediateState:
        self.prescoringScoredNotes = scoredNotes

      # Determine "valid" ratings
      with self.time_block("Compute valid ratings"):
        validRatings = note_ratings.get_valid_ratings(
          ratings[
            [
              c.noteIdKey,
              c.raterParticipantIdKey,
              c.helpfulNumKey,
              c.createdAtMillisKey,
            ]
          ],
          noteStatusHistory[
            [
              c.noteIdKey,
              c.createdAtMillisKey,
              c.timestampMillisOfNoteMostRecentNonNMRLabelKey,
            ]
          ],
          scoredNotes[
            [
              c.noteIdKey,
              c.currentlyRatedHelpfulBoolKey,
              c.currentlyRatedNotHelpfulBoolKey,
              c.awaitingMoreRatingsBoolKey,
            ]
          ],
        )
      if self._saveIntermediateState:
        self.validRatings = validRatings

      # Assigns contributor (author & rater) helpfulness bit based on (1) performance
      # authoring and reviewing previous and current notes.
      with self.time_block("Helpfulness scores pre-harassment "):
        helpfulnessScoresPreHarassmentFilter = (
          helpfulness_scores.compute_general_helpfulness_scores(
            scoredNotes[
              [
                c.noteAuthorParticipantIdKey,
                c.currentlyRatedHelpfulBoolKey,
                c.currentlyRatedNotHelpfulBoolKey,
                c.internalNoteInterceptKey,
              ]
            ],
            validRatings[
              [
                c.raterParticipantIdKey,
                c.ratingAgreesWithNoteStatusKey,
                c.ratingCountKey,
                c.successfulRatingNotHelpfulCount,
                c.successfulRatingHelpfulCount,
                c.unsuccessfulRatingNotHelpfulCount,
                c.unsuccessfulRatingHelpfulCount,
              ]
            ],
            self._minMeanNoteScore,
            self._minCRHVsCRNHRatio,
            self._minRaterAgreeRatio,
            ratingsForTraining[[c.noteIdKey, c.raterParticipantIdKey, c.helpfulNumKey]],
          )
        )
      if self._saveIntermediateState:
        self.prescoringHelpfulnessScores = helpfulnessScoresPreHarassmentFilter

      # Filters ratings matrix to include only rows (ratings) where the rater was
      # considered helpful.
      with self.time_block("Filtering by helpfulness score"):
        ratingsHelpfulnessScoreFilteredPreHarassmentFilter = (
          helpfulness_scores.filter_ratings_by_helpfulness_scores(
            ratingsForTraining[
              [
                c.noteIdKey,
                c.raterParticipantIdKey,
                c.notHelpfulSpamHarassmentOrAbuseTagKey,
                c.createdAtMillisKey,
                c.helpfulnessLevelKey,
                c.notHelpfulOtherTagKey,
              ]
            ],
            helpfulnessScoresPreHarassmentFilter,
          )
        )

      if self._saveIntermediateState:
        self.ratingsHelpfulnessScoreFilteredPreHarassmentFilter = (
          ratingsHelpfulnessScoreFilteredPreHarassmentFilter
        )

      with self.time_block("Harassment tag consensus"):
        harassmentAbuseNoteParams, _, _ = tag_consensus.train_tag_model(
          ratings=ratingsHelpfulnessScoreFilteredPreHarassmentFilter,
          tag=c.notHelpfulSpamHarassmentOrAbuseTagKey,
          helpfulModelNoteParams=noteParamsUnfiltered[
            [c.noteIdKey, c.internalNoteInterceptKey, c.internalNoteFactor1Key]
          ],
          helpfulModelRaterParams=raterParamsUnfiltered[
            [
              c.raterParticipantIdKey,
              c.internalRaterInterceptKey,
              c.internalRaterFactor1Key,
            ]
          ],
          name="harassment",
        )
      if not self._saveIntermediateState:
        del ratingsHelpfulnessScoreFilteredPreHarassmentFilter
        gc.collect()

      # Assigns contributor (author & rater) helpfulness bit based on (1) performance
      # authoring and reviewing previous and current notes, and (2) including an extra
      # penalty for rating a harassment/abuse note as helpful.
      with self.time_block("Helpfulness scores post-harassment"):
        helpfulnessScores = helpfulness_scores.compute_general_helpfulness_scores(
          scoredNotes[
            [
              c.noteAuthorParticipantIdKey,
              c.currentlyRatedHelpfulBoolKey,
              c.currentlyRatedNotHelpfulBoolKey,
              c.internalNoteInterceptKey,
            ]
          ],
          validRatings[
            [
              c.raterParticipantIdKey,
              c.ratingAgreesWithNoteStatusKey,
              c.ratingCountKey,
              c.successfulRatingNotHelpfulCount,
              c.successfulRatingHelpfulCount,
              c.unsuccessfulRatingNotHelpfulCount,
              c.unsuccessfulRatingHelpfulCount,
            ]
          ],
          self._minMeanNoteScore,
          self._minCRHVsCRNHRatio,
          self._minRaterAgreeRatio,
          ratings=ratingsForTraining[[c.noteIdKey, c.raterParticipantIdKey, c.helpfulNumKey]],
          tagConsensusHarassmentAbuseNotes=harassmentAbuseNoteParams,
          tagConsensusHarassmentHelpfulRatingPenalty=self.tagConsensusHarassmentHelpfulRatingPenalty,
          multiplyPenaltyByHarassmentScore=self.multiplyPenaltyByHarassmentScore,
          minimumHarassmentScoreToPenalize=self.minimumHarassmentScoreToPenalize,
        )
      if not self._saveIntermediateState:
        del validRatings
        gc.collect()
      if self._saveIntermediateState:
        self.helpfulnessScores = helpfulnessScores

      ## One extra final round!
      # Filter ratings based on prev helpfulness scores
      with c.time_block("Final round MF"):
        finalRoundRatings = helpfulness_scores.filter_ratings_by_helpfulness_scores(
          ratingsForTraining[
            [
              c.noteIdKey,
              c.raterParticipantIdKey,
              c.helpfulNumKey,
              c.notHelpfulIncorrectTagKey,
              c.notHelpfulSourcesMissingOrUnreliableTagKey,
              c.notHelpfulIrrelevantSourcesTagKey,
            ]
          ],
          helpfulnessScores[[c.raterParticipantIdKey, c.aboveHelpfulnessThresholdKey]],
        )
        noteParams, raterParams, globalBias = self._mfRanker.run_mf(
          ratings=finalRoundRatings[[c.noteIdKey, c.raterParticipantIdKey, c.helpfulNumKey]],
          noteInit=noteParamsUnfiltered[
            [c.noteIdKey, c.internalNoteInterceptKey, c.internalNoteFactor1Key]
          ],
          userInit=raterParamsUnfiltered[
            [
              c.raterParticipantIdKey,
              c.internalRaterInterceptKey,
              c.internalRaterFactor1Key,
            ]
          ],
          run_name=f"{self.get_name()}/final_round_mf",
        )

    # Run Diligence MF Prescoring, based on the final MF
    with self.time_block("Low Diligence MF"):
      # Initialize diligence rater factors with final round helpful MF rater factor
      raterParamsDiligenceInit = raterParams[
        [c.raterParticipantIdKey, c.internalRaterFactor1Key]
      ].rename({c.internalRaterFactor1Key: c.lowDiligenceRaterFactor1Key}, axis=1)
      logger.info(
        f"In {self.get_name()} prescoring, about to call diligence with {len(finalRoundRatings)} final round ratings."
      )
      (
        diligenceNoteParams,
        diligenceRaterParams,
        diligenceGlobalIntercept,
      ) = fit_low_diligence_model_prescoring(
        finalRoundRatings[
          [
            c.noteIdKey,
            c.raterParticipantIdKey,
            c.notHelpfulIncorrectTagKey,
            c.notHelpfulSourcesMissingOrUnreliableTagKey,
            c.notHelpfulIrrelevantSourcesTagKey,
          ]
        ],
        raterInitStateDiligence=raterParamsDiligenceInit,
      )
      noteParams = noteParams.merge(diligenceNoteParams, on=c.noteIdKey)

      noteParams = noteParams.merge(
        harassmentAbuseNoteParams[
          [c.noteIdKey, c.harassmentNoteInterceptKey, c.harassmentNoteFactor1Key]
        ],
        on=c.noteIdKey,
        how="left",
      )
      raterParams = raterParams.merge(diligenceRaterParams, on=c.raterParticipantIdKey)

    # Compute scored notes -- currently not returned; only used for downstream computation.
    scoredNotes = note_ratings.compute_scored_notes(
      ratings[
        [
          c.noteIdKey,
          c.raterParticipantIdKey,
          c.helpfulnessLevelKey,
          c.createdAtMillisKey,
        ]
        + c.notHelpfulTagsTSVOrder
        + c.helpfulTagsTSVOrder
      ],
      keep_columns(
        noteParamsUnfiltered,
        [
          c.noteIdKey,
          c.internalNoteInterceptKey,
          c.internalNoteFactor1Key,
        ]
        + c.noteParameterUncertaintyTSVColumns,
      ),
      raterParamsUnfiltered[
        [
          c.raterParticipantIdKey,
          c.internalRaterFactor1Key,
        ]
      ],
      noteStatusHistory[
        [
          c.noteIdKey,
          c.createdAtMillisKey,
          c.noteAuthorParticipantIdKey,
          c.classificationKey,
          c.currentLabelKey,
          c.lockedStatusKey,
        ]
      ],
      minRatingsNeeded=self._minRatingsNeeded,
      crhThreshold=self._crhThreshold,
      crnhThresholdIntercept=self._crnhThresholdIntercept,
      crnhThresholdNoteFactorMultiplier=self._crnhThresholdNoteFactorMultiplier,
      crnhThresholdNMIntercept=self._crnhThresholdNMIntercept,
      crnhThresholdUCBIntercept=self._crnhThresholdUCBIntercept,
      crhSuperThreshold=self._crhSuperThreshold,
      inertiaDelta=self._inertiaDelta,
      tagFilterThresholds=None,
      incorrectFilterThreshold=self._incorrectFilterThreshold,
      finalRound=False,
      factorThreshold=self._factorThreshold,
      firmRejectThreshold=self._firmRejectThreshold,
    )

    # Compute meta output
    metaOutput = c.PrescoringMetaScorerOutput(
      globalIntercept=globalBias,
      lowDiligenceGlobalIntercept=diligenceGlobalIntercept,
      tagFilteringThresholds=self.compute_tag_thresholds_for_percentile(
        scoredNotes=noteParams[[c.noteIdKey, c.internalNoteFactor1Key]].merge(
          scoredNotes[[c.noteIdKey, c.currentlyRatedHelpfulBoolKey]],
          on=c.noteIdKey,
          suffixes=("", "_dup"),
        ),
        raterParams=raterParams[[c.raterParticipantIdKey, c.internalRaterFactor1Key]],
        ratings=ratings[
          [
            c.noteIdKey,
            c.raterParticipantIdKey,
          ]
          + c.notHelpfulTagsTSVOrder
        ],
      ),
      finalRoundNumRatings=len(finalRoundRatings),
      finalRoundNumNotes=finalRoundRatings[c.noteIdKey].nunique(),
      finalRoundNumUsers=finalRoundRatings[c.raterParticipantIdKey].nunique(),
    )

    # Compute user incorrect tag aggregates
    userIncorrectTagUsageDf = get_user_incorrect_ratio(
      ratings[
        [
          c.noteIdKey,
          c.raterParticipantIdKey,
        ]
        + c.notHelpfulTagsTSVOrder
      ]
    )

    raterModelOutput = raterParams.merge(
      helpfulnessScores[
        [
          c.raterParticipantIdKey,
          c.crhCrnhRatioDifferenceKey,
          c.meanNoteScoreKey,
          c.raterAgreeRatioKey,
          c.aboveHelpfulnessThresholdKey,
          c.successfulRatingHelpfulCount,
          c.successfulRatingNotHelpfulCount,
          c.unsuccessfulRatingHelpfulCount,
          c.unsuccessfulRatingNotHelpfulCount,
          c.totalHelpfulHarassmentRatingsPenaltyKey,
          c.raterAgreeRatioWithHarassmentAbusePenaltyKey,
        ]
      ],
      on=c.raterParticipantIdKey,
      how="outer",
    ).merge(
      userIncorrectTagUsageDf,
      on=c.raterParticipantIdKey,
      how="left",
      unsafeAllowed={c.totalRatingsMadeByRaterKey, c.incorrectTagRatingsMadeByRaterKey},
    )

    noteModelOutput = noteParams
    # Returning should remove references to these, but manually trigger GC just to reclaim
    # resources as soon as possible.
    del ratings
    del ratingsForTraining
    del finalRoundRatings
    gc.collect()
    return noteModelOutput, raterModelOutput, metaOutput

    Fit initial matrix factorization model(s) on the ratings data in order to generate
    initial note and rater parameters (and rater helpfulness scores) that are passed to
    the final scoring step. The final scoring step will be able to run faster by
    using the rater helpfulness scores computed here, and also intializing its parameters
    with these parameters.


    Args:
        ratings (pd.DataFrame)
        noteStatusHistory (pd.DataFrame)
        userEnrollmentRaw (pd.DataFrame)

    Returns:
        Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
          noteParamsUnfiltered (pd.DataFrame)
          raterParamsUnfiltered (pd.DataFrame)
          helpfulnessScores (pd.DataFrame)
    
sourcecode/scoring/mf_base_scorer.py