Build an Insurance Underwriting Risk Check

"""Carrier underwriting risk check — CarrierOk API.

Illustrative starting framework — not underwriting, legal, or compliance
advice. Validate thresholds against your own book and model governance.

Usage:
    export CARRIEROK_API_KEY=sk_test_your_key_here
    python risk_check.py 1234567

Field-name notes (these are the API's real names — don't "fix" them):
  - BASIC suffixes: unsafe_driving, hours_of_service, driver_fitness,
    controlled_substance, vehicle_maintence, hazardous_materials,
    crash_indicator
  - iss_recommendation values are UPPERCASE: INSPECT / OPTIONAL / PASS
  - indicator_authority / indicator_insurance are TRI-STATE
    (true / false / null): null means "not applicable / nothing required"
    — normal for intrastate and private carriers — and must not be
    treated as a failure
  - Numeric counters arrive as strings; fields with no value for a carrier
    are omitted from the response entirely (not returned as null)
"""
import os
import sys
import requests

API_KEY = os.environ["CARRIEROK_API_KEY"]  # sk_test_... or sk_live_...
BASE = "https://api.carrierok.com/v2"

# FMCSA weights these three as the strongest crash predictors and applies
# lower intervention thresholds to them.
CRASH_PREDICTIVE = ("unsafe_driving", "hours_of_service", "crash_indicator")
ALL_BASICS = CRASH_PREDICTIVE + (
    "driver_fitness", "controlled_substance",
    "vehicle_maintence", "hazardous_materials",
)


def get_profile(dot_number: str) -> dict:
    resp = requests.get(
        f"{BASE}/profile",
        params={"dot_number": dot_number},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=15,
    )
    if resp.status_code == 429:
        raise RuntimeError(
            f"Rate limited — retry after {resp.headers.get('Retry-After')}s")
    resp.raise_for_status()
    items = resp.json().get("items", [])
    if not items:
        raise LookupError(f"No carrier found for DOT {dot_number}")
    return items[0]


def assess(p: dict) -> tuple[str, list[str]]:
    reasons = []

    # MC authority and FMCSA insurance filings are interstate for-hire
    # requirements. Intrastate carriers are regulated by their state —
    # for them the key registration check is an active USDOT.
    interstate_for_hire = (
        p.get("carrier_operation_desc") == "interstate"
        and p.get("operation_class_authorized_for_hire") is True
    )

    # ── Tier 1: disqualifiers ──────────────────────────────────────────
    if p.get("usdot_status") != "Active":
        reasons.append("USDOT registration not active "
                       f"(usdot_status={p.get('usdot_status')})")
    if p.get("out_of_service_flag"):
        reasons.append("Active out-of-service order")
    if interstate_for_hire and p.get("indicator_authority") is not True:
        reasons.append("Interstate for-hire without active operating "
                       f"authority (common={p.get('authority_common')}, "
                       f"contract={p.get('authority_contract')})")
    elif p.get("indicator_authority") is False:
        reasons.append("Operating authority explicitly inactive or revoked")
    if p.get("indicator_insurance") is False:
        # Only ever false where FMCSA requires a filing (null otherwise),
        # so this check self-gates for intrastate/private carriers.
        reasons.append("Insurance on file below FMCSA requirement "
                       f"(BIPD on file: {p.get('insurance_bipd_on_file', '0')}, "
                       f"required: {p.get('insurance_bipd_required', '?')})")
    if reasons:
        return "DECLINE / REFER", reasons

    # ── Tier 2: high-risk signals ──────────────────────────────────────
    if p.get("risk_score") == "High":
        reasons.append(f"CarrierOk composite risk score: High "
                       f"(p={p.get('risk_score_probability')})")
    if p.get("iss_recommendation") == "INSPECT":
        reasons.append(f"ISS recommendation: INSPECT "
                       f"(ISS {p.get('iss_value')})")
    if p.get("safety_rating_desc") in ("Conditional", "Unsatisfactory"):
        reasons.append(f"FMCSA safety rating: {p['safety_rating_desc']}")

    crash_alerts = [b for b in CRASH_PREDICTIVE
                    if p.get(f"basic_alert_{b}")]
    all_alerts = [b for b in ALL_BASICS if p.get(f"basic_alert_{b}")]
    if crash_alerts:
        reasons.append("Alert in crash-predictive BASIC(s): "
                       + ", ".join(crash_alerts))
    if len(all_alerts) >= 2:
        reasons.append(f"{len(all_alerts)} BASIC alerts: "
                       + ", ".join(all_alerts))
    if reasons:
        return "HIGH RISK", reasons

    # ── Tier 3: enhanced review (above standard, below high-risk) ─────
    if p.get("risk_score") == "Medium":
        reasons.append("CarrierOk composite risk score: Medium")
    if len(all_alerts) == 1:
        reasons.append(f"Single BASIC alert: {all_alerts[0]}")
    if int(p.get("dot_age") or "0") < 540:
        reasons.append("New entrant (DOT registered < 540 days)")
    if reasons:
        return "ENHANCED REVIEW", reasons

    # ── Tier 4: LOW RISK requires positive evidence; the unremarkable
    #    carrier (risk_score null/absent, no signals) is STANDARD REVIEW.
    if p.get("risk_score") == "Low":
        return "LOW RISK", ["Composite risk score: Low",
                            "No BASIC alerts, registration and filings clear"]
    return "STANDARD REVIEW", [
        f"Risk score: {p.get('risk_score', 'not computed')}",
        "No adverse signals — apply your standard workflow"]


if __name__ == "__main__":
    profile = get_profile(sys.argv[1])
    verdict, reasons = assess(profile)
    print(f"{profile.get('legal_name')} (DOT {profile.get('dot_number')})")
    print(f"Verdict: {verdict}")
    for r in reasons:
        print(f"  - {r}")

$ python risk_check.py 1234567
EXAMPLE TRANSPORT LLC (DOT 1234567)
Verdict: HIGH RISK
  - Alert in crash-predictive BASIC(s): hours_of_service