diff --git a/src/parent_child/preprocessing.py b/src/parent_child/preprocessing.py index bcce01d..720980b 100644 --- a/src/parent_child/preprocessing.py +++ b/src/parent_child/preprocessing.py @@ -870,10 +870,13 @@ def parent_child_preprocessing( one_to_one_df = update_group(one_to_one_df) one_to_one_df = standardize_provider_groups(one_to_one_df) - # Per-row fallback: when a row has no PROV_GROUP_* identifiers at all, - # promote PROV_OTHER_* into the PROV_GROUP_* slots so the downstream - # grouping-key / tier logic treats them as real identifiers instead of - # dropping the row to the filename-based fallback. + # Per-column fallback: when a specific PROV_GROUP_* field is blank, + # promote its PROV_OTHER_* counterpart independently. The previous + # all-or-nothing variant stranded TINs in PROV_OTHER_TIN whenever the + # row happened to have a name — e.g., multi-TIN amendments where the + # extractor put the contract's TIN list in PROV_OTHER_TIN but populated + # PROV_GROUP_NAME_FULL_cleaned from AARETE_DERIVED_PROVIDER_NAME. That + # broke cross-tier matching to the parent (Amedisys 270797096 case). other_cols = ("PROV_OTHER_TIN", "PROV_OTHER_NPI", "PROV_OTHER_NAME_FULL") if any(c in one_to_one_df.columns for c in other_cols): @@ -881,23 +884,19 @@ def parent_child_preprocessing( as_str = series.astype("string").fillna("").str.strip() return (as_str == "") | (as_str.str.lower() == "nan") - blank_group = ( - _is_blank(one_to_one_df["PROV_GROUP_TIN"]) - & _is_blank(one_to_one_df["PROV_GROUP_NPI"]) - & _is_blank(one_to_one_df["PROV_GROUP_NAME_FULL_cleaned"]) - ) + promoted_per_col = {} for src_col, dst_col in ( ("PROV_OTHER_TIN", "PROV_GROUP_TIN"), ("PROV_OTHER_NPI", "PROV_GROUP_NPI"), ("PROV_OTHER_NAME_FULL", "PROV_GROUP_NAME_FULL_cleaned"), ): - if src_col in one_to_one_df.columns: - one_to_one_df.loc[blank_group, dst_col] = one_to_one_df.loc[ - blank_group, src_col - ] - logging.info( - f"PROV_OTHER_* fallback applied to {int(blank_group.sum())} rows with no group identifiers" - ) + if src_col in one_to_one_df.columns and dst_col in one_to_one_df.columns: + blank = _is_blank(one_to_one_df[dst_col]) & ~_is_blank( + one_to_one_df[src_col] + ) + one_to_one_df.loc[blank, dst_col] = one_to_one_df.loc[blank, src_col] + promoted_per_col[dst_col] = int(blank.sum()) + logging.info(f"PROV_OTHER_* per-column fallback applied: {promoted_per_col}") # Determine which fields to consolidate based on available columns fields_to_consolidate = [] diff --git a/src/parent_child/qc.py b/src/parent_child/qc.py index c66c88f..9da44de 100644 --- a/src/parent_child/qc.py +++ b/src/parent_child/qc.py @@ -2158,12 +2158,12 @@ class ParentChildEngine: tmp.index, "child_rank_dest" ] - # Orphan ranks — bucketed by grouping_key so numbering resets per group. - # Standalone orphans (alone in their grouping_key) get no sequence suffix - # ("0."), so numbering doesn't balloon across unrelated rows. - # Multi-orphan groups keep the "0.." layout with seq - # resetting per group. When the grouping_key column isn't present (e.g. - # in isolated test harnesses), every orphan is treated as its own bucket. + # Orphan ranks — bucketed by the visible group (provider_group) so two + # orphans that share a Group label but live in different grouping_keys + # still get distinct rank strings. Previously bucketing on grouping_key + # let unrelated singleton orphans collide as "0..1" — see + # the Inspira Health Network case. Fall back to grouping_key when + # provider_group is absent (isolated test harnesses). pm = df["is_parent"] cm = df["assigned_parent_idx"].notna() orphan_mask = (~pm) & (~cm) @@ -2171,12 +2171,16 @@ class ParentChildEngine: if orphan_mask.any(): sort_cols = ["eff_date", "input_row_order"] tmp = df.loc[orphan_mask, sort_cols].copy() - if "grouping_key" in df.columns: + if "provider_group" in df.columns: + tmp["_gk"] = ( + df.loc[orphan_mask, "provider_group"].fillna("__none__").astype(str) + ) + elif "grouping_key" in df.columns: tmp["_gk"] = ( df.loc[orphan_mask, "grouping_key"].fillna("__none__").astype(str) ) else: - # No grouping info → every orphan is its own bucket (singleton). + # No grouping info → every orphan is its own bucket. tmp["_gk"] = [f"__row_{i}__" for i in range(len(tmp))] tmp["eff_sort"] = tmp["eff_date"].fillna(pd.Timestamp.max) tmp = tmp.sort_values( @@ -2184,7 +2188,6 @@ class ParentChildEngine: ) seq_int = tmp.groupby("_gk", sort=False).cumcount() + 1 - group_size = tmp.groupby("_gk", sort=False)["_gk"].transform("size") if "AARETE_DERIVED_AMENDMENT_NUM" in df.columns: amendment_str = df.loc[tmp.index, "AARETE_DERIVED_AMENDMENT_NUM"].apply( @@ -2194,10 +2197,7 @@ class ParentChildEngine: amendment_str = pd.Series(["0"] * len(tmp), index=tmp.index) seq_str = seq_int.astype("string") - multi_rank = ("0." + amendment_str + "." + seq_str).values - # Singletons always have order=1 (nothing to sequence against). - single_rank = ("0." + amendment_str + ".1").values - ranks = np.where(group_size.values == 1, single_rank, multi_rank) + ranks = ("0." + amendment_str + "." + seq_str).values df.loc[tmp.index, "orphan_rank_dest"] = ranks df.loc[tmp.index, "combined_rank_dest"] = ranks diff --git a/src/tests/test_parent_child.py b/src/tests/test_parent_child.py index bb8b6e0..61c3755 100644 --- a/src/tests/test_parent_child.py +++ b/src/tests/test_parent_child.py @@ -27,6 +27,8 @@ from src.parent_child.pipeline import ( assign_child_ranks, ) from src.parent_child.__main__ import extract_client_name +from src.parent_child.preprocessing import parent_child_preprocessing +from src.parent_child.column_mapper import ColumnMapper from src.parent_child.qc import ( to_string, to_datetime, @@ -972,6 +974,98 @@ class TestComputeRanksWithAmendmentNum(unittest.TestCase): # Standalone orphan: order fixed at 1, B=2. self.assertEqual(result.at[3, "orphan_rank_dest"], "0.2.1") + def test_orphans_in_same_provider_group_get_distinct_ranks(self): + """Two orphans in different grouping_keys but same provider_group must + get distinct rank strings. + + Regression: Inspira Health Network case — two singleton orphans across + different grouping_keys both ended up as "0.0.1", making them + indistinguishable to the downstream active-rate logic. + """ + engine = self._make_engine() + df = pd.DataFrame( + { + "is_parent": [False, False], + "parent_rank": [np.nan, np.nan], + "assigned_parent_idx": pd.array([pd.NA, pd.NA], dtype="Int64"), + "eff_date": pd.to_datetime(["2024-01-01", "2024-02-01"]), + "input_row_order": [0, 1], + "AARETE_DERIVED_AMENDMENT_NUM": [np.nan, np.nan], + "grouping_key": ["_tin:111", "_tin:222"], + "provider_group": ["Inspira Health Network", "Inspira Health Network"], + } + ) + df["parent_child_flag_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["parent_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["child_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["orphan_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["combined_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + + result = engine.compute_ranks(df) + + # Sorted by eff_date within the provider_group bucket: index 0, then 1. + self.assertEqual(result.at[0, "orphan_rank_dest"], "0.0.1") + self.assertEqual(result.at[1, "orphan_rank_dest"], "0.0.2") + self.assertNotEqual( + result.at[0, "orphan_rank_dest"], result.at[1, "orphan_rank_dest"] + ) + + def test_orphan_in_distinct_provider_groups_share_seq_1(self): + """Orphans in different provider_groups each get their own bucket, so + a leading sequence of 1 within each group is expected and OK.""" + engine = self._make_engine() + df = pd.DataFrame( + { + "is_parent": [False, False], + "parent_rank": [np.nan, np.nan], + "assigned_parent_idx": pd.array([pd.NA, pd.NA], dtype="Int64"), + "eff_date": pd.to_datetime(["2024-01-01", "2024-02-01"]), + "input_row_order": [0, 1], + "AARETE_DERIVED_AMENDMENT_NUM": [np.nan, np.nan], + "grouping_key": ["_tin:111", "_tin:222"], + "provider_group": ["Group A", "Group B"], + } + ) + df["parent_child_flag_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["parent_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["child_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["orphan_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["combined_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + + result = engine.compute_ranks(df) + + # Each is the sole orphan in its group → both correctly get seq 1. + # The Group column distinguishes them visually for the reviewer. + self.assertEqual(result.at[0, "orphan_rank_dest"], "0.0.1") + self.assertEqual(result.at[1, "orphan_rank_dest"], "0.0.1") + + def test_orphan_rank_falls_back_to_grouping_key_when_no_provider_group(self): + """When provider_group is absent (isolated test harness), bucketing + falls back to grouping_key so existing tests keep working.""" + engine = self._make_engine() + df = pd.DataFrame( + { + "is_parent": [False, False], + "parent_rank": [np.nan, np.nan], + "assigned_parent_idx": pd.array([pd.NA, pd.NA], dtype="Int64"), + "eff_date": pd.to_datetime(["2024-01-01", "2024-02-01"]), + "input_row_order": [0, 1], + "AARETE_DERIVED_AMENDMENT_NUM": [np.nan, np.nan], + "grouping_key": ["_tin:111", "_tin:111"], + } + ) + df["parent_child_flag_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["parent_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["child_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["orphan_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + df["combined_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string") + + result = engine.compute_ranks(df) + + # Same grouping_key, sequenced by date. + self.assertEqual(result.at[0, "orphan_rank_dest"], "0.0.1") + self.assertEqual(result.at[1, "orphan_rank_dest"], "0.0.2") + class TestNormalizeAmendmentVal(unittest.TestCase): """Unit tests for ParentChildEngine._normalize_amendment_val.""" @@ -1040,5 +1134,62 @@ class TestNormalizeAmendmentVal(unittest.TestCase): self.assertEqual(ParentChildEngine._normalize_amendment_val(-1), "-1") +class TestProvOtherPerColumnFallback(unittest.TestCase): + """parent_child_preprocessing promotes PROV_OTHER_* per-column. + + Regression: Hackensack/Amedisys amendments had PROV_GROUP_TIN blank but + PROV_GROUP_NAME_FULL populated (from AARETE_DERIVED_PROVIDER_NAME). The + previous all-or-nothing fallback skipped them, leaving their TINs + stranded in PROV_OTHER_TIN and breaking cross-tier match to the parent. + """ + + def _make_min_df(self): + # Minimal columns the preprocessing pipeline expects to see. + return pd.DataFrame( + { + "FILE_NAME": ["parent.pdf", "amendment.pdf"], + "CONTRACT_TITLE": ["Provider Services Agreement", "First Amendment"], + "AARETE_DERIVED_AMENDMENT_NUM": [None, 1], + "AARETE_DERIVED_EFFECTIVE_DT": ["2020-01-01", "2021-01-01"], + "AARETE_DERIVED_TERMINATION_DT": [None, None], + "PAYER_NAME": ["Clover", "Clover"], + "PAYER_STATE": ["NJ", "NJ"], + "PROV_GROUP_TIN": ['["270797096"]', None], + "PROV_GROUP_NPI": [None, None], + "PROV_GROUP_NAME_FULL": [ + "Home Health Services of Hackensack", + "Amedisys Holding", + ], + "PROV_OTHER_TIN": [None, '["208619703","270797096"]'], + "PROV_OTHER_NPI": [None, None], + "PROV_OTHER_NAME_FULL": [None, None], + "AARETE_DERIVED_PROVIDER_NAME": [ + "Home Health Services of Hackensack", + "Amedisys Holding", + ], + "AARETE_DERIVED_PAYER_NAME": ["Clover", "Clover"], + "LOB": ["Medicare Advantage", "Medicare Advantage"], + } + ) + + def test_blank_group_tin_picks_up_other_tin_even_when_name_present(self): + df_in = self._make_min_df() + col_mapper = ColumnMapper(df_in) + cleaned_df, _ = parent_child_preprocessing(df_in, col_mapper) + + # The amendment row's PROV_GROUP_TIN should now carry the OTHER_TIN + # value, despite PROV_GROUP_NAME_FULL_cleaned being non-blank + # ("Amedisys Holding"). The parent row's TIN should be untouched. + amendment_row = cleaned_df[ + cleaned_df["FILE_NAME"].str.contains("amendment", case=False, na=False) + ].iloc[0] + parent_row = cleaned_df[ + cleaned_df["FILE_NAME"].str.contains("parent", case=False, na=False) + ].iloc[0] + + self.assertIn("270797096", str(amendment_row["PROV_GROUP_TIN"])) + self.assertIn("270797096", str(parent_row["PROV_GROUP_TIN"])) + + if __name__ == "__main__": unittest.main()