How Insurance Companies Use Data Structures

# Real-world pattern: policy lookup by ID

policies = {}  # hashmap of policy_id -> policy

def load_policies():
    for row in db.query("SELECT id, premium, status FROM policies"):
        policies[row["id"]] = row

def get_policy(policy_id):
    # O(1) lookup — millions of times per day
    return policies.get(policy_id)

# Fast. Simple. Essential for real-time systems.

# Customer 360 view using nested hashmaps

customer_policies = {}  # customer_id -> list of policies

for policy in policies:
    customer_id = policy["customer_id"]
    customer_policies \
        .setdefault(customer_id, []) \
        .append(policy)

def total_premium(customer_id):
    return sum(p["premium"]
               for p in customer_policies.get(customer_id, []))

# Instant aggregation across all products for a customer.
# Powers dashboards, upsell offers, and renewal reminders.

# Simplified underwriting decision tree

def decide_underwriting(applicant):
    if applicant["age"] > 60:
        if applicant["smoker"]:
            return {"decision": "decline",
                    "reason": "age + smoking"}
        return {"decision": "approve",
                "loading": 1.2}
    if applicant["bmi"] > 35:
        return {"decision": "approve",
                "loading": 1.15}
    if applicant["history_claims"] > 3:
        return {"decision": "refer",
                "reason": "claims history"}
    return {"decision": "approve", "loading": 1.0}

# Real systems use this logic at scale.
# Trees make rules explicit + auditable.

# Balanced tree for fast range lookups
# e.g. "find all policies with premium between X and Y"

from sortedcontainers import SortedDict

premiums = SortedDict()  # (premium, policy_id) -> policy

for policy in policies:
    premiums[(policy["premium"], policy["id"])] = policy

def policies_in_range(low, high):
    return list(premiums.irange((low,), (high,)))

# Underwriting dashboards use this for
# segmentation, reporting, and audits.
# Balanced trees → O(log n) range queries.

# Building a simple fraud graph

import networkx as nx

g = nx.Graph()

# Add claims and link them by shared attributes
for claim in claims:
    cid = claim["claim_id"]
    g.add_node(cid, type="claim")

    for attr in ["address", "phone", "bank_account"]:
        if claim.get(attr):
            g.add_node(claim[attr], type=attr)
            g.add_edge(cid, claim[attr])

# Suspicious clusters: subgraphs where many claims
# share the same address / phone / bank account.
# Graph analysis makes fraud rings obvious.

# Detecting fraud rings via connected components

for component in nx.connected_components(g):
    claims_in_component = [
        n for n in component if g.nodes[n]["type"] == "claim"
    ]

    if len(claims_in_component) > 5:
        print(f"Potential fraud ring: {len(claims_in_component)} claims")
        for cid in claims_in_component:
            print(f"  Claim {cid}")
            # Refer to SIU (Special Investigation Unit)

# Real systems add graph algorithms:
# - PageRank to rank suspicious entities
# - Community detection to cluster
# - Betweenness to find orchestrators
# - Graph ML for predictive fraud scoring

# Simple FIFO claims intake (Python)

from collections import deque

claims_queue = deque()

def submit_claim(claim):
    claims_queue.append(claim)
    print(f"Claim {claim['id']} added to queue")

def process_next():
    if not claims_queue:
        return None
    claim = claims_queue.popleft()
    return claim

# Real systems use Kafka / SQS / RabbitMQ —
# but the underlying pattern is a queue.

# Priority queue for urgent claims

import heapq

priority_queue = []  # (score, claim)
counter = 0          # tie-breaker

def submit_claim(claim):
    global counter
    priority = compute_priority(claim)  # higher = more urgent
    heapq.heappush(priority_queue, (-priority, counter, claim))
    counter += 1

def process_next():
    if not priority_queue:
        return None
    return heapq.heappop(priority_queue)[2]

# High-value, fast-track, and urgent claims
# get processed first — no matter the arrival time.

# Priority scoring — real factors

def compute_priority(claim):
    score = 0

    # High value claims first
    if claim["amount"] > 1_000_000:
        score += 50

    # Fast-track lanes
    if claim["type"] == "auto-total-loss":
        score += 30

    # Elderly / vulnerable customers
    if claim["customer_age"] >= 65:
        score += 20

    # Repeat or known-fraud flags
    if claim.get("fraud_flag"):
        score += 40

    # Time-sensitive
    if claim["days_since_intake"] > 5:
        score += 25

    return score

# Rules → numbers → heap → fair, fast processing.