"""CRM Analytics API routes.
Endpoints that aggregate CRM data synced from HubSpot (or other CRM sources)
into actionable business metrics: pipeline value, deal velocity, contact trends.
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
from flask import Blueprint, request, jsonify, g
from flask_login import current_user
from ..models import db, CrmContact, CrmDeal, CrmCompany, Company, UserCompany, User, Goal, GoalMetric, AdMetric, AdCampaign, Project, RevenueRecord, Location
from ..services.forecast_service import calculate_run_rate
from .api_proxy import require_auth_json, require_tier
from app.utils.csrf import require_csrf
analytics_bp = Blueprint("analytics", __name__)
logger = logging.getLogger(__name__)
# -- helpers ----------------------------------------------------------------
# Map numeric goal levels (from seed data) to string levels used by cascade logic
_LEVEL_MAP: Dict[str, str] = {
"1": "org",
"2": "department",
"3": "team",
"4": "rep",
}
def _normalise_level(raw: Any) -> str:
"""Normalise a goal's level to one of: org, department, team, rep."""
if raw is None:
return "org"
s = str(raw).strip()
return _LEVEL_MAP.get(s, s.lower())
def _get_company_id() -> str:
"""Return the first company associated with the current user.
Super-admin proxy routes can inject a company_id via ``g.proxy_company_id``
to bypass the membership lookup.
"""
# Check for proxy override first (super-admin impersonation)
if getattr(g, 'proxy_company_id', None):
return g.proxy_company_id
user_companies = UserCompany.query.filter_by(user_id=current_user.id).all()
if not user_companies:
raise ValueError("No company found for current user")
return user_companies[0].company_id
def _get_user_company_ids() -> List[str]:
"""Return ALL company IDs associated with the current user."""
user_companies = UserCompany.query.filter_by(user_id=current_user.id).all()
if not user_companies:
raise ValueError("No company found for current user")
return [uc.company_id for uc in user_companies]
def _json_error(msg: str, code: int) -> tuple:
return jsonify({"error": msg}), code
# -- Pipeline overview ------------------------------------------------------
@analytics_bp.route("/api/analytics/pipeline", methods=["GET"])
@require_auth_json()
def pipeline_overview():
"""GET /api/analytics/pipeline
Returns deals grouped by stage with total count and value per stage,
plus aggregate pipeline metrics.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
# Group deals by stage
deals = CrmDeal.query.filter_by(company_id=company_id).all()
stages: Dict[str, Dict[str, Any]] = {}
total_value = 0.0
weighted_value = 0.0
total_deals = len(deals)
for deal in deals:
stage_name = deal.stage or "unknown"
amount = deal.amount or 0.0
probability = deal.probability or 0.0
if stage_name not in stages:
stages[stage_name] = {"stage": stage_name, "count": 0, "total_amount": 0.0, "weighted_amount": 0.0}
stages[stage_name]["count"] += 1
stages[stage_name]["total_amount"] += amount
stages[stage_name]["weighted_amount"] += amount * probability
total_value += amount
weighted_value += amount * probability
# Count open vs closed
closed_won = sum(1 for d in deals if d.stage in ("closedwon", "closed_won", "won"))
closed_lost = sum(1 for d in deals if d.stage in ("closedlost", "closed_lost", "lost"))
# Average deal size
avg_deal = total_value / total_deals if total_deals > 0 else 0.0
return jsonify({
"pipeline": {
"total_deals": total_deals,
"total_value": round(total_value, 2),
"weighted_value": round(weighted_value, 2),
"avg_deal_size": round(avg_deal, 2),
"closed_won": closed_won,
"closed_lost": closed_lost,
},
"stages": sorted(stages.values(), key=lambda s: -s["weighted_amount"]),
})
# -- Contact trends ---------------------------------------------------------
@analytics_bp.route("/api/analytics/contacts", methods=["GET"])
@require_auth_json()
def contact_analytics():
"""GET /api/analytics/contacts
Returns contact counts by lifecycle stage and a time-series of contact
growth over the past N days (default 30).
Query params:
period_days (int, default 30) — days of history to include in trend.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
period_days = request.args.get("period_days", 30, type=int)
contacts = CrmContact.query.filter_by(company_id=company_id).all()
# Count by lifecycle stage
stage_counts: Dict[str, int] = {}
for c in contacts:
stage = c.lifecycle_stage or "unknown"
stage_counts[stage] = stage_counts.get(stage, 0) + 1
# Daily trend: count contacts created per day over the last period_days
cutoff = datetime.now(timezone.utc) - timedelta(days=period_days)
trend: Dict[str, int] = {}
for day_offset in range(period_days):
day_start = cutoff + timedelta(days=day_offset)
day_end = day_start + timedelta(days=1)
day_key = day_start.strftime("%Y-%m-%d")
trend[day_key] = 0
for c in contacts:
if c.created_at and c.created_at >= cutoff:
day_key = c.created_at.strftime("%Y-%m-%d")
if day_key in trend:
trend[day_key] += 1
# Cumulative trend
cumulative = 0
trend_series: List[Dict[str, Any]] = []
for day_key in sorted(trend.keys()):
cumulative += trend[day_key]
trend_series.append({
"date": day_key,
"new_contacts": trend[day_key],
"cumulative": cumulative,
})
return jsonify({
"contacts": {
"total": len(contacts),
"by_lifecycle_stage": stage_counts,
},
"trend": trend_series,
"period_days": period_days,
})
# -- Deal velocity ----------------------------------------------------------
@analytics_bp.route("/api/analytics/deals/velocity", methods=["GET"])
@require_auth_json()
def deal_velocity():
"""GET /api/analytics/deals/velocity
Returns deal velocity metrics:
- Average days from creation to close for won deals.
- Breakdown by stage showing average days spent in each stage (approximate).
- Median and 90th-percentile deal duration.
Query params:
since_days (int, default 90) — look back this many days for deals.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
since_days = request.args.get("since_days", 90, type=int)
cutoff = datetime.now(timezone.utc) - timedelta(days=since_days)
deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.created_at >= cutoff
).all()
# Calculate deal durations for closed deals
durations: List[int] = []
stage_durations: Dict[str, List[int]] = {}
for deal in deals:
created = deal.created_at or datetime.now(timezone.utc)
# For closed deals, use expected_close_date or updated_at as proxy
closed_at: Optional[datetime] = None
if deal.stage in ("closedwon", "closed_won", "won", "closedlost", "closed_lost", "lost"):
closed_at = deal.expected_close_date or deal.updated_at
if closed_at:
delta = (closed_at - created).days
if delta > 0:
durations.append(delta)
# Statistics
if durations:
durations_sorted = sorted(durations)
avg_days = sum(durations) / len(durations)
median_days = durations_sorted[len(durations_sorted) // 2]
p90_idx = min(int(len(durations_sorted) * 0.9), len(durations_sorted) - 1)
p90_days = durations_sorted[p90_idx]
min_days = durations_sorted[0]
max_days = durations_sorted[-1]
else:
avg_days = median_days = p90_days = min_days = max_days = 0
# Won deals value
won_deals = [d for d in deals if d.stage in ("closedwon", "closed_won", "won")]
won_value = sum(d.amount or 0.0 for d in won_deals)
# Win rate
closed_deals = [d for d in deals if d.stage in ("closedwon", "closed_won", "won", "closedlost", "closed_lost", "lost")]
win_rate = (len(won_deals) / len(closed_deals) * 100) if closed_deals else 0.0
return jsonify({
"velocity": {
"avg_days_to_close": round(avg_days, 1),
"median_days_to_close": median_days,
"p90_days_to_close": p90_days,
"min_days_to_close": min_days,
"max_days_to_close": max_days,
"deals_analyzed": len(durations),
},
"win_metrics": {
"total_closed": len(closed_deals),
"won": len(won_deals),
"lost": len(closed_deals) - len(won_deals),
"win_rate_pct": round(win_rate, 1),
"won_value": round(won_value, 2),
},
"period_days": since_days,
})
# -- CRM summary dashboard --------------------------------------------------
@analytics_bp.route("/api/analytics/crm/summary", methods=["GET"])
@require_auth_json()
def crm_summary():
"""GET /api/analytics/crm/summary
High-level summary of all CRM data for the dashboard.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
contacts_count = CrmContact.query.filter_by(company_id=company_id).count()
companies_count = CrmCompany.query.filter_by(company_id=company_id).count()
deals_count = CrmDeal.query.filter_by(company_id=company_id).count()
pipeline_value = db.session.query(
db.func.sum(CrmDeal.amount)
).filter(CrmDeal.company_id == company_id).scalar() or 0.0
return jsonify({
"crm_summary": {
"contacts": contacts_count,
"companies": companies_count,
"deals": deals_count,
"pipeline_value": round(float(pipeline_value), 2),
}
})
# -- Revenue forecast -------------------------------------------------------
@analytics_bp.route("/api/analytics/forecast/revenue", methods=["GET"])
@require_auth_json()
def revenue_forecast():
"""GET /api/analytics/forecast/revenue
Monthly revenue forecast for the next 6 months based on open deals.
Distributes weighted deals into months by expected_close_date.
Returns:
- trailing_actuals: last 3 months of actual won-revenue (baseline)
- forecast_months: next 6 months with likely/best/conservative bands
- confidence_score: 0-100 based on data volume and deal coverage
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
now = datetime.now(timezone.utc)
today = now.date()
deals = CrmDeal.query.filter_by(company_id=company_id).all()
# Separate open vs won deals
closed_stages = {"closedwon", "closed_won", "won", "closedlost", "closed_lost", "lost"}
won_stages = {"closedwon", "closed_won", "won"}
open_deals = [d for d in deals if d.stage not in closed_stages]
won_deals = [d for d in deals if d.stage in won_stages]
# --- Trailing 3 months actuals ---
trailing_months: List[Dict[str, Any]] = []
for i in range(3, 0, -1):
month_start = today.replace(day=1) - timedelta(days=32 * i)
month_start = month_start.replace(day=1)
next_month = month_start + timedelta(days=32)
next_month = next_month.replace(day=1)
actual = 0.0
for d in won_deals:
close_date = d.expected_close_date
if close_date:
if close_date.tzinfo is None:
close_date = close_date.replace(tzinfo=timezone.utc)
if month_start <= close_date.date() < next_month:
actual += d.amount or 0.0
elif d.created_at:
if d.created_at.tzinfo is None:
d.created_at = d.created_at.replace(tzinfo=timezone.utc)
if month_start <= d.created_at.date() < next_month:
actual += d.amount or 0.0
trailing_months.append({
"month": month_start.strftime("%Y-%m"),
"label": month_start.strftime("%b %Y"),
"actual": round(actual, 2),
"likely": 0.0,
"best": 0.0,
"conservative": 0.0,
})
# --- Next 6 months forecast ---
forecast_months: List[Dict[str, Any]] = []
for i in range(6):
month_start = today.replace(day=1) + timedelta(days=32 * i)
month_start = month_start.replace(day=1)
next_month = month_start + timedelta(days=32)
next_month = next_month.replace(day=1)
likely_total = 0.0
best_total = 0.0
conservative_total = 0.0
for d in open_deals:
amount = d.amount or 0.0
probability = d.probability or 0.0
close_date = d.expected_close_date
if close_date:
if close_date.tzinfo is None:
close_date = close_date.replace(tzinfo=timezone.utc)
if month_start <= close_date.date() < next_month:
weighted = amount * probability
likely_total += weighted
best_total += amount
conservative_total += weighted * 0.5
forecast_months.append({
"month": month_start.strftime("%Y-%m"),
"label": month_start.strftime("%b %Y"),
"actual": 0.0,
"likely": round(likely_total, 2),
"best": round(best_total, 2),
"conservative": round(conservative_total, 2),
})
# --- Confidence score ---
# Based on: number of open deals with dates, data recency, deal coverage
scored = 0
if open_deals:
scored += 25 # Has open deals
dated = sum(1 for d in open_deals if d.expected_close_date)
if dated / max(len(open_deals), 1) > 0.5:
scored += 25 # Majority have close dates
if won_deals:
scored += 20 # Historical data
recent_wins = sum(1 for d in won_deals
if d.created_at
and (now - d.created_at).days < 180)
if recent_wins > 0:
scored += 15 # Recent win history
if len(open_deals) >= 5:
scored += 15 # Sufficient deal volume
# Average monthly revenue from actuals for context
avg_monthly_actual = (
sum(m["actual"] for m in trailing_months) / len(trailing_months)
if trailing_months and any(m["actual"] > 0 for m in trailing_months)
else 0.0
)
return jsonify({
"forecast": {
"trailing_actuals": trailing_months,
"forecast_months": forecast_months,
"confidence_score": scored,
"avg_monthly_actual": round(avg_monthly_actual, 2),
"total_pipeline_value": round(sum(d.amount or 0.0 for d in open_deals), 2),
"total_weighted_value": round(
sum((d.amount or 0.0) * (d.probability or 0.0) for d in open_deals), 2
),
"open_deals_count": len(open_deals),
}
})
# -- Pipeline health forecast ------------------------------------------------
@analytics_bp.route("/api/analytics/forecast/pipeline", methods=["GET"])
@require_auth_json()
def pipeline_forecast():
"""GET /api/analytics/forecast/pipeline
Pipeline health score and growth projection.
Returns:
- health_score: 0-100 composite score
- coverage_ratio: pipeline_value / avg_monthly_won_revenue
- growth_rate_pct: MoM new-deal growth
- at_risk_deals: past-due open deals
- stage_distribution: current deal counts per stage
- stage_conversion: estimated conversion between stages
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
now = datetime.now(timezone.utc)
today = now.date()
closed_stages = {"closedwon", "closed_won", "won", "closedlost", "closed_lost", "lost"}
won_stages = {"closedwon", "closed_won", "won"}
deals = CrmDeal.query.filter_by(company_id=company_id).all()
open_deals = [d for d in deals if d.stage not in closed_stages]
won_deals = [d for d in deals if d.stage in won_stages]
# --- Current pipeline metrics ---
total_value = sum(d.amount or 0.0 for d in open_deals)
weighted_value = sum((d.amount or 0.0) * (d.probability or 0.0) for d in open_deals)
# --- Monthly won revenue (avg) ---
won_last_6m = []
for d in won_deals:
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and (now - close).days <= 180:
won_last_6m.append(d)
avg_monthly_won = (
sum(d.amount or 0.0 for d in won_last_6m) / 6.0
if won_last_6m
else 0.0
)
coverage_ratio = total_value / avg_monthly_won if avg_monthly_won > 0 else 0.0
# --- Month-over-month growth ---
this_month_start = today.replace(day=1)
this_month_deals = sum(1 for d in open_deals
if d.created_at
and d.created_at.tzinfo is not None
and d.created_at.date() >= this_month_start)
last_month_start = this_month_start - timedelta(days=32)
last_month_start = last_month_start.replace(day=1)
last_month_deals = sum(1 for d in open_deals
if d.created_at
and d.created_at.tzinfo is not None
and last_month_start <= d.created_at.date() < this_month_start)
growth_rate = (
((this_month_deals - last_month_deals) / last_month_deals * 100)
if last_month_deals > 0 else 0.0
)
# --- At-risk deals (past expected_close_date and still open) ---
at_risk = []
for d in open_deals:
if d.expected_close_date:
close = d.expected_close_date
if close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close.date() < today:
at_risk.append({
"deal_name": d.name or "Unnamed deal",
"amount": round(d.amount or 0.0, 2),
"stage": d.stage or "unknown",
"expected_close_date": d.expected_close_date.strftime("%Y-%m-%d")
if d.expected_close_date else None,
"days_overdue": (today - close.date()).days,
})
at_risk.sort(key=lambda x: -x["days_overdue"])
# --- Stage distribution ---
stage_counts: Dict[str, int] = {}
stage_values: Dict[str, float] = {}
for d in open_deals:
s = d.stage or "unknown"
stage_counts[s] = stage_counts.get(s, 0) + 1
stage_values[s] = stage_values.get(s, 0.0) + (d.amount or 0.0)
stage_distribution = [
{"stage": k, "count": stage_counts[k], "value": round(stage_values[k], 2)}
for k in stage_counts
]
stage_distribution.sort(key=lambda x: -x["value"])
# --- Stage conversion estimates ---
# Approximate from stage distribution: if we have stages ordered,
# the ratio of count(stage N+1) / count(stage N) is a rough conversion
typical_stages = ["appointments_set", "qualification", "needs_analysis",
"value_proposition", "proposal_sent", "negotiation", "closedwon"]
stage_order_map = {s: i for i, s in enumerate(typical_stages)}
stage_conversion: List[Dict[str, Any]] = []
for stage_name, count in stage_counts.items():
if count == 0:
continue
order = stage_order_map.get(stage_name, -1)
# Find next stage in the typical flow
next_stage_name = None
if order >= 0 and order + 1 < len(typical_stages):
next_stage_name = typical_stages[order + 1]
next_count = stage_counts.get(next_stage_name, 0)
conversion = (next_count / count * 100) if count > 0 else 0.0
else:
conversion = 0.0
stage_conversion.append({
"stage": stage_name,
"count": count,
"next_stage": next_stage_name,
"conversion_pct": round(conversion, 1),
})
# --- Health score calculation ---
health = 0
# Pipeline coverage (0-30 points): 3x coverage is ideal
if coverage_ratio >= 3.0:
health += 30
elif coverage_ratio >= 2.0:
health += 24
elif coverage_ratio >= 1.0:
health += 18
elif coverage_ratio > 0:
health += 10
# Growth (0-20 points)
if growth_rate >= 10:
health += 20
elif growth_rate >= 0:
health += 15
elif growth_rate >= -10:
health += 10
else:
health += 5
# At-risk ratio (0-20 points)
at_risk_value = sum(d["amount"] for d in at_risk)
if total_value > 0:
at_risk_pct = at_risk_value / total_value
if at_risk_pct <= 0.05:
health += 20
elif at_risk_pct <= 0.15:
health += 15
elif at_risk_pct <= 0.30:
health += 10
else:
health += 5
else:
health += 10 # No pipeline = neutral
# Deal volume (0-15 points)
if len(open_deals) >= 10:
health += 15
elif len(open_deals) >= 5:
health += 10
elif len(open_deals) >= 1:
health += 5
# Won deal history (0-15 points)
if len(won_last_6m) >= 10:
health += 15
elif len(won_last_6m) >= 5:
health += 10
elif len(won_last_6m) >= 1:
health += 5
return jsonify({
"pipeline_health": {
"health_score": min(health, 100),
"total_pipeline_value": round(total_value, 2),
"weighted_pipeline_value": round(weighted_value, 2),
"open_deals_count": len(open_deals),
"coverage_ratio": round(coverage_ratio, 2),
"avg_monthly_won_revenue": round(avg_monthly_won, 2),
"growth_rate_pct": round(growth_rate, 1),
"this_month_new_deals": this_month_deals,
"last_month_new_deals": last_month_deals,
},
"at_risk_deals": at_risk[:20], # Cap at 20
"stage_distribution": stage_distribution,
"stage_conversion": stage_conversion,
})
# -- Key Performance Indicators ----------------------------------------------
@analytics_bp.route("/api/analytics/kpis", methods=["GET"])
@require_auth_json()
def kpis():
"""GET /api/analytics/kpis
Key performance indicators calculated from CRM data.
Query params:
period_days (int, default 90) — look-back period for calculations.
Returns:
All KPIs with current value, previous-period value, % change,
and trend direction (up/down/flat).
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
period_days = request.args.get("period_days", 90, type=int)
now = datetime.now(timezone.utc)
current_cutoff = now - timedelta(days=period_days)
# Previous period: same length before current
previous_cutoff = current_cutoff - timedelta(days=period_days)
deals = CrmDeal.query.filter_by(company_id=company_id).all()
contacts = CrmContact.query.filter_by(company_id=company_id).all()
companies = CrmCompany.query.filter_by(company_id=company_id).all()
closed_stages = {"closedwon", "closed_won", "won", "closedlost", "closed_lost", "lost"}
won_stages = {"closedwon", "closed_won", "won"}
def _in_current(deal):
"""Check if deal falls in the current period."""
dt = deal.created_at
if dt and dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt is not None and dt >= current_cutoff
def _in_previous(deal):
"""Check if deal falls in the previous period."""
dt = deal.created_at
if dt and dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt is not None and previous_cutoff <= dt < current_cutoff
def _contact_in_current(contact):
dt = contact.created_at
if dt and dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt is not None and dt >= current_cutoff
def _contact_in_previous(contact):
dt = contact.created_at
if dt and dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt is not None and previous_cutoff <= dt < current_cutoff
# --- Revenue metrics ---
current_won = [d for d in deals if d.stage in won_stages and _in_current(d)]
previous_won = [d for d in deals if d.stage in won_stages and _in_previous(d)]
current_revenue = sum(d.amount or 0.0 for d in current_won)
previous_revenue = sum(d.amount or 0.0 for d in previous_won)
# Avg monthly revenue
months_in_period = period_days / 30.0
avg_monthly_current = current_revenue / months_in_period if months_in_period > 0 else 0
avg_monthly_previous = previous_revenue / months_in_period if months_in_period > 0 else 0
# --- Deal metrics ---
current_closed = [d for d in deals if d.stage in closed_stages and _in_current(d)]
previous_closed = [d for d in deals if d.stage in closed_stages and _in_previous(d)]
current_win_rate = (len(current_won) / len(current_closed) * 100) if current_closed else 0
previous_win_rate = (len(previous_won) / len(previous_closed) * 100) if previous_closed else 0
current_avg_deal_size = (current_revenue / len(current_won)) if current_won else 0
previous_avg_deal_size = (previous_revenue / len(previous_won)) if previous_won else 0
open_deals = [d for d in deals if d.stage not in closed_stages]
# --- Efficiency metrics ---
# Avg days to close for won deals
current_durations = []
for d in current_won:
created = d.created_at
closed = d.expected_close_date or d.updated_at
if created and closed:
if created.tzinfo is None:
created = created.replace(tzinfo=timezone.utc)
if closed.tzinfo is None:
closed = closed.replace(tzinfo=timezone.utc)
days = (closed - created).days
if days > 0:
current_durations.append(days)
previous_durations = []
for d in previous_won:
created = d.created_at
closed = d.expected_close_date or d.updated_at
if created and closed:
if created.tzinfo is None:
created = created.replace(tzinfo=timezone.utc)
if closed.tzinfo is None:
closed = closed.replace(tzinfo=timezone.utc)
days = (closed - created).days
if days > 0:
previous_durations.append(days)
avg_days_current = sum(current_durations) / len(current_durations) if current_durations else 0
avg_days_previous = sum(previous_durations) / len(previous_durations) if previous_durations else 0
# Deals per month
current_deals_created = sum(1 for d in deals if _in_current(d))
previous_deals_created = sum(1 for d in deals if _in_previous(d))
deals_per_month_current = current_deals_created / months_in_period if months_in_period > 0 else 0
deals_per_month_previous = previous_deals_created / months_in_period if months_in_period > 0 else 0
# --- Customer metrics ---
total_contacts = len(contacts)
current_contacts = sum(1 for c in contacts if _contact_in_current(c))
previous_contacts = sum(1 for c in contacts if _contact_in_previous(c))
total_companies = len(companies)
# --- Helper for trend ---
def _pct_change(current_val: float, previous_val: float) -> float:
if previous_val == 0:
return 100.0 if current_val > 0 else 0.0
return ((current_val - previous_val) / abs(previous_val)) * 100
def _trend_direction(pct: float) -> str:
if abs(pct) < 1:
return "flat"
return "up" if pct > 0 else "down"
# Build KPI list
kpi_list: List[Dict[str, Any]] = [
{
"label": "Total Revenue",
"category": "revenue",
"value": round(current_revenue, 2),
"previous_value": round(previous_revenue, 2),
"unit": "currency",
"pct_change": round(_pct_change(current_revenue, previous_revenue), 1),
"trend": _trend_direction(_pct_change(current_revenue, previous_revenue)),
},
{
"label": "Avg Monthly Revenue",
"category": "revenue",
"value": round(avg_monthly_current, 2),
"previous_value": round(avg_monthly_previous, 2),
"unit": "currency",
"pct_change": round(_pct_change(avg_monthly_current, avg_monthly_previous), 1),
"trend": _trend_direction(_pct_change(avg_monthly_current, avg_monthly_previous)),
},
{
"label": "Win Rate",
"category": "deals",
"value": round(current_win_rate, 1),
"previous_value": round(previous_win_rate, 1),
"unit": "percent",
"pct_change": round(_pct_change(current_win_rate, previous_win_rate), 1),
"trend": _trend_direction(_pct_change(current_win_rate, previous_win_rate)),
},
{
"label": "Avg Deal Size",
"category": "deals",
"value": round(current_avg_deal_size, 2),
"previous_value": round(previous_avg_deal_size, 2),
"unit": "currency",
"pct_change": round(_pct_change(current_avg_deal_size, previous_avg_deal_size), 1),
"trend": _trend_direction(_pct_change(current_avg_deal_size, previous_avg_deal_size)),
},
{
"label": "Deals Won",
"category": "deals",
"value": len(current_won),
"previous_value": len(previous_won),
"unit": "count",
"pct_change": round(_pct_change(len(current_won), len(previous_won)), 1),
"trend": _trend_direction(_pct_change(len(current_won), len(previous_won))),
},
{
"label": "Deals in Pipeline",
"category": "deals",
"value": len(open_deals),
"previous_value": 0,
"unit": "count",
"pct_change": 0,
"trend": "flat",
},
{
"label": "Avg Days to Close",
"category": "efficiency",
"value": round(avg_days_current, 1),
"previous_value": round(avg_days_previous, 1),
"unit": "days",
"pct_change": round(_pct_change(avg_days_current, avg_days_previous), 1),
"trend": _trend_direction(_pct_change(avg_days_current, avg_days_previous)),
"invert_trend": True, # Lower is better for days to close
},
{
"label": "Deals Created/Month",
"category": "efficiency",
"value": round(deals_per_month_current, 1),
"previous_value": round(deals_per_month_previous, 1),
"unit": "count",
"pct_change": round(_pct_change(deals_per_month_current, deals_per_month_previous), 1),
"trend": _trend_direction(_pct_change(deals_per_month_current, deals_per_month_previous)),
},
{
"label": "Total Contacts",
"category": "customers",
"value": total_contacts,
"previous_value": total_contacts - previous_contacts if total_contacts > previous_contacts else 0,
"unit": "count",
"pct_change": round(_pct_change(current_contacts, previous_contacts), 1),
"trend": _trend_direction(_pct_change(current_contacts, previous_contacts)),
},
{
"label": "New Contacts",
"category": "customers",
"value": current_contacts,
"previous_value": previous_contacts,
"unit": "count",
"pct_change": round(_pct_change(current_contacts, previous_contacts), 1),
"trend": _trend_direction(_pct_change(current_contacts, previous_contacts)),
},
{
"label": "Unique Companies",
"category": "customers",
"value": total_companies,
"previous_value": total_companies,
"unit": "count",
"pct_change": 0,
"trend": "flat",
},
]
return jsonify({
"kpis": kpi_list,
"period_days": period_days,
"current_period": {
"start": current_cutoff.strftime("%Y-%m-%d"),
"end": now.strftime("%Y-%m-%d"),
},
"previous_period": {
"start": previous_cutoff.strftime("%Y-%m-%d"),
"end": current_cutoff.strftime("%Y-%m-%d"),
},
})
# -- Cascading Goals --------------------------------------------------------
def _goal_to_dict(goal: Goal, include_subgoals: bool = True) -> Dict[str, Any]:
"""Serialize a goal to dict with optional nested sub-goals."""
d: Dict[str, Any] = {
"id": goal.id,
"name": goal.name,
"description": goal.description or "",
"level": _normalise_level(goal.level),
"target_value": goal.target_value,
"current_value": goal.current_value,
"unit": goal.unit,
"status": goal.status,
"progress_percentage": round(goal.progress_percentage(), 1),
"parent_goal_id": goal.parent_goal_id,
"assigned_to": goal.assigned_to,
"weight": goal.weight,
"start_date": goal.start_date.strftime("%Y-%m-%dT%H:%M:%SZ") if goal.start_date else None,
"end_date": goal.end_date.strftime("%Y-%m-%dT%H:%M:%SZ") if goal.end_date else None,
"company_id": goal.company_id,
}
# Metrics
d["metrics"] = [
{
"id": m.id,
"metric_name": m.metric_name,
"target_value": m.target_value,
"actual_value": m.actual_value,
"unit": m.unit,
"period": m.period,
"progress_percentage": round(m.actual_value / m.target_value * 100, 1) if m.target_value and m.target_value != 0 else 0.0,
}
for m in goal.metrics
]
# Sub-goals
if include_subgoals:
d["sub_goals"] = [_goal_to_dict(g) for g in goal.sub_goals]
return d
def _calculate_cascade(goals_by_company: List[Goal]) -> Dict[str, Any]:
"""Calculate cascade summary from org → dept → team → rep.
Walks the parent_goal_id hierarchy to find descendants at each level.
This handles cases where sub-goals may have different company_ids than
their parent org goal.
"""
all_ids = {g.id for g in goals_by_company}
def _descendants(goal_id, level):
"""Find all goals at the given level that are descendants of goal_id.
Walks the entire subtree (all intermediate levels) then filters by level.
"""
result = []
current_parents = {goal_id}
visited = set()
while current_parents:
children = [
g for g in goals_by_company
if g.parent_goal_id in current_parents
and g.id not in visited
]
for c in children:
visited.add(c.id)
if _normalise_level(c.level) == level:
result.append(c)
current_parents = {g.id for g in children}
return result
org_goals = [g for g in goals_by_company if _normalise_level(g.level) == "org"]
# Collect all departments, teams, reps that are descendants of org goals
dept_goals: List[Goal] = []
team_goals: List[Goal] = []
rep_goals: List[Goal] = []
seen_ids = set()
for org in org_goals:
depts = _descendants(org.id, "department")
for d in depts:
if d.id not in seen_ids:
dept_goals.append(d)
seen_ids.add(d.id)
teams = _descendants(org.id, "team")
for t in teams:
if t.id not in seen_ids:
team_goals.append(t)
seen_ids.add(t.id)
reps = _descendants(org.id, "rep")
for r in reps:
if r.id not in seen_ids:
rep_goals.append(r)
seen_ids.add(r.id)
org_target = sum(g.target_value or 0 for g in org_goals)
org_current = sum(g.current_value or 0 for g in org_goals)
dept_targets = [
{"id": g.id, "name": g.name, "target_value": g.target_value, "current_value": g.current_value,
"progress_percentage": round(g.progress_percentage(), 1)}
for g in dept_goals
]
dept_total_target = sum(g.target_value or 0 for g in dept_goals)
dept_total_current = sum(g.current_value or 0 for g in dept_goals)
team_targets = [
{"id": g.id, "name": g.name, "target_value": g.target_value, "current_value": g.current_value,
"parent_goal_id": g.parent_goal_id, "progress_percentage": round(g.progress_percentage(), 1)}
for g in team_goals
]
team_total_target = sum(g.target_value or 0 for g in team_goals)
team_total_current = sum(g.current_value or 0 for g in team_goals)
rep_targets = [
{"id": g.id, "name": g.name, "target_value": g.target_value, "current_value": g.current_value,
"parent_goal_id": g.parent_goal_id, "progress_percentage": round(g.progress_percentage(), 1)}
for g in rep_goals
]
rep_total_target = sum(g.target_value or 0 for g in rep_goals)
rep_total_current = sum(g.current_value or 0 for g in rep_goals)
# Cascade variance: difference between org target and sum of all sub-goal targets
total_cascade = dept_total_target + team_total_target + rep_total_target
cascade_variance = round((total_cascade - org_target) / org_target * 100, 1) if org_target else 0.0
return {
"org_target": org_target,
"org_current": org_current,
"org_progress_percentage": round(org_current / org_target * 100, 1) if org_target else 0.0,
"dept_targets": dept_targets,
"dept_total_target": dept_total_target,
"dept_total_current": dept_total_current,
"team_targets": team_targets,
"team_total_target": team_total_target,
"team_total_current": team_total_current,
"rep_targets": rep_targets,
"rep_total_target": rep_total_target,
"rep_total_current": rep_total_current,
"total_cascade": total_cascade,
"cascade_variance": cascade_variance,
}
@analytics_bp.route("/api/analytics/goals", methods=["GET"])
@require_auth_json()
def get_goals():
"""GET /api/analytics/goals
Return all goals for ALL companies the user is linked to, with full
hierarchy and cascade summary.
Query params:
company_id — filter by specific company
level — filter by level (org, department, team, rep)
status — filter by status (active, completed, paused, cancelled)
parent_goal_id — filter by parent goal
start_date_gte — filter goals with start_date >= this value
end_date_lte — filter goals with end_date <= this value
include_cascade — "true" or "false" (default "true")
q — search by goal name (substring match, primary)
search_name — search by goal name (legacy alias for 'q')
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
query = Goal.query.filter(Goal.company_id.in_(company_ids))
# Apply filters
filter_company = request.args.get('company_id')
if filter_company and filter_company in company_ids:
query = query.filter(Goal.company_id == filter_company)
elif filter_company and filter_company not in company_ids:
return _json_error("Invalid company_id", 400)
filter_level = request.args.get('level')
if filter_level:
query = query.filter(Goal.level == filter_level)
filter_status = request.args.get('status')
if filter_status:
query = query.filter(Goal.status == filter_status)
filter_parent = request.args.get('parent_goal_id')
if filter_parent:
query = query.filter(Goal.parent_goal_id == filter_parent)
# Date range filters
start_date_gte = request.args.get('start_date_gte')
if start_date_gte:
try:
sd = datetime.fromisoformat(start_date_gte)
query = query.filter(Goal.start_date >= sd)
except (ValueError, TypeError):
pass
end_date_lte = request.args.get('end_date_lte')
if end_date_lte:
try:
ed = datetime.fromisoformat(end_date_lte)
query = query.filter(Goal.end_date <= ed)
except (ValueError, TypeError):
pass
# Name search — support both 'q' (primary) and 'search_name' (legacy)
q = request.args.get('q', '').strip()
if not q:
q = request.args.get('search_name', '').strip()
if q:
query = query.filter(Goal.name.ilike(f'%{q}%'))
goals = query.all()
# Build tree: if no specific filters, show org-level roots; otherwise show flat list
has_filters = bool(filter_level or filter_status or filter_parent or start_date_gte or end_date_lte or q)
if has_filters:
tree = [_goal_to_dict(g) for g in goals]
else:
org_goals = [g for g in goals if _normalise_level(g.level) == "org"]
tree = [_goal_to_dict(g) for g in org_goals]
# Cascade summary
include_cascade = request.args.get('include_cascade', 'true').lower() != 'false'
cascade = _calculate_cascade(goals) if include_cascade else None
return jsonify({
"goals": tree,
"cascade_summary": cascade,
"total_goals": len(goals),
})
@analytics_bp.route("/api/analytics/goals", methods=["POST"])
@require_auth_json()
@require_csrf
def create_goal():
"""POST /api/analytics/goals
Create a new goal.
Body: {
"name": str,
"description": str (optional),
"target_value": number,
"current_value": number (optional, defaults to 0),
"level": str ("org", "department", "team", "rep"),
"parent_goal_id": str (optional),
"company_id": str,
"status": str (optional, defaults to "active"),
"start_date": str ISO date (optional),
"end_date": str ISO date (optional),
"unit": str (optional, defaults to "$"),
}
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
data = request.get_json()
if not data or "name" not in data or "target_value" not in data:
return _json_error("name and target_value are required", 400)
company_id = data.get("company_id")
if not company_id or company_id not in company_ids:
return _json_error("Invalid company_id", 400)
parent_goal_id = data.get("parent_goal_id")
if parent_goal_id:
parent = db.session.get(Goal, parent_goal_id)
if not parent or parent.company_id not in company_ids:
return _json_error("Invalid parent_goal_id", 400)
goal = Goal(
company_id=company_id,
name=data["name"],
description=data.get("description", ""),
target_value=float(data["target_value"]),
current_value=float(data.get("current_value", 0)),
level=data.get("level", "org"),
parent_goal_id=parent_goal_id,
status=data.get("status", "active"),
unit=data.get("unit", "$"),
)
if data.get("start_date"):
goal.start_date = datetime.fromisoformat(data["start_date"])
if data.get("end_date"):
goal.end_date = datetime.fromisoformat(data["end_date"])
db.session.add(goal)
db.session.commit()
all_goals = Goal.query.filter(Goal.company_id.in_(company_ids)).all()
cascade = _calculate_cascade(all_goals)
return jsonify({
"goal": _goal_to_dict(goal),
"cascade_summary": cascade,
}), 201
@analytics_bp.route("/api/analytics/goals/<goal_id>", methods=["PUT"])
@require_auth_json()
@require_csrf
def update_goal(goal_id: str):
"""PUT /api/analytics/goals/<id>
Full goal update. Any subset of fields can be updated.
Body: {
"name": str (optional),
"description": str (optional),
"target_value": number (optional),
"current_value": number (optional),
"level": str (optional),
"status": str (optional),
"assigned_to": str (optional),
"start_date": str ISO date (optional),
"end_date": str ISO date (optional),
"unit": str (optional),
"weight": number (optional),
"parent_goal_id": str (optional),
}
Returns updated goal with cascade info.
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
data = request.get_json()
if not data:
return _json_error("Request body is required", 400)
# Validate parent_goal_id if provided
new_parent = data.get("parent_goal_id")
if new_parent is not None:
if new_parent:
parent = db.session.get(Goal, new_parent)
if not parent or parent.company_id not in company_ids:
return _json_error("Invalid parent_goal_id", 400)
# Prevent circular reference
if new_parent == goal_id:
return _json_error("Goal cannot be its own parent", 400)
# Prevent cycle: check that the parent is not a descendant of this goal
check_id = new_parent
while check_id:
if check_id == goal_id:
return _json_error("Circular parent reference detected", 400)
parent_goal = db.session.get(Goal, check_id)
check_id = parent_goal.parent_goal_id if parent_goal else None
# Update fields
updatable = ["name", "description", "target_value", "current_value", "level",
"status", "assigned_to", "unit", "weight", "parent_goal_id"]
for field in updatable:
if field in data:
setattr(goal, field, data[field])
# Handle dates
if "start_date" in data and data["start_date"]:
goal.start_date = datetime.fromisoformat(data["start_date"])
elif "start_date" in data and data["start_date"] is None:
goal.start_date = None
if "end_date" in data and data["end_date"]:
goal.end_date = datetime.fromisoformat(data["end_date"])
elif "end_date" in data and data["end_date"] is None:
goal.end_date = None
db.session.commit()
# Recalculate cascade
all_goals = Goal.query.filter(Goal.company_id.in_(company_ids)).all()
cascade = _calculate_cascade(all_goals)
return jsonify({
"goal": _goal_to_dict(goal),
"cascade_summary": cascade,
})
@analytics_bp.route("/api/analytics/goals/<goal_id>", methods=["DELETE"])
@require_auth_json()
@require_csrf
def delete_goal(goal_id: str):
"""DELETE /api/analytics/goals/<id>
Delete a goal and all its sub-goals.
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
# Delete children recursively
children = Goal.query.filter_by(parent_goal_id=goal.id).all()
for child in children:
db.session.delete(child)
# Also delete grandchildren
grandchildren = Goal.query.filter_by(parent_goal_id=child.id).all()
for gc in grandchildren:
db.session.delete(gc)
db.session.delete(goal)
db.session.commit()
all_goals = Goal.query.filter(Goal.company_id.in_(company_ids)).all()
cascade = _calculate_cascade(all_goals)
return jsonify({
"deleted": goal_id,
"cascade_summary": cascade,
})
@analytics_bp.route("/api/analytics/goals/<goal_id>/progress", methods=["PUT"])
@require_auth_json()
@require_csrf
def update_goal_progress(goal_id: str):
"""PUT /api/analytics/goals/<id>/progress
Update current_value on a goal. Optionally roll up to parent.
Body: { "current_value": float }
Returns updated goal with cascade info.
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
data = request.get_json()
if not data or "current_value" not in data:
return _json_error("current_value is required", 400)
new_value = data["current_value"]
if not isinstance(new_value, (int, float)):
return _json_error("current_value must be a number", 400)
goal.current_value = float(new_value)
# Optional roll-up: update parent's current_value based on weighted children
if data.get("roll_up", False) and goal.parent_goal_id:
parent = db.session.get(Goal, goal.parent_goal_id)
if parent:
siblings = Goal.query.filter_by(parent_goal_id=parent.id).all()
if siblings:
total_weight = sum(s.weight or 1.0 for s in siblings)
if total_weight > 0:
# Just sum weighted current values
parent.current_value = sum(
(s.current_value or 0) * (s.weight or 1.0) / total_weight
for s in siblings
)
db.session.commit()
# Recalculate cascade
all_goals = Goal.query.filter(Goal.company_id.in_(company_ids)).all()
cascade = _calculate_cascade(all_goals)
return jsonify({
"goal": _goal_to_dict(goal),
"cascade_summary": cascade,
})
# -- Goal Metrics CRUD ----------------------------------------------------
@analytics_bp.route("/api/analytics/goals/<goal_id>/metrics", methods=["GET"])
@require_auth_json()
def get_goal_metrics(goal_id: str):
"""GET /api/analytics/goals/<id>/metrics
List all metrics for a specific goal.
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
metrics = [
{
"id": m.id,
"goal_id": m.goal_id,
"metric_name": m.metric_name,
"target_value": m.target_value,
"actual_value": m.actual_value,
"unit": m.unit,
"period": m.period,
"period_start": m.period_start.strftime("%Y-%m-%dT%H:%M:%SZ") if m.period_start else None,
"period_end": m.period_end.strftime("%Y-%m-%dT%H:%M:%SZ") if m.period_end else None,
"progress_percentage": round(m.actual_value / m.target_value * 100, 1) if m.target_value and m.target_value != 0 else 0.0,
}
for m in goal.metrics
]
return jsonify({"metrics": metrics})
@analytics_bp.route("/api/analytics/goals/<goal_id>/metrics", methods=["POST"])
@require_auth_json()
@require_csrf
def create_goal_metric(goal_id: str):
"""POST /api/analytics/goals/<id>/metrics
Create a new metric for a goal.
Body: {
"metric_name": str,
"target_value": number,
"actual_value": number (optional, defaults to 0),
"unit": str (optional),
"period": str (optional, defaults to "monthly"),
"period_start": str ISO date (optional),
"period_end": str ISO date (optional),
}
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
data = request.get_json()
if not data or "metric_name" not in data or "target_value" not in data:
return _json_error("metric_name and target_value are required", 400)
metric = GoalMetric(
goal_id=goal_id,
metric_name=data["metric_name"],
target_value=float(data["target_value"]),
actual_value=float(data.get("actual_value", 0)),
unit=data.get("unit", goal.unit or ""),
period=data.get("period", "monthly"),
)
if data.get("period_start"):
metric.period_start = datetime.fromisoformat(data["period_start"])
if data.get("period_end"):
metric.period_end = datetime.fromisoformat(data["period_end"])
db.session.add(metric)
db.session.commit()
return jsonify({
"metric": {
"id": metric.id,
"goal_id": metric.goal_id,
"metric_name": metric.metric_name,
"target_value": metric.target_value,
"actual_value": metric.actual_value,
"unit": metric.unit,
"period": metric.period,
"period_start": metric.period_start.strftime("%Y-%m-%dT%H:%M:%SZ") if metric.period_start else None,
"period_end": metric.period_end.strftime("%Y-%m-%dT%H:%M:%SZ") if metric.period_end else None,
"progress_percentage": round(metric.actual_value / metric.target_value * 100, 1) if metric.target_value and metric.target_value != 0 else 0.0,
}
}), 201
@analytics_bp.route("/api/analytics/goals/<goal_id>/metrics/<metric_id>", methods=["PUT"])
@require_auth_json()
@require_csrf
def update_goal_metric(goal_id: str, metric_id: str):
"""PUT /api/analytics/goals/<id>/metrics/<metric_id>
Update a metric for a goal.
Body: any subset of { metric_name, target_value, actual_value, unit, period, period_start, period_end }
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
metric = GoalMetric.query.filter_by(id=metric_id, goal_id=goal_id).first()
if not metric:
return _json_error("Metric not found", 404)
data = request.get_json()
if not data:
return _json_error("Request body is required", 400)
updatable = ["metric_name", "target_value", "actual_value", "unit", "period"]
for field in updatable:
if field in data:
setattr(metric, field, data[field])
if "period_start" in data:
metric.period_start = datetime.fromisoformat(data["period_start"]) if data["period_start"] else None
if "period_end" in data:
metric.period_end = datetime.fromisoformat(data["period_end"]) if data["period_end"] else None
db.session.commit()
return jsonify({
"metric": {
"id": metric.id,
"goal_id": metric.goal_id,
"metric_name": metric.metric_name,
"target_value": metric.target_value,
"actual_value": metric.actual_value,
"unit": metric.unit,
"period": metric.period,
"period_start": metric.period_start.strftime("%Y-%m-%dT%H:%M:%SZ") if metric.period_start else None,
"period_end": metric.period_end.strftime("%Y-%m-%dT%H:%M:%SZ") if metric.period_end else None,
"progress_percentage": round(metric.actual_value / metric.target_value * 100, 1) if metric.target_value and metric.target_value != 0 else 0.0,
}
})
@analytics_bp.route("/api/analytics/goals/<goal_id>/metrics/<metric_id>", methods=["DELETE"])
@require_auth_json()
@require_csrf
def delete_goal_metric(goal_id: str, metric_id: str):
"""DELETE /api/analytics/goals/<id>/metrics/<metric_id>
Delete a metric for a goal.
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
goal = Goal.query.filter(Goal.id == goal_id, Goal.company_id.in_(company_ids)).first()
if not goal:
return _json_error("Goal not found", 404)
metric = GoalMetric.query.filter_by(id=metric_id, goal_id=goal_id).first()
if not metric:
return _json_error("Metric not found", 404)
db.session.delete(metric)
db.session.commit()
return jsonify({"deleted": metric_id})
# -- Users list for dropdowns -----------------------------------------------
@analytics_bp.route("/api/analytics/users", methods=["GET"])
@require_auth_json()
def get_analytics_users():
"""GET /api/analytics/users
Return list of users for the companies the current user has access to.
Used for assigned_to dropdown in goal forms.
"""
try:
company_ids = _get_user_company_ids()
except ValueError as exc:
return _json_error(str(exc), 400)
users = (
User.query
.join(UserCompany)
.filter(UserCompany.company_id.in_(company_ids))
.all()
)
items = []
for u in users:
uc = UserCompany.query.filter_by(user_id=u.id, company_id=company_ids[0]).first()
items.append({
"id": u.id,
"email": u.email,
"full_name": u.full_name or u.email.split("@")[0],
"role": u.role or "user",
"company_role": uc.role if uc else None,
})
return jsonify({"users": items})
# -- Strategic Intelligence: FP&A Waterfall + Root Cause --------------------
WON_STAGES = {"closedwon", "closed_won", "won"}
CLOSED_STAGES = {"closedwon", "closed_won", "won", "closedlost", "closed_lost", "lost"}
def _month_start(dt: datetime) -> datetime:
"""Return the first day of the month for the given datetime."""
return dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
def _month_key(dt: datetime) -> str:
"""Return YYYY-MM string."""
return dt.strftime("%Y-%m")
def _label_month(dt: datetime) -> str:
"""Return 'Mon YYYY' string."""
return dt.strftime("%b %Y")
def _monthly_won_revenue(deals: list, month_start: datetime, month_end: datetime) -> tuple:
"""Return (total_won_revenue, won_deals) for a given month window."""
won = 0.0
won_deal_list: list = []
for d in deals:
if d.stage not in WON_STAGES:
continue
close = d.expected_close_date or d.created_at
if close is None:
continue
if close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if month_start <= close < month_end:
won += d.amount or 0.0
won_deal_list.append(d)
return won, won_deal_list
def _budget_per_month(company_id: str) -> float:
"""Estimate monthly budget from company target_revenue or org-level goal."""
# Try company target_revenue
uc = UserCompany.query.filter_by(user_id=current_user.id).first()
if uc:
company = db.session.get(Company, uc.company_id)
if company and company.target_revenue and company.target_revenue > 0:
return company.target_revenue / 12.0
# Try org-level goal
goals = Goal.query.filter_by(company_id=company_id, level="org", status="active").all()
if goals:
total = sum(g.target_value or 0 for g in goals)
if total > 0:
return total / 12.0
return 0.0
@analytics_bp.route("/api/analytics/strategic-intelligence/overview", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='growth')
def strategic_intelligence_overview():
"""GET /api/analytics/strategic-intelligence/overview
Returns the high-level variance summary for the current month.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
now = datetime.now(timezone.utc)
today = now.date()
current_month_start = _month_start(now)
# End of current month
if today.month == 12:
current_month_end = current_month_start.replace(year=current_month_start.year + 1, month=1)
else:
current_month_end = current_month_start.replace(month=current_month_start.month + 1)
deals = CrmDeal.query.filter_by(company_id=company_id).all()
open_stages_complement = WON_STAGES | {"closedlost", "closed_lost", "lost"}
# Budget (monthly)
budget = _budget_per_month(company_id)
# Actual won revenue this month
actual, _ = _monthly_won_revenue(deals, current_month_start, current_month_end)
# Forecast (weighted pipeline value for open deals)
forecast = 0.0
for d in deals:
if d.stage not in open_stages_complement:
amount = d.amount or 0.0
probability = d.probability or 0.0
forecast += amount * probability
variance = actual - budget
variance_pct = (variance / budget * 100) if budget > 0 else 0.0
# Trend: last 6 months actual vs budget
trend: list = []
for i in range(6):
# Go back i months
month_idx = today.month - i
year = today.year
while month_idx <= 0:
month_idx += 12
year -= 1
ms = today.replace(year=year, month=month_idx, day=1)
ms_dt = datetime(ms.year, ms.month, ms.day, tzinfo=timezone.utc)
if ms.month == 12:
me_dt = ms_dt.replace(year=ms_dt.year + 1, month=1)
else:
me_dt = ms_dt.replace(month=ms_dt.month + 1)
m_actual, _ = _monthly_won_revenue(deals, ms_dt, me_dt)
trend.append({
"month": _month_key(ms_dt),
"label": _label_month(ms_dt),
"actual": round(m_actual, 2),
"budget": round(budget, 2),
})
trend.reverse() # Oldest first
return jsonify({
"period": current_month_start.strftime("%Y-%m"),
"period_label": current_month_start.strftime("%b %Y"),
"budget": round(budget, 2),
"actual": round(actual, 2),
"forecast": round(forecast, 2),
"variance": round(variance, 2),
"variance_pct": round(variance_pct, 1),
"trend": trend,
})
@analytics_bp.route("/api/analytics/strategic-intelligence/waterfall", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='growth')
def strategic_intelligence_waterfall():
"""GET /api/analytics/strategic-intelligence/waterfall
Returns waterfall chart data decomposing the gap between budget and actual.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
now = datetime.now(timezone.utc)
today = now.date()
current_month_start = _month_start(now)
if today.month == 12:
current_month_end = current_month_start.replace(year=current_month_start.year + 1, month=1)
else:
current_month_end = current_month_start.replace(month=current_month_start.month + 1)
# Previous month
prev_month_start = current_month_start
if prev_month_start.month == 1:
prev_month_start = prev_month_start.replace(month=12, year=prev_month_start.year - 1)
else:
prev_month_start = prev_month_start.replace(month=prev_month_start.month - 1)
if prev_month_start.month == 12:
prev_month_end = prev_month_start.replace(year=prev_month_start.year + 1, month=1)
else:
prev_month_end = prev_month_start.replace(month=prev_month_start.month + 1)
deals = CrmDeal.query.filter_by(company_id=company_id).all()
budget = _budget_per_month(company_id)
# Actual won this month
actual, current_won = _monthly_won_revenue(deals, current_month_start, current_month_end)
# Previous month won
prev_actual, prev_won = _monthly_won_revenue(deals, prev_month_start, prev_month_end)
# --- Waterfall components ---
# 1. Pipeline contribution: deals that were already open at start of month and closed
pipeline_contribution = 0.0
for d in current_won:
created = d.created_at
if created and created.tzinfo is None:
created = created.replace(tzinfo=timezone.utc)
if created and created < current_month_start:
pipeline_contribution += d.amount or 0.0
# 2. New deals: deals created this month that closed
new_deals_value = 0.0
for d in current_won:
created = d.created_at
if created and created.tzinfo is None:
created = created.replace(tzinfo=timezone.utc)
if created and current_month_start <= created < current_month_end:
new_deals_value += d.amount or 0.0
# 3. Deal size variance: difference in avg deal size vs previous period
avg_this = (actual / len(current_won)) if current_won else 0.0
avg_prev = (prev_actual / len(prev_won)) if prev_won else 0.0
deal_size_variance = (avg_this - avg_prev) * len(current_won) if current_won else 0.0
# 4. Conversion rate variance: impact of win rate change
# Current month win rate
current_closed = [d for d in deals if d.stage in CLOSED_STAGES]
current_closed_this_month = []
for d in current_closed:
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and current_month_start <= close < current_month_end:
current_closed_this_month.append(d)
prev_closed_this_month = []
for d in current_closed:
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and prev_month_start <= close < prev_month_end:
prev_closed_this_month.append(d)
current_win_rate = len(current_won) / len(current_closed_this_month) if current_closed_this_month else 0.0
prev_win_rate = len(prev_won) / len(prev_closed_this_month) if prev_closed_this_month else 0.0
# Impact of win rate change: (current_rate - prev_rate) * total pipeline available
open_pipeline = sum(d.amount or 0.0 for d in deals if d.stage not in CLOSED_STAGES)
conversion_variance = (current_win_rate - prev_win_rate) * open_pipeline
# 5. Churn / cancellations: lost deals value this month
lost_this_month = 0.0
lost_stages = {"closedlost", "closed_lost", "lost"}
for d in deals:
if d.stage not in lost_stages:
continue
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and current_month_start <= close < current_month_end:
lost_this_month += d.amount or 0.0
# 6. Market mix: breakdown by source/channel
source_breakdown: dict = {}
for d in current_won:
props = d.properties_json or {}
source = props.get("source", "unknown")
if isinstance(source, str):
source = source.title()
else:
source = "Unknown"
source_breakdown[source] = source_breakdown.get(source, 0.0) + (d.amount or 0.0)
# Build waterfall steps
steps: list = [
{"label": "Budget", "value": round(budget, 2), "cumulative": round(budget, 2), "color": "#00A846"},
]
# Calculate total actual from components to see the gap
cumulative = budget
factors = [
{"label": "Pipeline Revenue", "value": round(pipeline_contribution, 2)},
{"label": "New Deals Revenue", "value": round(new_deals_value, 2)},
]
# Add variance factors
if deal_size_variance != 0:
if deal_size_variance > 0:
factors.append({"label": "Deal Size Variance", "value": round(deal_size_variance, 2)})
else:
factors.append({"label": "Deal Size Variance", "value": round(deal_size_variance, 2)})
if conversion_variance != 0:
if conversion_variance > 0:
factors.append({"label": "Conversion Rate Impact", "value": round(conversion_variance, 2)})
else:
factors.append({"label": "Conversion Rate Impact", "value": round(conversion_variance, 2)})
if lost_this_month > 0:
factors.append({"label": "Lost Deals", "value": round(-lost_this_month, 2)})
for f in factors:
cumulative += f["value"]
color = "#22C55E" if f["value"] >= 0 else "#EF4444"
steps.append({
"label": f["label"],
"value": f["value"],
"cumulative": round(cumulative, 2),
"color": color,
})
# Final step: Actual
steps.append({"label": "Actual", "value": round(actual, 2), "cumulative": round(actual, 2), "color": "#00A846"})
return jsonify({
"waterfall": steps,
"budget": round(budget, 2),
"actual": round(actual, 2),
"components": {
"pipeline_contribution": round(pipeline_contribution, 2),
"new_deals": round(new_deals_value, 2),
"deal_size_variance": round(deal_size_variance, 2),
"conversion_rate_variance": round(conversion_variance, 2),
"churn_cancellations": round(-lost_this_month, 2),
},
"market_mix": [{
"source": k, "value": round(v, 2)
} for k, v in sorted(source_breakdown.items(), key=lambda x: -x[1])],
})
@analytics_bp.route("/api/analytics/strategic-intelligence/root-cause", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='growth')
def strategic_intelligence_root_cause():
"""GET /api/analytics/strategic-intelligence/root-cause
Returns a drill-down tree structure explaining the revenue variance.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
now = datetime.now(timezone.utc)
today = now.date()
current_month_start = _month_start(now)
if today.month == 12:
current_month_end = current_month_start.replace(year=current_month_start.year + 1, month=1)
else:
current_month_end = current_month_start.replace(month=current_month_start.month + 1)
# Previous month
prev_month_start = current_month_start
if prev_month_start.month == 1:
prev_month_start = prev_month_start.replace(month=12, year=prev_month_start.year - 1)
else:
prev_month_start = prev_month_start.replace(month=prev_month_start.month - 1)
if prev_month_start.month == 12:
prev_month_end = prev_month_start.replace(year=prev_month_start.year + 1, month=1)
else:
prev_month_end = prev_month_start.replace(month=prev_month_start.month + 1)
deals = CrmDeal.query.filter_by(company_id=company_id).all()
budget = _budget_per_month(company_id)
# Current and previous month won deals
actual, current_won = _monthly_won_revenue(deals, current_month_start, current_month_end)
prev_actual, prev_won = _monthly_won_revenue(deals, prev_month_start, prev_month_end)
variance = actual - budget
# Current closed deals this month
current_closed = []
for d in deals:
if d.stage not in CLOSED_STAGES:
continue
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and current_month_start <= close < current_month_end:
current_closed.append(d)
prev_closed = []
for d in deals:
if d.stage not in CLOSED_STAGES:
continue
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and prev_month_start <= close < prev_month_end:
prev_closed.append(d)
current_win_rate = len(current_won) / len(current_closed) if current_closed else 0.0
prev_win_rate = len(prev_won) / len(prev_closed) if prev_closed else 0.0
avg_this = (actual / len(current_won)) if current_won else 0.0
avg_prev = (prev_actual / len(prev_won)) if prev_won else 0.0
open_pipeline = sum(d.amount or 0.0 for d in deals if d.stage not in CLOSED_STAGES)
# Build root cause tree
children: list = []
# 1. Win Rate factor
win_rate_impact = (current_win_rate - prev_win_rate) * open_pipeline
if abs(win_rate_impact) > 0.01:
# Break down by stage for current won deals
stage_impact: dict = {}
for d in current_won:
stage = d.stage or "unknown"
stage_impact[stage] = stage_impact.get(stage, 0.0) + (d.amount or 0.0)
# Also look at lost deals by stage
lost_stages = {"closedlost", "closed_lost", "lost"}
stage_lost_impact: dict = {}
for d in current_closed:
if d.stage not in lost_stages:
continue
stage = d.stage or "unknown"
stage_lost_impact[stage] = stage_lost_impact.get(stage, 0.0) + (d.amount or 0.0)
stage_children: list = []
# Show stages where there were losses
for stage, val in sorted(stage_lost_impact.items(), key=lambda x: -x[1]):
stage_children.append({
"label": f"Stage: {stage}",
"value": round(-val, 2),
"children": [],
})
# If no lost stages, show won stages
if not stage_children:
for stage, val in sorted(stage_impact.items(), key=lambda x: -x[1]):
stage_children.append({
"label": f"Stage: {stage}",
"value": round(val, 2),
"children": [],
})
impact_deals = len([d for d in current_closed if d.stage in lost_stages])
children.append({
"label": "Win Rate Change" if win_rate_impact >= 0 else "Win Rate Decline",
"value": round(win_rate_impact, 2),
"details": {
"current_rate": round(current_win_rate, 2),
"previous_rate": round(prev_win_rate, 2),
"impact_deals": impact_deals,
},
"children": stage_children,
})
# 2. Deal Size Variance
if current_won:
deal_size_impact = (avg_this - avg_prev) * len(current_won)
if abs(deal_size_impact) > 0.01:
children.append({
"label": "Deal Size Variance",
"value": round(deal_size_impact, 2),
"details": {
"avg_this_period": round(avg_this, 2),
"avg_last_period": round(avg_prev, 2),
"impact_deals": len(current_won),
},
"children": [],
})
# 3. Volume factor (number of deals)
volume_diff = len(current_won) - len(prev_won)
if abs(volume_diff) > 0 and avg_prev > 0:
volume_impact = volume_diff * avg_prev
children.append({
"label": "Volume Change" if volume_impact >= 0 else "Volume Decline",
"value": round(volume_impact, 2),
"details": {
"deals_this_period": len(current_won),
"deals_last_period": len(prev_won),
"change": volume_diff,
},
"children": [],
})
# 4. Lost deals (churn)
lost_value = sum(d.amount or 0.0 for d in current_closed if d.stage in lost_stages)
if lost_value > 0:
lost_by_stage: dict = {}
for d in current_closed:
if d.stage not in lost_stages:
continue
stage = d.stage or "unknown"
lost_by_stage[stage] = lost_by_stage.get(stage, 0.0) + (d.amount or 0.0)
lost_stage_children = [
{"label": f"Stage: {stage}", "value": round(-val, 2), "children": []}
for stage, val in sorted(lost_by_stage.items(), key=lambda x: -x[1])
]
children.append({
"label": "Lost Deals / Churn",
"value": round(-lost_value, 2),
"details": {
"lost_deals_count": len([d for d in current_closed if d.stage in lost_stages]),
"total_lost_value": round(lost_value, 2),
},
"children": lost_stage_children,
})
# 5. Market mix
source_impact: dict = {}
for d in current_won:
props = d.properties_json or {}
source = props.get("source", "unknown")
if isinstance(source, str):
source = source.title()
else:
source = "Unknown"
source_impact[source] = source_impact.get(source, 0.0) + (d.amount or 0.0)
if source_impact:
mix_children = [
{"label": source, "value": round(val, 2), "children": []}
for source, val in sorted(source_impact.items(), key=lambda x: -x[1])
]
children.append({
"label": "Market Mix",
"value": round(actual, 2),
"details": {
"sources": len(source_impact),
"top_source": max(source_impact, key=source_impact.get) if source_impact else "N/A",
},
"children": mix_children,
})
return jsonify({
"root": {
"label": "Revenue Variance",
"value": round(variance, 2),
"budget": round(budget, 2),
"actual": round(actual, 2),
"children": children,
}
})
@analytics_bp.route("/api/analytics/strategic-intelligence/contribution", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='growth')
def strategic_intelligence_contribution():
"""GET /api/analytics/strategic-intelligence/contribution
Returns factor decomposition — what drove the variance.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
now = datetime.now(timezone.utc)
today = now.date()
current_month_start = _month_start(now)
if today.month == 12:
current_month_end = current_month_start.replace(year=current_month_start.year + 1, month=1)
else:
current_month_end = current_month_start.replace(month=current_month_start.month + 1)
# Previous month
prev_month_start = current_month_start
if prev_month_start.month == 1:
prev_month_start = prev_month_start.replace(month=12, year=prev_month_start.year - 1)
else:
prev_month_start = prev_month_start.replace(month=prev_month_start.month - 1)
if prev_month_start.month == 12:
prev_month_end = prev_month_start.replace(year=prev_month_start.year + 1, month=1)
else:
prev_month_end = prev_month_start.replace(month=prev_month_start.month + 1)
deals = CrmDeal.query.filter_by(company_id=company_id).all()
budget = _budget_per_month(company_id)
# Current and previous month data
actual, current_won = _monthly_won_revenue(deals, current_month_start, current_month_end)
prev_actual, prev_won = _monthly_won_revenue(deals, prev_month_start, prev_month_end)
# Current closed deals this month
current_closed = []
for d in deals:
if d.stage not in CLOSED_STAGES:
continue
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and current_month_start <= close < current_month_end:
current_closed.append(d)
prev_closed = []
for d in deals:
if d.stage not in CLOSED_STAGES:
continue
close = d.expected_close_date or d.created_at
if close and close.tzinfo is None:
close = close.replace(tzinfo=timezone.utc)
if close and prev_month_start <= close < prev_month_end:
prev_closed.append(d)
# Metrics
current_count = len(current_won)
prev_count = len(prev_won)
avg_this = (actual / current_count) if current_count > 0 else 0.0
avg_prev = (prev_actual / prev_count) if prev_count > 0 else 0.0
current_win_rate = current_count / len(current_closed) if current_closed else 0.0
prev_win_rate = prev_count / len(prev_closed) if prev_closed else 0.0
open_pipeline = sum(d.amount or 0.0 for d in deals if d.stage not in CLOSED_STAGES)
# --- Volume effect: more/fewer deals ---
volume_effect = (current_count - prev_count) * avg_prev if avg_prev > 0 else 0.0
volume_direction = "positive" if volume_effect > 0 else ("negative" if volume_effect < 0 else "neutral")
volume_desc = (
f"{current_count} deals won vs {prev_count} last month"
f" ({'+' if current_count >= prev_count else ''}{current_count - prev_count} deals)"
)
volume_recommendation = (
"Deal volume is increasing — focus on maintaining pipeline intake."
if volume_effect > 0
else "Deal volume declined — review lead generation and pipeline coverage."
)
# --- Price effect: average deal size change ---
price_effect = (avg_this - avg_prev) * current_count if current_count > 0 else 0.0
price_direction = "positive" if price_effect > 0 else ("negative" if price_effect < 0 else "neutral")
price_desc = (
f"Avg deal size ${avg_this:,.0f} vs ${avg_prev:,.0f} last month"
)
price_recommendation = (
"Higher deal sizes — consider upselling strategies."
if price_effect > 0
else "Deal sizes shrinking — review pricing and package offerings."
)
# --- Rate effect: win rate change ---
rate_effect = (current_win_rate - prev_win_rate) * open_pipeline
rate_direction = "positive" if rate_effect > 0 else ("negative" if rate_effect < 0 else "neutral")
rate_desc = (
f"Win rate {current_win_rate:.0%} vs {prev_win_rate:.0%} last month"
)
rate_recommendation = (
"Win rate improved — document what's working and scale."
if rate_effect > 0
else "Win rate dropped — analyze lost deal reasons and qualification criteria."
)
# --- Mix effect: source/channel contribution change ---
current_source_rev: dict = {}
for d in current_won:
props = d.properties_json or {}
source = (props.get("source", "unknown") or "unknown")
if isinstance(source, str):
source = source.title()
else:
source = "Unknown"
current_source_rev[source] = current_source_rev.get(source, 0.0) + (d.amount or 0.0)
prev_source_rev: dict = {}
for d in prev_won:
props = d.properties_json or {}
source = (props.get("source", "unknown") or "unknown")
if isinstance(source, str):
source = source.title()
else:
source = "Unknown"
prev_source_rev[source] = prev_source_rev.get(source, 0.0) + (d.amount or 0.0)
# Mix effect: sum of changes in source contribution
all_sources = set(list(current_source_rev.keys()) + list(prev_source_rev.keys()))
mix_effect = sum(
(current_source_rev.get(s, 0) - prev_source_rev.get(s, 0))
for s in all_sources
)
# The mix effect is just the revenue difference not explained by volume/price/rate
mix_effect = actual - prev_actual - volume_effect - price_effect
mix_direction = "positive" if mix_effect > 0 else ("negative" if mix_effect < 0 else "neutral")
mix_desc = f"Source mix shifted by ${mix_effect:,.0f}"
mix_recommendation = (
"Channel mix favorable — double down on top-performing sources."
if mix_effect > 0
else "Channel mix shifted unfavorably — rebalance marketing across channels."
)
source_details = []
for source in all_sources:
curr_val = current_source_rev.get(source, 0.0)
prev_val = prev_source_rev.get(source, 0.0)
source_details.append({
"source": source,
"current": round(curr_val, 2),
"previous": round(prev_val, 2),
"change": round(curr_val - prev_val, 2),
})
source_details.sort(key=lambda x: -abs(x["change"]))
return jsonify({
"total_variance": round(actual - budget, 2),
"period": current_month_start.strftime("%Y-%m"),
"factors": [
{
"name": "volume_effect",
"label": "Volume Effect",
"value": round(volume_effect, 2),
"direction": volume_direction,
"description": volume_desc,
"recommendation": volume_recommendation,
"percentage": round(volume_effect / actual * 100, 1) if actual > 0 else 0.0,
},
{
"name": "price_effect",
"label": "Price Effect",
"value": round(price_effect, 2),
"direction": price_direction,
"description": price_desc,
"recommendation": price_recommendation,
"percentage": round(price_effect / actual * 100, 1) if actual > 0 else 0.0,
},
{
"name": "rate_effect",
"label": "Rate Effect",
"value": round(rate_effect, 2),
"direction": rate_direction,
"description": rate_desc,
"recommendation": rate_recommendation,
"percentage": round(rate_effect / actual * 100, 1) if actual > 0 else 0.0,
},
{
"name": "mix_effect",
"label": "Mix Effect",
"value": round(mix_effect, 2),
"direction": mix_direction,
"description": mix_desc,
"recommendation": mix_recommendation,
"percentage": round(mix_effect / actual * 100, 1) if actual > 0 else 0.0,
},
],
"source_breakdown": source_details,
})
# -- Scale Optimization routes ----------------------------------------------
@analytics_bp.route("/api/analytics/scale-optimization/overview", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='command')
def scale_optimization_overview():
"""GET /api/analytics/scale-optimization/overview
Returns summary metrics: total ad spend (30d/90d), conversions,
attributed revenue, overall ROAS, avg LTV:CAC, active campaigns by channel.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 403)
now = datetime.now(timezone.utc)
days_30 = now - timedelta(days=30)
days_90 = now - timedelta(days=90)
# AdMetric aggregates
metrics_30 = db.session.query(
db.func.sum(AdMetric.spend).label("total_spend"),
db.func.sum(AdMetric.impressions).label("total_impressions"),
db.func.sum(AdMetric.clicks).label("total_clicks"),
db.func.sum(AdMetric.conversions).label("total_conversions"),
db.func.avg(AdMetric.roas).label("avg_roas"),
).filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= days_30,
).first()
metrics_90 = db.session.query(
db.func.sum(AdMetric.spend).label("total_spend"),
db.func.sum(AdMetric.conversions).label("total_conversions"),
).filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= days_90,
).first()
total_spend_30d = float(metrics_30.total_spend or 0)
total_spend_90d = float(metrics_90.total_spend or 0)
total_conversions = float(metrics_30.total_conversions or 0)
# Attributed revenue from won deals in the same period
won_deals = db.session.query(db.func.sum(CrmDeal.amount)).filter(
CrmDeal.company_id == company_id,
CrmDeal.status.in_(['closed_won', 'Closed Won']),
CrmDeal.created_at >= days_90,
).scalar()
attributed_revenue = float(won_deals or 0)
# Overall ROAS
overall_roas = round(attributed_revenue / total_spend_90d, 2) if total_spend_90d > 0 else 0.0
# LTV:CAC
avg_deal_amount = 0.0
if total_conversions > 0:
avg_deal_amount = attributed_revenue / max(total_conversions, 1)
ltv = avg_deal_amount * 12 # default 12 month lifetime
cac = total_spend_90d / max(total_conversions, 1) if total_conversions > 0 else 0
ltv_cac = round(ltv / cac, 2) if cac > 0 else 0.0
# Active campaigns by channel
channel_campaigns = db.session.query(
AdCampaign.source_service,
db.func.count(AdCampaign.id).label("count"),
).filter(
AdCampaign.company_id == company_id,
AdCampaign.status == 'active',
).group_by(AdCampaign.source_service).all()
active_campaigns = {row.source_service or 'other': row.count for row in channel_campaigns}
total_active_campaigns = sum(active_campaigns.values())
return jsonify({
"total_spend_30d": round(total_spend_30d, 2),
"total_spend_90d": round(total_spend_90d, 2),
"total_conversions": total_conversions,
"attributed_revenue": round(attributed_revenue, 2),
"overall_roas": overall_roas,
"avg_ltv_cac": ltv_cac,
"total_active_campaigns": total_active_campaigns,
"campaigns_by_channel": active_campaigns,
"period": days_30.strftime("%Y-%m-%d") + " to " + now.strftime("%Y-%m-%d"),
})
@analytics_bp.route("/api/analytics/scale-optimization/run-rate", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='command')
def scale_optimization_run_rate():
"""GET /api/analytics/scale-optimization/run-rate
Returns YTD revenue, annual target, run rate, pace status,
required monthly pace, and remaining needed from Company.target_revenue.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 403)
# Import here to avoid circular imports
from ..services.forecast_service import calculate_run_rate
result = calculate_run_rate(company_id)
return jsonify({
"ytd_revenue": round(result.get("ytd_revenue", 0), 2),
"annual_target": result.get("annual_target"),
"run_rate": result.get("run_rate"),
"months_elapsed": result.get("months_elapsed", 0),
"months_remaining": result.get("months_remaining", 0),
"pace": result.get("pace"),
"required_monthly": result.get("required_monthly"),
"remaining_needed": result.get("remaining_needed"),
})
@analytics_bp.route("/api/analytics/scale-optimization/channels", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='command')
def scale_optimization_channels():
"""GET /api/analytics/scale-optimization/channels
Returns per-channel breakdown: spend, impressions, clicks, conversions,
CTR, CPC, attributed revenue, ROAS, LTV:CAC, trend.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 403)
now = datetime.now(timezone.utc)
days_30 = now - timedelta(days=30)
days_60 = now - timedelta(days=60)
# Current period (last 30 days)
current_rows = db.session.query(
AdMetric.source_service,
db.func.sum(AdMetric.spend).label("total_spend"),
db.func.sum(AdMetric.impressions).label("total_impressions"),
db.func.sum(AdMetric.clicks).label("total_clicks"),
db.func.sum(AdMetric.conversions).label("total_conversions"),
db.func.avg(AdMetric.ctr).label("avg_ctr"),
db.func.avg(AdMetric.cpc).label("avg_cpc"),
db.func.avg(AdMetric.roas).label("avg_roas"),
).filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= days_30,
).group_by(AdMetric.source_service).all()
# Previous period (31-60 days ago) for trend
prev_rows = db.session.query(
AdMetric.source_service,
db.func.avg(AdMetric.roas).label("avg_roas"),
).filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= days_60,
AdMetric.metric_date < days_30,
).group_by(AdMetric.source_service).all()
prev_roas_map = {row.source_service: float(row.avg_roas or 0) for row in prev_rows}
# Revenue by channel from won deals (use source field on CrmDeal)
won_deals_by_source = db.session.query(
CrmDeal.source,
db.func.sum(CrmDeal.amount).label("total_revenue"),
db.func.count(CrmDeal.id).label("deal_count"),
).filter(
CrmDeal.company_id == company_id,
CrmDeal.status.in_(['closed_won', 'Closed Won']),
CrmDeal.created_at >= days_60,
).group_by(CrmDeal.source).all()
revenue_by_source = {row.source or 'other': float(row.total_revenue or 0) for row in won_deals_by_source}
deals_by_source = {row.source or 'other': row.deal_count for row in won_deals_by_source}
# Channel name mapping
channel_names = {
'google_ads': 'Google Ads',
'facebook_ads': 'Facebook Ads',
'google': 'Google Ads',
'facebook': 'Facebook Ads',
}
# Build channel performance list
channels = []
for row in current_rows:
source = row.source_service or 'other'
spend = float(row.total_spend or 0)
impressions = int(row.total_impressions or 0)
clicks = float(row.total_clicks or 0)
conversions = float(row.total_conversions or 0)
ctr = float(row.avg_ctr or 0)
cpc = float(row.avg_cpc or 0)
roas = float(row.avg_roas or 0)
# Map ad source to deal source for revenue attribution
deal_source = source
for key, val in channel_names.items():
if key in source.lower():
# find matching deal source
for ds in revenue_by_source:
if val.lower() in ds.lower():
deal_source = ds
break
revenue = revenue_by_source.get(deal_source, revenue_by_source.get(source, 0))
deal_count = deals_by_source.get(deal_source, deals_by_source.get(source, 0))
# LTV:CAC per channel
cac = spend / max(conversions, 1) if conversions > 0 else 0
avg_deal = revenue / max(deal_count, 1) if deal_count > 0 else 0
ltv = avg_deal * 12
ltv_cac = round(ltv / cac, 2) if cac > 0 else 0.0
# Trend
prev_channel_roas = prev_roas_map.get(source, 0)
if roas > 0 and prev_channel_roas > 0:
trend = "up" if roas > prev_channel_roas * 1.05 else ("down" if roas < prev_channel_roas * 0.95 else "flat")
elif roas > 0:
trend = "up"
else:
trend = "flat"
channels.append({
"channel": channel_names.get(source, source.replace('_', ' ').title()),
"source_service": source,
"spend": round(spend, 2),
"impressions": impressions,
"clicks": round(clicks, 0),
"conversions": round(conversions, 0),
"ctr": round(ctr, 4),
"cpc": round(cpc, 2),
"attributed_revenue": round(revenue, 2),
"roas": round(roas, 2),
"ltv_cac": ltv_cac,
"cac": round(cac, 2),
"trend": trend,
})
channels.sort(key=lambda c: c.get('roas', 0), reverse=True)
return jsonify({"channels": channels})
@analytics_bp.route("/api/analytics/scale-optimization/matrix", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='command')
def scale_optimization_matrix():
"""GET /api/analytics/scale-optimization/matrix
Returns product × channel × market data with spend, revenue, ROAS per cell.
Uses metadata_json from CrmDeal for product/market categorization.
Falls back to deal source field if no product/market metadata.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 403)
now = datetime.now(timezone.utc)
days_90 = now - timedelta(days=90)
# Get all won deals with metadata
won_deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.status.in_(['closed_won', 'Closed Won']),
CrmDeal.created_at >= days_90,
).all()
# Extract product and market from metadata
product_revenue: Dict[str, float] = {}
market_revenue: Dict[str, float] = {}
product_market_revenue: Dict[str, Dict[str, float]] = {}
for deal in won_deals:
metadata = deal.metadata_json or {}
product = metadata.get('product') or metadata.get('product_name') or 'General'
market = metadata.get('market') or metadata.get('market_segment') or deal.source or 'General'
amount = float(deal.amount or 0)
product_revenue[product] = product_revenue.get(product, 0) + amount
market_revenue[market] = market_revenue.get(market, 0) + amount
if product not in product_market_revenue:
product_market_revenue[product] = {}
product_market_revenue[product][market] = product_market_revenue[product].get(market, 0) + amount
# Get spend by channel
channel_spend = db.session.query(
AdMetric.source_service,
db.func.sum(AdMetric.spend).label("total_spend"),
db.func.sum(AdMetric.conversions).label("total_conversions"),
db.func.avg(AdMetric.roas).label("avg_roas"),
).filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= days_90,
).group_by(AdMetric.source_service).all()
channel_data = {}
for row in channel_spend:
source = row.source_service or 'other'
channel_data[source] = {
"spend": float(row.total_spend or 0),
"conversions": float(row.total_conversions or 0),
"avg_roas": float(row.avg_roas or 0),
}
# Build matrix: rows = products/markets, columns = channels
products = sorted(product_revenue.keys())
markets = sorted(market_revenue.keys())
channels_list = sorted(channel_data.keys())
# Build cells
rows = []
for product in products:
row: Dict[str, Any] = {"dimension": product, "type": "product"}
for channel in channels_list:
cdata = channel_data.get(channel, {})
spend = cdata.get("spend", 0)
# Distribute revenue proportionally across channels
total_channel_spend = sum(cd.get("spend", 0) for cd in channel_data.values())
revenue_share = spend / total_channel_spend if total_channel_spend > 0 else 0
revenue = product_revenue.get(product, 0) * revenue_share
cell_roas = round(revenue / spend, 2) if spend > 0 else 0.0
row[channel] = {
"spend": round(spend * revenue_share, 2) if products else round(spend / len(products), 2),
"revenue": round(revenue, 2),
"roas": cell_roas,
}
rows.append(row)
# Also add market rows
for market in markets:
row: Dict[str, Any] = {"dimension": market, "type": "market"}
for channel in channels_list:
cdata = channel_data.get(channel, {})
spend = cdata.get("spend", 0)
total_channel_spend = sum(cd.get("spend", 0) for cd in channel_data.values())
revenue_share = spend / total_channel_spend if total_channel_spend > 0 else 0
revenue = market_revenue.get(market, 0) * revenue_share
cell_roas = round(revenue / (spend * revenue_share), 2) if spend > 0 and revenue_share > 0 else 0.0
row[channel] = {
"spend": round(spend * revenue_share, 2),
"revenue": round(revenue, 2),
"roas": cell_roas,
}
rows.append(row)
return jsonify({
"rows": rows,
"channels": channels_list,
"channel_names": {
'google_ads': 'Google Ads',
'facebook_ads': 'Facebook Ads',
},
})
@analytics_bp.route("/api/analytics/scale-optimization/recommendations", methods=["GET"])
@require_auth_json()
@require_tier(min_tier='command')
def scale_optimization_recommendations():
"""GET /api/analytics/scale-optimization/recommendations
Returns budget reallocation suggestions based on ROAS comparison
across channels, prioritized by projected revenue lift.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 403)
now = datetime.now(timezone.utc)
days_30 = now - timedelta(days=30)
# Get current channel performance
channel_rows = db.session.query(
AdMetric.source_service,
db.func.sum(AdMetric.spend).label("total_spend"),
db.func.sum(AdMetric.conversions).label("total_conversions"),
db.func.avg(AdMetric.roas).label("avg_roas"),
).filter(
AdMetric.company_id == company_id,
AdMetric.metric_date >= days_30,
).group_by(AdMetric.source_service).all()
# Revenue by source from won deals
won_deals = db.session.query(
CrmDeal.source,
db.func.sum(CrmDeal.amount).label("total_revenue"),
).filter(
CrmDeal.company_id == company_id,
CrmDeal.status.in_(['closed_won', 'Closed Won']),
CrmDeal.created_at >= days_30,
).group_by(CrmDeal.source).all()
revenue_by_source = {row.source or 'other': float(row.total_revenue or 0) for row in won_deals}
channel_names = {
'google_ads': 'Google Ads',
'facebook_ads': 'Facebook Ads',
}
channels = []
for row in channel_rows:
source = row.source_service or 'other'
spend = float(row.total_spend or 0)
roas = float(row.avg_roas or 0)
# Find attributed revenue
revenue = 0.0
for ds in revenue_by_source:
if source in ds.lower() or ds in source.lower():
revenue = revenue_by_source[ds]
break
calculated_roas = round(revenue / spend, 2) if spend > 0 else roas
channels.append({
"source": source,
"name": channel_names.get(source, source.replace('_', ' ').title()),
"spend": spend,
"roas": calculated_roas if calculated_roas > 0 else roas,
"revenue": revenue,
})
channels.sort(key=lambda c: c["roas"], reverse=True)
# Generate recommendations
recommendations = []
if len(channels) >= 2:
total_spend = sum(c["spend"] for c in channels)
overall_avg_roas = sum(c["revenue"] for c in channels) / total_spend if total_spend > 0 else 0
for i, low_channel in enumerate(channels):
for j, high_channel in enumerate(channels):
if i == j:
continue
if low_channel["roas"] >= high_channel["roas"]:
continue
# Only recommend if the gap is significant (>20% ROAS difference)
if low_channel["roas"] > 0 and high_channel["roas"] > 0:
roas_ratio = high_channel["roas"] / low_channel["roas"]
if roas_ratio < 1.2:
continue
# Suggest moving a portion of budget (20-50% of low channel spend)
reallocate_pct = min(0.5, max(0.2, (roas_ratio - 1) * 0.3))
move_amount = round(low_channel["spend"] * reallocate_pct)
if move_amount < 100: # Minimum $100 reallocation
continue
# Projected revenue impact
current_revenue = low_channel["revenue"] + high_channel["revenue"]
# After reallocation:
new_low_spend = low_channel["spend"] - move_amount
new_high_spend = high_channel["spend"] + move_amount
new_low_revenue = new_low_spend * low_channel["roas"] if low_channel["roas"] > 0 else 0
new_high_revenue = new_high_spend * high_channel["roas"] if high_channel["roas"] > 0 else 0
projected_revenue = new_low_revenue + new_high_revenue
revenue_lift = projected_revenue - current_revenue
recommendations.append({
"id": f"move_{low_channel['source']}_to_{high_channel['source']}",
"action": "reallocate",
"description": f"Move ${move_amount:,.0f}/month from {low_channel['name']} to {high_channel['name']}",
"from_channel": low_channel["name"],
"from_channel_key": low_channel["source"],
"to_channel": high_channel["name"],
"to_channel_key": high_channel["source"],
"move_amount": move_amount,
"current_roas_from": round(low_channel["roas"], 2),
"current_roas_to": round(high_channel["roas"], 2),
"projected_revenue": round(projected_revenue, 2),
"current_revenue": round(current_revenue, 2),
"revenue_lift": round(revenue_lift, 2),
"revenue_lift_pct": round((revenue_lift / current_revenue * 100) if current_revenue > 0 else 0, 1),
"confidence": "high" if roas_ratio > 1.5 else ("medium" if roas_ratio > 1.2 else "low"),
"reason": f"{high_channel['name']} delivers {roas_ratio:.1f}x better ROAS than {low_channel['name']}.",
})
# Sort by projected revenue lift
recommendations.sort(key=lambda r: r["revenue_lift"], reverse=True)
return jsonify({
"recommendations": recommendations,
"total_channels": len(channels),
"overall_avg_roas": round(overall_avg_roas, 2) if channels else 0,
"channel_summary": channels,
"generated_at": now.isoformat(),
})
# -- P3: Project Analytics ---------------------------------------------------
@analytics_bp.route("/api/analytics/projects/schedule", methods=["GET"])
@require_auth_json()
def project_schedule_analytics():
"""GET /api/analytics/projects/schedule
Project schedule variance analytics. Returns a summary of project
timelines, slippage rates, and on-time completion metrics.
Query params:
since_days (int, default 180) — look back window for project data.
include_active (bool, default true) — include active projects.
Returns:
- total_projects: count of projects in window.
- on_time_count: projects completed within planned window.
- slipped_count: projects exceeding planned window.
- avg_variance_pct: average schedule variance %.
- slipped_projects: list of slipped projects with details.
- on_time_rate_pct: percentage of on-time completions.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
since_days = request.args.get("since_days", 180, type=int)
include_active = request.args.get("include_active", "true").lower() in ("true", "1", "yes")
cutoff = datetime.now(timezone.utc) - timedelta(days=since_days)
projects = Project.query.filter(
Project.company_id == company_id,
Project.start_date >= cutoff
).all()
total = len(projects)
on_time = 0
slipped = 0
variances: List[float] = []
slipped_list: List[Dict[str, Any]] = []
for p in projects:
start = p.start_date
end = p.end_date
if not start or not end:
continue
if start.tzinfo is None:
start = start.replace(tzinfo=timezone.utc)
if end.tzinfo is None:
end = end.replace(tzinfo=timezone.utc)
planned_span = (end - start).days
if planned_span <= 0:
continue
if p.completed_date:
actual = p.completed_date
if actual.tzinfo is None:
actual = actual.replace(tzinfo=timezone.utc)
actual_span = (actual - start).days
variance_pct = ((actual_span - planned_span) / planned_span) * 100
elif include_active and p.status in ("active", "in_progress"):
# Active project — check if already behind schedule
now = datetime.now(timezone.utc)
actual_span = (now - start).days
elapsed_pct = (now - start).days / planned_span if planned_span > 0 else 0
# Estimate variance based on progress
variance_pct = 0.0
else:
continue
variances.append(variance_pct)
if variance_pct > 10:
slipped += 1
slipped_list.append({
"id": p.id,
"name": p.name or "Unnamed Project",
"status": p.status,
"planned_days": planned_span,
"actual_days": actual_span if p.completed_date else None,
"variance_pct": round(variance_pct, 1),
"start_date": start.strftime("%Y-%m-%d"),
"end_date": end.strftime("%Y-%m-%d"),
"completed_date": p.completed_date.strftime("%Y-%m-%d") if p.completed_date else None,
"severity": "critical" if variance_pct > 50 else "high" if variance_pct > 30 else "medium",
})
else:
on_time += 1
slipped_list.sort(key=lambda x: -x["variance_pct"])
avg_variance = sum(variances) / len(variances) if variances else 0.0
on_time_rate = (on_time / (on_time + slipped) * 100) if (on_time + slipped) > 0 else 0.0
return jsonify({
"schedule": {
"total_projects": total,
"on_time_count": on_time,
"slipped_count": slipped,
"on_time_rate_pct": round(on_time_rate, 1),
"avg_variance_pct": round(avg_variance, 1),
"period_days": since_days,
},
"slipped_projects": slipped_list[:20], # Cap at 20
})
@analytics_bp.route("/api/analytics/projects/margin", methods=["GET"])
@require_auth_json()
def project_margin_analytics():
"""GET /api/analytics/projects/margin
Project margin analytics. Returns margin distribution, low-margin
alerts, and profitability trends.
Query params:
since_days (int, default 180) — look back window for project data.
include_active (bool, default false) — include active projects.
threshold_pct (float, default 15) — margin threshold for alerts.
min_project_value (float, default 5000) — minimum project value to analyze.
Returns:
- total_projects: count of projects in window.
- avg_margin_pct: average margin across projects.
- low_margin_count: projects below threshold.
- negative_margin_count: projects losing money.
- total_revenue: sum of project revenue.
- total_profit: sum of (revenue - actual_cost).
- margin_distribution: margin brackets.
- low_margin_projects: list of projects below threshold.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
since_days = request.args.get("since_days", 180, type=int)
include_active = request.args.get("include_active", "false").lower() in ("true", "1", "yes")
threshold_pct = request.args.get("threshold_pct", 15, type=float)
min_value = request.args.get("min_project_value", 5000, type=float)
cutoff = datetime.now(timezone.utc) - timedelta(days=since_days)
projects = Project.query.filter(
Project.company_id == company_id,
Project.start_date >= cutoff
).all()
total_revenue = 0.0
total_cost = 0.0
total_profit = 0.0
margins: List[float] = []
low_margin = 0
negative_margin = 0
low_margin_list: List[Dict[str, Any]] = []
# Margin distribution brackets
distribution = {
"excellent": {"label": "≥25%", "count": 0, "revenue": 0.0},
"good": {"label": "15-25%", "count": 0, "revenue": 0.0},
"low": {"label": "7.5-15%", "count": 0, "revenue": 0.0},
"critical": {"label": "0-7.5%", "count": 0, "revenue": 0.0},
"negative": {"label": "<0%", "count": 0, "revenue": 0.0},
}
for p in projects:
# Skip active projects unless requested
if p.status in ("active", "in_progress") and not include_active:
continue
revenue = p.revenue or p.budget or 0.0
cost = p.actual_cost or 0.0
# Skip projects below minimum value
if revenue < min_value:
continue
total_revenue += revenue
total_cost += cost
margin_pct = ((revenue - cost) / revenue * 100) if revenue > 0 else 0.0
profit = revenue - cost
total_profit += profit
margins.append(margin_pct)
# Categorize
if margin_pct < 0:
distribution["negative"]["count"] += 1
distribution["negative"]["revenue"] += revenue
negative_margin += 1
low_margin += 1
low_margin_list.append({
"id": p.id,
"name": p.name or "Unnamed Project",
"revenue": round(revenue, 2),
"cost": round(cost, 2),
"margin_pct": round(margin_pct, 1),
"profit": round(profit, 2),
"severity": "critical",
})
elif margin_pct < 7.5:
distribution["critical"]["count"] += 1
distribution["critical"]["revenue"] += revenue
low_margin += 1
low_margin_list.append({
"id": p.id,
"name": p.name or "Unnamed Project",
"revenue": round(revenue, 2),
"cost": round(cost, 2),
"margin_pct": round(margin_pct, 1),
"profit": round(profit, 2),
"severity": "high",
})
elif margin_pct < threshold_pct:
distribution["low"]["count"] += 1
distribution["low"]["revenue"] += revenue
low_margin += 1
low_margin_list.append({
"id": p.id,
"name": p.name or "Unnamed Project",
"revenue": round(revenue, 2),
"cost": round(cost, 2),
"margin_pct": round(margin_pct, 1),
"profit": round(profit, 2),
"severity": "medium",
})
elif margin_pct < 25:
distribution["good"]["count"] += 1
distribution["good"]["revenue"] += revenue
else:
distribution["excellent"]["count"] += 1
distribution["excellent"]["revenue"] += revenue
avg_margin = sum(margins) / len(margins) if margins else 0.0
analyzed = len(margins)
low_margin_list.sort(key=lambda x: x["margin_pct"])
return jsonify({
"margin": {
"total_projects": total,
"analyzed": analyzed,
"avg_margin_pct": round(avg_margin, 1),
"low_margin_count": low_margin,
"negative_margin_count": negative_margin,
"total_revenue": round(total_revenue, 2),
"total_cost": round(total_cost, 2),
"total_profit": round(total_profit, 2),
"overall_margin_pct": round((total_profit / total_revenue * 100) if total_revenue > 0 else 0, 1),
"threshold_pct": threshold_pct,
"period_days": since_days,
},
"distribution": distribution,
"low_margin_projects": low_margin_list[:20], # Cap at 20
})
@analytics_bp.route("/api/analytics/projects/overview", methods=["GET"])
@require_auth_json()
def project_overview():
"""GET /api/analytics/projects/overview
Combined project health dashboard. Aggregates schedule and margin
analytics into a single overview for the dashboard.
Query params:
since_days (int, default 180) — look back window.
include_active (bool, default false) — include active projects.
Returns:
- summary: high-level health indicators.
- schedule: on-time rate, slipped count.
- margin: average margin, low margin count.
- at_risk: projects flagged by either detector.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
since_days = request.args.get("since_days", 180, type=int)
include_active = request.args.get("include_active", "false").lower() in ("true", "1", "yes")
cutoff = datetime.now(timezone.utc) - timedelta(days=since_days)
projects = Project.query.filter(
Project.company_id == company_id,
Project.start_date >= cutoff
).all()
total = len(projects)
active = 0
completed = 0
total_revenue = 0.0
total_cost = 0.0
total_profit = 0.0
margins: List[float] = []
# Schedule tracking
on_time = 0
slipped = 0
# At-risk list
at_risk: List[Dict[str, Any]] = []
for p in projects:
start = p.start_date
end = p.end_date
if p.status in ("active", "in_progress"):
active += 1
elif p.status in ("completed", "closed"):
completed += 1
revenue = p.revenue or p.budget or 0.0
cost = p.actual_cost or 0.0
total_revenue += revenue
total_cost += cost
margin_pct = ((revenue - cost) / revenue * 100) if revenue > 0 else 0.0
total_profit += revenue - cost
margins.append(margin_pct)
# Schedule variance
if start and end:
if start.tzinfo is None:
start = start.replace(tzinfo=timezone.utc)
if end.tzinfo is None:
end = end.replace(tzinfo=timezone.utc)
planned_span = (end - start).days
if planned_span > 0:
if p.completed_date:
actual = p.completed_date
if actual.tzinfo is None:
actual = actual.replace(tzinfo=timezone.utc)
actual_span = (actual - start).days
variance_pct = ((actual_span - planned_span) / planned_span) * 100
elif include_active and p.status in ("active", "in_progress"):
variance_pct = 0.0
else:
variance_pct = None
if variance_pct is not None:
if variance_pct > 10:
slipped += 1
else:
on_time += 1
# At-risk: low margin or schedule slippage
reasons = []
if margin_pct < 15:
reasons.append("low_margin")
if variance_pct is not None and variance_pct > 10:
reasons.append("schedule_slip")
if reasons:
at_risk.append({
"id": p.id,
"name": p.name or "Unnamed Project",
"status": p.status,
"revenue": round(revenue, 2),
"margin_pct": round(margin_pct, 1),
"variance_pct": round(variance_pct, 1) if variance_pct is not None else None,
"reasons": reasons,
"severity": "critical" if (margin_pct < 0 or (variance_pct or 0) > 50) else "high" if (margin_pct < 7.5 or (variance_pct or 0) > 30) else "medium",
})
at_risk.sort(key=lambda x: x["severity"] == "critical" and 0 or x["severity"] == "high" and 1 or 2)
avg_margin = sum(margins) / len(margins) if margins else 0.0
on_time_rate = (on_time / (on_time + slipped) * 100) if (on_time + slipped) > 0 else 0.0
# Health score (0-100)
health = 50 # Start at neutral
health += min(on_time_rate - 50, 25) # Up to +25 for on-time rate
health += min(avg_margin - 15, 15) # Up to +15 for margin
health -= min(slipped * 3, 15) # -3 per slipped project
health -= min(sum(1 for r in at_risk if r["severity"] == "critical") * 5, 20) # -5 per critical
health = max(0, min(100, health))
return jsonify({
"overview": {
"health_score": round(health, 0),
"total_projects": total,
"active_projects": active,
"completed_projects": completed,
"total_revenue": round(total_revenue, 2),
"total_profit": round(total_profit, 2),
"avg_margin_pct": round(avg_margin, 1),
"on_time_rate_pct": round(on_time_rate, 1),
"slipped_count": slipped,
"at_risk_count": len(at_risk),
"period_days": since_days,
},
"at_risk": at_risk[:15], # Cap at 15
})
# -- Multi-Location Rollup Analytics (P4) -----------------------------------
@analytics_bp.route("/api/analytics/locations", methods=["GET"])
@require_auth_json()
def location_rollup():
"""GET /api/analytics/locations
Multi-location rollup analytics. Returns per-location revenue metrics,
variance analysis, and concentration alerts.
Query params:
period_days (int, default 90) — look back period for revenue.
min_variance_pct (float, default 20) — variance threshold for alerts.
concentration_pct (float, default 60) — concentration alert threshold.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
period_days = request.args.get("period_days", 90, type=int)
min_variance_pct = request.args.get("min_variance_pct", 20, type=float)
concentration_pct = request.args.get("concentration_pct", 60, type=float)
cutoff = datetime.now(timezone.utc) - timedelta(days=period_days)
# Get locations for this company
locations = Location.query.filter_by(company_id=company_id).all()
# Also find denormalized locations from revenue records
records = RevenueRecord.query.filter(
RevenueRecord.company_id == company_id,
RevenueRecord.transaction_date >= cutoff,
RevenueRecord.status == "completed"
).all()
# Group revenue by location
location_stats: Dict[str, Dict[str, Any]] = {}
# Initialize from formal locations
for loc in locations:
key = loc.id
location_stats[key] = {
"location_id": loc.id,
"location_name": loc.name,
"revenue": 0.0,
"transaction_count": 0,
"avg_transaction": 0.0,
}
for rr in records:
# Use location_id if present, otherwise fall back to location_name hash
if rr.location_id:
key = rr.location_id
elif rr.location_name:
# Create a synthetic key for denormalized locations
key = rr.location_name
if key not in location_stats:
location_stats[key] = {
"location_id": key,
"location_name": rr.location_name,
"revenue": 0.0,
"transaction_count": 0,
"avg_transaction": 0.0,
"denormalized": True,
}
else:
continue # Skip records with no location
if key not in location_stats:
location_stats[key] = {
"location_id": key,
"location_name": rr.location_name or "Unknown",
"revenue": 0.0,
"transaction_count": 0,
"avg_transaction": 0.0,
}
location_stats[key]["revenue"] += rr.amount or 0.0
location_stats[key]["transaction_count"] += 1
# Calculate averages and percentages
total_revenue = sum(s["revenue"] for s in location_stats.values())
location_list = []
for key, stats in location_stats.items():
count = stats["transaction_count"]
revenue = stats["revenue"]
avg = revenue / count if count > 0 else 0.0
pct_of_total = (revenue / total_revenue * 100) if total_revenue > 0 else 0.0
location_list.append({
"location_id": stats["location_id"],
"location_name": stats["location_name"],
"revenue": round(revenue, 2),
"transaction_count": count,
"avg_transaction": round(avg, 2),
"pct_of_total": round(pct_of_total, 1),
})
# Sort by revenue descending
location_list.sort(key=lambda x: x["revenue"], reverse=True)
# Calculate overall metrics
avg_revenue_per_location = total_revenue / len(location_list) if location_list else 0.0
variance_from_avg = 0.0
if len(location_list) > 1 and avg_revenue_per_location > 0:
variances = [
abs(loc["revenue"] - avg_revenue_per_location) / avg_revenue_per_location * 100
for loc in location_list
]
variance_from_avg = max(variances) if variances else 0.0
# Generate alerts
alerts: List[Dict[str, Any]] = []
# Concentration alert
if location_list and location_list[0]["pct_of_total"] > concentration_pct:
alerts.append({
"type": "concentration",
"severity": "high" if location_list[0]["pct_of_total"] > 75 else "medium",
"location_id": location_list[0]["location_id"],
"location_name": location_list[0]["location_name"],
"message": f"{location_list[0]['location_name']} accounts for {location_list[0]['pct_of_total']}% of revenue (threshold: {concentration_pct}%)",
"pct_of_total": location_list[0]["pct_of_total"],
})
# Variance alert
if variance_from_avg > min_variance_pct:
underperformers = [
loc for loc in location_list
if loc["revenue"] < avg_revenue_per_location * (1 - min_variance_pct / 100)
]
for up in underperformers:
alerts.append({
"type": "underperformer",
"severity": "high" if up["revenue"] < avg_revenue_per_location * 0.5 else "medium",
"location_id": up["location_id"],
"location_name": up["location_name"],
"message": f"{up['location_name']} is significantly below average ({up['revenue']} vs avg {round(avg_revenue_per_location, 2)})",
"revenue": up["revenue"],
"avg_revenue": round(avg_revenue_per_location, 2),
})
return jsonify({
"locations": {
"total_locations": len(location_list),
"total_revenue": round(total_revenue, 2),
"avg_revenue_per_location": round(avg_revenue_per_location, 2),
"max_variance_pct": round(variance_from_avg, 1),
"period_days": period_days,
},
"breakdown": location_list,
"alerts": alerts,
})
@analytics_bp.route("/api/analytics/locations/comparison", methods=["GET"])
@require_auth_json()
def location_comparison():
"""GET /api/analytics/locations/comparison
Side-by-side comparison of location performance with benchmarking.
Query params:
period_days (int, default 90) — look back period.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
period_days = request.args.get("period_days", 90, type=int)
cutoff = datetime.now(timezone.utc) - timedelta(days=period_days)
records = RevenueRecord.query.filter(
RevenueRecord.company_id == company_id,
RevenueRecord.transaction_date >= cutoff,
RevenueRecord.status == "completed"
).all()
# Group by location
location_data: Dict[str, Dict[str, Any]] = {}
for rr in records:
if not rr.location_id and not rr.location_name:
continue
key = rr.location_id or rr.location_name or "unknown"
if key not in location_data:
location_data[key] = {
"location_id": key,
"location_name": rr.location_name or "Unknown",
"revenue": 0.0,
"count": 0,
"amounts": [],
}
location_data[key]["revenue"] += rr.amount or 0.0
location_data[key]["count"] += 1
location_data[key]["amounts"].append(rr.amount or 0.0)
# Compute benchmarks
total = sum(v["revenue"] for v in location_data.values())
avg_revenue = total / len(location_data) if location_data else 0.0
median_revenue = sorted(v["revenue"] for v in location_data.values())[len(location_data) // 2] if location_data else 0.0
comparison = []
for key, data in location_data.items():
avg_tx = data["revenue"] / data["count"] if data["count"] > 0 else 0.0
vs_avg = ((data["revenue"] - avg_revenue) / avg_revenue * 100) if avg_revenue > 0 else 0.0
vs_median = ((data["revenue"] - median_revenue) / median_revenue * 100) if median_revenue > 0 else 0.0
pct_share = (data["revenue"] / total * 100) if total > 0 else 0.0
comparison.append({
"location_id": data["location_id"],
"location_name": data["location_name"],
"revenue": round(data["revenue"], 2),
"transaction_count": data["count"],
"avg_transaction": round(avg_tx, 2),
"pct_of_total": round(pct_share, 1),
"vs_avg_pct": round(vs_avg, 1),
"vs_median_pct": round(vs_median, 1),
"ranking": 0, # Set below
})
# Rank by revenue
comparison.sort(key=lambda x: x["revenue"], reverse=True)
for i, item in enumerate(comparison):
item["ranking"] = i + 1
return jsonify({
"comparison": {
"total_locations": len(comparison),
"total_revenue": round(total, 2),
"avg_revenue": round(avg_revenue, 2),
"median_revenue": round(median_revenue, 2),
"period_days": period_days,
},
"locations": comparison,
})
@analytics_bp.route("/api/analytics/locations/detector", methods=["POST"])
@require_auth_json()
def location_detector_run():
"""POST /api/analytics/locations/detector
Run the MultiLocationRollupDetector for the authenticated company.
Returns leak candidates for concentration and underperformance issues.
Request body (optional):
period_days (int, default 90) — look back period.
min_variance_pct (float, default 20) — variance threshold.
concentration_pct (float, default 60) — concentration threshold.
"""
try:
company_id = _get_company_id()
except ValueError as exc:
return _json_error(str(exc), 400)
data = request.get_json(silent=True) or {}
period_days = data.get("period_days", 90)
min_variance_pct = data.get("min_variance_pct", 20)
concentration_pct = data.get("concentration_pct", 60)
from ..services.leak_detectors import MultiLocationRollupDetector
detector = MultiLocationRollupDetector()
# Apply custom params if provided (overrides default_params)
if "period_days" in data:
detector.default_params["lookback_days"] = data["period_days"]
if "min_variance_pct" in data:
detector.default_params["min_variance_pct"] = data["min_variance_pct"]
if "concentration_pct" in data:
detector.default_params["concentration_threshold"] = data["concentration_pct"] / 100.0
candidates = detector.check(company_id=company_id)
# Format results
results = []
for c in candidates:
results.append({
"dedupe_key": c.dedupe_key(),
"detector": c.detector_id,
"source": c.source,
"severity": c.severity,
"description": c.description,
"estimated_loss": c.estimated_loss,
"metadata": c.metadata_json or {},
"rule_params": c.rule_params or {},
})
return jsonify({
"detector": "multi_location_rollup",
"company_id": company_id,
"candidates_found": len(results),
"candidates": results,
})