Skip to content

Expansion Domain API Reference

Auto-generated from Google-style docstrings in src/domain/expansion/.

Value Objects

src.domain.expansion.value_objects

Value objects for the Expansion domain.

UpgradePropensity: P(upgrade to next tier in 90 days), analogous to ChurnProbability. TargetTier: Encapsulates tier-ladder logic and ARR uplift multipliers.

Classes

UpgradePropensity dataclass

Calibrated P(upgrade to next plan tier within 90 days).

Business Context: Analogous to ChurnProbability in the prediction domain. The 0.5 threshold separates accounts to actively pursue vs. nurture. Tier boundaries mirror ChurnProbability for consistent CS tooling language.

Source code in src/domain/expansion/value_objects.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@dataclass(frozen=True)
class UpgradePropensity:
    """Calibrated P(upgrade to next plan tier within 90 days).

    Business Context: Analogous to ChurnProbability in the prediction domain.
    The 0.5 threshold separates accounts to actively pursue vs. nurture.
    Tier boundaries mirror ChurnProbability for consistent CS tooling language.
    """

    value: float

    def __post_init__(self) -> None:
        if not (0.0 <= self.value <= 1.0):
            raise ValueError(f"UpgradePropensity must be in [0, 1], got {self.value}")

    @property
    def tier(self) -> RiskTier:
        """Propensity tier — maps to RiskTier for consistent tooling language.

        Thresholds:
            CRITICAL  ≥ 0.75  — Immediate expansion outreach
            HIGH      ≥ 0.50  — Include in next QBR
            MEDIUM    ≥ 0.25  — Monitor, increase feature-adoption marketing
            LOW       < 0.25  — Maintain current service levels
        """
        if self.value >= 0.75:
            return RiskTier.CRITICAL
        if self.value >= 0.5:
            return RiskTier.HIGH
        if self.value >= 0.25:
            return RiskTier.MEDIUM
        return RiskTier.LOW
Attributes
tier property 
tier: RiskTier

Propensity tier — maps to RiskTier for consistent tooling language.

Thresholds

CRITICAL ≥ 0.75 — Immediate expansion outreach HIGH ≥ 0.50 — Include in next QBR MEDIUM ≥ 0.25 — Monitor, increase feature-adoption marketing LOW < 0.25 — Maintain current service levels

TargetTier dataclass

Encapsulates the tier ladder and ARR uplift multipliers.

Business Context: Determines what the next upgrade step is and how much net-new ARR that upgrade is worth (probability-weighted). Used by ExpansionResult.expected_arr_uplift to produce a defensible dollar figure for VP/Sales reviews — not a raw ACV, but a probability-weighted delta.

Tier ladder

STARTER → GROWTH → ENTERPRISE → CUSTOM (seat/add-on expansion) → None

Source code in src/domain/expansion/value_objects.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@dataclass(frozen=True)
class TargetTier:
    """Encapsulates the tier ladder and ARR uplift multipliers.

    Business Context: Determines what the next upgrade step is and how much
    net-new ARR that upgrade is worth (probability-weighted). Used by
    ExpansionResult.expected_arr_uplift to produce a defensible dollar figure
    for VP/Sales reviews — not a raw ACV, but a probability-weighted delta.

    Tier ladder:
        STARTER → GROWTH → ENTERPRISE → CUSTOM (seat/add-on expansion) → None
    """

    current_tier: PlanTier

    @property
    def next_tier(self) -> PlanTier | None:
        """The next plan tier in the upgrade sequence, or None at the ceiling."""
        mapping: dict[PlanTier, PlanTier | None] = {
            PlanTier.FREE: PlanTier.STARTER,  # freemium → first paid tier
            PlanTier.STARTER: PlanTier.GROWTH,
            PlanTier.GROWTH: PlanTier.ENTERPRISE,
            PlanTier.ENTERPRISE: PlanTier.CUSTOM,  # seat/add-on expansion, not full tier flip
            PlanTier.CUSTOM: None,  # ceiling reached
        }
        return mapping.get(self.current_tier)

    @property
    def arr_uplift_multiplier(self) -> float:
        """MRR multiplier representing the target tier's typical ACV vs current ACV.

        Business Context: Multipliers derived from median ACV jumps observed between
        tiers. Enterprise → CUSTOM is 1.2× because growth at that tier comes from
        seat additions and add-on modules (Dash, Sign, AI), not a full tier change.
        FREE uses 0.0 as a sentinel — the guard clause in calculate_expected_uplift
        handles the FREE → STARTER conversion using the Starter floor ARR instead.
        """
        multipliers: dict[PlanTier, float] = {
            PlanTier.FREE: 0.0,  # sentinel — guard clause handles FREE conversion
            PlanTier.STARTER: 3.0,  # ~$1k → ~$3k MRR
            PlanTier.GROWTH: 5.0,  # ~$5k → ~$25k MRR
            PlanTier.ENTERPRISE: 1.2,  # seat/add-on upsell, not a tier flip
            PlanTier.CUSTOM: 0.0,  # no automated propensity above this
        }
        return multipliers.get(self.current_tier, 0.0)

    def calculate_expected_uplift(self, current_mrr: float, propensity: float) -> float:
        """Probability-weighted net annual revenue opportunity from this upgrade.

        Formula (paid tiers): (MRR × 12) × (multiplier - 1) × propensity
        Formula (FREE tier):  500 × 12 × propensity  (Starter floor ARR)

        Business Context: FREE tier customers have zero MRR so the standard
        (MRR × 12 × ...) formula would always yield $0. Instead, the Starter floor
        ARR ($500/mo = $6k/yr) is used as the conversion value. This surfaces
        free-to-paid conversion as a real revenue opportunity in Sales prioritisation.
        The $500/mo floor matches the Starter tier minimum (per tier ladder).

        Args:
            current_mrr: Customer's current Monthly Recurring Revenue (USD).
                         Ignored for FREE tier — Starter floor is used instead.
            propensity: UpgradePropensity.value — calibrated probability in [0, 1].

        Returns:
            Expected net ARR uplift in USD, rounded to 2 decimal places.
            Returns 0.0 if there is no higher tier, multiplier is zero, or
            propensity is zero.
        """
        free_to_starter_floor_mrr = 500.0
        # Guard: FREE tier uses Starter floor, not current MRR (which is always 0)
        if self.current_tier == PlanTier.FREE:
            return round(free_to_starter_floor_mrr * 12 * propensity, 2)
        if not self.next_tier or self.arr_uplift_multiplier == 0:
            return 0.0
        return round(current_mrr * 12 * max(0.0, self.arr_uplift_multiplier - 1) * propensity, 2)
Attributes
next_tier property 
next_tier: PlanTier | None

The next plan tier in the upgrade sequence, or None at the ceiling.

arr_uplift_multiplier property 
arr_uplift_multiplier: float

MRR multiplier representing the target tier's typical ACV vs current ACV.

Business Context: Multipliers derived from median ACV jumps observed between tiers. Enterprise → CUSTOM is 1.2× because growth at that tier comes from seat additions and add-on modules (Dash, Sign, AI), not a full tier change. FREE uses 0.0 as a sentinel — the guard clause in calculate_expected_uplift handles the FREE → STARTER conversion using the Starter floor ARR instead.

Functions
calculate_expected_uplift 
calculate_expected_uplift(current_mrr: float, propensity: float) -> float

Probability-weighted net annual revenue opportunity from this upgrade.

Formula (paid tiers): (MRR × 12) × (multiplier - 1) × propensity Formula (FREE tier): 500 × 12 × propensity (Starter floor ARR)

Business Context: FREE tier customers have zero MRR so the standard (MRR × 12 × ...) formula would always yield $0. Instead, the Starter floor ARR ($500/mo = $6k/yr) is used as the conversion value. This surfaces free-to-paid conversion as a real revenue opportunity in Sales prioritisation. The $500/mo floor matches the Starter tier minimum (per tier ladder).

Parameters:

Name Type Description Default
current_mrr float

Customer's current Monthly Recurring Revenue (USD). Ignored for FREE tier — Starter floor is used instead.

 required
propensity float

UpgradePropensity.value — calibrated probability in [0, 1].

 required

Returns:

Type Description
float

Expected net ARR uplift in USD, rounded to 2 decimal places.
float

Returns 0.0 if there is no higher tier, multiplier is zero, or
float

propensity is zero.
Source code in src/domain/expansion/value_objects.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def calculate_expected_uplift(self, current_mrr: float, propensity: float) -> float:
    """Probability-weighted net annual revenue opportunity from this upgrade.

    Formula (paid tiers): (MRR × 12) × (multiplier - 1) × propensity
    Formula (FREE tier):  500 × 12 × propensity  (Starter floor ARR)

    Business Context: FREE tier customers have zero MRR so the standard
    (MRR × 12 × ...) formula would always yield $0. Instead, the Starter floor
    ARR ($500/mo = $6k/yr) is used as the conversion value. This surfaces
    free-to-paid conversion as a real revenue opportunity in Sales prioritisation.
    The $500/mo floor matches the Starter tier minimum (per tier ladder).

    Args:
        current_mrr: Customer's current Monthly Recurring Revenue (USD).
                     Ignored for FREE tier — Starter floor is used instead.
        propensity: UpgradePropensity.value — calibrated probability in [0, 1].

    Returns:
        Expected net ARR uplift in USD, rounded to 2 decimal places.
        Returns 0.0 if there is no higher tier, multiplier is zero, or
        propensity is zero.
    """
    free_to_starter_floor_mrr = 500.0
    # Guard: FREE tier uses Starter floor, not current MRR (which is always 0)
    if self.current_tier == PlanTier.FREE:
        return round(free_to_starter_floor_mrr * 12 * propensity, 2)
    if not self.next_tier or self.arr_uplift_multiplier == 0:
        return 0.0
    return round(current_mrr * 12 * max(0.0, self.arr_uplift_multiplier - 1) * propensity, 2)

Entities

src.domain.expansion.entities

ExpansionResult entity — output of the Expansion domain services.

Symmetric mirror of PredictionResult in the prediction domain. Immutable (frozen=True) because results are computed once at inference time.

Classes

ExpansionResult dataclass

The complete output of an expansion propensity prediction for one customer.

Business Context: Pairs with PredictionResult to form the full NRR lifecycle view — Retain (churn) + Expand (upgrade). The expected_arr_uplift property produces a probability-weighted dollar figure for VP/Sales prioritisation.

Parameters:

Name Type Description Default
customer_id str

The customer this prediction belongs to.

 required
current_mrr float

Customer's current MRR at prediction time (USD).

 required
propensity UpgradePropensity

Calibrated P(upgrade to next tier within 90 days).

 required
target TargetTier

TargetTier encapsulating next step and uplift multiplier.

 required
top_features list[ShapFeature]

Top SHAP drivers sorted by |shap_impact| descending.

 list()
model_version str

Semantic version of the expansion model artifact.

 '1.0.0'
predicted_at datetime

UTC timestamp of when the prediction was generated.

 (lambda: now(UTC))()
Source code in src/domain/expansion/entities.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@dataclass(frozen=True)
class ExpansionResult:
    """The complete output of an expansion propensity prediction for one customer.

    Business Context: Pairs with PredictionResult to form the full NRR lifecycle
    view — Retain (churn) + Expand (upgrade). The expected_arr_uplift property
    produces a probability-weighted dollar figure for VP/Sales prioritisation.

    Args:
        customer_id: The customer this prediction belongs to.
        current_mrr: Customer's current MRR at prediction time (USD).
        propensity: Calibrated P(upgrade to next tier within 90 days).
        target: TargetTier encapsulating next step and uplift multiplier.
        top_features: Top SHAP drivers sorted by |shap_impact| descending.
        model_version: Semantic version of the expansion model artifact.
        predicted_at: UTC timestamp of when the prediction was generated.
    """

    customer_id: str
    current_mrr: float
    propensity: UpgradePropensity
    target: TargetTier
    top_features: list[ShapFeature] = field(default_factory=list)
    model_version: str = "1.0.0"
    predicted_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    @property
    def expected_arr_uplift(self) -> float:
        """Probability-weighted net annual revenue opportunity from this upgrade.

        Business Context: Delegates to TargetTier.calculate_expected_uplift()
        to avoid duplicating the (MRR × 12 × (multiplier - 1) × propensity)
        formula. This property is the primary input for Sales prioritisation:
        sort the expansion list by expected_arr_uplift DESC to find accounts
        with the highest ROI on an upgrade conversation.

        Returns:
            Expected net ARR uplift in USD.
        """
        return self.target.calculate_expected_uplift(
            current_mrr=self.current_mrr,
            propensity=self.propensity.value,
        )

    @property
    def is_high_value_target(self) -> bool:
        """True if this account warrants Senior AE (or CSM) attention.

        Business Context: A $50 ARR uplift at 85% propensity is not worth a
        Senior AE's time. This property separates 'interesting signal' from
        'actionable Sales motion' — resource allocation logic baked into the
        domain, not the dashboard filters. Threshold: > $10k expected uplift
        AND propensity tier is High or Critical.

        FREE-tier override: free-to-paid max uplift is $6k (below the $10k
        threshold), so free-tier customers at CRITICAL propensity (≥0.75)
        always return True — the conversion event is high-priority regardless.

        Returns:
            True if the account should be escalated for active outreach.
        """
        if self.target.current_tier == PlanTier.FREE and self.propensity.tier == RiskTier.CRITICAL:
            return True
        return self.expected_arr_uplift > 10_000 and self.propensity.tier in (RiskTier.HIGH, RiskTier.CRITICAL)

    def recommended_action(self, churn_probability: float | None = None) -> str:
        """Deterministic GTM playbook routing based on propensity tier.

        Business Context: When both churn probability and expansion propensity
        are available, the conflict matrix resolves the correct CS motion.
        High Churn + High Expansion = 'Flight Risk' — upselling a churning
        customer accelerates churn rather than preventing it.

        Args:
            churn_probability: Optional churn probability from PredictionResult.
                               When provided, enables conflict-matrix resolution.

        Returns:
            Human-readable action string for CS/Sales tooling.
        """
        next_plan = self.target.next_tier.value.upper() if self.target.next_tier else "UPSELL"

        # Conflict matrix: when both scores are available, churn risk takes precedence
        if churn_probability is not None:
            high_churn = churn_probability >= 0.50
            high_expansion = self.propensity.value >= 0.50
            if not high_churn and high_expansion:
                return (
                    f"Growth Engine — schedule {next_plan} upgrade conversation. "
                    f"Expected ARR uplift: ${self.expected_arr_uplift:,.0f}."
                )
            if high_churn and high_expansion:
                return (
                    f"Flight Risk — Senior Exec intervention required before any "
                    f"upsell motion. Restore health first, then revisit {next_plan} migration."
                )
            if high_churn and not high_expansion:
                return "Retention priority — defer expansion until account health is restored."
            return "Stable base — nurture via product-led expansion signals."

        # Single-score routing (no churn context)
        if self.propensity.tier == RiskTier.CRITICAL:
            return f"EXPANSION PRIORITY: High intent detected. Immediate outreach for {next_plan} migration."
        if self.propensity.tier == RiskTier.HIGH:
            return f"NURTURE: Strong usage signals. Highlight {next_plan} features in next QBR."
        if self.propensity.tier == RiskTier.MEDIUM:
            return "MONITOR: Early signals detected. Increase feature-adoption marketing."
        return "STABLE: Maintain current service levels."

    def to_summary_context(self) -> dict[str, object]:
        """Produces a verified, grounded-facts dict for the LLM PromptBuilder.

        Business Context: Only surfaces confirmed model outputs to the LLM.
        This is the 'clean hands' hallucination guardrail — the model can only
        fabricate facts we explicitly passed to it, not invent new signals.

        Returns:
            Dict of scalar values safe to inject into a prompt [CONTEXT] block.
        """
        return {
            "customer_id": self.customer_id,
            "propensity_score": f"{self.propensity.value:.2%}",
            "propensity_tier": self.propensity.tier.value,
            "expected_uplift": f"${self.expected_arr_uplift:,.2f}",
            "target_tier": self.target.next_tier.value if self.target.next_tier else "N/A",
            "top_signals": [f.feature_name for f in self.top_features[:3]],
        }
Attributes
expected_arr_uplift property 
expected_arr_uplift: float

Probability-weighted net annual revenue opportunity from this upgrade.

Business Context: Delegates to TargetTier.calculate_expected_uplift() to avoid duplicating the (MRR × 12 × (multiplier - 1) × propensity) formula. This property is the primary input for Sales prioritisation: sort the expansion list by expected_arr_uplift DESC to find accounts with the highest ROI on an upgrade conversation.

Returns:

Type Description
float

Expected net ARR uplift in USD.
is_high_value_target property 
is_high_value_target: bool

True if this account warrants Senior AE (or CSM) attention.

Business Context: A $50 ARR uplift at 85% propensity is not worth a Senior AE's time. This property separates 'interesting signal' from 'actionable Sales motion' — resource allocation logic baked into the domain, not the dashboard filters. Threshold: > $10k expected uplift AND propensity tier is High or Critical.

FREE-tier override: free-to-paid max uplift is $6k (below the $10k threshold), so free-tier customers at CRITICAL propensity (≥0.75) always return True — the conversion event is high-priority regardless.

Returns:

Type Description
bool

True if the account should be escalated for active outreach.
Functions
recommended_action 
recommended_action(churn_probability: float | None = None) -> str

Deterministic GTM playbook routing based on propensity tier.

Business Context: When both churn probability and expansion propensity are available, the conflict matrix resolves the correct CS motion. High Churn + High Expansion = 'Flight Risk' — upselling a churning customer accelerates churn rather than preventing it.

Parameters:

Name Type Description Default
churn_probability float | None

Optional churn probability from PredictionResult. When provided, enables conflict-matrix resolution.

 None

Returns:

Type Description
str

Human-readable action string for CS/Sales tooling.
Source code in src/domain/expansion/entities.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def recommended_action(self, churn_probability: float | None = None) -> str:
    """Deterministic GTM playbook routing based on propensity tier.

    Business Context: When both churn probability and expansion propensity
    are available, the conflict matrix resolves the correct CS motion.
    High Churn + High Expansion = 'Flight Risk' — upselling a churning
    customer accelerates churn rather than preventing it.

    Args:
        churn_probability: Optional churn probability from PredictionResult.
                           When provided, enables conflict-matrix resolution.

    Returns:
        Human-readable action string for CS/Sales tooling.
    """
    next_plan = self.target.next_tier.value.upper() if self.target.next_tier else "UPSELL"

    # Conflict matrix: when both scores are available, churn risk takes precedence
    if churn_probability is not None:
        high_churn = churn_probability >= 0.50
        high_expansion = self.propensity.value >= 0.50
        if not high_churn and high_expansion:
            return (
                f"Growth Engine — schedule {next_plan} upgrade conversation. "
                f"Expected ARR uplift: ${self.expected_arr_uplift:,.0f}."
            )
        if high_churn and high_expansion:
            return (
                f"Flight Risk — Senior Exec intervention required before any "
                f"upsell motion. Restore health first, then revisit {next_plan} migration."
            )
        if high_churn and not high_expansion:
            return "Retention priority — defer expansion until account health is restored."
        return "Stable base — nurture via product-led expansion signals."

    # Single-score routing (no churn context)
    if self.propensity.tier == RiskTier.CRITICAL:
        return f"EXPANSION PRIORITY: High intent detected. Immediate outreach for {next_plan} migration."
    if self.propensity.tier == RiskTier.HIGH:
        return f"NURTURE: Strong usage signals. Highlight {next_plan} features in next QBR."
    if self.propensity.tier == RiskTier.MEDIUM:
        return "MONITOR: Early signals detected. Increase feature-adoption marketing."
    return "STABLE: Maintain current service levels."
to_summary_context 
to_summary_context() -> dict[str, object]

Produces a verified, grounded-facts dict for the LLM PromptBuilder.

Business Context: Only surfaces confirmed model outputs to the LLM. This is the 'clean hands' hallucination guardrail — the model can only fabricate facts we explicitly passed to it, not invent new signals.

Returns:

Type Description
dict[str, object]

Dict of scalar values safe to inject into a prompt [CONTEXT] block.
Source code in src/domain/expansion/entities.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def to_summary_context(self) -> dict[str, object]:
    """Produces a verified, grounded-facts dict for the LLM PromptBuilder.

    Business Context: Only surfaces confirmed model outputs to the LLM.
    This is the 'clean hands' hallucination guardrail — the model can only
    fabricate facts we explicitly passed to it, not invent new signals.

    Returns:
        Dict of scalar values safe to inject into a prompt [CONTEXT] block.
    """
    return {
        "customer_id": self.customer_id,
        "propensity_score": f"{self.propensity.value:.2%}",
        "propensity_tier": self.propensity.tier.value,
        "expected_uplift": f"${self.expected_arr_uplift:,.2f}",
        "target_tier": self.target.next_tier.value if self.target.next_tier else "N/A",
        "top_signals": [f.feature_name for f in self.top_features[:3]],
    }

Summary Result Entity

src.domain.expansion.summary_entities

ExpansionSummaryResult entity — output of GenerateExpansionSummaryUseCase.

Immutable dataclass that carries the LLM-generated AE brief, optional email draft, guardrail validation outcome, and full provenance for audit logging.

Classes

ExpansionSummaryResult dataclass

Complete output of the expansion narrative generation pipeline.

Business Context: Closes the "last mile" between a high-propensity score in DuckDB and an actionable, personalised pitch in the AE's hands. The correlation_id links each brief to its downstream outcome in expansion_outreach_log, enabling the data team to measure whether high fact_confidence briefs close at higher rates (V2 fine-tuning flywheel).

Parameters:

Name Type Description Default
customer_id str

UUID of the account this brief is about.

 required
propensity_summary str

Plain-English propensity tier + score sentence.

 required
key_narrative_drivers list[str]

Top 3 signals in business language (from SHAP).

 required
ae_tactical_brief str

LLM-generated brief with guardrail watermark appended.

 required
email_draft str | None

Optional 3-sentence email with CTA (AE audience only).

 required
guardrail_status Literal['PASSED', 'FLAGGED', 'REJECTED']

'PASSED' / 'FLAGGED' (1 issue) / 'REJECTED' (2+ issues).

 required
fact_confidence float

1.0 − (0.25 × n_flags), floored at 0.0.

 required
generated_at datetime

UTC timestamp when the summary was created.

 required
model_used str

LLM model identifier (e.g. 'llama-3.1-8b-instant').

 required
llm_provider str

Inference provider — 'groq' or 'ollama'.

 required
propensity_score float

Raw calibrated probability in [0, 1].

 required
propensity_tier str

Human-readable tier — 'low' / 'medium' / 'high' / 'critical'.

 required
target_tier str | None

Next upgrade target (e.g. 'enterprise'), or None at ceiling.

 required
expected_arr_uplift float

Probability-weighted net ARR opportunity (USD).

 required
correlation_id str

UUID hex linking this brief to GTM outreach log for V2.

 required
Source code in src/domain/expansion/summary_entities.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass(frozen=True)
class ExpansionSummaryResult:
    """Complete output of the expansion narrative generation pipeline.

    Business Context: Closes the "last mile" between a high-propensity score
    in DuckDB and an actionable, personalised pitch in the AE's hands.
    The correlation_id links each brief to its downstream outcome in
    expansion_outreach_log, enabling the data team to measure whether
    high fact_confidence briefs close at higher rates (V2 fine-tuning flywheel).

    Args:
        customer_id: UUID of the account this brief is about.
        propensity_summary: Plain-English propensity tier + score sentence.
        key_narrative_drivers: Top 3 signals in business language (from SHAP).
        ae_tactical_brief: LLM-generated brief with guardrail watermark appended.
        email_draft: Optional 3-sentence email with CTA (AE audience only).
        guardrail_status: 'PASSED' / 'FLAGGED' (1 issue) / 'REJECTED' (2+ issues).
        fact_confidence: 1.0 − (0.25 × n_flags), floored at 0.0.
        generated_at: UTC timestamp when the summary was created.
        model_used: LLM model identifier (e.g. 'llama-3.1-8b-instant').
        llm_provider: Inference provider — 'groq' or 'ollama'.
        propensity_score: Raw calibrated probability in [0, 1].
        propensity_tier: Human-readable tier — 'low' / 'medium' / 'high' / 'critical'.
        target_tier: Next upgrade target (e.g. 'enterprise'), or None at ceiling.
        expected_arr_uplift: Probability-weighted net ARR opportunity (USD).
        correlation_id: UUID hex linking this brief to GTM outreach log for V2.
    """

    customer_id: str
    propensity_summary: str
    key_narrative_drivers: list[str]
    ae_tactical_brief: str
    email_draft: str | None
    guardrail_status: Literal["PASSED", "FLAGGED", "REJECTED"]
    fact_confidence: float
    generated_at: datetime
    model_used: str
    llm_provider: str
    propensity_score: float
    propensity_tier: str
    target_tier: str | None
    expected_arr_uplift: float
    correlation_id: str

Domain Service

src.domain.expansion.expansion_service

ExpansionModelService — domain service for upgrade propensity prediction.

Symmetric mirror of ChurnModelService in the prediction domain. Ports and protocols follow the same pattern: domain has no direct knowledge of infrastructure (no file I/O, no DuckDB, no pickle files here).

Classes

ExpansionFeatureVector

Bases: Protocol

Protocol for expansion feature extraction — implemented in infrastructure layer.

Queries the mart_customer_expansion_features dbt mart for the 20-feature vector (15 churn features + 5 expansion signals).

Source code in src/domain/expansion/expansion_service.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class ExpansionFeatureVector(Protocol):
    """Protocol for expansion feature extraction — implemented in infrastructure layer.

    Queries the mart_customer_expansion_features dbt mart for the 20-feature
    vector (15 churn features + 5 expansion signals).
    """

    def extract(self, customer: Customer) -> dict[str, float | str]:
        """Extract the 20-feature expansion vector for a customer.

        Args:
            customer: Active Customer entity with plan_tier and MRR.

        Returns:
            Flat dict of feature_name → value (numeric as float, categorical as str).
            All feature engineering lives in mart_customer_expansion_features.

        Raises:
            ValueError: If the customer is not found in the mart or has already upgraded.
        """
        ...
Functions
extract 
extract(customer: Customer) -> dict[str, float | str]

Extract the 20-feature expansion vector for a customer.

Parameters:

Name Type Description Default
customer Customer

Active Customer entity with plan_tier and MRR.

 required

Returns:

Type Description
dict[str, float | str]

Flat dict of feature_name → value (numeric as float, categorical as str).
dict[str, float | str]

All feature engineering lives in mart_customer_expansion_features.

Raises:

Type Description
ValueError

If the customer is not found in the mart or has already upgraded.
Source code in src/domain/expansion/expansion_service.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def extract(self, customer: Customer) -> dict[str, float | str]:
    """Extract the 20-feature expansion vector for a customer.

    Args:
        customer: Active Customer entity with plan_tier and MRR.

    Returns:
        Flat dict of feature_name → value (numeric as float, categorical as str).
        All feature engineering lives in mart_customer_expansion_features.

    Raises:
        ValueError: If the customer is not found in the mart or has already upgraded.
    """
    ...
ExpansionModelPort

Bases: ABC

Abstract port for the underlying expansion ML model.

Concrete implementations in src/infrastructure/ml/ load the trained XGBoost expansion model artifact.

Source code in src/domain/expansion/expansion_service.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class ExpansionModelPort(ABC):
    """Abstract port for the underlying expansion ML model.

    Concrete implementations in src/infrastructure/ml/ load the trained
    XGBoost expansion model artifact.
    """

    @abstractmethod
    def predict_proba(self, features: dict[str, float | str]) -> float:
        """Return P(upgrade to next tier within 90 days)."""
        ...

    @abstractmethod
    def explain(self, features: dict[str, float | str]) -> list[ShapFeature]:
        """Return SHAP feature contributions for explainability."""
        ...

    @property
    @abstractmethod
    def version(self) -> str:
        """Semantic version of the loaded model artifact."""
        ...
Attributes
version abstractmethod property 
version: str

Semantic version of the loaded model artifact.

Functions
predict_proba abstractmethod 
predict_proba(features: dict[str, float | str]) -> float

Return P(upgrade to next tier within 90 days).

Source code in src/domain/expansion/expansion_service.py 
49
50
51
52
@abstractmethod
def predict_proba(self, features: dict[str, float | str]) -> float:
    """Return P(upgrade to next tier within 90 days)."""
    ...
explain abstractmethod 
explain(features: dict[str, float | str]) -> list[ShapFeature]

Return SHAP feature contributions for explainability.

Source code in src/domain/expansion/expansion_service.py 
54
55
56
57
@abstractmethod
def explain(self, features: dict[str, float | str]) -> list[ShapFeature]:
    """Return SHAP feature contributions for explainability."""
    ...
ExpansionModelService

Orchestrates feature extraction → model inference → result construction.

Business Context: Mirrors ChurnModelService exactly. Feature extraction, model inference, and SHAP computation are all delegated to injected dependencies — this service only owns the assembly logic.

Parameters:

Name Type Description Default
model ExpansionModelPort

Concrete expansion ML model (injected from infrastructure layer).

 required
feature_extractor ExpansionFeatureVector

Queries the dbt expansion mart for the feature vector.

 required
Source code in src/domain/expansion/expansion_service.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class ExpansionModelService:
    """Orchestrates feature extraction → model inference → result construction.

    Business Context: Mirrors ChurnModelService exactly. Feature extraction,
    model inference, and SHAP computation are all delegated to injected
    dependencies — this service only owns the assembly logic.

    Args:
        model: Concrete expansion ML model (injected from infrastructure layer).
        feature_extractor: Queries the dbt expansion mart for the feature vector.
    """

    def __init__(
        self,
        model: ExpansionModelPort,
        feature_extractor: ExpansionFeatureVector,
    ) -> None:
        self._model = model
        self._feature_extractor = feature_extractor

    def predict(self, customer: Customer) -> ExpansionResult:
        """Generate a full ExpansionResult for a customer.

        Business Context: Expansion depends on usage signals and GTM intent,
        not compliance risk — so no RiskScore is needed here. The domain
        service is intentionally leaner than ChurnModelService.

        Args:
            customer: Active Customer entity that has not yet upgraded.

        Returns:
            ExpansionResult with calibrated upgrade propensity, target tier,
            SHAP explanations, and a deterministic GTM action recommendation.
        """
        features = self._feature_extractor.extract(customer)
        propensity_value = self._model.predict_proba(features)
        shap_features = self._model.explain(features)

        return ExpansionResult(
            customer_id=customer.customer_id,
            current_mrr=float(customer.mrr.amount),
            propensity=UpgradePropensity(value=propensity_value),
            target=TargetTier(current_tier=customer.plan_tier),
            top_features=sorted(shap_features, key=lambda f: abs(f.shap_impact), reverse=True)[:5],
            model_version=self._model.version,
        )
Functions
predict 
predict(customer: Customer) -> ExpansionResult

Generate a full ExpansionResult for a customer.

Business Context: Expansion depends on usage signals and GTM intent, not compliance risk — so no RiskScore is needed here. The domain service is intentionally leaner than ChurnModelService.

Parameters:

Name Type Description Default
customer Customer

Active Customer entity that has not yet upgraded.

 required

Returns:

Type Description
ExpansionResult

ExpansionResult with calibrated upgrade propensity, target tier,
ExpansionResult

SHAP explanations, and a deterministic GTM action recommendation.
Source code in src/domain/expansion/expansion_service.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def predict(self, customer: Customer) -> ExpansionResult:
    """Generate a full ExpansionResult for a customer.

    Business Context: Expansion depends on usage signals and GTM intent,
    not compliance risk — so no RiskScore is needed here. The domain
    service is intentionally leaner than ChurnModelService.

    Args:
        customer: Active Customer entity that has not yet upgraded.

    Returns:
        ExpansionResult with calibrated upgrade propensity, target tier,
        SHAP explanations, and a deterministic GTM action recommendation.
    """
    features = self._feature_extractor.extract(customer)
    propensity_value = self._model.predict_proba(features)
    shap_features = self._model.explain(features)

    return ExpansionResult(
        customer_id=customer.customer_id,
        current_mrr=float(customer.mrr.amount),
        propensity=UpgradePropensity(value=propensity_value),
        target=TargetTier(current_tier=customer.plan_tier),
        top_features=sorted(shap_features, key=lambda f: abs(f.shap_impact), reverse=True)[:5],
        model_version=self._model.version,
    )