Build a Carrier Vetting View

import requests

API_KEY = "YOUR_API_KEY"

def fetch_carrier(dot_number: str) -> dict | None:
    """Fetch a full carrier profile by DOT number."""
    resp = requests.get(
        "https://api.carrierok.com/v2/profile",
        params={"dot_number": dot_number},
        headers={"Authorization": f"Bearer {API_KEY}"},
    )
    resp.raise_for_status()
    items = resp.json()["items"]
    return items[0] if items else None

carrier = fetch_carrier("568253")
print(f"Evaluating: {carrier['legal_name']}")

def check_authority(c: dict) -> dict:
    """Check authority status, age, and revocation history."""
    issues = []

    # Active authority required
    if c["authority_common"] != "Active":
        issues.append(f"Common authority: {c['authority_common']}")
    if c["authority_contract"] not in ("Active", "None"):
        issues.append(f"Contract authority: {c['authority_contract']}")

    # Authority age — flag new entrants (< 18 months)
    age_days = int(c.get("authority_age_common", 0))
    age_months = age_days // 30

    # Revocation history
    revocations = int(c.get("total_revocations", 0))
    if revocations > 0:
        issues.append(
            f"{revocations} prior revocation(s), "
            f"last {c.get('days_since_last_revocation', '?')} days ago"
        )

    return {
        "status": c["authority_common"],
        "age_days": age_days,
        "age_months": age_months,
        "is_new_entrant": age_days < 540,
        "revocations": revocations,
        "pass": c["authority_common"] == "Active" and len(issues) == 0,
        "issues": issues,
    }

def check_insurance(c: dict) -> dict:
    """Validate BIPD and cargo insurance coverage."""
    issues = []

    bipd_on_file = int(c.get("insurance_bipd_on_file", 0))
    bipd_required = int(c.get("insurance_bipd_required", 0))
    cargo_on_file = int(c.get("insurance_cargo_on_file", 0))

    # BIPD must meet or exceed required minimum
    bipd_adequate = bipd_on_file >= bipd_required if bipd_required > 0 else True
    if not bipd_adequate:
        issues.append(
            f"BIPD on file (${bipd_on_file:,}) "
            f"below required (${bipd_required:,})"
        )

    # Cargo insurance (not always required, but good to flag if absent)
    cargo_required = int(c.get("insurance_cargo_required", 0))
    if cargo_required > 0 and cargo_on_file < cargo_required:
        issues.append(
            f"Cargo on file (${cargo_on_file:,}) "
            f"below required (${cargo_required:,})"
        )

    return {
        "bipd_on_file": bipd_on_file,
        "bipd_required": bipd_required,
        "bipd_adequate": bipd_adequate,
        "cargo_on_file": cargo_on_file,
        "indicator": c.get("indicator_insurance", False),
        "pass": bipd_adequate and len(issues) == 0,
        "issues": issues,
    }

from datetime import datetime, timedelta

def check_fleet(c: dict) -> dict:
    """Analyze fleet size, composition, and MCS-150 freshness."""
    issues = []

    power_units = int(c.get("total_power_units", 0))
    drivers = int(c.get("total_drivers", 0))
    cdl_drivers = int(c.get("total_drivers_cdl", 0))

    # MCS-150 freshness — should be updated every 2 years
    mcs150_date = c.get("mcs150_date", "")
    mcs150_stale = False
    if mcs150_date:
        last_update = datetime.strptime(mcs150_date, "%Y-%m-%d")
        if datetime.now() - last_update > timedelta(days=730):
            mcs150_stale = True
            issues.append(
                f"MCS-150 last updated {mcs150_date} "
                f"(over 2 years ago)"
            )

    # Fleet size category
    if power_units == 0:
        fleet_size = "unknown"
        issues.append("No power units reported")
    elif power_units <= 5:
        fleet_size = "owner-operator"
    elif power_units <= 50:
        fleet_size = "small"
    elif power_units <= 500:
        fleet_size = "mid-size"
    else:
        fleet_size = "large"

    return {
        "power_units": power_units,
        "drivers": drivers,
        "cdl_drivers": cdl_drivers,
        "fleet_size": fleet_size,
        "cargo_carried": c.get("cargo_carried", ""),
        "mcs150_date": mcs150_date,
        "mcs150_stale": mcs150_stale,
        "pass": not mcs150_stale and power_units > 0,
        "issues": issues,
    }

def check_fraud(c: dict) -> dict:
    """Detect fraud indicators and chameleon carrier patterns."""
    issues = []

    risk = c.get("risk_score", "Unknown")
    risk_prob = c.get("risk_score_probability", 0)

    if risk in ("High", "Very High"):
        issues.append(f"Risk score: {risk} ({risk_prob:.0%} probability)")

    # Contact churn — frequent changes suggest identity manipulation
    name_changes = int(c.get("name_change_count", 0))
    phone_changes = int(c.get("phone_change_count", 0))
    address_changes = int(c.get("address_change_count", 0))
    email_changes = int(c.get("email_change_count", 0))

    total_changes = name_changes + phone_changes + address_changes
    if total_changes > 10:
        issues.append(
            f"High contact churn: {name_changes} name, "
            f"{phone_changes} phone, {address_changes} address changes"
        )

    # Revocation + young authority = possible chameleon
    dot_age = int(c.get("dot_age", 9999))
    revocations = int(c.get("total_revocations", 0))
    if dot_age < 540 and revocations > 0:
        issues.append(
            "Possible chameleon: new entity with prior revocations"
        )

    return {
        "risk_score": risk,
        "risk_probability": risk_prob,
        "contact_changes": total_changes,
        "name_changes": name_changes,
        "phone_changes": phone_changes,
        "address_changes": address_changes,
        "pass": risk not in ("High", "Very High") and len(issues) == 0,
        "issues": issues,
    }

==================================================
CARRIER: HANSEN & ADKINS AUTO TRANSPORT INC
DOT:     568253
VERDICT: REVIEW
==================================================

Issues (2):
  - 2 prior revocation(s), last 7954 days ago
  - Vehicle OOS rate 24.0% exceeds 1.5x national avg (22.3%)

Authority: Active (264 months)
Insurance: BIPD $1,000,000 / $1,000,000
Safety:    ISS 45 (moderate)
Fleet:     897 power units, 897 drivers
Fraud:     Low risk