Customer Domain

The customer_domain bounded context owns the customer entity lifecycle — from signup through churn. It is the anchor that all other domains reference via customer_id.

Entities

src.domain.customer.entities

Customer entity – core of the Customer bounded context.

An Entity has a unique identity (customer_id) and mutable lifecycle state. Business rules about customers live here, not in the API or infrastructure layers.

Classes

Customer dataclass

Represents a B2B SaaS customer account.

The customer entity owns lifecycle state (active vs churned) and exposes domain methods that encapsulate churn-relevant business rules.

Parameters:

Name	Type	Description	Default
customer_id	str	Unique identifier (UUID string).	required
industry	Industry	Vertical segment, used for cohort analysis.	required
plan_tier	PlanTier	Commercial tier (starter / growth / enterprise).	required
signup_date	date	Date of first contract activation.	required
mrr	MRR	Monthly Recurring Revenue value object.	required
churn_date	date | None	Date of cancellation; None means still active (right-censored).	None

Source code in src/domain/customer/entities.py

@dataclass
class Customer:
    """Represents a B2B SaaS customer account.

    The customer entity owns lifecycle state (active vs churned) and exposes
    domain methods that encapsulate churn-relevant business rules.

    Args:
        customer_id: Unique identifier (UUID string).
        industry: Vertical segment, used for cohort analysis.
        plan_tier: Commercial tier (starter / growth / enterprise).
        signup_date: Date of first contract activation.
        mrr: Monthly Recurring Revenue value object.
        churn_date: Date of cancellation; None means still active (right-censored).
    """

    customer_id: str
    industry: Industry
    plan_tier: PlanTier
    signup_date: date
    mrr: MRR
    churn_date: date | None = field(default=None)

    @property
    def is_active(self) -> bool:
        """True if the customer has not churned."""
        return self.churn_date is None

    @property
    def tenure_days(self) -> int:
        """Days from signup to churn (or today if still active).

        Used as the time axis in survival analysis models.
        """
        end_date = self.churn_date if self.churn_date else date.today()
        return (end_date - self.signup_date).days

    @property
    def is_early_stage(self) -> bool:
        """True if customer is within the critical first-90-day onboarding window.

        20–25% of voluntary churn occurs in this window (per Forrester data).
        """
        return self.tenure_days <= 90

    @property
    def annual_revenue_at_risk(self) -> str:
        """Human-readable annual revenue that would be lost if this customer churns."""
        return str(self.mrr.revenue_at_risk)

    def mark_churned(self, churn_date: date) -> None:
        """Record the churn event.

        Args:
            churn_date: The date cancellation was confirmed.

        Raises:
            ValueError: If churn_date precedes signup_date or customer already churned.
        """
        if not self.is_active:
            raise ValueError(f"Customer {self.customer_id} has already churned.")
        if churn_date < self.signup_date:
            raise ValueError(f"churn_date {churn_date} cannot precede signup_date {self.signup_date}")
        self.churn_date = churn_date

Attributes

is_active property

is_active: bool

True if the customer has not churned.

tenure_days property

tenure_days: int

Days from signup to churn (or today if still active).

Used as the time axis in survival analysis models.

is_early_stage property

is_early_stage: bool

True if customer is within the critical first-90-day onboarding window.

20–25% of voluntary churn occurs in this window (per Forrester data).

annual_revenue_at_risk property

annual_revenue_at_risk: str

Human-readable annual revenue that would be lost if this customer churns.

Functions

mark_churned

mark_churned(churn_date: date) -> None

Record the churn event.

Parameters:

Name	Type	Description	Default
churn_date	date	The date cancellation was confirmed.	required

Raises:

Type	Description
ValueError	If churn_date precedes signup_date or customer already churned.

Source code in src/domain/customer/entities.py

def mark_churned(self, churn_date: date) -> None:
    """Record the churn event.

    Args:
        churn_date: The date cancellation was confirmed.

    Raises:
        ValueError: If churn_date precedes signup_date or customer already churned.
    """
    if not self.is_active:
        raise ValueError(f"Customer {self.customer_id} has already churned.")
    if churn_date < self.signup_date:
        raise ValueError(f"churn_date {churn_date} cannot precede signup_date {self.signup_date}")
    self.churn_date = churn_date

Value Objects

src.domain.customer.value_objects

Value objects for the Customer domain.

Value objects are immutable and compared by value, not identity. They encapsulate business rules about what constitutes a valid value.

Classes

PlanTier

Bases: StrEnum

The commercial tier a customer is on.

Tier correlates with feature access, support SLA, and churn risk profile. Enterprise customers have dedicated CSMs and lower base churn rates. CUSTOM represents bespoke enterprise contracts — the ceiling of the tier ladder. FREE represents the freemium tier (zero MRR); the highest-leverage conversion event in SaaS is free → paid when a customer hits a feature data-sharing limit.

Source code in src/domain/customer/value_objects.py

class PlanTier(StrEnum):
    """The commercial tier a customer is on.

    Tier correlates with feature access, support SLA, and churn risk profile.
    Enterprise customers have dedicated CSMs and lower base churn rates.
    CUSTOM represents bespoke enterprise contracts — the ceiling of the tier ladder.
    FREE represents the freemium tier (zero MRR); the highest-leverage conversion
    event in SaaS is free → paid when a customer hits a feature data-sharing limit.
    """

    FREE = "free"
    STARTER = "starter"
    GROWTH = "growth"
    ENTERPRISE = "enterprise"
    CUSTOM = "custom"

Industry

Bases: StrEnum

Vertical industry segment.

Used for cohort segmentation and industry-specific churn benchmarking.

Source code in src/domain/customer/value_objects.py

class Industry(StrEnum):
    """Vertical industry segment.

    Used for cohort segmentation and industry-specific churn benchmarking.
    """

    FINTECH = "FinTech"
    HEALTHTECH = "HealthTech"
    LEGALTECH = "LegalTech"
    HR_TECH = "HR Tech"
    EDTECH = "EdTech"
    INSURTECH = "InsurTech"
    PROPTECH = "PropTech"
    RETAILTECH = "RetailTech"
    OTHER = "Other"

MRR dataclass

Monthly Recurring Revenue in USD.

Enforces non-negative constraint. Used for business impact calculations (e.g. revenue at risk = MRR × churn_probability).

Source code in src/domain/customer/value_objects.py

@dataclass(frozen=True)
class MRR:
    """Monthly Recurring Revenue in USD.

    Enforces non-negative constraint. Used for business impact calculations
    (e.g. revenue at risk = MRR × churn_probability).
    """

    amount: Decimal

    def __post_init__(self) -> None:
        if self.amount < Decimal("0"):
            raise ValueError(f"MRR cannot be negative, got {self.amount}")

    @classmethod
    def from_float(cls, value: float) -> MRR:
        """Create MRR from a float, rounding to 2 decimal places."""
        return cls(amount=Decimal(str(round(value, 2))))

    @property
    def revenue_at_risk(self) -> Decimal:
        """Annual revenue at risk if customer churns (MRR × 12)."""
        return self.amount * Decimal("12")

    def __str__(self) -> str:
        return f"${self.amount:,.2f}"

Attributes

revenue_at_risk property

revenue_at_risk: Decimal

Annual revenue at risk if customer churns (MRR × 12).

Functions

from_float classmethod

from_float(value: float) -> MRR

Create MRR from a float, rounding to 2 decimal places.

Source code in src/domain/customer/value_objects.py

@classmethod
def from_float(cls, value: float) -> MRR:
    """Create MRR from a float, rounding to 2 decimal places."""
    return cls(amount=Decimal(str(round(value, 2))))

Repository Interface

src.domain.customer.repository

CustomerRepository – abstract port (interface) for the Customer domain.

Infrastructure implementations live in src/infrastructure/repositories/. The domain layer depends only on this interface, never on DuckDB directly.

Classes

CustomerRepository

Bases: ABC

Port (interface) for persisting and retrieving Customer entities.

Source code in src/domain/customer/repository.py

class CustomerRepository(ABC):
    """Port (interface) for persisting and retrieving Customer entities."""

    @abstractmethod
    def get_by_id(self, customer_id: str) -> Customer | None:
        """Retrieve a customer by their unique ID.

        Returns:
            Customer entity, or None if not found.
        """
        ...

    @abstractmethod
    def get_all_active(self) -> Sequence[Customer]:
        """Return all customers who have not yet churned.

        Used for batch churn scoring runs.
        """
        ...

    @abstractmethod
    def get_sample(self, n: int) -> Sequence[Customer]:
        """Return a random sample of n customers (any churn status).

        Business Context:
            Used by the demo list endpoint to seed load-test tooling and
            the UI customer picker. Deterministic seed (REPEATABLE 42) ensures
            the same customers are returned across requests for reproducibility.

        Args:
            n: Number of customers to sample (caller must clamp to safe max).
        """
        ...

    @abstractmethod
    def save(self, customer: Customer) -> None:
        """Persist a new or updated Customer entity."""
        ...

Functions

get_by_id abstractmethod

get_by_id(customer_id: str) -> Customer | None

Retrieve a customer by their unique ID.

Returns:

Type	Description
Customer | None	Customer entity, or None if not found.

Source code in src/domain/customer/repository.py

@abstractmethod
def get_by_id(self, customer_id: str) -> Customer | None:
    """Retrieve a customer by their unique ID.

    Returns:
        Customer entity, or None if not found.
    """
    ...

get_all_active abstractmethod

get_all_active() -> Sequence[Customer]

Return all customers who have not yet churned.

Used for batch churn scoring runs.

Source code in src/domain/customer/repository.py

@abstractmethod
def get_all_active(self) -> Sequence[Customer]:
    """Return all customers who have not yet churned.

    Used for batch churn scoring runs.
    """
    ...

get_sample abstractmethod

get_sample(n: int) -> Sequence[Customer]

Return a random sample of n customers (any churn status).

Business Context

Used by the demo list endpoint to seed load-test tooling and the UI customer picker. Deterministic seed (REPEATABLE 42) ensures the same customers are returned across requests for reproducibility.

Parameters:

Name	Type	Description	Default
n	int	Number of customers to sample (caller must clamp to safe max).	required

Source code in src/domain/customer/repository.py

@abstractmethod
def get_sample(self, n: int) -> Sequence[Customer]:
    """Return a random sample of n customers (any churn status).

    Business Context:
        Used by the demo list endpoint to seed load-test tooling and
        the UI customer picker. Deterministic seed (REPEATABLE 42) ensures
        the same customers are returned across requests for reproducibility.

    Args:
        n: Number of customers to sample (caller must clamp to safe max).
    """
    ...

save abstractmethod

save(customer: Customer) -> None

Persist a new or updated Customer entity.

Source code in src/domain/customer/repository.py

@abstractmethod
def save(self, customer: Customer) -> None:
    """Persist a new or updated Customer entity."""
    ...