def score_notes(
    self, noteStats: pd.DataFrame, currentLabels: pd.DataFrame, statusColumn: str
  ) -> Tuple[pd.DataFrame, Optional[pd.DataFrame]]:
    """Flip notes from NMR to CRH based on group models and subject to core/expansion model safeguards."""
    # Identify notes blocked from CRH status due to FR/CRNH status in core or expansion
    coreRejects = noteStats[c.coreRatingStatusKey].isin({c.firmReject, c.currentlyRatedNotHelpful})
    expansionRejects = noteStats[c.expansionRatingStatusKey].isin(
      {c.firmReject, c.currentlyRatedNotHelpful}
    )
    blocked = coreRejects | (noteStats[c.coreRatingStatusKey].isna() & expansionRejects)
    noteStats = noteStats[~blocked]
    # Generate the set of note status updates
    probationaryCRHOrNYHNotes = noteStats[
      (noteStats[c.groupRatingStatusKey].isin({c.currentlyRatedHelpful, c.needsYourHelp}))
      & (noteStats[c.modelingGroupKey] == self._groupNumber)
    ][[c.noteIdKey, c.groupRatingStatusKey]]
    # Identify notes which are currently NMR or NYH.
    currentNMROrNYHNotes = currentLabels[
      currentLabels[statusColumn].isin({c.needsMoreRatings, c.needsYourHelp})
    ][[c.noteIdKey, statusColumn]]
    # Identify candidate note status updates.  This requires pruning to notes that have a NMR->NYH, NMR->CRH
    # or NYH->CRH transition - in other words dropping NYH->NYH transitions.
    noteStatusUpdates = probationaryCRHOrNYHNotes.merge(
      currentNMROrNYHNotes, on=c.noteIdKey, how="inner"
    )
    noteStatusUpdates = noteStatusUpdates[
      (noteStatusUpdates[c.groupRatingStatusKey] != c.needsYourHelp)
      | (noteStatusUpdates[statusColumn] != c.needsYourHelp)
    ][[c.noteIdKey, c.groupRatingStatusKey]]
    # If necessary, identify notes which pass score bound checks for expansion and core models.
    # Apply min and max threhsolds to core and expansion intercepts
    noteStats = noteStats[[c.noteIdKey, c.coreNoteInterceptKey, c.expansionNoteInterceptKey]].copy()
    noteStats["core"] = noteStats[c.coreNoteInterceptKey] > self._minSafeguardThreshold
    if self._coreCrhThreshold is not None:
      noteStats["core"] = noteStats["core"] & (
        noteStats[c.coreNoteInterceptKey] < self._coreCrhThreshold
      )
    noteStats.loc[noteStats[c.coreNoteInterceptKey].isna(), "core"] = np.nan
    noteStats["expansion"] = noteStats[c.expansionNoteInterceptKey] > self._minSafeguardThreshold
    if self._expansionCrhThreshold is not None:
      noteStats["expansion"] = noteStats["expansion"] & (
        noteStats[c.expansionNoteInterceptKey] < self._expansionCrhThreshold
      )
    noteStats.loc[noteStats[c.expansionNoteInterceptKey].isna(), "expansion"] = np.nan

    # Prioritize core over expansion intercepts when available
    def _get_value(row):
      idx = row.first_valid_index()
      # If either core or expansion had an intercept then return whether it was in the valid
      # range.  If neither had an intercept, return False.  Preference is given to core due
      # to the ordering when selecting columns from noteStats below.
      if idx is None:
        return False
      elif row[idx] == 1.0:
        return True
      elif row[idx] == 0.0:
        return False
      else:
        assert False, f"unexpected value: {row[idx]}"

    with c.time_block("Get value apply for group model"):
      noteStats["actionable"] = noteStats[["core", "expansion"]].apply(_get_value, axis=1)

    # Filter set of note status updates to only include actionable notes
    actionableNotes = noteStats[noteStats["actionable"]][[c.noteIdKey]]
    noteStatusUpdates = noteStatusUpdates.merge(actionableNotes, on=c.noteIdKey, how="inner")

    # Set note status and return
    noteStatusUpdates = noteStatusUpdates.rename(columns={c.groupRatingStatusKey: statusColumn})

    return (noteStatusUpdates, None)
Flip notes from NMR to CRH based on group models and subject to core/expansion model safeguards.
sourcecode/scoring/scoring_rules.py