blocks-transformer/block.py

import logging
import requests

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
)
logger = logging.getLogger(__name__)

url = "http://centurion-h2o-server.default.svc.cluster.local:8000/scoring"
headers = {
    "accept": "application/json",
    "Content-Type": "application/json"
}

def __main__(pti: float, score_results: int, revolving_amount_monthly_payment: int, AT31S: int, total_amount_high_credit: int, AT20S: int, BALMAG01: int, record_counts_revolving_trade_count: int, PAYMNT10: int, closed_with_balance_amount_current_balance: int, REV83: int, AGG102: int, BC21S: int, record_counts_total_trade_count: int, revolving_amount_current_balance: int, total_amount_current_balance: int, revolving_amount_high_credit: int, US01S: int, AGG101: int, closed_with_balance_amount_monthly_payment: int, revolving_amount_credit_limit: int, revolving_amount_percent_available_credit: int, AT09S: int) -> dict:
    input_data = {
        "pti": pti, "score_results": score_results, "revolving_amount_monthly_payment": revolving_amount_monthly_payment,
        "AT31S": AT31S, "total_amount_high_credit": total_amount_high_credit, "AT20S": AT20S, "BALMAG01": BALMAG01,
        "record_counts_revolving_trade_count": record_counts_revolving_trade_count, "PAYMNT10": PAYMNT10,
        "closed_with_balance_amount_current_balance": closed_with_balance_amount_current_balance, "REV83": REV83,
        "AGG102": AGG102, "BC21S": BC21S, "record_counts_total_trade_count": record_counts_total_trade_count,
        "revolving_amount_current_balance": revolving_amount_current_balance, "total_amount_current_balance": total_amount_current_balance,
        "revolving_amount_high_credit": revolving_amount_high_credit, "US01S": US01S, "AGG101": AGG101,
        "closed_with_balance_amount_monthly_payment": closed_with_balance_amount_monthly_payment,
        "revolving_amount_credit_limit": revolving_amount_credit_limit, "revolving_amount_percent_available_credit": revolving_amount_percent_available_credit,
        "AT09S": AT09S
    }

    filtered_data = {key: value for key, value in input_data.items() if value is not None}
    data = {
            "model_name": "pd_v1",
            "features": filtered_data
            }
    try:
        try:
            response = requests.post(url, headers=headers, json=data)
            response.raise_for_status()  # Raise an error for HTTP issues

            # Parse the response and return the "p1" value
            response_data = response.json()
            print("response json", response_data)
            p1_value = response_data.get("prediction", {}).get("p1")
        except requests.RequestException as error:
            logger.error(f"Error while predicting: {error}")
            return None
        
        if p1_value is not None:
            logger.info(f"Prediction p1: {p1_value}")
            return {'probability': float(p1_value)}
        else:
            logger.error("Response does not contain 'p1'")
            return None

    except Exception as error:
        logger.error(f"Error while predicting: {error}")
        return {'error': str(error)}