Underwriting AI: inżynieria cech i punktacja ryzyka w nowoczesnych ubezpieczeniach
Underwriting to bijące serce każdej firmy ubezpieczeniowej: i proces, dzięki któremu to robi decyduje, czy podjąć ryzyko, za jaką cenę i na jakich warunkach. Przez dziesięciolecia ten proces zależało to od wiedzy ubezpieczycieli, którzy analizowali dokumenty papierowe i składali wnioski zasady aktuarialne skodyfikowane w tabelach. Wynik? Decyzja w 3-5 dni roboczych, koszty wysokie koszty operacyjne i subiektywna zmienność wśród ubezpieczycieli.
Sztuczna inteligencja radykalnie przepisuje te zasady. Zdaniem McKinsey’a, globalne inwestycje w rozwiązania AI dla ubezpieczeń przekroczą 6 miliardów dolarów w 2025 roku, przy czym BCG szacuje, że 36% całkowitej wartości ubezpieczenia AI koncentruje się właśnie na funkcji gwarantowania emisji. Liczby operacyjne są równie imponujące: średni czas podejmowania decyzji ubezpieczeniowej spadł z 3-5 dni do 12,4 minuty dla nich standardowych polis, zachowując dokładność oceny ryzyka na poziomie 99,3%.
Ale jak właściwie działa system gwarantowania AI? Ten przewodnik dekonstruuje całość stos techniczny: od gromadzenia i inżynierii cech, po modele scoringowe ryzyka, aż do po interpretowalność i zarządzanie stronniczością — z gotowymi do produkcji przykładami prawdziwego kodu.
Czego się nauczysz
- Architektura kompleksowego systemu underwritingu AI
- Inżynieria funkcji specyficzna dla domeny ubezpieczeniowej
- Modele ML do oceny ryzyka: XGBoost, dwa etapy częstotliwości/dotkliwości
- Możliwość interpretacji za pomocą SHAP w przypadku decyzji podlegających audytowi i zapewniających zgodność z przepisami
- Wykrywanie stronniczości i łagodzenie uczciwości w kontekście regulacyjnym UE
- MLOps dla modeli ubezpieczeniowych w produkcji za pomocą MLflow
- Monitorowanie dryfu danych za pomocą wskaźnika stabilności populacji (PSI)
Proces underwritingu: od starszego do natywnego AI
Przed zaprojektowaniem systemu AI istotne jest zrozumienie tradycyjnego przepływu pracy, z jakim mamy do czynienia automatyzacja. Proces underwritingu dzieli się na cztery podstawowe fazy:
- Zbieranie informacji: Wnioskodawca podaje dane o sobie i ryzyku (kwestionariusz, dokumenty, ewentualna fizyczna kontrola środka trwałego)
- Analiza ryzyka: Ubezpieczyciel ocenia prawdopodobieństwo i wagę wszelkich przyszłych roszczeń
- Wycena: Ustalenie składki na podstawie oszacowanego ryzyka i celów współczynnika mieszanego portfela
- Decyzja: Akceptacja, odrzucenie lub akceptacja z warunkami (wyłączenia, franczyza, premium)
System oparty na sztucznej inteligencji nie eliminuje tych faz, ale głęboko je przekształca: gromadzenie danych analiza ryzyka staje się automatyczna z heterogenicznych źródeł (otwarte dane, telematyka, biuro informacji kredytowej). i wykonywane przez modele ML w milisekundach, ceny są dynamiczne i spersonalizowane dla każdego wnioskodawcy, a decyzja jest zautomatyzowana w standardowych przypadkach pod nadzorem człowieka, m.in przypadki złożone lub graniczne.
Ramy regulacyjne: Ustawa UE o sztucznej inteligencji i gwarantowanie emisji
Europejska ustawa o sztucznej inteligencji (w pełni obowiązująca od sierpnia 2027 r.) klasyfikuje systemy punktacja kredyt i ubezpieczenie Jak Sztuczna inteligencja wysokiego ryzyka (załącznik III). To pociąga za sobą określone obowiązki: przejrzystość zautomatyzowanych decyzji, prawo do kontroli przez człowieka, szczegółową dokumentację techniczną i ocenę zgodności przed wprowadzeniem na rynek. Projekt Systemy underwritingu AI muszą uwzględniać te wymagania już na etapie architektury, a nie w jaki sposób późniejsza modernizacja.
Inżynieria funkcji w zakresie ubezpieczeń
Jakość inżynierii cech jest czynnikiem, który najbardziej wyróżnia model underwritingu doskonałe od przeciętnego. W przeciwieństwie do dziedzin takich jak wizja komputerowa, gdzie znajdują się te funkcje automatycznie wyodrębniane z warstw splotowych, wymagane są dane tabelaryczne ubezpieczenia głęboka inżynieria manualna oparta na wiedzy z dziedziny aktuarialnej.
Funkcje dla sektora motoryzacyjnego pogrupowano w pięć głównych kategorii:
- Charakterystyka demograficzna: wiek, stan cywilny, rodzaj zamieszkania
- Funkcje jazdy: lata posiadania prawa jazdy, wiek, w którym po raz pierwszy je uzyskano, historia wypadków i wykroczeń
- Cechy pojazdu: marka, model, rok, wartość, moc, rok rejestracji
- Cechy geograficzne: gęstość zaludnienia, wskaźnik przestępczości obszaru, ryzyko pogodowe
- Cechy ekonomiczne: zdolność kredytowa, wymagany rodzaj umowy ubezpieczenia
import pandas as pd
import numpy as np
from typing import Dict, Optional
from dataclasses import dataclass
from datetime import date
@dataclass
class PolicyApplicant:
"""Rappresenta i dati grezzi di un richiedente polizza auto."""
applicant_id: str
birth_date: date
license_date: date
zip_code: str
vehicle_make: str
vehicle_year: int
vehicle_value: float
annual_mileage: int
claims_3yr: int
violations_3yr: int
credit_score: Optional[int] = None
marital_status: str = "single"
housing_type: str = "tenant"
class AutoInsuranceFeatureEngineer:
"""
Feature engineering per underwriting auto.
Produce 40+ feature da dati grezzi del richiedente,
includendo feature derivate, interazioni e encoding
domain-specific.
"""
VEHICLE_MAKE_RISK: Dict[str, int] = {
"Ferrari": 5, "Lamborghini": 5, "Porsche": 4,
"BMW": 3, "Mercedes": 3, "Audi": 3,
"Toyota": 1, "Honda": 1, "Volkswagen": 2,
"Ford": 2, "Fiat": 2, "Renault": 2,
}
def __init__(self, reference_date: Optional[date] = None):
self.reference_date = reference_date or date.today()
def engineer_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
features: Dict[str, float] = {}
features.update(self._demographic_features(applicant))
features.update(self._driving_experience_features(applicant))
features.update(self._vehicle_features(applicant))
features.update(self._claims_features(applicant))
features.update(self._geographic_features(applicant))
if applicant.credit_score is not None:
features.update(self._credit_features(applicant))
features.update(self._interaction_features(features))
return features
def _demographic_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
age = (self.reference_date - applicant.birth_date).days / 365.25
return {
"age": age,
"age_squared": age ** 2,
"age_under_25": float(age < 25),
"age_over_70": float(age > 70),
"age_risk_young": max(0.0, (25 - age) / 25) if age < 25 else 0.0,
"age_risk_senior": max(0.0, (age - 70) / 20) if age > 70 else 0.0,
"is_married": float(applicant.marital_status == "married"),
"is_homeowner": float(applicant.housing_type == "owner"),
}
def _driving_experience_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
years_licensed = (self.reference_date - applicant.license_date).days / 365.25
age = (self.reference_date - applicant.birth_date).days / 365.25
age_at_license = age - years_licensed
return {
"years_licensed": years_licensed,
"years_licensed_squared": years_licensed ** 2,
"age_at_first_license": age_at_license,
"late_license_ratio": max(0.0, (age_at_license - 18) / 10),
"is_new_driver": float(years_licensed < 2),
"is_experienced_driver": float(years_licensed > 10),
}
def _vehicle_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
vehicle_age = self.reference_date.year - applicant.vehicle_year
make_risk = self.VEHICLE_MAKE_RISK.get(applicant.vehicle_make, 2)
return {
"vehicle_age": float(vehicle_age),
"vehicle_value": applicant.vehicle_value,
"vehicle_value_log": np.log1p(applicant.vehicle_value),
"vehicle_make_risk_score": float(make_risk),
"is_high_performance": float(make_risk >= 4),
"is_new_vehicle": float(vehicle_age <= 2),
"is_old_vehicle": float(vehicle_age > 10),
"annual_mileage": float(applicant.annual_mileage),
"annual_mileage_log": np.log1p(applicant.annual_mileage),
"high_mileage": float(applicant.annual_mileage > 20000),
}
def _claims_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
claims = applicant.claims_3yr
violations = applicant.violations_3yr
return {
"claims_3yr": float(claims),
"violations_3yr": float(violations),
"has_any_claim": float(claims > 0),
"has_multiple_claims": float(claims > 1),
"has_violations": float(violations > 0),
# Score combinato ponderato: sinistri pesano 3x rispetto a infrazioni
"incident_score": claims * 3.0 + violations * 1.5,
"claims_x_violations": float(claims * violations),
}
def _geographic_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
# In produzione: lookup su DB geografici (ISTAT, OpenStreetMap, criminalita)
zip_hash = hash(applicant.zip_code) % 100
urban_score = (zip_hash % 5) / 4.0
crime_index = (zip_hash % 3) / 2.0
weather_risk = (zip_hash % 4) / 3.0
return {
"urban_density_score": urban_score,
"area_crime_index": crime_index,
"area_weather_risk": weather_risk,
"composite_geo_risk": (urban_score + crime_index + weather_risk) / 3,
}
def _credit_features(self, applicant: PolicyApplicant) -> Dict[str, float]:
score = applicant.credit_score or 0
return {
"credit_score": float(score),
"credit_score_normalized": (score - 300) / (850 - 300),
"poor_credit": float(score < 580),
"fair_credit": float(580 <= score < 670),
"good_credit": float(670 <= score < 740),
"excellent_credit": float(score >= 740),
}
def _interaction_features(self, features: Dict[str, float]) -> Dict[str, float]:
return {
# Giovane + auto sportiva = rischio molto alto
"young_high_perf": (
features.get("age_risk_young", 0) *
features.get("is_high_performance", 0)
),
# Sinistri + area ad alto crimine amplificano il rischio
"claims_urban": (
features.get("claims_3yr", 0) *
features.get("urban_density_score", 0)
),
# Mileage alto + veicolo vecchio = rischio meccanico aumentato
"mileage_old_vehicle": (
features.get("annual_mileage_log", 0) *
features.get("is_old_vehicle", 0)
),
}Modele punktacji ryzyka: podejścia i kompromisy
Wybór modelu uczenia maszynowego do punktacji ryzyka musi równoważyć dokładność przewidywalność, interpretowalność (podstawowa dla zgodności), szybkość wnioskowania i łatwość konserwacji. Oto główne podejścia branży ubezpieczeniowej:
Porównanie modeli punktacji ryzyka ubezpieczeniowego
| Model | Dokładność | Interpretowalność | Idealny przypadek użycia |
|---|---|---|---|
| GLM (Poissona/gamma) | Przeciętny | Bardzo wysoki | Wartość bazowa aktuarialna, akceptacja organów regulacyjnych |
| Losowy las | Wysoki | Przeciętny | Znaczenie funkcji, odporność na wartości odstające |
| XGBoost/LightGBM | Bardzo wysoki | Przeciętny | Produkcja standardowa, SOTA na danych tabelarycznych |
| Tabelaryczna sieć neuronowa | Wysoki | Niski | Złożone funkcje z osadzeniem kategorycznym |
Najbardziej ugruntowaną metodą w branży jest tzw model dwustopniowy: modelka częstotliwości (prawdopodobieństwo wystąpienia co najmniej jednego wypadku) i dotkliwości (oczekiwany koszt wypadku biorąc pod uwagę, że to nastąpi). Oczekiwana składka czysta wynosi: Częstotliwość x istotność.
import xgboost as xgb
from sklearn.metrics import mean_absolute_error, mean_squared_error
import numpy as np
import pandas as pd
from typing import Dict, Optional
class TwoStageRiskScorer:
"""
Modello a due stadi per pricing assicurativo auto.
Stage 1: Frequency model (Poisson regression con XGBoost)
Target = numero sinistri per polizza
Stage 2: Severity model (Tweedie/Gamma con XGBoost)
Target = importo sinistro, addestrato solo su polizze con sinistri
Pure Premium = E[Frequency] * E[Severity | has_claim]
"""
FREQUENCY_PARAMS: Dict = {
"objective": "count:poisson",
"eval_metric": "poisson-nloglik",
"max_depth": 6,
"learning_rate": 0.05,
"n_estimators": 500,
"min_child_weight": 50, # stabilità attuariale: min sinistri per leaf
"subsample": 0.8,
"colsample_bytree": 0.8,
"reg_alpha": 0.1,
"reg_lambda": 1.0,
"tree_method": "hist",
"early_stopping_rounds": 50,
}
SEVERITY_PARAMS: Dict = {
"objective": "reg:tweedie",
"tweedie_variance_power": 1.5, # 1=Poisson, 2=Gamma
"eval_metric": "tweedie-nloglik@1.5",
"max_depth": 5,
"learning_rate": 0.05,
"n_estimators": 300,
"min_child_weight": 30,
"subsample": 0.8,
"colsample_bytree": 0.7,
"reg_alpha": 0.1,
"reg_lambda": 1.0,
"tree_method": "hist",
"early_stopping_rounds": 30,
}
def __init__(self) -> None:
self.frequency_model = xgb.XGBRegressor(**self.FREQUENCY_PARAMS)
self.severity_model = xgb.XGBRegressor(**self.SEVERITY_PARAMS)
self.feature_names: list = []
def fit(
self,
X: pd.DataFrame,
y_claims: pd.Series,
y_amounts: pd.Series,
exposure: pd.Series,
eval_fraction: float = 0.2,
) -> "TwoStageRiskScorer":
"""
Addestra entrambi i modelli.
IMPORTANTE: usa split temporale, non random shuffle.
I dati assicurativi sono autocorrelati nel tempo.
"""
self.feature_names = X.columns.tolist()
split_idx = int(len(X) * (1 - eval_fraction))
X_train, X_val = X.iloc[:split_idx], X.iloc[split_idx:]
freq_train = y_claims.iloc[:split_idx]
freq_val = y_claims.iloc[split_idx:]
# Stage 1: Frequency
self.frequency_model.fit(
X_train, freq_train,
sample_weight=exposure.iloc[:split_idx],
eval_set=[(X_val, freq_val)],
verbose=50,
)
# Stage 2: Severity - solo su polizze con sinistri
has_claim = y_amounts > 0
X_sev = X[has_claim]
y_sev = y_amounts[has_claim]
sev_split = int(len(X_sev) * (1 - eval_fraction))
self.severity_model.fit(
X_sev.iloc[:sev_split], y_sev.iloc[:sev_split],
eval_set=[(X_sev.iloc[sev_split:], y_sev.iloc[sev_split:])],
verbose=30,
)
return self
def predict_pure_premium(
self, X: pd.DataFrame, exposure: float = 1.0
) -> np.ndarray:
"""Calcola il pure premium: E[Freq] * E[Severity]."""
freq = self.frequency_model.predict(X) * exposure
sev = self.severity_model.predict(X)
return freq * sev
def evaluate(self, X: pd.DataFrame, y_claims: pd.Series) -> Dict[str, float]:
pred = self.frequency_model.predict(X)
mae = mean_absolute_error(y_claims, pred)
rmse = float(np.sqrt(mean_squared_error(y_claims, pred)))
gini = self._gini_coefficient(y_claims.values, pred)
lift = self._lift_at_decile(y_claims.values, pred, 0.1)
return {
"mae": round(mae, 6),
"rmse": round(rmse, 6),
"gini_coefficient": round(gini, 4),
"lift_top_decile": round(lift, 4),
}
def _gini_coefficient(self, actual: np.ndarray, predicted: np.ndarray) -> float:
"""Gini coefficient: metrica attuariale standard per modelli di frequenza."""
idx = np.argsort(predicted)
cum = np.cumsum(actual[idx])
cum_norm = cum / cum[-1]
n = len(actual)
lorenz_area = float(np.sum(cum_norm)) / n
return 2 * (lorenz_area - 0.5)
def _lift_at_decile(
self, actual: np.ndarray, predicted: np.ndarray, decile: float
) -> float:
k = max(1, int(len(actual) * decile))
top_idx = np.argsort(predicted)[-k:]
base_rate = actual.mean()
if base_rate == 0:
return 0.0
return float(actual[top_idx].mean() / base_rate)Interpretowalność za pomocą SHAP: decyzje podlegające audytowi
W kontekście regulowanym, takim jak ubezpieczeniowy, model czarnej skrzynki nie jest wystarczający. Przepisy wymagają, aby decyzje ubezpieczeniowe były zrozumiałe: dla klienta (prawo do wyjaśnienia RODO), dla underwriterów (przegląd przypadków granicznych) i dla regulatorów (Filar 3 Wypłacalność II, ORSA). Narzędziem referencyjnym jest SHAP (SHapley Additive exPlanations). branży w zakresie możliwości interpretacji post-hoc modeli zespołowych.
import shap
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
class UnderwritingExplainer:
"""
Spiegazioni SHAP per decisioni underwriting.
Genera output a tre livelli: cliente, underwriter, compliance.
"""
FEATURE_LABELS: Dict[str, str] = {
"age": "eta del guidatore",
"years_licensed": "anni di patente",
"claims_3yr": "sinistri negli ultimi 3 anni",
"violations_3yr": "infrazioni negli ultimi 3 anni",
"vehicle_make_risk_score": "categoria rischio veicolo",
"vehicle_age": "anzianita del veicolo",
"vehicle_value": "valore del veicolo",
"annual_mileage": "chilometraggio annuo dichiarato",
"composite_geo_risk": "rischio della zona geografica",
"credit_score": "score creditizio",
"young_high_perf": "combinazione giovane + veicolo sportivo",
}
def __init__(self, model, feature_names: List[str]) -> None:
self.feature_names = feature_names
self.explainer = shap.TreeExplainer(model)
def explain(
self, X_row: pd.DataFrame, risk_score: float
) -> Dict:
"""Spiegazione completa per una singola valutazione."""
shap_values = self.explainer.shap_values(X_row)
impacts: List[Tuple[str, float]] = sorted(
zip(self.feature_names, shap_values[0]),
key=lambda x: abs(x[1]),
reverse=True
)
return {
"risk_score": round(risk_score, 2),
"decision": self._score_to_decision(risk_score),
"customer_message": self._customer_message(impacts, risk_score),
"top_risk_factors": [
{
"name": name,
"label": self.FEATURE_LABELS.get(name, name),
"direction": "aumenta rischio" if shap > 0 else "riduce rischio",
"magnitude": round(abs(shap), 4),
}
for name, shap in impacts[:5]
],
"audit_trail": {
"base_expected_value": float(self.explainer.expected_value),
"all_shap_values": {
n: round(float(s), 6)
for n, s in zip(self.feature_names, shap_values[0])
},
"input_features": X_row.to_dict(orient="records")[0],
},
}
def _customer_message(
self, impacts: List[Tuple[str, float]], score: float
) -> str:
high = [(n, v) for n, v in impacts if abs(v) > 0.1]
if not high:
return "Il tuo profilo rientra nella fascia di rischio standard."
positivi = [self.FEATURE_LABELS.get(n, n) for n, v in high[:3] if v < 0]
negativi = [self.FEATURE_LABELS.get(n, n) for n, v in high[:3] if v > 0]
parts = []
if negativi:
parts.append(f"Fattori che aumentano il profilo di rischio: {', '.join(negativi)}.")
if positivi:
parts.append(f"Fattori a tuo favore: {', '.join(positivi)}.")
return " ".join(parts)
def _score_to_decision(self, score: float) -> str:
if score < 30:
return "ACCEPT_PREFERRED"
elif score < 60:
return "ACCEPT_STANDARD"
elif score < 80:
return "ACCEPT_SUBSTANDARD"
return "DECLINE_OR_MANUAL_REVIEW"Wykrywanie uczciwości i uprzedzeń w kontekście UE
Stosowanie zmiennych zastępczych (kod pocztowy, ocena zdolności kredytowej) może wprowadzić dyskryminację pośrednią zabronione przez prawo. W Europie dyrektywa dotycząca równości płci (potwierdzona przez wyrok Trybunału Sprawiedliwości UE w sprawie Test-Achats z 2011 r.) zabrania stosowania płci przy ustalaniu cen ubezpieczenie. Ustawa o sztucznej inteligencji dodaje ograniczenia dla systemów wysokiego ryzyka sklasyfikowanych w załączniku III, wymaganie obowiązkowych ocen zgodności przed wdrożeniem.
import pandas as pd
import numpy as np
from sklearn.metrics import confusion_matrix
from typing import Dict, List
class FairnessAuditor:
"""
Auditor di fairness per modelli underwriting (EU-compliant).
Metriche implementate:
- Disparate Impact (regola 80%)
- Demographic Parity Gap
- Equal Opportunity (TPR parity)
- Calibration by group
"""
DISPARATE_IMPACT_THRESHOLD = 0.8 # EEOC 80% rule
MAX_DP_GAP = 0.1 # linee guida EIOPA
def __init__(
self,
predictions: np.ndarray,
true_labels: np.ndarray,
sensitive_df: pd.DataFrame,
) -> None:
self.predictions = predictions
self.true_labels = true_labels
self.sensitive_df = sensitive_df
def full_audit(self) -> Dict:
results: Dict = {}
for attr in self.sensitive_df.columns:
groups = self.sensitive_df[attr].unique()
attr_results: Dict = {}
for group in groups:
mask = self.sensitive_df[attr] == group
g_pred = self.predictions[mask]
g_true = self.true_labels[mask]
attr_results[str(group)] = {
"count": int(mask.sum()),
"acceptance_rate": float((g_pred < 0.6).mean()),
"avg_score": round(float(g_pred.mean()), 4),
"tpr": self._tpr(g_true, g_pred),
}
di = self._disparate_impact(attr_results)
dp = self._dp_gap(attr_results)
attr_results["_metrics"] = {
"disparate_impact": round(di, 4),
"demographic_parity_gap": round(dp, 4),
"passes_di_rule": di >= self.DISPARATE_IMPACT_THRESHOLD,
"passes_dp_rule": dp <= self.MAX_DP_GAP,
"overall_fair": di >= self.DISPARATE_IMPACT_THRESHOLD and dp <= self.MAX_DP_GAP,
}
results[attr] = attr_results
return results
def _tpr(self, labels: np.ndarray, preds: np.ndarray, thr: float = 0.5) -> float:
if len(labels) < 10:
return float("nan")
binary = (preds >= thr).astype(int)
try:
tn, fp, fn, tp = confusion_matrix(labels, binary, labels=[0, 1]).ravel()
return round(tp / (tp + fn), 4) if (tp + fn) > 0 else 0.0
except ValueError:
return float("nan")
def _disparate_impact(self, groups: Dict) -> float:
rates = [v["acceptance_rate"] for k, v in groups.items()
if not k.startswith("_") and isinstance(v, dict)]
if not rates or max(rates) == 0:
return 1.0
return min(rates) / max(rates)
def _dp_gap(self, groups: Dict) -> float:
rates = [v["acceptance_rate"] for k, v in groups.items()
if not k.startswith("_") and isinstance(v, dict)]
return (max(rates) - min(rates)) if rates else 0.0MLOps i monitorowanie w produkcji
Modele underwritingu podlegają dryf koncepcji częste: profil zmiany wnioskodawców (nowe modele pojazdów elektrycznych, zmiany demograficzne), koszty napraw cierpi na inflację, ekstremalne zjawiska klimatyczne zmieniają schemat strat. System ciągłego monitorowania z Wskaźnik stabilności populacji (PSI) tj konieczne jest określenie, kiedy model wymaga ponownego holowania.
from scipy import stats
import numpy as np
import pandas as pd
from typing import Dict, List
from datetime import datetime
class DriftMonitor:
"""
Monitora data drift per modelli underwriting.
Usa PSI (Population Stability Index) come metrica primaria.
PSI interpretation:
- PSI < 0.1: Nessun cambiamento significativo
- PSI 0.1-0.25: Cambiamento moderato, monitorare
- PSI > 0.25: Cambiamento significativo, retraining consigliato
"""
def __init__(self, reference_df: pd.DataFrame, features: List[str]) -> None:
self.reference_df = reference_df
self.features = features
def check_drift(self, current_df: pd.DataFrame) -> Dict:
feature_results: Dict = {}
critical_features = []
for feat in self.features:
if feat not in current_df.columns:
continue
psi = self._psi(self.reference_df[feat], current_df[feat])
ks_stat, ks_p = stats.ks_2samp(
self.reference_df[feat].dropna(),
current_df[feat].dropna()
)
status = "ok" if psi < 0.1 else ("warning" if psi < 0.25 else "critical")
feature_results[feat] = {
"psi": round(psi, 4),
"ks_statistic": round(ks_stat, 4),
"ks_pvalue": round(ks_p, 4),
"status": status,
}
if status == "critical":
critical_features.append(feat)
avg_psi = float(np.mean([v["psi"] for v in feature_results.values()]))
return {
"checked_at": datetime.now().isoformat(),
"overall_psi": round(avg_psi, 4),
"retraining_recommended": avg_psi > 0.1,
"critical_features": critical_features,
"feature_details": feature_results,
}
def _psi(self, ref: pd.Series, cur: pd.Series, bins: int = 10) -> float:
ref_clean = ref.dropna().values
cur_clean = cur.dropna().values
edges = np.percentile(ref_clean, np.linspace(0, 100, bins + 1))
edges = np.unique(edges)
ref_counts, _ = np.histogram(ref_clean, bins=edges)
cur_counts, _ = np.histogram(cur_clean, bins=edges)
ref_pct = (ref_counts + 1e-10) / len(ref_clean)
cur_pct = (cur_counts + 1e-10) / len(cur_clean)
return float(np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct)))Najlepsze praktyki i anty-wzorce
Najlepsze praktyki w zakresie underwritingu AI
- Architektura dwustopniowa (częstotliwość/dotkliwość): oraz standard aktuarialny i pozwala na dokładniejszą wycenę kwoty roszczenia niż w przypadku pojedynczego modelu
- Obowiązkowy podział czasu: dane ubezpieczeniowe są autokorelowane w czasie; nigdy nie używaj losowego losowania do podziału pociągu/testu
- Ekspozycja jako offset: zawsze używaj czasu trwania polisy (ekspozycja w latach) jako przesunięcia w modelu Poissona, aby znormalizować liczbę roszczeń
- Utrzymuj bazowy GLM: uogólnione modele liniowe są łatwiej weryfikowane przez organy regulacyjne i zapewniają punkty odniesienia do oceny wartości dodanej prania pieniędzy
- Tryb cienia przed uruchomieniem: Uruchom model równolegle z underwritingiem ludzkim przez 30–90 dni, porównując decyzje przed automatyzacją
- Cotygodniowe monitorowanie PSI: Dryf w sektorze samochodowym jest częsty ze względu na nowe modele pojazdów, inflację kosztów napraw i zmiany regulacyjne
Anty-wzorce, których należy unikać
- Wyciek funkcji: nigdy nie używaj zmiennych dostępnych dopiero po złożeniu wniosku (kwota roszczenia, rezerwa) jako cech szkoleniowych modelu częstotliwości
- Optymalizuj tylko AUC: w sektorze ubezpieczeń odpowiednimi miarami są współczynnik Giniego, współczynnik mieszany i wzrost w górnym decylu ryzyka
- Modele z ponad 500 funkcjami: niemożliwe do zatwierdzenia aktuarialnego i uzasadnienia przed regulatorem; preferują rygorystyczny wybór funkcji (maks. 40-60 funkcji)
- Ignorowanie koncentracji portfela: model, który akceptuje jedynie bardzo niskie profile ryzyka, tworzy portfel zapobiegający selekcji i niezrównoważony
- Dyskryminacja zastępcza: zmienne takie jak kod pocztowy mogą być wskaźnikami pochodzenia etnicznego; zawsze sprawdzaj odmienny wpływ przed wdrożeniem
Wnioski i dalsze kroki
Underwriting AI nie zastępuje ludzkiego gwaranta, ale go wzmacnia: decyzje dotyczące Polisy standardowe (80-90% wolumenu) można z dużą dokładnością w pełni zautomatyzować wyższa niż średnia ludzka, uwalniając specjalistów od skomplikowanych przypadków, w których doświadczenie panowanie i niezastąpione.
Klucze do udanego systemu to: oparta na wiedzy głęboka inżynieria funkcji aktuarialna, dwustopniowa architektura częstotliwości/dotkliwości, interpretowalność SHAP pod kątem zgodności, obowiązkowy audyt uczciwości i ciągłe monitorowanie za pomocą PSI w celu zarządzania znoszeniem.
Kolejny artykuł z tej serii poświęcony jestautomatyzacja roszczeń za pomocą Computer Vision i NLP: od cyfrowego FNOL po automatyczną fotograficzną ocenę uszkodzeń, aż do przyspieszone rozliczenie typu end-to-end.
Seria inżynieryjna InsurTech
- 01 - Domena ubezpieczeniowa dla programistów: produkty, aktorzy i modele danych
- 02 — Zarządzanie polityką w chmurze: architektura oparta na interfejsie API
- 03 – Rurociąg telematyczny: przetwarzanie danych UBI na dużą skalę
- 04 – Underwriting AI: Inżynieria cech i punktacja ryzyka (ten artykuł)
- 05 - Automatyzacja roszczeń: wizja komputerowa i NLP
- 06 - Wykrywanie oszustw: analiza wykresów i sygnał behawioralny
- 07 - Integracja standardu ACORD i API ubezpieczeń
- 08 - Inżynieria zgodności: Wypłacalność II i MSSF 17