blocks-transformer/score_processing.py
Ankur Malik 20b2f4afdb
All checks were successful
Build and Push Docker Image / test (push) Successful in 8s
Build and Push Docker Image / build_and_push (push) Successful in 17s
Update Scores block processing and schema
2025-11-23 23:41:46 -05:00

116 lines
3.4 KiB
Python

import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
)
logger = logging.getLogger(__name__)
def processing(data: dict) -> dict:
score_threshold = 1200
score_threshold2 = 1225
# ---- Compute hd_score1, hd_score2, final hd_score ----
try:
hd_score1 = max([
data["hd_score_s1"],
data["hd_score_s2"],
data["hd_score_g1"],
data["hd_score_s3"],
data["hd_score_m1"],
])
hd_score2 = data["hd_score_iso_m2"]
if hd_score1 > score_threshold:
hd_score = hd_score1
elif hd_score2 >= score_threshold2:
hd_score = hd_score2
else:
hd_score = max(hd_score1, hd_score2)
# logging.info(f"Computed hd_score1={hd_score1}, hd_score2={hd_score2}, final hd_score={hd_score}")
except Exception as e:
logging.error(f"Error calculating hd_score values: {e}")
return {}
# ---- recommended_action ----
try:
if hd_score1 > score_threshold or hd_score2 >= score_threshold2:
recommended_action = "Decline Application"
else:
recommended_action = "Pass Application"
logging.info(f"recommended_action: {recommended_action}")
except Exception as e:
logging.error(f"Error assigning recommended_action: {e}")
return {}
# ---- description ----
try:
if hd_score1 > score_threshold:
description = (
f"HD Fraud Score is above the risk threshold {score_threshold}, "
f"Recommended action: Decline Application"
)
elif hd_score2 >= score_threshold2:
description = (
f"HD Fraud Score is above the risk threshold {score_threshold2}, "
f"Recommended action: Decline Application"
)
else:
description = (
"HD Fraud Score is below the risk threshold, Recommended action: Pass Application"
)
logging.info(f"description: {description}")
except Exception as e:
logging.error(f"Error creating description: {e}")
return {}
# ---- hd_action_reasoncode ----
try:
reason_hierarchy = [
("hd_score_s1", "S1 Triggered"),
("hd_score_s2", "S2 Triggered"),
("hd_score_g1", "G1 Triggered"),
("hd_score_s3", "S3 Triggered"),
("hd_score_m1", "M1 Triggered"),
("hd_score_iso_m2", "M2 Triggered"), # NEW
]
hd_action_reasoncode = "Pass"
for key, reason in reason_hierarchy:
score = data[key]
if key == "hd_score_iso_m2":
if score >= score_threshold2:
hd_action_reasoncode = reason
break
else:
if score > score_threshold:
hd_action_reasoncode = reason
break
logging.info(f"hd_action_reasoncode: {hd_action_reasoncode}")
except Exception as e:
logging.error(f"Error determining action_reasoncode: {e}")
return {}
# ---- Return only original 4 fields ----
return {
"hd_score": hd_score,
"recommended_action": recommended_action,
"description": description,
"hd_action_reasoncode": hd_action_reasoncode,
}