Products - Analytics Service

Location: api/nextango/apps/products/analytics_service.py Last Updated: 2026-07-06

Overview

Provides aggregated product catalog and inventory metrics for the analytics dashboard. Two service classes cover separate concerns: ProductAnalyticsService for sales-driven product metrics (top sellers, category, brand) and InventoryAnalyticsService for stock health (levels, turnover, dead stock, movements, trends). All methods read from the replica database when USE_REPLICA_DATABASE=True and cache their results (see Caching).

Class: ProductAnalyticsService

Aggregation methods for product catalog analytics derived from SaleLineItem records on completed transactions.

get_top_products()

Purpose: Returns the top N products by revenue or units sold, optionally scoped to a store and date range.

Parameters:

Returns:

{
    'data': [
        {
            'product_name': str,   # Product name
            'sku': str,            # Variant SKU snapshot (skuSnapshot)
            'units_sold': int,
            'revenue': Decimal,
        },
        ...
    ],
    'sort_by': str,
    'period': {'start': str | None, 'end': str | None},
}

Example:

from nextango.apps.products.analytics_service import ProductAnalyticsService
from datetime import date

result = ProductAnalyticsService.get_top_products(
    store_id='store-uuid',
    start_date=date(2026, 1, 1),
    end_date=date(2026, 1, 31),
    limit=5,
    sort_by='revenue',
)

get_category_performance()

Purpose: Returns revenue, units sold, and product count grouped by product category.

Parameters:

Returns:

{
    'data': [
        {
            'category': str,        # Category name from Product.organization JSON
            'units_sold': int,
            'revenue': Decimal,
            'product_count': int,   # Distinct products in category
        },
        ...
    ]
}

Products where organization.category is null are excluded. Rows are ordered by revenue, highest first.

get_brand_performance()

Purpose: Returns revenue, units sold, and product count grouped by brand.

Parameters:

Returns:

{
    'data': [
        {
            'brand': str,           # Brand name from Product.organization JSON
            'units_sold': int,
            'revenue': Decimal,
            'product_count': int,
        },
        ...
    ]
}

Products where organization.brand is null are excluded. Rows are ordered by revenue, highest first.

Class: InventoryAnalyticsService

Aggregation methods for inventory health, movement, and trend analytics, derived from InventoryLevel, InventoryMovement, and DailyInventorySnapshot records.

get_inventory_health()

Purpose: Returns a snapshot of inventory health across all active SKUs, optionally filtered to a single store. Stock thresholds use available quantity, computed in the database as max(0, stock_on_hand - stock_reserved) to match the InventoryLevel.stock_available property.

Parameters:

Returns:

{
    'total_sku_count': int,
    'total_units_on_hand': int,          # Sum of stock_on_hand
    'total_reserved': int,               # Sum of stock_reserved
    'total_inventory_value': Decimal,    # stock_on_hand x current_cost; NULL costs count as zero
    'low_stock_skus': int,               # Available > 0 and <= reorder_point
    'out_of_stock_skus': int,            # Available <= 0
    'overstock_skus': int,               # max_stock_level > 0 and stock_on_hand > max_stock_level
    'reorder_needed': int,               # reorder_point > 0 and available <= reorder_point
}

Example:

from nextango.apps.products.analytics_service import InventoryAnalyticsService

health = InventoryAnalyticsService.get_inventory_health(store_id='store-uuid')
print(f"Out of stock: {health['out_of_stock_skus']}")

get_inventory_turnover()

Purpose: Returns the top N SKUs by total transaction count (a proxy for turnover rate).

Parameters:

Returns:

{
    'data': [
        {
            'variant_sku': str,
            'variant_name': str,
            'product_name': str,
            'total_sold': int,          # total_transaction_count from InventoryLevel
            'on_hand': int,             # stock_on_hand
            'last_sale': datetime | None,
            'total_rev': Decimal,
        },
        ...
    ]
}

Only includes active SKUs with stock_on_hand > 0 and total_transaction_count > 0.

get_dead_stock()

Purpose: Returns active SKUs with stock on hand but no transaction activity in the past N days (dead stock).

Parameters:

A SKU qualifies when stock_on_hand > 0 and its last_transaction_date is older than the threshold or has never been set.

Returns:

{
    'data': [
        {
            'variant_sku': str,
            'variant_name': str,
            'stock_on_hand': int,
            'last_transaction_date': datetime | None,
            'product_name': str,
            'inventory_value': Decimal,  # stock_on_hand x current_cost; NULL costs count as zero
        },
        ...
    ],
    'total_dead_stock_value': Decimal,
    'total_dead_stock_units': int,
    'days_threshold': int,
}

Rows are ordered by inventory_value, highest first.

get_inventory_movements_summary()

Purpose: Returns a breakdown of inventory movements by type for a given period. Movement types come from InventoryMovement.movement_type: receiving, sale, return, adjustment, transfer, commit, void, restore.

Parameters:

Returns:

{
    'data': [
        {
            'movement_type': str,
            'total_quantity': int,
            'count': int,
        },
        ...
    ],
    'net_movement': int,    # Sum of all total_quantity values
}

get_inventory_trends()

Purpose: Returns daily inventory trend rows from DailyInventorySnapshot for the last N days, ordered by date ascending. The rollup_daily_inventory_snapshot task populates the snapshots this method reads.

Parameters:

Returns:

{
    'data': [
        {
            'date': str,                      # ISO date string
            'total_units_on_hand': int,
            'total_inventory_value': float,
            'low_stock_count': int,
            'out_of_stock_count': int,
        },
        ...
    ]
}

No products endpoint exposes this method yet; callers invoke the service directly.

Caching

Every method caches its result, keyed by method and parameters. The TTL is 300 seconds for all methods except get_inventory_movements_summary, which uses 120 seconds. All methods read from the replica database when USE_REPLICA_DATABASE=True.

Related Documentation

• Models: Product, ProductVariant, InventoryLevel, InventoryMovement, DailyInventorySnapshot
• Views: the /products/analytics/ actions call ProductAnalyticsService; the /inventory-levels/analytics/ actions call InventoryAnalyticsService
• Services: inventory_valuation_expression(), the shared NULL-safe valuation expression both inventory value figures use
• Tasks: rollup_daily_inventory_snapshot builds the DailyInventorySnapshot rows that get_inventory_trends reads