Project Structure

import logging.config
import yaml

def setup_logging():
    """Configure logging from YAML file."""
    with open('config/logs_configs/logging_config.yaml', 'r') as f:
        config = yaml.safe_load(f)
    logging.config.dictConfig(config)

import torch.nn as nn
from dataclasses import dataclass

@dataclass
class ModelConfig:
    input_size: int
    output_size: int
    hidden_layers: list[int]
    activation_functions: list[str]
    dropout_rate: float
    learning_rate: float
    epochs: int
    batch_size: int

class CreditScoreModel(nn.Module):
    def __init__(self, config: ModelConfig):
        super().__init__()
        # Dynamic layer construction based on config
        self.layers = self._build_layers(config)
    
    def forward(self, x):
        return self.layers(x)

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.pipeline import Pipeline
import joblib

def preprocess_data(df: pd.DataFrame, save_path: str = None):
    """Create and fit preprocessing pipeline."""
    
    # Numerical features: imputation + scaling
    numerical_transformer = Pipeline([
        ('imputer', SimpleImputer(strategy='mean')),
        ('scaler', StandardScaler())
    ])
    
    # Categorical features: imputation + encoding
    categorical_transformer = Pipeline([
        ('imputer', SimpleImputer(strategy='constant', fill_value='unknown')),
        ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False))
    ])
    
    # Combine transformers
    preprocessor = ColumnTransformer([
        ('num', numerical_transformer, numerical_features),
        ('cat', categorical_transformer, categorical_features)
    ])
    
    # Fit and save
    X_processed = preprocessor.fit_transform(X)
    joblib.dump(preprocessor, save_path)
    
    return train_test_split(X_processed, y, test_size=0.2, random_state=42)

def train(args):
    config = load_config(args.config)
    
    mlflow.set_experiment("Credit Score Training")
    with mlflow.start_run(run_name=config_name):
        # Log configuration
        mlflow.log_params(config)
        
        # Load and preprocess data
        df = load_data(dataset_path)
        X_train, X_test, y_train, y_test = preprocess_data(df)
        
        # Initialize model
        model = CreditScoreModel(model_config)
        
        # Training loop
        for epoch in range(epochs):
            # ... training code ...
            mlflow.log_metric("train_loss", loss, step=epoch)
        
        # Evaluation and artifact logging
        mlflow.log_metric("test_roc_auc", roc_auc)
        mlflow.log_figure(confusion_matrix_plot, "confusion_matrix.png")
        mlflow.log_artifact(model_save_path)

class Predictor:
    """Singleton pattern for model inference."""
    _instance = None
    _initialized = False
    
    def __init__(self, model_path: str, config_path: str, preprocessor_path: str):
        if Predictor._initialized:
            return
        
        # Load configuration
        with open(config_path, 'r') as f:
            yaml_config = yaml.safe_load(f)
        
        # Load preprocessor
        self.preprocessor = joblib.load(preprocessor_path)
        
        # Load model
        self.model = CreditScoreModel(model_config)
        self.model.load_state_dict(torch.load(model_path))
        self.model.eval()
        
        Predictor._initialized = True
    
    def inference(self, data: CreditRiskInput) -> dict:
        # Preprocess input
        df = pd.DataFrame([data.model_dump(by_alias=True)])
        processed = self.preprocessor.transform(df)
        
        # Predict
        with torch.no_grad():
            probs = self.model.predict_probability(torch.FloatTensor(processed))
            prediction = "good" if probs[0][1] > 0.5 else "bad"
        
        return {"prediction": prediction, "probability": probs[0][1].item()}

# Global singleton instance
predictor = Predictor(
    model_path=DEFAULT_MODEL_PATH,
    config_path=DEFAULT_CONFIG_PATH,
    preprocessor_path=DEFAULT_PREPROCESSOR_PATH
)

from fastapi import FastAPI, HTTPException
from server.schemas import CreditRiskInput, CreditRiskOutput
from inference.inference import predictor

app = FastAPI(
    title="Credit Score Prediction API",
    version="1.0.0"
)

@app.post("/credit_score_prediction", response_model=CreditRiskOutput)
async def predict_credit_risk(data: CreditRiskInput) -> CreditRiskOutput:
    try:
        result = predictor.inference(data)
        return CreditRiskOutput(
            prediction=result["prediction"],
            probability=result["probability"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))