767 lines
35 KiB
Python
767 lines
35 KiB
Python
"""Commercial target and revenue forecast for ClientFlow.
|
||
|
||
The service deliberately separates:
|
||
- realised value for the selected target metric and month;
|
||
- committed backlog not yet realised;
|
||
- probabilistic open pipeline;
|
||
- value already realised outside the selected period.
|
||
|
||
It is a management forecast, not accounting recognition or cash-flow advice.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from calendar import monthrange
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from typing import Any, Dict
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.db import engine
|
||
|
||
|
||
DEFAULT_STAGE_PROBABILITIES: dict[str, float] = {
|
||
"NEW_LEAD": 0.10,
|
||
"INFO_REQUESTED": 0.14,
|
||
"INFO_SENT": 0.18,
|
||
"QUOTE_REQUESTED": 0.24,
|
||
"QUOTE_SENT": 0.40,
|
||
"PROFORMA_REQUESTED": 0.48,
|
||
"PROFORMA_SENT": 0.58,
|
||
"INVOICE_REQUESTED": 0.62,
|
||
"INVOICE_SENT": 0.70,
|
||
"WAITING_PAYMENT": 0.76,
|
||
"PAYMENT_CONFIRMED": 0.95,
|
||
"ODOO_ORDER_CREATED": 0.97,
|
||
"IN_PRODUCTION": 0.98,
|
||
"ORDER_PREPARATION": 0.98,
|
||
"READY_TO_SHIP": 0.99,
|
||
"INVOICED": 0.99,
|
||
"SHIPMENT_CREATED": 0.995,
|
||
"SHIPPED": 0.995,
|
||
"TRACKING_SENT": 0.995,
|
||
"DELIVERED": 1.0,
|
||
"WON": 1.0,
|
||
"REVIEW": 0.08,
|
||
}
|
||
|
||
STAGE_EXPECTED_DAYS: dict[str, int] = {
|
||
"PAYMENT_CONFIRMED": 7,
|
||
"ODOO_ORDER_CREATED": 14,
|
||
"IN_PRODUCTION": 21,
|
||
"ORDER_PREPARATION": 14,
|
||
"READY_TO_SHIP": 7,
|
||
"INVOICED": 7,
|
||
"SHIPMENT_CREATED": 7,
|
||
"SHIPPED": 7,
|
||
"TRACKING_SENT": 7,
|
||
"DELIVERED": 3,
|
||
"WAITING_PAYMENT": 21,
|
||
"INVOICE_REQUESTED": 21,
|
||
"INVOICE_SENT": 21,
|
||
"PROFORMA_REQUESTED": 30,
|
||
"PROFORMA_SENT": 30,
|
||
"QUOTE_SENT": 45,
|
||
"QUOTE_REQUESTED": 60,
|
||
"INFO_SENT": 75,
|
||
"INFO_REQUESTED": 90,
|
||
"NEW_LEAD": 90,
|
||
"REVIEW": 120,
|
||
}
|
||
|
||
COMMITTED_STAGES = {
|
||
"PAYMENT_CONFIRMED",
|
||
"ODOO_ORDER_CREATED",
|
||
"IN_PRODUCTION",
|
||
"ORDER_PREPARATION",
|
||
"READY_TO_SHIP",
|
||
"INVOICED",
|
||
"SHIPMENT_CREATED",
|
||
"SHIPPED",
|
||
"TRACKING_SENT",
|
||
"DELIVERED",
|
||
"WON",
|
||
}
|
||
|
||
ADVANCED_VALUE_STAGES = {
|
||
"QUOTE_SENT",
|
||
"PROFORMA_REQUESTED",
|
||
"PROFORMA_SENT",
|
||
"INVOICE_REQUESTED",
|
||
"INVOICE_SENT",
|
||
"WAITING_PAYMENT",
|
||
*COMMITTED_STAGES,
|
||
}
|
||
|
||
TARGET_METRICS = {
|
||
"invoiced": "Faturação emitida",
|
||
"cash_received": "Pagamentos recebidos",
|
||
"won_sales": "Vendas ganhas",
|
||
}
|
||
|
||
_SCHEMA_READY = False
|
||
|
||
|
||
def _float(value: Any, default: float = 0.0) -> float:
|
||
try:
|
||
return float(value or 0)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def ensure_revenue_forecast_schema() -> None:
|
||
global _SCHEMA_READY
|
||
if _SCHEMA_READY:
|
||
return
|
||
with engine.begin() as conn:
|
||
conn.execute(text("""
|
||
CREATE TABLE IF NOT EXISTS sales_targets (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
month_start DATE NOT NULL,
|
||
metric TEXT NOT NULL DEFAULT 'invoiced',
|
||
target_amount NUMERIC(14,2) NOT NULL DEFAULT 0,
|
||
currency TEXT NOT NULL DEFAULT 'EUR',
|
||
updated_by TEXT NOT NULL DEFAULT 'operator',
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
UNIQUE(month_start, metric)
|
||
)
|
||
"""))
|
||
conn.execute(text("CREATE INDEX IF NOT EXISTS idx_sales_targets_month ON sales_targets(month_start, metric)"))
|
||
_SCHEMA_READY = True
|
||
|
||
|
||
def _normalise_metric(metric: str | None) -> str:
|
||
key = str(metric or "invoiced").strip().lower()
|
||
return key if key in TARGET_METRICS else "invoiced"
|
||
|
||
|
||
def _parse_month(value: str | date | datetime | None, *, now: datetime | None = None) -> date:
|
||
now = now or datetime.now(timezone.utc)
|
||
if isinstance(value, datetime):
|
||
return value.date().replace(day=1)
|
||
if isinstance(value, date):
|
||
return value.replace(day=1)
|
||
raw = str(value or "").strip()
|
||
if raw:
|
||
try:
|
||
return datetime.strptime(raw[:7], "%Y-%m").date().replace(day=1)
|
||
except ValueError:
|
||
pass
|
||
return now.date().replace(day=1)
|
||
|
||
|
||
def month_bounds(value: str | date | datetime | None = None, *, now: datetime | None = None) -> tuple[date, date]:
|
||
start = _parse_month(value, now=now)
|
||
return start, date(start.year, start.month, monthrange(start.year, start.month)[1])
|
||
|
||
|
||
def get_sales_target(*, month: str | date | datetime | None = None, metric: str = "invoiced") -> dict[str, Any]:
|
||
ensure_revenue_forecast_schema()
|
||
month_start, _ = month_bounds(month)
|
||
metric = _normalise_metric(metric)
|
||
with engine.begin() as conn:
|
||
row = conn.execute(text("""
|
||
SELECT month_start, metric, target_amount, currency, updated_by, updated_at
|
||
FROM sales_targets
|
||
WHERE month_start = :month_start AND metric = :metric
|
||
"""), {"month_start": month_start, "metric": metric}).mappings().first()
|
||
if not row:
|
||
return {
|
||
"month_start": month_start.isoformat(),
|
||
"metric": metric,
|
||
"metric_label": TARGET_METRICS[metric],
|
||
"target_amount": 0.0,
|
||
"currency": "EUR",
|
||
"configured": False,
|
||
}
|
||
result = dict(row)
|
||
result["month_start"] = str(result.get("month_start"))
|
||
result["target_amount"] = round(_float(result.get("target_amount")), 2)
|
||
result["metric_label"] = TARGET_METRICS[metric]
|
||
result["configured"] = result["target_amount"] > 0
|
||
return result
|
||
|
||
|
||
def set_sales_target(
|
||
*,
|
||
month: str | date | datetime,
|
||
metric: str,
|
||
target_amount: float,
|
||
currency: str = "EUR",
|
||
updated_by: str = "operator",
|
||
) -> dict[str, Any]:
|
||
ensure_revenue_forecast_schema()
|
||
month_start, _ = month_bounds(month)
|
||
metric = _normalise_metric(metric)
|
||
amount = max(0.0, round(_float(target_amount), 2))
|
||
with engine.begin() as conn:
|
||
conn.execute(text("""
|
||
INSERT INTO sales_targets(month_start, metric, target_amount, currency, updated_by)
|
||
VALUES (:month_start, :metric, :target_amount, :currency, :updated_by)
|
||
ON CONFLICT (month_start, metric)
|
||
DO UPDATE SET target_amount = EXCLUDED.target_amount,
|
||
currency = EXCLUDED.currency,
|
||
updated_by = EXCLUDED.updated_by,
|
||
updated_at = now()
|
||
"""), {
|
||
"month_start": month_start,
|
||
"metric": metric,
|
||
"target_amount": amount,
|
||
"currency": str(currency or "EUR").upper()[:3],
|
||
"updated_by": str(updated_by or "operator")[:120],
|
||
})
|
||
return get_sales_target(month=month_start, metric=metric)
|
||
|
||
|
||
def stage_probability(stage: str, historical: dict[str, dict[str, Any]] | None = None) -> tuple[float, str]:
|
||
stage_key = str(stage or "NEW_LEAD").strip().upper()
|
||
default = DEFAULT_STAGE_PROBABILITIES.get(stage_key, 0.15)
|
||
sample = (historical or {}).get(stage_key) or {}
|
||
resolved = int(sample.get("resolved") or 0)
|
||
rate = _float(sample.get("win_rate"), default)
|
||
if resolved >= 8:
|
||
# Bayesian-style shrinkage avoids replacing the prior with a noisy sample.
|
||
prior_weight = 8
|
||
blended = (rate * resolved + default * prior_weight) / (resolved + prior_weight)
|
||
return max(0.02, min(blended, 1.0)), "historical_blended"
|
||
return default, "stage_default"
|
||
|
||
|
||
def activity_factor(*, updated_at: datetime | None, overdue_tasks: int = 0, completed_recent: int = 0, has_conflict: bool = False) -> float:
|
||
now = datetime.now(timezone.utc)
|
||
if updated_at is None:
|
||
recency = 0.65
|
||
else:
|
||
if updated_at.tzinfo is None:
|
||
updated_at = updated_at.replace(tzinfo=timezone.utc)
|
||
age_days = max((now - updated_at).total_seconds() / 86400.0, 0.0)
|
||
if age_days <= 7:
|
||
recency = 1.0
|
||
elif age_days <= 14:
|
||
recency = 0.92
|
||
elif age_days <= 30:
|
||
recency = 0.78
|
||
elif age_days <= 60:
|
||
recency = 0.62
|
||
else:
|
||
recency = 0.48
|
||
factor = recency
|
||
if overdue_tasks:
|
||
factor *= max(0.65, 1.0 - min(int(overdue_tasks), 4) * 0.08)
|
||
# Small positive cap: routine task completion must not inflate sales probability.
|
||
if completed_recent:
|
||
factor *= min(1.03, 1.0 + min(int(completed_recent), 3) * 0.01)
|
||
if has_conflict:
|
||
factor *= 0.60
|
||
return round(max(0.20, min(factor, 1.03)), 4)
|
||
|
||
|
||
def forecast_bucket(expected_date: datetime, *, now: datetime | None = None) -> str:
|
||
now = now or datetime.now(timezone.utc)
|
||
if expected_date.tzinfo is None:
|
||
expected_date = expected_date.replace(tzinfo=timezone.utc)
|
||
days = (expected_date - now).days
|
||
if days <= 30:
|
||
return "0_30"
|
||
if days <= 60:
|
||
return "31_60"
|
||
if days <= 90:
|
||
return "61_90"
|
||
return "90_plus"
|
||
|
||
|
||
def management_status(*, target: float, forecast_total: float, quality_score: float, recoverable_total: float = 0.0) -> dict[str, str]:
|
||
target = max(_float(target), 0.0)
|
||
forecast_total = max(_float(forecast_total), 0.0)
|
||
recoverable_total = max(_float(recoverable_total), 0.0)
|
||
quality_score = max(0.0, min(_float(quality_score), 1.0))
|
||
if target <= 0:
|
||
return {"code": "no_target", "label": "Meta não configurada", "tone": "gray", "message": "Define a meta mensal para calcular o desvio e o cumprimento previsto."}
|
||
ratio = forecast_total / target
|
||
if ratio >= 1.0 and quality_score >= 0.60:
|
||
return {"code": "supported", "label": "Meta suportada", "tone": "green", "message": "O realizado, comprometido e pipeline provável suportam a meta atual."}
|
||
if ratio >= 1.0:
|
||
return {"code": "supported_low_confidence", "label": "Meta suportada com baixa confiança", "tone": "orange", "message": "O valor suporta a meta, mas a cobertura ou qualidade dos dados é insuficiente para tratar a previsão como segura."}
|
||
if forecast_total + recoverable_total >= target and recoverable_total > 0:
|
||
return {"code": "recoverable", "label": "Meta recuperável", "tone": "orange", "message": "A previsão base está abaixo da meta, mas pagamentos e conversões já existentes podem cobrir o desvio se forem acelerados neste mês."}
|
||
if ratio >= 0.85:
|
||
return {"code": "at_risk", "label": "Meta em risco moderado", "tone": "orange", "message": "Acelera oportunidades existentes, pagamentos e valorização antes de aumentar a prospeção."}
|
||
return {"code": "not_supported", "label": "Meta sem cobertura suficiente", "tone": "red", "message": "Mesmo acelerando o pipeline recuperável conhecido, continua a faltar valor; reforça conversão e novo pipeline."}
|
||
|
||
|
||
def _historical_stage_rates(conn: Any) -> dict[str, dict[str, Any]]:
|
||
rows = conn.execute(text("""
|
||
WITH resolved AS (
|
||
SELECT id,
|
||
CASE WHEN upper(stage) IN ('WON','DELIVERED') THEN 1 ELSE 0 END AS won
|
||
FROM opportunities
|
||
WHERE lower(status) IN ('closed','won','lost','no_interest')
|
||
OR upper(stage) IN ('WON','LOST','NO_INTEREST','DELIVERED')
|
||
), visits AS (
|
||
SELECT DISTINCT e.opportunity_id, upper(e.to_stage) AS stage
|
||
FROM opportunity_events e
|
||
WHERE e.to_stage IS NOT NULL AND btrim(e.to_stage) <> ''
|
||
)
|
||
SELECT v.stage,
|
||
COUNT(*)::int AS resolved,
|
||
SUM(r.won)::int AS won,
|
||
AVG(r.won::numeric)::float AS win_rate
|
||
FROM visits v
|
||
JOIN resolved r ON r.id = v.opportunity_id
|
||
GROUP BY v.stage
|
||
""")).mappings().all()
|
||
return {
|
||
str(row.get("stage") or "").upper(): {
|
||
"resolved": int(row.get("resolved") or 0),
|
||
"won": int(row.get("won") or 0),
|
||
"win_rate": _float(row.get("win_rate")),
|
||
}
|
||
for row in rows
|
||
}
|
||
|
||
|
||
def _realised_for_period(conn: Any, *, metric: str, period_start: date, period_end: date) -> dict[str, Any]:
|
||
if metric == "cash_received":
|
||
rows = conn.execute(text("""
|
||
WITH latest_doc AS (
|
||
SELECT DISTINCT ON (opportunity_id) opportunity_id, COALESCE(total_amount, amount, 0) AS amount
|
||
FROM commercial_documents
|
||
WHERE opportunity_id IS NOT NULL AND COALESCE(is_active, TRUE) = TRUE
|
||
ORDER BY opportunity_id,
|
||
CASE WHEN document_kind = 'invoice' THEN 0 ELSE 1 END,
|
||
COALESCE(document_date, created_at::date) DESC,
|
||
created_at DESC
|
||
)
|
||
SELECT ol.opportunity_id::text,
|
||
COALESCE(CASE WHEN COALESCE(ol.payload->>'amount','') ~ '^[0-9]+([.,][0-9]+)?$' THEN replace(ol.payload->>'amount', ',', '.')::numeric END, ld.amount, o.value_amount, 0) AS amount,
|
||
COALESCE(ol.updated_at, ol.created_at) AS realised_at,
|
||
COALESCE(ol.external_name, 'Pagamento confirmado') AS reference
|
||
FROM operation_links ol
|
||
JOIN opportunities o ON o.id = ol.opportunity_id
|
||
LEFT JOIN latest_doc ld ON ld.opportunity_id = ol.opportunity_id
|
||
WHERE ol.system = 'clientflow'
|
||
AND ol.external_type = 'payment'
|
||
AND lower(ol.status) = 'confirmed'
|
||
AND COALESCE(ol.updated_at, ol.created_at)::date BETWEEN :period_start AND :period_end
|
||
"""), {"period_start": period_start, "period_end": period_end}).mappings().all()
|
||
elif metric == "won_sales":
|
||
rows = conn.execute(text("""
|
||
SELECT o.id::text AS opportunity_id,
|
||
COALESCE(o.value_amount, ld.amount, 0) AS amount,
|
||
o.updated_at AS realised_at,
|
||
COALESCE(o.customer_name, o.title, 'Venda ganha') AS reference
|
||
FROM opportunities o
|
||
LEFT JOIN LATERAL (
|
||
SELECT COALESCE(total_amount, amount, 0) AS amount
|
||
FROM commercial_documents d
|
||
WHERE d.opportunity_id = o.id AND COALESCE(d.is_active, TRUE) = TRUE
|
||
ORDER BY CASE WHEN d.document_kind = 'invoice' THEN 0 ELSE 1 END,
|
||
COALESCE(d.document_date, d.created_at::date) DESC
|
||
LIMIT 1
|
||
) ld ON TRUE
|
||
WHERE (upper(o.stage) IN ('WON','DELIVERED') OR lower(o.status) IN ('won','closed'))
|
||
AND o.updated_at::date BETWEEN :period_start AND :period_end
|
||
"""), {"period_start": period_start, "period_end": period_end}).mappings().all()
|
||
else:
|
||
rows = conn.execute(text("""
|
||
SELECT d.opportunity_id::text,
|
||
COALESCE(d.total_amount, d.amount, 0) AS amount,
|
||
COALESCE(d.document_date::timestamp, d.created_at) AS realised_at,
|
||
COALESCE(d.document_number, d.external_id, 'Fatura') AS reference
|
||
FROM commercial_documents d
|
||
WHERE lower(d.document_kind) = 'invoice'
|
||
AND COALESCE(d.is_active, TRUE) = TRUE
|
||
AND COALESCE(d.role, 'current') IN ('current','accepted')
|
||
AND COALESCE(d.document_date, d.created_at::date) BETWEEN :period_start AND :period_end
|
||
"""), {"period_start": period_start, "period_end": period_end}).mappings().all()
|
||
items = [dict(r) for r in rows]
|
||
return {
|
||
"amount": round(sum(_float(r.get("amount")) for r in items), 2),
|
||
"count": len(items),
|
||
"opportunity_ids": {str(r.get("opportunity_id")) for r in items if r.get("opportunity_id")},
|
||
"items": items,
|
||
}
|
||
|
||
|
||
def _already_realised_ids(conn: Any, *, metric: str) -> set[str]:
|
||
if metric == "cash_received":
|
||
sql = """
|
||
SELECT DISTINCT opportunity_id::text
|
||
FROM operation_links
|
||
WHERE system = 'clientflow' AND external_type = 'payment' AND lower(status) = 'confirmed'
|
||
"""
|
||
elif metric == "won_sales":
|
||
sql = """
|
||
SELECT DISTINCT id::text
|
||
FROM opportunities
|
||
WHERE upper(stage) IN ('WON','DELIVERED') OR lower(status) IN ('won','closed')
|
||
"""
|
||
else:
|
||
sql = """
|
||
SELECT DISTINCT opportunity_id::text
|
||
FROM commercial_documents
|
||
WHERE opportunity_id IS NOT NULL
|
||
AND lower(document_kind) = 'invoice'
|
||
AND COALESCE(is_active, TRUE) = TRUE
|
||
AND COALESCE(role, 'current') IN ('current','accepted')
|
||
"""
|
||
return {str(r[0]) for r in conn.execute(text(sql)).all() if r[0]}
|
||
|
||
|
||
def _stage_summary(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
grouped: dict[str, dict[str, Any]] = {}
|
||
for item in items:
|
||
stage = str(item.get("stage") or "NEW_LEAD")
|
||
row = grouped.setdefault(stage, {"stage": stage, "count": 0, "valued": 0, "gross": 0.0, "weighted": 0.0})
|
||
row["count"] += 1
|
||
if _float(item.get("amount")) > 0:
|
||
row["valued"] += 1
|
||
row["gross"] += _float(item.get("amount"))
|
||
row["weighted"] += _float(item.get("weighted_amount"))
|
||
result = list(grouped.values())
|
||
for row in result:
|
||
row["gross"] = round(row["gross"], 2)
|
||
row["weighted"] = round(row["weighted"], 2)
|
||
return sorted(result, key=lambda r: (r["weighted"], r["gross"]), reverse=True)
|
||
|
||
|
||
def get_revenue_forecast(*, limit: int = 1000, month: str | None = None, metric: str = "invoiced") -> Dict[str, Any]:
|
||
"""Return management target, month and 30/60/90-day pipeline forecasts."""
|
||
ensure_revenue_forecast_schema()
|
||
limit = max(1, min(int(limit or 1000), 5000))
|
||
metric = _normalise_metric(metric)
|
||
now = datetime.now(timezone.utc)
|
||
period_start, period_end = month_bounds(month, now=now)
|
||
target = get_sales_target(month=period_start, metric=metric)
|
||
|
||
with engine.begin() as conn:
|
||
historical = _historical_stage_rates(conn)
|
||
realised = _realised_for_period(conn, metric=metric, period_start=period_start, period_end=period_end)
|
||
already_realised_ids = _already_realised_ids(conn, metric=metric)
|
||
rows = conn.execute(text("""
|
||
WITH latest_doc AS (
|
||
SELECT DISTINCT ON (opportunity_id)
|
||
opportunity_id,
|
||
total_amount,
|
||
document_number,
|
||
document_kind,
|
||
document_date
|
||
FROM commercial_documents
|
||
WHERE COALESCE(is_active, TRUE) = TRUE
|
||
AND COALESCE(role, 'current') IN ('current','accepted','historical','history')
|
||
ORDER BY opportunity_id,
|
||
CASE WHEN COALESCE(is_primary, FALSE) THEN 0 ELSE 1 END,
|
||
CASE document_kind WHEN 'invoice' THEN 1 WHEN 'quotation' THEN 2 WHEN 'proforma' THEN 3 ELSE 4 END,
|
||
COALESCE(document_date, created_at::date) DESC,
|
||
created_at DESC
|
||
), item_totals AS (
|
||
SELECT opportunity_id, SUM(COALESCE(total_price, 0)) AS total
|
||
FROM opportunity_items
|
||
GROUP BY opportunity_id
|
||
), task_stats AS (
|
||
SELECT opportunity_id,
|
||
COUNT(*) FILTER (WHERE status = 'pending' AND due_at IS NOT NULL AND due_at < now())::int AS overdue_tasks,
|
||
COUNT(*) FILTER (WHERE status IN ('done','completed') AND COALESCE(done_at, updated_at) >= now() - interval '14 days')::int AS completed_recent
|
||
FROM tasks
|
||
WHERE opportunity_id IS NOT NULL
|
||
GROUP BY opportunity_id
|
||
), payment_flags AS (
|
||
SELECT opportunity_id,
|
||
BOOL_OR(system = 'clientflow' AND external_type = 'payment' AND lower(status) = 'confirmed') AS payment_confirmed
|
||
FROM operation_links
|
||
GROUP BY opportunity_id
|
||
)
|
||
SELECT o.id::text,
|
||
o.title,
|
||
o.customer_name,
|
||
o.stage,
|
||
o.status,
|
||
o.value_amount,
|
||
o.currency,
|
||
o.updated_at,
|
||
o.metadata,
|
||
ld.total_amount AS document_amount,
|
||
ld.document_number,
|
||
ld.document_kind,
|
||
ld.document_date,
|
||
it.total AS item_amount,
|
||
COALESCE(ts.overdue_tasks, 0) AS overdue_tasks,
|
||
COALESCE(ts.completed_recent, 0) AS completed_recent,
|
||
COALESCE(pf.payment_confirmed, FALSE) AS payment_confirmed
|
||
FROM opportunities o
|
||
LEFT JOIN latest_doc ld ON ld.opportunity_id = o.id
|
||
LEFT JOIN item_totals it ON it.opportunity_id = o.id
|
||
LEFT JOIN task_stats ts ON ts.opportunity_id = o.id
|
||
LEFT JOIN payment_flags pf ON pf.opportunity_id = o.id
|
||
WHERE lower(o.status) = 'open'
|
||
AND upper(o.stage) NOT IN ('LOST','NO_INTEREST','ARCHIVED')
|
||
AND NOT COALESCE(lower(o.metadata->>'exclude_from_funnel') IN ('true','1','yes','sim'), FALSE)
|
||
ORDER BY o.updated_at DESC
|
||
LIMIT :limit
|
||
"""), {"limit": limit}).mappings().all()
|
||
|
||
buckets = {
|
||
"0_30": {"label": "Até 30 dias", "gross": 0.0, "weighted": 0.0, "count": 0, "valued": 0},
|
||
"31_60": {"label": "31–60 dias", "gross": 0.0, "weighted": 0.0, "count": 0, "valued": 0},
|
||
"61_90": {"label": "61–90 dias", "gross": 0.0, "weighted": 0.0, "count": 0, "valued": 0},
|
||
"90_plus": {"label": "Mais de 90 dias", "gross": 0.0, "weighted": 0.0, "count": 0, "valued": 0},
|
||
}
|
||
items: list[dict[str, Any]] = []
|
||
valued = stale = conflicts = 0
|
||
|
||
for row in rows:
|
||
data = dict(row)
|
||
metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
|
||
document_amount = _float(data.get("document_amount"))
|
||
item_amount = _float(data.get("item_amount"))
|
||
opportunity_amount = _float(data.get("value_amount"))
|
||
if document_amount > 0:
|
||
amount, value_source = document_amount, "document"
|
||
elif item_amount > 0:
|
||
amount, value_source = item_amount, "items"
|
||
else:
|
||
amount, value_source = opportunity_amount, "opportunity"
|
||
if amount > 0:
|
||
valued += 1
|
||
|
||
stage = str(data.get("stage") or "NEW_LEAD").upper()
|
||
probability, probability_source = stage_probability(stage, historical)
|
||
has_conflict = bool(metadata.get("identity_conflict") or metadata.get("fiscal_conflict") or metadata.get("has_nif_conflict"))
|
||
conflicts += int(has_conflict)
|
||
factor = activity_factor(
|
||
updated_at=data.get("updated_at"),
|
||
overdue_tasks=int(data.get("overdue_tasks") or 0),
|
||
completed_recent=int(data.get("completed_recent") or 0),
|
||
has_conflict=has_conflict,
|
||
)
|
||
effective_probability = max(0.0, min(probability * factor, 1.0))
|
||
weighted = round(amount * effective_probability, 2)
|
||
days = STAGE_EXPECTED_DAYS.get(stage, 90)
|
||
expected_date = now + timedelta(days=days)
|
||
bucket_key = forecast_bucket(expected_date, now=now)
|
||
bucket = buckets[bucket_key]
|
||
bucket["gross"] = round(bucket["gross"] + amount, 2)
|
||
bucket["weighted"] = round(bucket["weighted"] + weighted, 2)
|
||
bucket["count"] += 1
|
||
bucket["valued"] += int(amount > 0)
|
||
updated_at = data.get("updated_at")
|
||
if updated_at and updated_at.tzinfo is None:
|
||
updated_at = updated_at.replace(tzinfo=timezone.utc)
|
||
is_stale = bool(updated_at and (now - updated_at).days > 30)
|
||
stale += int(is_stale)
|
||
opp_id = str(data.get("id") or "")
|
||
realised_in_period = opp_id in realised["opportunity_ids"]
|
||
already_realised = opp_id in already_realised_ids
|
||
if realised_in_period:
|
||
forecast_class = "realised"
|
||
elif already_realised:
|
||
forecast_class = "realised_other_period"
|
||
elif stage in COMMITTED_STAGES:
|
||
forecast_class = "committed"
|
||
else:
|
||
forecast_class = "probable"
|
||
items.append({
|
||
"id": opp_id,
|
||
"title": data.get("title"),
|
||
"customer_name": data.get("customer_name"),
|
||
"stage": stage,
|
||
"amount": round(amount, 2),
|
||
"currency": data.get("currency") or "EUR",
|
||
"value_source": value_source,
|
||
"document_number": data.get("document_number"),
|
||
"document_kind": data.get("document_kind"),
|
||
"document_date": str(data.get("document_date") or ""),
|
||
"payment_confirmed": bool(data.get("payment_confirmed")),
|
||
"probability": round(probability, 4),
|
||
"probability_source": probability_source,
|
||
"probability_sample": historical.get(stage) or {},
|
||
"activity_factor": factor,
|
||
"effective_probability": round(effective_probability, 4),
|
||
"weighted_amount": weighted,
|
||
"expected_date": expected_date.isoformat(),
|
||
"bucket": bucket_key,
|
||
"forecast_class": forecast_class,
|
||
"overdue_tasks": int(data.get("overdue_tasks") or 0),
|
||
"completed_recent": int(data.get("completed_recent") or 0),
|
||
"has_conflict": has_conflict,
|
||
"is_stale": is_stale,
|
||
"updated_at": str(data.get("updated_at") or ""),
|
||
})
|
||
|
||
total = len(items)
|
||
gross = round(sum(x["amount"] for x in items), 2)
|
||
weighted = round(sum(x["weighted_amount"] for x in items), 2)
|
||
value_coverage = round(valued / total, 4) if total else 0.0
|
||
quality_score = round(max(0.0, min(1.0, value_coverage * 0.65 + (1 - stale / total if total else 0) * 0.20 + (1 - conflicts / total if total else 0) * 0.15)), 4)
|
||
|
||
month_end_dt = datetime.combine(period_end, datetime.max.time(), tzinfo=timezone.utc)
|
||
next_30_end = now + timedelta(days=30)
|
||
|
||
def eligible_future(item: dict[str, Any]) -> bool:
|
||
return item["forecast_class"] in {"committed", "probable"} and item["amount"] > 0
|
||
|
||
month_items = [i for i in items if eligible_future(i) and datetime.fromisoformat(i["expected_date"]) <= month_end_dt]
|
||
next_30_items = [i for i in items if eligible_future(i) and datetime.fromisoformat(i["expected_date"]) <= next_30_end]
|
||
|
||
def horizon_summary(scope_items: list[dict[str, Any]], *, include_realised: bool) -> dict[str, Any]:
|
||
committed_items = [i for i in scope_items if i["forecast_class"] == "committed"]
|
||
probable_items = [i for i in scope_items if i["forecast_class"] == "probable"]
|
||
committed_amount = round(sum(i["amount"] for i in committed_items), 2)
|
||
probable_weighted = round(sum(i["weighted_amount"] for i in probable_items), 2)
|
||
probable_gross = round(sum(i["amount"] for i in probable_items), 2)
|
||
realised_amount = realised["amount"] if include_realised else 0.0
|
||
return {
|
||
"realised": realised_amount,
|
||
"committed": committed_amount,
|
||
"probable": probable_weighted,
|
||
"probable_gross": probable_gross,
|
||
"forecast_total": round(realised_amount + committed_amount + probable_weighted, 2),
|
||
"future_total": round(committed_amount + probable_weighted, 2),
|
||
"committed_count": len(committed_items),
|
||
"probable_count": len(probable_items),
|
||
"count": len(scope_items),
|
||
}
|
||
|
||
month_horizon = horizon_summary(month_items, include_realised=True)
|
||
next_30_horizon = horizon_summary(next_30_items, include_realised=False)
|
||
target_amount = _float(target.get("target_amount"))
|
||
forecast_total = month_horizon["forecast_total"]
|
||
gap = round(max(target_amount - forecast_total, 0.0), 2)
|
||
surplus = round(max(forecast_total - target_amount, 0.0), 2)
|
||
attainment = round(forecast_total / target_amount, 4) if target_amount > 0 else 0.0
|
||
|
||
# Existing WAITING_PAYMENT pipeline expected after month-end can sometimes
|
||
# be accelerated into the selected month. It remains separate from the base
|
||
# forecast so the dashboard does not overstate certainty.
|
||
month_item_ids = {i["id"] for i in month_items}
|
||
recovery_items = [
|
||
i for i in items
|
||
if i["forecast_class"] == "probable"
|
||
and i["stage"] == "WAITING_PAYMENT"
|
||
and i["amount"] > 0
|
||
and i["id"] not in month_item_ids
|
||
]
|
||
recoverable_weighted = round(sum(i["weighted_amount"] for i in recovery_items), 2)
|
||
recoverable_gross = round(sum(i["amount"] for i in recovery_items), 2)
|
||
accelerated_total = round(forecast_total + recoverable_weighted, 2)
|
||
maximum_known_total = round(forecast_total + recoverable_gross, 2)
|
||
residual_gap = round(max(target_amount - accelerated_total, 0.0), 2)
|
||
accelerated_attainment = round(accelerated_total / target_amount, 4) if target_amount > 0 else 0.0
|
||
status = management_status(
|
||
target=target_amount,
|
||
forecast_total=forecast_total,
|
||
quality_score=quality_score,
|
||
recoverable_total=recoverable_weighted,
|
||
)
|
||
|
||
probable_valued = [i for i in items if i["forecast_class"] == "probable" and i["amount"] > 0]
|
||
avg_conversion = (
|
||
sum(i["effective_probability"] for i in probable_valued) / len(probable_valued)
|
||
if probable_valued else 0.35
|
||
)
|
||
avg_conversion = max(0.15, min(avg_conversion, 0.75))
|
||
new_pipeline_required = round(residual_gap / avg_conversion, 2) if residual_gap > 0 else 0.0
|
||
average_value = round(sum(i["amount"] for i in probable_valued) / len(probable_valued), 2) if probable_valued else 0.0
|
||
new_opportunities_required = int(-(-new_pipeline_required // average_value)) if new_pipeline_required > 0 and average_value > 0 else 0
|
||
|
||
conservative = round(realised["amount"] + month_horizon["committed"] * 0.90 + month_horizon["probable"] * 0.75, 2)
|
||
probable = forecast_total
|
||
optimistic = accelerated_total
|
||
|
||
future_items = [i for i in items if i["forecast_class"] in {"committed", "probable"}]
|
||
valued_future = [i for i in future_items if i["amount"] > 0]
|
||
top4 = sum(i["weighted_amount"] if i["forecast_class"] == "probable" else i["amount"] for i in sorted(valued_future, key=lambda x: (x["weighted_amount"], x["amount"]), reverse=True)[:4])
|
||
future_expected_total = sum(i["weighted_amount"] if i["forecast_class"] == "probable" else i["amount"] for i in valued_future)
|
||
concentration = round(top4 / future_expected_total, 4) if future_expected_total else 0.0
|
||
|
||
waiting_payment = [i for i in future_items if i["stage"] == "WAITING_PAYMENT" and i["amount"] > 0]
|
||
operational_blocked = [i for i in future_items if i["forecast_class"] == "committed" and i["amount"] > 0]
|
||
zero_value_advanced = [i for i in future_items if i["stage"] in ADVANCED_VALUE_STAGES and i["amount"] <= 0]
|
||
overdue_items = [i for i in future_items if i["overdue_tasks"] > 0]
|
||
|
||
diagnostics: list[dict[str, Any]] = []
|
||
if total - valued:
|
||
diagnostics.append({"severity": "high" if value_coverage < 0.5 else "medium", "label": f"{total - valued} oportunidades sem valor", "detail": f"Cobertura monetária de {value_coverage * 100:.0f}%."})
|
||
if waiting_payment:
|
||
diagnostics.append({"severity": "medium", "label": f"{len(waiting_payment)} pagamentos pendentes", "detail": f"Valor conhecido: {sum(i['amount'] for i in waiting_payment):.2f} EUR."})
|
||
if overdue_items:
|
||
diagnostics.append({"severity": "medium", "label": f"{len(overdue_items)} oportunidades com tasks vencidas", "detail": "Podem atrasar conversão ou execução."})
|
||
if concentration >= 0.60:
|
||
diagnostics.append({"severity": "medium", "label": "Previsão concentrada", "detail": f"As 4 maiores oportunidades representam {concentration * 100:.0f}% do valor esperado futuro."})
|
||
if gap > 0 and recoverable_weighted > 0:
|
||
diagnostics.append({"severity": "medium", "label": f"Meta recuperável: {recoverable_weighted:.2f} EUR ponderados", "detail": f"Acelera {len(recovery_items)} pagamento(s) existente(s). Desvio residual após recuperação: {residual_gap:.2f} EUR."})
|
||
if residual_gap > 0:
|
||
diagnostics.append({"severity": "high", "label": f"Desvio residual de {residual_gap:.2f} EUR", "detail": f"Novo pipeline estimado necessário: {new_pipeline_required:.2f} EUR."})
|
||
|
||
actions: list[dict[str, Any]] = []
|
||
for item in sorted(operational_blocked, key=lambda x: x["amount"], reverse=True)[:5]:
|
||
actions.append({**item, "action_group": "Executar receita comprometida", "recommended_action": "Concluir a próxima ação operacional", "impact_amount": item["amount"], "priority": 1})
|
||
for item in sorted(waiting_payment, key=lambda x: x["amount"], reverse=True)[:5]:
|
||
actions.append({**item, "action_group": "Acelerar pagamento", "recommended_action": "Contactar e confirmar data de pagamento", "impact_amount": item["weighted_amount"], "priority": 2})
|
||
for item in sorted(overdue_items, key=lambda x: (x["weighted_amount"], x["amount"]), reverse=True)[:5]:
|
||
actions.append({**item, "action_group": "Recuperar atraso", "recommended_action": "Resolver task vencida", "impact_amount": item["weighted_amount"], "priority": 3})
|
||
for item in sorted(zero_value_advanced, key=lambda x: x["updated_at"], reverse=True)[:5]:
|
||
actions.append({**item, "action_group": "Valorizar oportunidade", "recommended_action": "Associar documento ou definir valor", "impact_amount": 0.0, "priority": 4})
|
||
deduped_actions: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for action in sorted(actions, key=lambda x: (x["priority"], -_float(x["impact_amount"]))):
|
||
if action["id"] in seen:
|
||
continue
|
||
seen.add(action["id"])
|
||
deduped_actions.append(action)
|
||
|
||
items.sort(key=lambda x: (x["weighted_amount"], x["amount"]), reverse=True)
|
||
return {
|
||
"generated_at": now.isoformat(),
|
||
"model": "sales_target_management_forecast_v2",
|
||
"scope": "management_forecast_not_accounting_or_cashflow",
|
||
"period": {"month_start": period_start.isoformat(), "month_end": period_end.isoformat(), "next_30_end": next_30_end.date().isoformat()},
|
||
"target": target,
|
||
"management": {
|
||
"metric": metric,
|
||
"metric_label": TARGET_METRICS[metric],
|
||
"month": month_horizon,
|
||
"next_30_days": next_30_horizon,
|
||
"target_amount": round(target_amount, 2),
|
||
"gap": gap,
|
||
"surplus": surplus,
|
||
"attainment": attainment,
|
||
"status": status,
|
||
"recoverable": {
|
||
"weighted": recoverable_weighted,
|
||
"gross": recoverable_gross,
|
||
"count": len(recovery_items),
|
||
"accelerated_total": accelerated_total,
|
||
"maximum_known_total": maximum_known_total,
|
||
"residual_gap": residual_gap,
|
||
"accelerated_attainment": accelerated_attainment,
|
||
},
|
||
"scenarios": {"conservative": conservative, "probable": probable, "optimistic": optimistic, "maximum_known": maximum_known_total},
|
||
"new_pipeline_conversion": round(avg_conversion, 4),
|
||
"new_pipeline_required": new_pipeline_required,
|
||
"average_opportunity_value": average_value,
|
||
"new_opportunities_required": new_opportunities_required,
|
||
"concentration_top4": concentration,
|
||
},
|
||
"summary": {
|
||
"opportunities": total,
|
||
"gross_pipeline": gross,
|
||
"weighted_pipeline": weighted,
|
||
"valued_opportunities": valued,
|
||
"unvalued_opportunities": total - valued,
|
||
"value_coverage": value_coverage,
|
||
"stale_opportunities": stale,
|
||
"identity_conflicts": conflicts,
|
||
"quality_score": quality_score,
|
||
"realised_amount": realised["amount"],
|
||
"realised_count": realised["count"],
|
||
},
|
||
"buckets": buckets,
|
||
"historical_stage_rates": historical,
|
||
"stage_summary": _stage_summary(items),
|
||
"diagnostics": diagnostics,
|
||
"priority_actions": deduped_actions[:10],
|
||
"realised_items": realised["items"],
|
||
"items": items,
|
||
}
|