"""
KPI Service - Dashboard KPI aggregation.
Implements: get_kpi_dashboard
"""
from datetime import datetime, timezone, timedelta
from sqlalchemy import func
from ..models import db, KPIValue, Company
def get_kpi_dashboard(company_id, period='monthly'):
"""Get aggregated KPIs for dashboard view."""
now = datetime.now(timezone.utc)
# Get the last 12 months of KPI data
cutoff = now - timedelta(days=365)
kpi_values = KPIValue.query.filter_by(
company_id=company_id,
).filter(
KPIValue.period_start >= cutoff
).order_by(KPIValue.kpi_name, KPIValue.period_start).all()
# Aggregate by KPI name
kpis_by_name = {}
for kv in kpi_values:
name = kv.kpi_name
if name not in kpis_by_name:
kpis_by_name[name] = {
'name': name,
'category': kv.category,
'unit': kv.unit,
'values': [],
'current': None,
'trend': None,
}
kpis_by_name[name]['values'].append({
'period_start': kv.period_start,
'value': kv.value,
})
kpis_by_name[name]['current'] = kv.value
# Calculate trends
for name, kpi_data in kpis_by_name.items():
values = [v['value'] for v in kpi_data['values']]
if len(values) >= 2:
# Simple trend: compare last 2 values
recent = values[-1]
previous = values[-2]
if previous > 0:
kpi_data['trend'] = ((recent - previous) / previous) * 100
else:
kpi_data['trend'] = 0
else:
kpi_data['trend'] = 0
return list(kpis_by_name.values())
def get_kpi_history(company_id, kpi_name, months=12):
"""Get time-series history for a specific KPI."""
now = datetime.now(timezone.utc)
cutoff = now - timedelta(days=30 * months)
kpi_values = KPIValue.query.filter_by(
company_id=company_id,
kpi_name=kpi_name,
).filter(
KPIValue.period_start >= cutoff
).order_by(KPIValue.period_start).all()
return {
'name': kpi_name,
'values': [
{
'period_start': kv.period_start,
'value': kv.value,
'unit': kv.unit,
}
for kv in kpi_values
],
}
def get_revenue_kpis(company_id):
"""Get revenue-specific KPIs for the dashboard."""
kpis = get_kpi_dashboard(company_id)
revenue_kpis = [k for k in kpis if k['category'] in ['revenue', 'marketing', 'sales']]
return revenue_kpis
def get_operational_kpis(company_id):
"""Get operational KPIs for the dashboard."""
kpis = get_kpi_dashboard(company_id)
ops_kpis = [k for k in kpis if k['category'] in ['operations', 'efficiency', 'customer']]
return ops_kpis