syedkhizarrayaz/BM-AI-Analysis-And-Alert-Prioritization-Agent
0
1# imports2import json3import pandas as pd4from sqlalchemy import create_engine5from datetime import datetime6from sklearn.preprocessing import OneHotEncoder, StandardScaler7from sklearn.compose import ColumnTransformer8from sklearn.pipeline import Pipeline9from sklearn.impute import SimpleImputer10from imblearn.pipeline import Pipeline as ImbPipeline11import joblib12import os13import xmltodict14from fastapi import APIRouter, HTTPException, Request15from config import DB_CONNECTION_STR, MODEL_FILE, ACCUMULATED_DATA, DATA_TABLE16from pydantic import BaseModel, ValidationError, Field17from typing import Optional, List18from datetime import datetime19import logging20from huggingface_hub import hf_hub_download21 22# API router instance23router = APIRouter()24 25# **Setup logging** (add this section to configure the logging)26logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[27 logging.FileHandler("runmodelapi.log"),28 logging.StreamHandler()29])30 31logger = logging.getLogger(__name__)32 33MODEL_REPO = "syedkhizarrayaz/Alert-Prioritization"34MODEL_FILE = os.environ.get("MODEL_FILE", "AMLClassificationModel2.pkl")35MODEL_DIR = "/app/models"36MODEL_PATH = f"{MODEL_DIR}/{MODEL_FILE}"37 38def load_model():39 # Ensure MODEL_DIR exists40 os.makedirs(MODEL_DIR, exist_ok=True)41 42 if not os.path.exists(MODEL_PATH):43 print("⬇️ Downloading model from Hugging Face...")44 logger.info(f"Downloading model from repository: {MODEL_REPO}")45 46 # Get HF_TOKEN from environment (should be set in Hugging Face Space secrets)47 hf_token = os.environ.get("HF_TOKEN")48 if not hf_token:49 error_msg = (50 "HF_TOKEN environment variable is not set. "51 "Please set HF_TOKEN in Hugging Face Space secrets to access the private model repository."52 )53 logger.error(error_msg)54 raise ValueError(error_msg)55 56 logger.info("HF_TOKEN found, proceeding with model download...")57 58 try:59 # Download model file - hf_hub_download returns the full path to the downloaded file60 # Don't use local_dir to avoid path issues - download to default cache then copy61 logger.info(f"Attempting to download file '{MODEL_FILE}' from repository '{MODEL_REPO}'")62 downloaded_path = hf_hub_download(63 repo_id=MODEL_REPO,64 filename=MODEL_FILE,65 token=hf_token,66 )67 logger.info(f"Model downloaded successfully to cache: {downloaded_path}")68 print(f"✅ Model downloaded to: {downloaded_path}")69 70 # Verify the file exists71 if not os.path.exists(downloaded_path):72 raise FileNotFoundError(f"Downloaded file not found at: {downloaded_path}")73 74 # Copy to our desired location75 import shutil76 shutil.copy2(downloaded_path, MODEL_PATH)77 logger.info(f"Model copied to: {MODEL_PATH}")78 print(f"✅ Model copied to: {MODEL_PATH}")79 model_path = MODEL_PATH80 except Exception as e:81 error_msg = f"Error downloading model from {MODEL_REPO}: {str(e)}"82 logger.error(error_msg)83 # Provide helpful error message for common issues84 if "401" in str(e) or "Unauthorized" in str(e) or "authentication" in str(e).lower():85 raise ValueError(86 f"{error_msg}\n"87 "This might be due to:\n"88 "- Invalid or expired HF_TOKEN\n"89 "- HF_TOKEN not set in Hugging Face Space secrets\n"90 "- Insufficient permissions to access the private repository"91 )92 elif "404" in str(e) or "Not Found" in str(e):93 # Provide helpful suggestions for 404 errors94 suggestions = (95 f"Model file '{MODEL_FILE}' not found in repository '{MODEL_REPO}'.\n\n"96 "Please verify:\n"97 f"1. The file exists in the repository root (filename: {MODEL_FILE})\n"98 f"2. If the file is in a subdirectory, update MODEL_FILE environment variable\n"99 f" Example: If file is at 'models/AMLClassificationModel2.pkl' in repo,\n"100 f" set MODEL_FILE='models/AMLClassificationModel2.pkl'\n"101 f"3. The filename matches exactly (case-sensitive)\n"102 f"4. You have access to the private repository\n"103 f"5. HF_TOKEN has read permissions for the repository"104 )105 raise FileNotFoundError(f"{error_msg}\n\n{suggestions}")106 else:107 raise RuntimeError(error_msg)108 else:109 print("✅ Model already exists")110 logger.info(f"Model already exists at: {MODEL_PATH}")111 model_path = MODEL_PATH112 113 if not os.path.exists(model_path):114 raise FileNotFoundError(f"Model file not found at {model_path}")115 116 logger.info(f"Loading model from: {model_path}")117 return joblib.load(model_path)118 119 120# API request model121class AlertDataRequest(BaseModel):122 # Basic Alert Details123 AlertID: Optional[int] = Field(None, description="Unique alert identifier", example=1001)124 ScenarioClassID: Optional[int] = Field(None, description="Scenario class identifier")125 FocusTypeID: Optional[int] = Field(None, description="Focus type identifier")126 FocusColumnValue: Optional[str] = Field(None, description="Customer/entity identifier", example="PK-42101-1234567-1")127 AlertDate: Optional[datetime] = Field(None, description="Alert date timestamp")128 AlertScore: Optional[float] = Field(None, description="Alert risk score (0-100)", example=85.5)129 AlertType: Optional[str] = Field(None, description="Type of alert")130 Comments: Optional[str] = Field(None, description="Additional comments")131 wfstatus: Optional[str] = Field(None, description="Workflow status")132 AssignTo: Optional[str] = Field(None, description="Assigned to user/team")133 AlertDueDate: Optional[datetime] = Field(None, description="Alert due date timestamp")134 ActivityStatus: Optional[str] = Field(None, description="Activity status")135 Suppress: Optional[bool] = Field(None, description="Suppress flag")136 LastActivityDate: Optional[datetime] = Field(None, description="Last activity date timestamp")137 Hcode: Optional[str] = Field(None, description="H code")138 ServiceTypeID: Optional[int] = Field(None, description="Service type identifier")139 LogID: Optional[int] = Field(None, description="Log identifier")140 141 # Address and Demographic Information142 AddressCityID: Optional[int] = Field(None, description="Address city ID")143 AddressCityValue: Optional[str] = Field(None, description="Address city name")144 AddressCountryID: Optional[int] = Field(None, description="Address country ID")145 AddressRegionID: Optional[int] = Field(None, description="Address region ID")146 AddressRegionValue: Optional[str] = Field(None, description="Address region name")147 AddressStreet: Optional[str] = Field(None, description="Street address")148 Age: Optional[int] = Field(None, description="Customer age")149 BranchID: Optional[int] = Field(None, description="Branch identifier")150 BranchValue: Optional[str] = Field(None, description="Branch name")151 CountryOfBirth: Optional[str] = Field(None, description="Country of birth")152 CreateDate: Optional[object] = Field(None, description="Account creation date", example="2025-06-15T10:30:00")153 CustomerID: Optional[int] = Field(None, description="Customer identifier")154 CustomerName: Optional[str] = Field(None, description="Customer name")155 CustomerStatusID: Optional[int] = Field(None, description="Customer status ID")156 CustomerStatusValue: Optional[str] = Field(None, description="Customer status")157 CutomerGroupID: Optional[int] = Field(None, description="Customer group ID")158 DateOfBirth: Optional[datetime] = Field(None, description="Date of birth timestamp")159 GenderID: Optional[int] = Field(None, description="Gender ID")160 GenderValue: Optional[str] = Field(None, description="Gender value")161 IdentityExpiryDate: Optional[datetime] = Field(None, description="Identity expiry date timestamp")162 IdentityNumber: Optional[str] = Field(None, description="Identity number")163 IdentityTypeID: Optional[int] = Field(None, description="Identity type ID")164 IdentityTypeValue: Optional[str] = Field(None, description="Identity type")165 IncorporationDate: Optional[datetime] = Field(None, description="Incorporation date timestamp")166 IndustryID: Optional[int] = Field(None, description="Industry ID")167 IndustryValue: Optional[str] = Field(None, description="Industry value")168 IsDeceased: Optional[bool] = Field(None, description="Is deceased flag")169 170 # KYC and Financial Information171 KYCAvgAccBalance: Optional[float] = Field(None, description="KYC average account balance")172 KYCLimit: Optional[float] = Field(None, description="KYC limit")173 KYCMonthlyIncome: Optional[float] = Field(None, description="KYC monthly income")174 KYCNoOfCredits: Optional[int] = Field(None, description="KYC number of credits")175 KYCNoOfDebits: Optional[int] = Field(None, description="KYC number of debits")176 KYCRiskCategoryID: Optional[int] = Field(None, description="KYC risk category ID")177 riskLevel: Optional[str] = Field(None, description="Risk level (Low/Medium/High)", example="Low")178 KYCValueOfCredits: Optional[float] = Field(None, description="KYC value of credits")179 KYCValueOfDebits: Optional[float] = Field(None, description="KYC value of debits")180 ModifiedDate: Optional[datetime] = Field(None, description="Modified date timestamp")181 182 # Nationality and Occupation Information183 NationalityID1: Optional[int] = Field(None, description="Primary nationality ID")184 NationalityID2: Optional[int] = Field(None, description="Secondary nationality ID")185 OccupationID: Optional[int] = Field(None, description="Occupation ID")186 OccupationValue: Optional[str] = Field(None, description="Occupation value")187 RelationshipManagerID: Optional[int] = Field(None, description="Relationship manager ID")188 RelationshipManagerValue: Optional[str] = Field(None, description="Relationship manager name")189 SegmentID: Optional[int] = Field(None, description="Segment ID")190 SegmentValue: Optional[str] = Field(None, description="Segment value")191 192 # Additional Columns193 ScenarioID: Optional[int] = Field(None, description="Scenario ID")194 FocusTypeValue: Optional[str] = Field(None, description="Focus type value")195 MatchDate: Optional[datetime] = Field(None, description="Match date timestamp")196 Score: Optional[float] = Field(None, description="Score value")197 IsSuppressed: Optional[bool] = Field(None, description="Is suppressed flag")198 MatchDetails: Optional[str] = Field(None, description="JSON string with match details", example='{"id": "PK-42101-1234567-1", "scenario": "Unusually large installment", "score": 85.5, "riskLevel": "Low"}')199 MatchInfoJson: Optional[str] = Field(None, description="JSON array string with transaction details", example='[{"ID": "PK-42101-1234567-1", "TRANSACTIONAMOUNT": 150000, "CURRENCY": "PKR", "INSTALLMENTNUMBER": 1}]')200 ScenarioName: Optional[str] = Field(None, description="Alert scenario name", example="Unusually large installment")201 Status: Optional[str] = Field(None, description="Status")202 workflow: Optional[str] = Field(None, description="Current workflow status", example="Unassigned")203 desc: Optional[str] = Field(None, alias="_desc", description="Description")204 approvalOpt: Optional[str] = Field(None, description="Approval option")205 profileID: Optional[int] = Field(None, description="Profile ID")206 hometext: Optional[str] = Field(None, description="Home text")207 208 class Config:209 json_schema_extra = {210 "example": {211 "AlertID": 1001,212 "FocusColumnValue": "PK-42101-1234567-1",213 "AlertScore": 85.5,214 "CreateDate": "2025-06-15T10:30:00",215 "riskLevel": "Low",216 "MatchDetails": '{"id": "PK-42101-1234567-1", "scenario": "Unusually large installment", "score": 85.5, "riskLevel": "Low"}',217 "MatchInfoJson": '[{"ID": "PK-42101-1234567-1", "TRANSACTIONAMOUNT": 150000, "CURRENCY": "PKR", "INSTALLMENTNUMBER": 1}, {"ID": "PK-42101-1234567-1", "TRANSACTIONAMOUNT": 180000, "CURRENCY": "PKR", "INSTALLMENTNUMBER": 2}, {"ID": "PK-42101-1234567-1", "TRANSACTIONAMOUNT": 200000, "CURRENCY": "PKR", "INSTALLMENTNUMBER": 3}]',218 "ScenarioName": "Unusually large installment",219 "workflow": "Unassigned"220 }221 }222 223# Response models224class PredictionItem(BaseModel):225 AlertID: int = Field(..., description="Alert identifier", example=1001)226 FocusColumnValue: str = Field(..., description="Customer/entity identifier", example="PK-42101-1234567-1")227 STRScenario: str = Field(..., description="Scenario name", example="Unusually large installment")228 Prediction: str = Field(..., description="Predicted priority (High/Medium/Low)", example="High")229 230class PredictionResponse(BaseModel):231 status: int = Field(..., description="HTTP status code", example=200)232 message: str = Field(..., description="Response message", example="Success")233 data: List[PredictionItem] = Field(..., description="List of predictions")234 235 class Config:236 json_schema_extra = {237 "example": {238 "status": 200,239 "message": "Success",240 "data": [241 {242 "AlertID": 1001,243 "FocusColumnValue": "PK-42101-1234567-1",244 "STRScenario": "Unusually large installment",245 "Prediction": "High"246 }247 ]248 }249 }250 251# ML Model class252class AMLModelProcessor:253 # constructor254 def __init__(self, model_filename, db_connection_str, table_name=None):255 256 # **Log the initialization** 257 logger.info("Initializing AMLModelProcessor")258 # Load the saved model259 try:260 self.rf_reduced = load_model()261 logger.info(f"Model loaded from {model_filename}") # **Add info log for model loading**262 except Exception as e:263 logger.error(f"Failed to load model: {e}") # **Log error if model fails to load**264 raise e265 # Database connection details (optional - only if provided)266 self.db_connection_str = db_connection_str267 if db_connection_str:268 self.engine = create_engine(db_connection_str)269 logger.info(f"Database connection established with {db_connection_str}") # **Log DB connection success**270 else:271 self.engine = None272 logger.info("No database connection string provided - running in database-free mode") # **Log no DB mode**273 274 if table_name is not None and self.engine is not None:275 # Query data from the database table and load it into a pandas DataFrame276 try:277 self.query = f"SELECT * FROM {table_name}"278 self.pp_df = pd.read_sql(self.query, self.engine)279 logger.info(f"Data loaded from table {table_name}") # **Log data load success**280 281 # Save DataFrame to a temporary Excel file282 self.temp_excel = "temp_data.xlsx"283 self.pp_df.to_excel(self.temp_excel, index=False)284 except Exception as e:285 logger.error(f"Failed to load data from table {table_name}: {e}") # **Log error on data load failure**286 raise e287 elif table_name is not None and self.engine is None:288 logger.warning(f"Table name provided but no database connection - skipping table data load")289 290 # calculation to decide to send email or not291 def calculate_percentage_info_send_email(self, df4, original_df):292 percentage_info = []293 294 logger.info("Calculating percentage info for sending emails") # **Log start of calculation**295 296 for _, row in df4.iterrows():297 focus_value = row['FocusColumnValue']298 relevant_rows = original_df[original_df['FocusColumnValue'] == focus_value]299 count_reverted_aml = relevant_rows[relevant_rows['workflow'] == 'In Progress, Reverted - AML'].shape[0]300 total_count = relevant_rows.shape[0]301 percentage = (count_reverted_aml / total_count) * 100 if total_count > 0 else 0302 percentage_info.append(percentage)303 304 df4['PercentageInfoSendEmail'] = percentage_info305 return df4306 307 # total str raised for each customer308 def calculate_str_count(self, row, original_df):309 relevant_rows = original_df[310 (original_df['FocusColumnValue'] == row['FocusColumnValue']) &311 (original_df['AlertID'] != row['AlertID']) &312 (original_df['workflow'] == 'Selected for Reporting')313 ]314 315 logger.info(f"STR count for FocusColumnValue {row['FocusColumnValue']}: {len(relevant_rows)}") # **Log STR count calculation**316 317 return len(relevant_rows)318 319 # for processing match info xml transaction data320 def xml_to_dict(self, data):321 try:322 data = data.strip()323 data = "<ALERT>" + data + "</ALERT>"324 logger.info("Converting XML data to dictionary") # **Log XML conversion start**325 return xmltodict.parse(data)326 except Exception as e:327 logger.error(f"Error parsing XML: {e}") # **Log XML parsing error**328 return None329 330 # for processing MatchDetails or MatchInfoJSON transaction data331 def json_to_dict(self, data):332 try:333 data = data.strip()334 logger.info("Converting JSON data to dictionary") # **Log JSON conversion start**335 logger.info(f"JSON data: {data}")336 return json.loads(data)337 except json.JSONDecodeError as e:338 logger.error(f"Error parsing JSON: {e}") # **Log JSON parsing error**339 return None340 341 # create dictionary of transaction history342 def filter_transaction_data(self, transactions):343 keys_to_keep = ['DATE', 'TIME', 'TRXNTYPE', 'AMOUNT', 'DRCR']344 filtered_transactions = []345 logger.info("Filtering transaction data") # **Log transaction data filtering start**346 347 if isinstance(transactions, list):348 for transaction in transactions:349 filtered_transaction = {key: transaction[key] for key in keys_to_keep if key in transaction}350 filtered_transactions.append(filtered_transaction)351 elif isinstance(transactions, dict):352 filtered_transaction = {key: transactions[key] for key in keys_to_keep if key in transactions}353 filtered_transactions.append(filtered_transaction)354 355 return filtered_transactions356 357 # map the filtered transaction dictionary358 def fill_match_info_xml(self, row):359 if pd.notna(row['MatchInfoJson']):360 # xml_dict = self.xml_to_dict(row['MatchInfoJson'])361 xml_dict = self.json_to_dict(row['MatchInfoJson'])362 if xml_dict is not None:363 # alert_matches = xml_dict.get('ALERT', {}).get('ALERTMATCH', [])364 alert_matches = xml_dict365 row['FilteredTransactions'] = self.filter_transaction_data(alert_matches)366 else:367 row['FilteredTransactions'] = None368 else:369 row['FilteredTransactions'] = None370 return row371 372 # pre-process databased data373 def pp_preprocess_data(self, excel_file):374 try:375 logger.info("Preprocessing database data from Excel file") # **Log start of preprocessing**376 pp_df = pd.read_excel(excel_file)377 pp_df = pp_df.drop_duplicates(subset='AlertID', keep='first')378 unique_scenarios = pp_df['ScenarioName'].unique()379 380 for scenario in unique_scenarios:381 pp_df[scenario] = 0382 383 for index, row in pp_df.iterrows():384 scenario_name = row['ScenarioName']385 pp_df.at[index, scenario_name] = 1386 387 pp_df['CreateDate'] = pd.to_datetime(pp_df['CreateDate'], errors='coerce')388 pp_df = pp_df.dropna(subset=['CreateDate'])389 pp_current_date = pd.to_datetime(datetime.now())390 pp_df['DaysWithBank'] = (pp_current_date - pp_df['CreateDate']).dt.days391 392 pp_days_with_bank = pp_df.groupby('FocusColumnValue')['DaysWithBank'].max().reset_index()393 pp_num_alerts = pp_df.groupby('FocusColumnValue')['AlertID'].count().reset_index()394 pp_num_alerts.columns = ['FocusColumnValue', 'NumAlerts']395 396 pp_proportion = pd.merge(pp_num_alerts, pp_days_with_bank, on='FocusColumnValue')397 pp_proportion['AlertsPerDay'] = pp_proportion['NumAlerts'] / pp_proportion['DaysWithBank']398 399 def pp_categorize_days_with_bank(pp_days):400 if pp_days < 365:401 return 'High'402 elif pp_days < 5 * 365:403 return 'Medium'404 else:405 return 'Low'406 407 pp_days_with_bank['RelationshipCategory'] = pp_days_with_bank['DaysWithBank'].apply(pp_categorize_days_with_bank)408 pp_average_score = pp_df.groupby('FocusColumnValue')['AlertScore'].mean().reset_index()409 pp_average_score.columns = ['FocusColumnValue', 'AvgAlertScore']410 411 pp_customer_metrics = pd.merge(pp_num_alerts, pp_days_with_bank, on='FocusColumnValue')412 pp_customer_metrics = pd.merge(pp_customer_metrics, pp_proportion[['FocusColumnValue', 'AlertsPerDay']], on='FocusColumnValue')413 pp_customer_metrics = pd.merge(pp_customer_metrics, pp_average_score, on='FocusColumnValue')414 415 pp_merged_df = pd.merge(pp_df, pp_customer_metrics, on='FocusColumnValue')416 417 def pp_categorize_workflow(pp_val):418 return 1 if pp_val == "Selected for Reporting" else 0419 420 pp_merged_df['Workflow'] = pp_merged_df['workflow'].apply(pp_categorize_workflow)421 422 pp_merged_df = pp_merged_df.drop(columns=['AlertID', 'FocusColumnValue', 'workflow', "CreateDate", 423 'DaysWithBank_x', 'DaysWithBank_y', "NumAlerts"])424 425 X_pp = pp_merged_df.drop('Workflow', axis=1)426 y_pp = pp_merged_df['Workflow']427 428 numeric_features = X_pp.select_dtypes(include=['float64', 'int64']).columns.tolist()429 categorical_features = X_pp.select_dtypes(include=['object', 'bool']).columns.tolist()430 431 numeric_transformer = Pipeline(steps=[432 ('imputer', SimpleImputer(strategy='median')),433 ('scaler', StandardScaler())434 ])435 categorical_transformer = Pipeline(steps=[436 ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),437 ('onehot', OneHotEncoder(handle_unknown='ignore'))438 ])439 440 # pipeline441 preprocessor = ColumnTransformer(442 transformers=[443 ('num', numeric_transformer, numeric_features),444 ('cat', categorical_transformer, categorical_features)445 ]446 )447 448 # fit the pipeline on data449 imb_pipeline = ImbPipeline(steps=[('preprocessor', preprocessor), ('classifier', self.rf_reduced)])450 imb_pipeline.fit(X_pp, y_pp)451 452 logger.info("Making predictions on preprocessed data") # **Log prediction process**453 454 # prediction455 y_pp_pred = imb_pipeline.predict(X_pp)456 457 logger.info("Retrieving data for predicted transaction data") # **Log retrieval step**458 459 # retrive data for predicted transaction data460 filtered_rows = pp_df[(pp_df['workflow'] == 'Selected for Reporting')]461 focus_column_value_scenario_dict = filtered_rows.set_index('FocusColumnValue')['ScenarioName'].to_dict()462 pp_df['STRScenarioHistory'] = pp_df['FocusColumnValue'].map(focus_column_value_scenario_dict)463 pp_df = pp_df.apply(self.fill_match_info_xml, axis=1)464 pp_df = pp_df.drop(columns=['MatchDetails', 'MatchInfoJson'], errors='ignore')465 466 pp_df['STRCount'] = 0467 pp_df['STRCount'] = pp_df.apply(lambda row: self.calculate_str_count(row, pp_df), axis=1)468 pp_df = self.calculate_percentage_info_send_email(pp_df, pp_df)469 pp_df = pp_df.drop_duplicates(subset='AlertID', keep='first')470 471 # output data frame472 pp_predictions_df = pd.DataFrame({473 'AlertID': pp_df['AlertID'],474 'ScenarioClassID': pp_df['ScenarioClassID'],475 'FocusColumnValue': pp_df['FocusColumnValue'],476 'KYCMonthlyIncome': pp_df['KYCMonthlyIncome'],477 'KYCNoOfCredits': pp_df['KYCNoOfCredits'],478 'KYCNoOfDebits': pp_df['KYCNoOfDebits'],479 'riskLevel': pp_df['riskLevel'],480 'KYCValueOfCredits': pp_df['KYCValueOfCredits'],481 'KYCValueOfDebits': pp_df['KYCValueOfDebits'],482 'OccupationValue': pp_df['OccupationValue'],483 'STRScenario': pp_df['ScenarioName'],484 'STRScenarioHistory': pp_df['STRScenarioHistory'],485 'FilteredTransactions': pp_df['FilteredTransactions'],486 'STRCount': pp_df["STRCount"],487 'Prediction': y_pp_pred,488 'TrueResult': pp_df['workflow'],489 "PercentageInfoSendEmail": pp_df['PercentageInfoSendEmail']490 })491 logger.info("Preprocessing completed and predictions generated") # **Log preprocessing completion**492 493 except Exception as e:494 logger.error(f"Error in data preprocessing: {e}") # **Log preprocessing error**495 raise e496 497 return pp_predictions_df498 499 # pre-process json api data500 def pp_preprocess_data_json(self, json_data):501 """502 Preprocesses input data from JSON format, adapting it to handle JSON input as if it were a DataFrame.503 504 Parameters:505 - json_data: JSON-like input (dict or list of dicts), which contains the same structure as a DataFrame.506 """507 try:508 logger.info("Preprocessing data from the json data") # **Log start of preprocessing**509 pp_df = json_data510 pp_df.columns = pp_df.columns.astype(str)511 512 pp_df = pp_df.drop_duplicates(subset='AlertID', keep='first')513 unique_scenarios = pp_df['ScenarioName'].unique()514 515 for scenario in unique_scenarios:516 pp_df[scenario] = 0517 518 for index, row in pp_df.iterrows():519 scenario_name = row['ScenarioName']520 pp_df.at[index, scenario_name] = 1521 522 pp_df['CreateDate'] = pd.to_datetime(pp_df['CreateDate'], errors='coerce')523 pp_df = pp_df.dropna(subset=['CreateDate'])524 pp_current_date = pd.to_datetime(datetime.now())525 pp_df['DaysWithBank'] = (pp_current_date - pp_df['CreateDate']).dt.days526 527 pp_days_with_bank = pp_df.groupby('FocusColumnValue')['DaysWithBank'].max().reset_index()528 pp_num_alerts = pp_df.groupby('FocusColumnValue')['AlertID'].count().reset_index()529 pp_num_alerts.columns = ['FocusColumnValue', 'NumAlerts']530 531 pp_proportion = pd.merge(pp_num_alerts, pp_days_with_bank, on='FocusColumnValue')532 pp_proportion['AlertsPerDay'] = pp_proportion['NumAlerts'] / pp_proportion['DaysWithBank']533 534 def pp_categorize_days_with_bank(pp_days):535 if pp_days < 365:536 return 'High'537 elif pp_days < 5 * 365:538 return 'Medium'539 else:540 return 'Low'541 542 pp_days_with_bank['RelationshipCategory'] = pp_days_with_bank['DaysWithBank'].apply(pp_categorize_days_with_bank)543 pp_average_score = pp_df.groupby('FocusColumnValue')['AlertScore'].mean().reset_index()544 pp_average_score.columns = ['FocusColumnValue', 'AvgAlertScore']545 546 pp_customer_metrics = pd.merge(pp_num_alerts, pp_days_with_bank, on='FocusColumnValue')547 pp_customer_metrics = pd.merge(pp_customer_metrics, pp_proportion[['FocusColumnValue', 'AlertsPerDay']], on='FocusColumnValue')548 pp_customer_metrics = pd.merge(pp_customer_metrics, pp_average_score, on='FocusColumnValue')549 550 pp_merged_df = pd.merge(pp_df, pp_customer_metrics, on='FocusColumnValue')551 552 def pp_categorize_workflow(pp_val):553 return 1 if pp_val == "Selected for Reporting" else 0554 555 pp_merged_df['Workflow'] = pp_merged_df['workflow'].apply(pp_categorize_workflow)556 y_pp = pp_merged_df['Workflow']557 pp_merged_df = pp_merged_df.drop(columns=['AlertID', 'FocusColumnValue', 'workflow', "Workflow","CreateDate", 558 'DaysWithBank_x', 'DaysWithBank_y', "NumAlerts", "MatchDetails", "MatchInfoJson"])559 560 # X_pp = pp_merged_df.drop('Workflow', axis=1)561 X_pp = pp_merged_df562 X_pp.columns = X_pp.columns.astype(str) # Ensure all feature names are strings563 # y_pp = pp_merged_df['Workflow']564 565 print(X_pp.columns)566 numeric_features = X_pp.select_dtypes(include=['float64', 'int64']).columns.tolist()567 categorical_features = X_pp.select_dtypes(include=['object', 'bool']).columns.tolist()568 569 logger.info(f"Numeric features: {numeric_features}") # **Log numeric features**570 logger.info(f"Categorical features: {categorical_features}") # **Log categorical features**571 572 numeric_transformer = Pipeline(steps=[573 ('imputer', SimpleImputer(strategy='median')),574 ('scaler', StandardScaler())575 ])576 categorical_transformer = Pipeline(steps=[577 ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),578 ('onehot', OneHotEncoder(handle_unknown='ignore'))579 ])580 # pipeline581 preprocessor = ColumnTransformer(582 transformers=[583 ('num', numeric_transformer, numeric_features),584 ('cat', categorical_transformer, categorical_features)585 ]586 )587 logger.info("preprocessor",preprocessor)588 # fit the piepline on data589 imb_pipeline = ImbPipeline(steps=[('preprocessor', preprocessor), ('classifier', self.rf_reduced)])590 imb_pipeline.fit(X_pp, y_pp)591 592 logger.info("Making predictions on preprocessed data") # **Log prediction process**593 # with open("preprocessed_data_log.txt", "a") as f:594 # f.write("Preprocessed data:\n")595 # f.write(X_pp.to_string())596 # f.write("\n\n") # optional spacing597 # f.write(y_pp.to_string())598 599 # prediction600 y_pp_pred = imb_pipeline.predict(X_pp)601 602 logger.info("Retrieving data for predicted transaction data") # **Log retrieval step**603 604 # retrive data for predicted transactions605 filtered_rows = pp_df[(pp_df['workflow'] == 'Selected for Reporting')]606 focus_column_value_scenario_dict = filtered_rows.set_index('FocusColumnValue')['ScenarioName'].to_dict()607 pp_df['STRScenarioHistory'] = pp_df['FocusColumnValue'].map(focus_column_value_scenario_dict)608 pp_df = pp_df.apply(self.fill_match_info_xml, axis=1)609 pp_df = pp_df.drop(columns=['MatchDetails', 'MatchInfoJson'], errors='ignore')610 611 pp_df['STRCount'] = 0612 pp_df['STRCount'] = pp_df.apply(lambda row: self.calculate_str_count(row, pp_df), axis=1)613 pp_df = self.calculate_percentage_info_send_email(pp_df, pp_df)614 pp_df = pp_df.drop_duplicates(subset='AlertID', keep='first')615 616 # Replace 1 with 'High' and 0 with 'Low' in predictions617 y_pp_pred = ['High' if pred == 1 else 'Low' for pred in y_pp_pred]618 619 # Adjust predictions based on riskLevel (KYCRiskCategoryValue)620 # If pred is Low but riskLevel is High, set pred to Medium621 # If pred is High but riskLevel is Low, set pred to Medium622 if 'riskLevel' in pp_df.columns:623 for idx, (pred, risk_level) in enumerate(zip(y_pp_pred, pp_df['riskLevel'])):624 if pd.notna(risk_level):625 risk_level_str = str(risk_level).strip()626 if pred == 'Low' and risk_level_str in ['High', 'HIGH', 'high']:627 y_pp_pred[idx] = 'Medium'628 elif pred == 'High' and risk_level_str in ['Low', 'LOW', 'low']:629 y_pp_pred[idx] = 'Medium'630 631 # output data frame to be returned632 pp_predictions_df = pd.DataFrame({633 'AlertID': pp_df['AlertID'],634 'FocusColumnValue': pp_df['FocusColumnValue'],635 'STRScenario': pp_df['ScenarioName'],636 'Prediction': y_pp_pred,637 })638 639 # Convert pp_predictions_df to JSON format640 pp_predictions_json = pp_predictions_df.to_dict(orient='records')641 logger.info("Preprocessing completed and predictions generated") # **Log preprocessing completion**642 return {"status": 200, "message": "Success", "data": pp_predictions_json}643 644 except Exception as e:645 # Return an error JSON response if conversion fails646 logger.error(f"Error in data preprocessing: {e}") # **Log preprocessing error**647 return {"status": "error", "message": str(e)}648 649 # runner for database based api functions650 def run(self):651 try:652 pp_processed_data = self.pp_preprocess_data(self.temp_excel)653 # Only insert to database if engine is available654 if self.engine is not None and DATA_TABLE:655 try:656 pp_processed_data.to_sql(DATA_TABLE, con=self.engine, if_exists='append', index=False)657 print("Data inserted successfully.")658 except Exception as e:659 logger.warning(f"Failed to insert data to database: {e}")660 else:661 logger.info("Skipping database insert (no database connection or DATA_TABLE not configured)")662 663 # Save to Excel for testing/debugging (optional)664 test = "test.xlsx"665 try:666 pp_processed_data.to_excel(test, index=False)667 except Exception as e:668 logger.warning(f"Failed to save to Excel: {e}")669 print(pp_processed_data)670 # Return a JSON response indicating success671 return {"status": "success", "message": "Data inserted successfully", "file": test}672 except Exception as e:673 # Print error and return JSON response indicating failure674 print(f"Error inserting data: {e}")675 return {"status": "error", "message": str(e)}676 finally:677 if os.path.exists(self.temp_excel):678 os.remove(self.temp_excel)679 680# json based prediction api route681@router.post(682 "/api/ai-service/predictalertpriority",683 response_model=PredictionResponse,684 summary="Predict Alert Priority",685 description="Predict whether an alert should be escalated or closed using the ML model. Returns prediction (High/Medium/Low) for each alert."686)687async def predict_alert_priority(alert_data: AlertDataRequest):688 try:689 logger.info("API call to /predictalertpriority started")690 # Convert Pydantic model to dict691 json_data = alert_data.model_dump(exclude_none=True)692 693 # Convert the JSON data to a Pandas DataFrame694 df = pd.DataFrame([json_data])695 696 # Define the column order697 column_order = [698 'AlertID','FocusColumnValue','AlertScore', 'CreateDate','riskLevel', 'MatchDetails', 'MatchInfoJson', 'ScenarioName','workflow'699 ]700 701 # Define the dtype mapping702 dtype_mapping = {703 'AlertID': 'int64', 'FocusColumnValue': 'O',704 'AlertScore': 'float64', 'CreateDate': '<M8[ns]',705 'riskLevel': 'O','MatchDetails': 'O', 'MatchInfoJson': 'O','ScenarioName': 'O', 'workflow': 'O'706 }707 708 # Apply the dtype mapping to the DataFrame709 for column, dtype in dtype_mapping.items():710 if column in df.columns:711 if dtype == '<M8[ns]': # Handle datetime columns712 df[column] = pd.to_datetime(df[column], errors='coerce')713 else:714 df[column] = df[column].astype(dtype)715 716 # Reorder the columns in the DataFrame717 df = df[column_order]718 719 print(df.dtypes.tolist()) # Output the dtypes list to verify720 721 # Now, use the validated data for further processing722 model_filename = MODEL_FILE723 db_connection_str = DB_CONNECTION_STR724 processor = AMLModelProcessor(model_filename, db_connection_str)725 726 # Pass the validated data to your processor's preprocessing method727 # Assuming pp_preprocess_data_json accepts the validated data as a dictionary or object728 logger.info("API call to /predictalertpriority completed successfully")729 return processor.pp_preprocess_data_json(df)730 731 except ValidationError as e:732 # Handle Pydantic validation errors and return an appropriate response733 logger.error(f"Error in /predictalertpriority validation error: {e.errors()}")734 raise HTTPException(status_code=422, detail=f"Validation error: {e.errors()}")735 except Exception as e:736 # Handle other types of errors737 logger.error(f"Error in /predictalertpriority: {e}")738 raise HTTPException(status_code=500, detail=f"An error occurred while predicting alert priority")739