BFD-4267: Use negative values for synthetic data + some logging fixes

.github/workflows/ci-pipeline-synthetic.yml

@@ -24,12 +24,24 @@ jobs:
       with:
         node-version: '22'
     - name: Install dependencies
-      run: |
-        npm install -g fsh-sushi
+      run: npm install -g fsh-sushi
+    - name: Build sushi
+      run:  |
         cd ${{ github.workspace }}/apps/bfd-model-idr
         sushi build ./sushi
+    - name: Ensure synthetic data pipeline runs
+      run: |
+        cd ${{ github.workspace }}/apps/bfd-model-idr
         uv run patient_generator.py
         uv run claims_generator.py --benes ./out/SYNTHETIC_BENE_HSTRY.csv
         cd ../bfd-pipeline/bfd-pipeline-idr
         ./run-db.sh
         uv run pipeline.py local
+    - name: Ensure synthetic data pipeline runs with sandbox benes
+      run: |
+        cd ${{ github.workspace }}/apps/bfd-model-idr
+        uv run patient_generator.py --benes ./Benes-Sandbox.csv
+        uv run claims_generator.py --benes ./out/SYNTHETIC_BENE_HSTRY.csv
+        cd ../bfd-pipeline/bfd-pipeline-idr
+        ./run-db.sh
+        uv run pipeline.py local

apps/bfd-model-idr/Benes-Sandbox.csv

@@ -19998,4 +19998,4 @@ bene_id,BENE_MBI_ID,BENE_LAST_NAME,BENE_1ST_NAME,BENE_MIDL_NAME,BENE_BRTH_DT,BEN
 ,1S00E00FY96,,,,,,,
 ,1S00E00FY97,,,,,,,
 ,1S00E00FY98,,,,,,,
-,1S00E00FY99,,,,,,,
+,1S00E00FY99,,,,,,,

apps/bfd-model-idr/README.md

@@ -85,6 +85,15 @@ To utilize it:
 uv run patient_generator.py
 ```
 
+Or, to load the specific MBIs used in the sandbox env:
+
+```sh
+uv run patient_generator.py --benes Benes-Sandbox.csv
+```
+
+**NOTE**: the bene_id column in this file is to reference the bene_id field used in V1/V2.
+It's not used for sample data generation here.
+
 The script supports several options:
 - `--benes <csv_file>`: Use a CSV file containing beneficiary data to populate fields. Empty fields will be filled with random data.
 - `--claims`: Automatically generate claims after patient generation using the generated SYNTHETIC_BENE_HSTRY.csv file.

apps/bfd-model-idr/claims_generator.py

@@ -87,9 +87,6 @@ def save_output_files(
     shutil.copy("sample-data/SYNTHETIC_CLM_ANSI_SGNTR.csv", "out/SYNTHETIC_CLM_ANSI_SGNTR.csv")
 
 
-claims_to_generate_per_person = 5
-
-
 fiss_clm_type_cds = [
     1011,
     1041,
@@ -408,7 +405,7 @@ def gen_claim(bene_sk="-1", min_date="2018-01-01", max_date=str(now)):
     clm_dt_sgntr = {}
     clm_dt_sgntr["CLM_DT_SGNTR_SK"] = "".join(random.choices(string.digits, k=12))
     claim["CLM"]["CLM_DT_SGNTR_SK"] = clm_dt_sgntr["CLM_DT_SGNTR_SK"]
-    claim["CLM"]["CLM_UNIQ_ID"] = "".join(random.choices(string.digits, k=13))
+    claim["CLM"]["CLM_UNIQ_ID"] = "-" + "".join(random.choices(string.digits, k=13))
     # clm_type_cd = 60
     clm_type_cd = random.choice([1, 2, 3, 4, 10, 20, 30, 40, 50, 60, 71, 72, 81, 82])
     claim["CLM"]["CLM_TYPE_CD"] = clm_type_cd
@@ -1150,15 +1147,21 @@ def main():
         "--sushi",
         "-s",
         action="store_true",
-        help="Generate new StructureDefinitions. Use when testing locally if new .fsh files \
-            have been added.",
+        help="Generate new StructureDefinitions. Use when testing locally if new .fsh files "
+        "have been added.",
     )
     parser.add_argument(
         "--benes",
         "-b",
         type=str,
-        help="Pull BENE_SKs from the input file. Expected format is that of \
-            SYNTHETIC_BENE_HSTRY.csv",
+        help="Pull BENE_SKs from the input file. Expected format is that of "
+        "SYNTHETIC_BENE_HSTRY.csv",
+    )
+    parser.add_argument(
+        "--min-claims", type=int, default=5, help="Minimum number of claims to generate per person"
+    )
+    parser.add_argument(
+        "--max-claims", type=int, default=5, help="Maximum number of claims to generate per person"
     )
 
     args = parser.parse_args()
@@ -1187,14 +1190,16 @@ def main():
     CLM_LINE_PRFNL = []
     CLM_LINE_RX = []
     pt_complete = 0
+    min_claims = args.min_claims
+    max_claims = args.max_claims
     max_date = str(date.today())
     for pt_complete, pt_bene_sk in enumerate(bene_sk_list):
         if (pt_complete) % 1000 == 0 and pt_complete > 0:
             print(
-                f"Completed {pt_complete} patients with {claims_to_generate_per_person} \
-                    claims per patient."
+                f"Completed {pt_complete} patients with between {min_claims} and {max_claims} "
+                "claims per patient."
             )
-        for _ in range(claims_to_generate_per_person):
+        for _ in range(random.randint(min_claims, max_claims)):
             clm_from_dt_min = "2018-01-01"
             claim = gen_claim(bene_sk=pt_bene_sk, min_date=clm_from_dt_min, max_date=max_date)
             CLM.append(claim["CLM"])

apps/bfd-model-idr/generator_util.py

@@ -109,7 +109,7 @@ def gen_mbi(self):
         return mbi
 
     def gen_bene_sk(self):
-        bene_sk = random.randint(1000, 1000000000)
+        bene_sk = random.randint(-1000000000, -1000)
         if bene_sk in self.used_bene_sk:
             return self.gen_bene_sk()
         return bene_sk
@@ -455,39 +455,38 @@ def save_output_files(self):
         Path("out").mkdir(exist_ok=True)
 
         df = pd.json_normalize(self.bene_hstry_table)
-        if df.size > 0:
-            df = df[
-                [
-                    "BENE_SK",
-                    "BENE_XREF_EFCTV_SK",
-                    "BENE_XREF_SK",
-                    "BENE_MBI_ID",
-                    "BENE_LAST_NAME",
-                    "BENE_1ST_NAME",
-                    "BENE_MIDL_NAME",
-                    "BENE_BRTH_DT",
-                    "BENE_DEATH_DT",
-                    "BENE_VRFY_DEATH_DAY_SW",
-                    "BENE_SEX_CD",
-                    "BENE_RACE_CD",
-                    "BENE_LINE_1_ADR",
-                    "BENE_LINE_2_ADR",
-                    "BENE_LINE_3_ADR",
-                    "BENE_LINE_4_ADR",
-                    "BENE_LINE_5_ADR",
-                    "BENE_LINE_6_ADR",
-                    "GEO_ZIP_PLC_NAME",
-                    "GEO_ZIP5_CD",
-                    "GEO_USPS_STATE_CD",
-                    "CNTCT_LANG_CD",
-                    "IDR_LTST_TRANS_FLG",
-                    "IDR_TRANS_EFCTV_TS",
-                    "IDR_INSRT_TS",
-                    "IDR_UPDT_TS",
-                    "IDR_TRANS_OBSLT_TS",
-                ]
+        df = df[
+            [
+                "BENE_SK",
+                "BENE_XREF_EFCTV_SK",
+                "BENE_XREF_SK",
+                "BENE_MBI_ID",
+                "BENE_LAST_NAME",
+                "BENE_1ST_NAME",
+                "BENE_MIDL_NAME",
+                "BENE_BRTH_DT",
+                "BENE_DEATH_DT",
+                "BENE_VRFY_DEATH_DAY_SW",
+                "BENE_SEX_CD",
+                "BENE_RACE_CD",
+                "BENE_LINE_1_ADR",
+                "BENE_LINE_2_ADR",
+                "BENE_LINE_3_ADR",
+                "BENE_LINE_4_ADR",
+                "BENE_LINE_5_ADR",
+                "BENE_LINE_6_ADR",
+                "GEO_ZIP_PLC_NAME",
+                "GEO_ZIP5_CD",
+                "GEO_USPS_STATE_CD",
+                "CNTCT_LANG_CD",
+                "IDR_LTST_TRANS_FLG",
+                "IDR_TRANS_EFCTV_TS",
+                "IDR_INSRT_TS",
+                "IDR_UPDT_TS",
+                "IDR_TRANS_OBSLT_TS",
             ]
-            df.to_csv("out/SYNTHETIC_BENE_HSTRY.csv", index=False)
+        ]
+        df.to_csv("out/SYNTHETIC_BENE_HSTRY.csv", index=False)
 
         arr = [{"BENE_MBI_ID": mbi, **self.mbi_table[mbi]} for mbi in self.mbi_table]
         df = pd.json_normalize(arr)
@@ -502,6 +501,7 @@ def save_output_files(self):
                 "IDR_TRANS_OBSLT_TS",
             ]
         ]
+
         df.to_csv("out/SYNTHETIC_BENE_MBI_ID.csv", index=False)
 
         df = pd.json_normalize(self.mdcr_stus)

apps/bfd-model-idr/patient_generator.py

@@ -18,8 +18,8 @@
 parser.add_argument(
     "--claims",
     action="store_true",
-    help="Automatically generate claims after patient generation using the generated \
-        SYNTHETIC_BENE_HSTRY.csv file",
+    help="Automatically generate claims after patient generation using the generated "
+    "SYNTHETIC_BENE_HSTRY.csv file",
 )
 args = parser.parse_args()
 
@@ -47,7 +47,7 @@
 csv_data = None
 if args.benes:
     try:
-        csv_data = pd.read_csv(args.benes)
+        csv_data = pd.read_csv(args.benes, dtype={"BENE_SEX_CD": "Int64", "BENE_RACE_CD": "Int64"})
         print(f"Loaded {len(csv_data)} rows from CSV file: {args.benes}")
 
         patients_to_generate = len(csv_data)

apps/bfd-pipeline/bfd-pipeline-idr/extractor.py

@@ -15,7 +15,6 @@
 from model import LoadProgress, T
 from timer import Timer
 
-idr_query_timer = Timer("idr_query")
 cursor_execute_timer = Timer("cursor_execute")
 cursor_fetch_timer = Timer("cursor_fetch")
 transform_timer = Timer("transform")
@@ -26,7 +25,6 @@
 
 
 def print_timers() -> None:
-    idr_query_timer.print_results()
     cursor_execute_timer.print_results()
     cursor_fetch_timer.print_results()
     transform_timer.print_results()
@@ -73,18 +71,15 @@ def extract_idr_data(
         min_transaction_date = get_min_transaction_date()
         logger.info("extracting %s", cls.table())
         if progress is None:
-            idr_query_timer.start()
             # No saved progress, process the whole table from the beginning
-            res = self.extract_many(
+            return self.extract_many(
                 cls,
                 fetch_query.replace(
                     "{WHERE_CLAUSE}",
                     f"WHERE {batch_timestamp_clause} >= '{min_transaction_date}'",
                 ).replace("{ORDER_BY}", f"ORDER BY {batch_timestamp_clause}"),
                 {},
             )
-            idr_query_timer.stop()
-            return res
 
         previous_batch_complete = progress.batch_complete_ts >= progress.batch_start_ts
         # If we've completed the last batch, there shouldn't be any additional records
@@ -94,9 +89,8 @@ def extract_idr_data(
         # Snowflake, so we should always start loading from the most recent timestamp
         # that we've already fetched
         compare_timestamp = max(min_transaction_date, progress.last_ts)
-        idr_query_timer.start()
         # Saved progress found, start processing from where we left off
-        res = self.extract_many(
+        return self.extract_many(
             cls,
             fetch_query.replace(
                 "{WHERE_CLAUSE}",
@@ -106,8 +100,6 @@ def extract_idr_data(
             ).replace("{ORDER_BY}", f"ORDER BY {batch_timestamp_clause}"),
             {"timestamp": compare_timestamp},
         )
-        idr_query_timer.stop()
-        return res
 
 
 class PostgresExtractor(Extractor):

apps/bfd-pipeline/bfd-pipeline-idr/loader.py

@@ -9,6 +9,7 @@
 from model import LoadProgress, T
 from timer import Timer
 
+idr_query_timer = Timer("idr_query")
 temp_table_timer = Timer("temp_table")
 copy_timer = Timer("copy")
 insert_timer = Timer("insert")
@@ -18,6 +19,7 @@
 
 
 def print_timers() -> None:
+    idr_query_timer.print_results()
     temp_table_timer.print_results()
     copy_timer.print_results()
     insert_timer.print_results()
@@ -81,7 +83,15 @@ def load(
             num_rows = 0
 
             # load each batch in a separate transaction
-            for results in fetch_results:
+            while True:
+                idr_query_timer.start()
+                # We unfortunately need to use a while true loop here since we need to wrap the
+                # iterator with the timer calls.
+                results = next(fetch_results, None)
+                idr_query_timer.stop()
+                if results is None:
+                    break
+
                 data_loaded = True
                 logger.info("loading next %s results", len(results))
                 num_rows += len(results)

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/controller/GlobalExceptionController.java

@@ -1,5 +1,6 @@
 package gov.cms.bfd.server.ng.controller;
 
+import gov.cms.bfd.server.ng.filter.MdcFilter;
 import jakarta.servlet.RequestDispatcher;
 import jakarta.servlet.http.HttpServletRequest;
 import org.slf4j.Logger;
@@ -25,23 +26,33 @@ public String handleError(HttpServletRequest request) {
     var message = request.getAttribute(RequestDispatcher.ERROR_MESSAGE);
     // Note: this doesn't include the query string
     var originalUri = request.getAttribute(RequestDispatcher.ERROR_REQUEST_URI);
+    var logBuilder = LOGGER.atError();
+    // Unexpected errors could contain anything, so we should not return the text to the user.
+    var responseMessage = "an unknown error occurred";
 
     if (status != null) {
       var statusCode = Integer.parseInt(status.toString());
-      // For "expected" errors, we can assume the supplied error message is user-friendly, if
-      // present
+
       if (statusCode < 500) {
-        var responseMessage = message == null ? "error" : message.toString();
-        LOGGER
-            .atWarn()
-            .setMessage(responseMessage)
-            .addKeyValue("statusCode", statusCode)
-            .addKeyValue("originalUri", originalUri)
-            .log();
-        return responseMessage;
+        if (message != null) {
+          // For "expected" errors, we can assume the supplied error message is user-friendly, if
+          // present
+          responseMessage = message.toString();
+        }
+        // Sub-500 response codes indicate a bad request rather than an internal failure
+        logBuilder = LOGGER.atWarn();
       }
+      logBuilder = logBuilder.addKeyValue("statusCode", statusCode);
     }
-    // Unexpected errors could contain anything, so we should not return the text to the user.
-    return "an unknown error occurred";
+    // Since this is controller is hit from a redirect, we don't have the context from the normal
+    // MDC filter, so we need to add them to the log here explicitly,
+    logBuilder
+        .setMessage(responseMessage)
+        .addKeyValue(MdcFilter.URI_KEY, originalUri)
+        .addKeyValue(MdcFilter.REQUEST_ID_KEY, request.getRequestId())
+        .addKeyValue(MdcFilter.REMOTE_ADDRESS_KEY, request.getRemoteAddr())
+        .log();
+
+    return responseMessage;
   }
 }