"""The pipeline — deals, the stages they move through, and the forecast they add up to.
No database, no HTTP, no identity. The design is a general CRM opportunity model, held to the CRM spec
(modules/crm/module.yaml: CRM-04, 08, 09, 15, 16):
* STAGES ARE DATA. Their names, order, probability and forecast category are the thing a customer
changes most, so they are rows; what stays in code is that a closing stage demands EVIDENCE —
won needs an accepted quote, lost needs a reason from the agreed list, so "lost on price" can be
counted rather than guessed from free text.
* DERIVED, NEVER STORED: the weighted value (amount × probability), the forecast, and whether a
deal is at risk. Each is computed from figures already stored; a stored copy is a third number
that can disagree with the two it came from, or a nightly job that is late.
* A WON DEAL'S MONEY AND CUSTOMER ARE THE CONTRACT'S. Its amount, currency and customer are locked —
a won deal whose amount can still be typed over disagrees with the agreement it summarises. Its
name and owner are not: people leave and deals get renamed.
* (0.3.0, the CRM audit) WON TAKES THE ACCEPTED TOTAL — whatever estimate was typed — so "won this quarter" is
what customers signed. A CLOSED DEAL IS FINAL: won or lost, it is not moved, and its close date and amount are
frozen; a manager REOPENS one with a reason, and reopening a won deal VOIDS the quotes that won it, so the
same acceptance cannot win it twice.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Iterable, Optional
from erp_platform.money import Money, pct, totals_by_currency
CATEGORIES = ("pipeline", "best_case", "commit", "closed", "omitted")
HUNDRED = Decimal("100")
class PipelineRuleBroken(Exception):
"""A pipeline rule said no. → 409."""
class EvidenceMissing(PipelineRuleBroken):
"""The stage demands evidence the deal cannot offer yet (an accepted quote, a loss reason). → 422."""
# ── configuration: the customer's own sales process ────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class Stage:
name: str
position: int
probability: Decimal # 0–100
category: str
is_won: bool = False
is_closed: bool = False
def __post_init__(self):
object.__setattr__(self, "probability", pct(self.probability))
if not (0 <= self.probability <= 100):
raise PipelineRuleBroken(f"stage {self.name}: probability is a percentage")
if self.category not in CATEGORIES:
raise PipelineRuleBroken(f"stage {self.name}: category is one of {', '.join(CATEGORIES)}")
if self.is_won and not self.is_closed:
raise PipelineRuleBroken(f"stage {self.name}: a won stage is a closed stage")
@dataclass(frozen=True, slots=True)
class Pipeline:
"""The stages, in order, and the agreed reasons a deal is lost."""
stages: tuple
loss_reasons: tuple = ("price", "no_budget", "competitor", "no_decision", "timing", "other")
def __post_init__(self):
names = [s.name for s in self.stages]
if len(set(names)) != len(names):
raise PipelineRuleBroken("two stages share a name")
if not any(not s.is_closed for s in self.stages):
raise PipelineRuleBroken("a pipeline needs at least one open stage")
if sum(1 for s in self.stages if s.is_won) != 1:
raise PipelineRuleBroken("a pipeline has exactly one won stage")
if not any(s.is_closed and not s.is_won for s in self.stages):
raise PipelineRuleBroken("a pipeline needs a lost stage")
object.__setattr__(self, "stages", tuple(sorted(self.stages, key=lambda s: s.position)))
def stage(self, name: str) -> Stage:
for s in self.stages:
if s.name == name:
return s
raise PipelineRuleBroken(f"there is no stage {name!r}; the stages are "
f"{', '.join(s.name for s in self.stages)}")
@property
def first_open(self) -> Stage:
return next(s for s in self.stages if not s.is_closed)
DEFAULT_PIPELINE = Pipeline((
Stage("discovery", 10, Decimal("10"), "pipeline"),
Stage("qualification", 20, Decimal("25"), "pipeline"),
Stage("proposal", 30, Decimal("50"), "best_case"),
Stage("negotiation", 40, Decimal("75"), "commit"),
Stage("closed_won", 90, Decimal("100"), "closed", is_won=True, is_closed=True),
Stage("closed_lost", 99, Decimal("0"), "omitted", is_closed=True),
))
@dataclass(frozen=True, slots=True)
class RiskPolicy:
"""CRM-16, as a row: a deal at or past `from_probability` with nothing logged for `quiet_days`."""
quiet_days: int = 21
from_probability: Decimal = Decimal("50")
def period_of(close_date: Optional[str]) -> Optional[str]:
"""The forecast period a close date falls in: a calendar quarter, '2026-Q4'."""
if not close_date:
return None
d = date.fromisoformat(close_date[:10])
return f"{d.year}-Q{(d.month - 1) // 3 + 1}"
# ── the deal ───────────────────────────────────────────────────────────────────────────────
@dataclass(slots=True)
class Opportunity:
opportunity_id: str
bp_id: int # the customer: a business partner (modules/shared.yaml)
name: str
owner: str
stage: str
currency: str
amount: Optional[Money] = None
close_date: Optional[str] = None
probability: Optional[Decimal] = None # set only when THIS deal overrides its stage
category_override: Optional[str] = None
category_reason: Optional[str] = None
loss_reason: Optional[str] = None
last_activity_at: Optional[str] = None # ISO; the deal's own changes and logged activity
won_quote_ids: tuple = () # 0.3: the accepted quotes this deal was won with
voided_quote_ids: tuple = () # 0.3: quotes whose win was undone by a reopen
version: int = 0 # 0.3: saves so far — two people at once, one wins (409)
SETTLED_WHEN_WON = ("amount", "currency", "bp_id")
@classmethod
def open(cls, opportunity_id: str, bp_id: int, name: str, owner: str, currency: str,
pipeline: Pipeline, *, amount=None, close_date: Optional[str] = None) -> "Opportunity":
if not name.strip():
raise PipelineRuleBroken("a deal needs a name")
cur = Money.zero(currency).currency
o = cls(opportunity_id, int(bp_id), name.strip(), owner, pipeline.first_open.name, cur,
close_date=_iso_date(close_date))
if amount is not None:
o.set_amount(amount, pipeline)
return o
# ── value ────────────────────────────────────────────────────────────────────────────
def probability_for(self, pipeline: Pipeline) -> Decimal:
"""The stage's probability unless this deal carries its own. An open deal's override survives a
stage move; reaching a closed stage takes the stage's figure (won is 100, lost is 0)."""
s = pipeline.stage(self.stage)
if self.probability is not None and not s.is_closed:
return self.probability
return s.probability
def weighted(self, pipeline: Pipeline) -> Optional[Money]:
if self.amount is None:
return None
return self.amount * (self.probability_for(pipeline) / HUNDRED)
def category(self, pipeline: Pipeline) -> str:
s = pipeline.stage(self.stage)
if s.is_closed:
return s.category # a closed deal's category is a fact
return self.category_override or s.category
def is_open(self, pipeline: Pipeline) -> bool:
return not pipeline.stage(self.stage).is_closed
def is_won(self, pipeline: Pipeline) -> bool:
return pipeline.stage(self.stage).is_won
# ── editing ──────────────────────────────────────────────────────────────────────────
def _not_settled(self, what: str, pipeline: Pipeline) -> None:
if self.is_won(pipeline):
raise PipelineRuleBroken(f"this deal is closed as won, so its {what} is the contract's — amend the "
f"contract to change it")
if not self.is_open(pipeline):
raise PipelineRuleBroken(f"this deal is closed, so its {what} is what it closed with — a manager "
f"reopens it to change that")
def set_amount(self, amount, pipeline: Pipeline) -> None:
"""A closed deal's amount is frozen: a won one's is the contract's, a lost one's what was lost."""
self._not_settled("amount", pipeline)
m = Money.of(amount, self.currency)
if m.amount < 0:
raise PipelineRuleBroken("a deal's amount cannot be negative")
self.amount = m
def set_probability(self, value: Optional[Decimal], pipeline: Pipeline) -> None:
"""Override the stage's probability for this deal; None goes back to the stage's."""
if not self.is_open(pipeline):
raise PipelineRuleBroken("a closed deal's probability is its stage's")
if value is not None and not (0 <= pct(value) <= 100):
raise PipelineRuleBroken("probability is a percentage")
self.probability = None if value is None else pct(value)
def override_category(self, category: Optional[str], reason: Optional[str],
pipeline: Pipeline) -> None:
"""CRM-09: a manager calls a deal commit (or omits it) against its stage, saying why."""
if not self.is_open(pipeline):
raise PipelineRuleBroken("a closed deal's category is a fact, not a call")
if category is None:
self.category_override = self.category_reason = None
return
if category not in CATEGORIES or category == "closed":
raise PipelineRuleBroken(f"an open deal's category is one of pipeline, best_case, commit, "
f"omitted — not {category!r}")
if not (reason or "").strip():
raise EvidenceMissing("overriding a forecast category needs a reason — it is the number "
"leadership was given")
self.category_override, self.category_reason = category, reason.strip()
def set_close_date(self, close_date: Optional[str], pipeline: Optional[Pipeline] = None) -> None:
"""A closed deal's close date is when it closed — frozen, or the forecast's history moves under it."""
if pipeline is not None:
self._not_settled("close date", pipeline)
self.close_date = _iso_date(close_date)
def rename(self, name: str) -> None:
if not name.strip():
raise PipelineRuleBroken("a deal needs a name")
self.name = name.strip()
# ── moving ───────────────────────────────────────────────────────────────────────────
def move(self, to: str, pipeline: Pipeline, *, has_accepted_quote: Optional[bool],
loss_reason: Optional[str] = None, accepted_total: Optional[Money] = None,
won_quotes: tuple = ()) -> str:
"""Move the deal to a stage (CRM-04). Returns the stage it left.
`has_accepted_quote` is passed IN — the quote is its own aggregate in its own brick, and the
application layer asks it (for THIS customer and currency, and not a quote voided by a reopen). None
means "nobody can say" (no quote module installed): won is then recorded without that evidence, and
the caller says so in the history.
Won TAKES the accepted total (H1): the amount is what the customer accepted, not the estimate. A
closed deal is final (H3): moving it anywhere is refused — `reopen` is how it opens again.
"""
target = pipeline.stage(to)
before = self.stage
if before == target.name:
return before
if not self.is_open(pipeline):
raise PipelineRuleBroken(f"this deal is {before} — a closed deal is final. A manager can reopen it "
f"(with a reason) if it really is back in play")
if target.is_won:
if has_accepted_quote is False:
raise EvidenceMissing("a deal is won when the customer has accepted a quote — record "
"the acceptance on the quote first")
if accepted_total is not None:
if accepted_total.currency != self.currency:
raise PipelineRuleBroken(f"the accepted quote's currency is {accepted_total.currency}; this "
f"deal is in {self.currency}")
self.amount = accepted_total
self.won_quote_ids = tuple(won_quotes)
if target.is_closed and not target.is_won:
reason = (loss_reason or "").strip().lower()
if reason not in pipeline.loss_reasons:
raise EvidenceMissing(f"closing a deal as lost needs a reason from the list — "
f"{', '.join(pipeline.loss_reasons)} — so losses can be counted")
self.loss_reason = reason
if target.is_closed:
self.category_override = self.category_reason = None
self.stage = target.name
return before
def reopen(self, to: str, reason: str, pipeline: Pipeline) -> tuple:
"""H3: a manager opens a closed deal again, saying why. Returns (the stage it left, the quotes voided).
Reopening a WON deal voids the quotes that won it: they stay accepted in the quote brick (the customer
did sign them), but they no longer count as this deal's evidence — winning it again needs a new
acceptance, or the same signature would win the same deal twice."""
if self.is_open(pipeline):
raise PipelineRuleBroken(f"this deal is open ({self.stage}); only a closed deal is reopened")
if not (reason or "").strip():
raise EvidenceMissing("reopening a closed deal needs a reason — the forecast it was in changes")
target = pipeline.stage(to)
if target.is_closed:
raise PipelineRuleBroken(f"a deal is reopened into an open stage, not {target.name}")
before, voided = self.stage, self.won_quote_ids
self.voided_quote_ids = tuple(dict.fromkeys(self.voided_quote_ids + voided))
self.won_quote_ids, self.loss_reason, self.stage = (), None, target.name
return before, voided
def counts_as_evidence(self, quote_id: str) -> bool:
"""Whether an accepted quote may win this deal — not one whose win a reopen voided."""
return quote_id not in self.voided_quote_ids
# ── CRM-16: gone quiet ───────────────────────────────────────────────────────────────
def risk(self, pipeline: Pipeline, policy: RiskPolicy, today: Optional[str] = None) -> Optional[str]:
"""Why this deal is at risk, or None. Derived on every read, so no job can be late."""
if not self.is_open(pipeline) or self.probability_for(pipeline) < policy.from_probability:
return None
today_d = date.fromisoformat((today or date.today().isoformat())[:10])
if self.last_activity_at:
quiet = (today_d - date.fromisoformat(self.last_activity_at[:10])).days
if quiet >= policy.quiet_days:
return f"nothing logged for {quiet} days in {self.stage}"
if self.close_date and self.close_date < today_d.isoformat():
return f"the close date {self.close_date} has passed and the deal is still open"
return None
def _iso_date(v: Optional[str]) -> Optional[str]:
if not v:
return None
try:
return date.fromisoformat(str(v)[:10]).isoformat()
except ValueError:
raise PipelineRuleBroken(f"{v!r} is not a date (YYYY-MM-DD)") from None
# ── the forecast (CRM-09), derived ─────────────────────────────────────────────────────────
def forecast(deals: Iterable[Opportunity], pipeline: Pipeline) -> list:
"""Open, weighted and committed amounts per rep, period and currency.
open every open deal's amount
weighted open amount × probability
commit open deals called commit, plus deals already won in the period
won closed-won in the period
Per currency: summing SEK and EUR would need a rate nobody agreed. A deal with no amount or no
close date cannot be forecast and is counted in `unforecastable`, so the gap is visible.
"""
rows: dict = {}
for d in deals:
period = period_of(d.close_date)
key = (d.owner, period, d.currency)
r = rows.setdefault(key, {"owner": d.owner, "period": period, "currency": d.currency,
"open": [], "weighted": [], "commit": [], "won": [],
"deals": 0, "unforecastable": 0})
r["deals"] += 1
if d.amount is None or period is None:
r["unforecastable"] += 1
continue
if d.is_won(pipeline):
r["won"].append(d.amount)
r["commit"].append(d.amount)
elif d.is_open(pipeline):
r["open"].append(d.amount)
r["weighted"].append(d.weighted(pipeline))
if d.category(pipeline) == "commit":
r["commit"].append(d.amount)
out = []
for r in rows.values():
z = Money.zero(r["currency"])
for k in ("open", "weighted", "commit", "won"):
r[k] = totals_by_currency(r[k]).get(r["currency"], z).amount
out.append(r)
return sorted(out, key=lambda r: (r["period"] or "", r["owner"], r["currency"]))