Phase 11: PRGO Generation

Duration: 2 days Goal: Implement Provisioning Review Generating Order (PRGO) for demand forecasting

Overview

PRGO is Legacy ERP’s core forecasting mechanism:

Analyzes MSTC (Monthly Stock Taking Card) data
Calculates sufficiency (stock days)
Generates recommended order quantities per depot/item
Converts to Purchase Orders after approval

1. PRGO DocType

class PRGO(BaseDocType):
    """Provisioning Review Generating Order."""

    prgo_id: str = Field(max_length=50, unique=True)
    prgo_date: date  # Stock taking date
    run_date: datetime  # When PRGO was generated

    # Parameters
    review_factor_months: int = 2  # Review period
    blocking_factor_months: int = 1  # Blocking period

    prgo_type: str  # Normal, Liquor, Leh

    # Filters
    category: str | None = None  # I1, I2, I3
    group_from: str | None = None
    group_to: str | None = None
    generic_from: str | None = None
    generic_to: str | None = None

    description: str | None = None
    status: str = "Draft"  # Draft, Generated, Finalized, Converted

    class Meta:
        table_name = "prgo"

1.1 PRGO Item Detail

class PRGOItemDetail(BaseDocType):
    """Item-level PRGO recommendation."""

    prgo: str  # FK to PRGO
    item: str  # FK to Item
    depot: str  # FK to Depot

    # MSTC Data
    opening_stock: Decimal = Decimal("0")
    receipts: Decimal = Decimal("0")
    sales: Decimal = Decimal("0")
    closing_stock: Decimal = Decimal("0")

    # Denials/Substitutes
    qty_denied: Decimal = Decimal("0")
    qty_substituted: Decimal = Decimal("0")
    qty_forced_sale: Decimal = Decimal("0")

    # Calculated
    avg_monthly_sales: Decimal = Decimal("0")
    dues_in: Decimal = Decimal("0")
    in_transit: Decimal = Decimal("0")

    # Sufficiency
    actual_sufficiency_days: Decimal = Decimal("0")
    recommended_sufficiency_days: Decimal = Decimal("0")

    # PRGO Qty
    prgo_qty: Decimal = Decimal("0")  # System calculated
    bg_qty: Decimal = Decimal("0")  # For I3 items (Bank Guarantee)
    recommended_qty: Decimal = Decimal("0")  # Modified by user

    remarks: str | None = None

    class Meta:
        table_name = "prgo_item_detail"
        indexes = ["prgo", "item", "depot"]

2. Sufficiency Calculation

class SufficiencyCalculator:
    """Calculate stock sufficiency metrics."""

    async def calculate_sufficiency(
        self,
        item: str,
        depot: str,
        as_of: date,
    ) -> dict:
        """
        Sufficiency = Current Stock / Average Monthly Consumption

        Returns days of stock available.
        """
        # Get current stock
        stock = await self.stock_ledger.get_balance(item, depot)

        # Get 6-month sales average
        sales_data = await self.get_sales_history(item, depot, months=6)
        avg_monthly = sum(s.qty for s in sales_data) / 6

        if avg_monthly == 0:
            return {
                "actual_sufficiency_days": 999,  # Infinite if no sales
                "avg_monthly_sales": 0,
            }

        daily_consumption = avg_monthly / 30
        sufficiency_days = stock / daily_consumption

        return {
            "actual_sufficiency_days": sufficiency_days,
            "avg_monthly_sales": avg_monthly,
            "current_stock": stock,
        }

    async def get_recommended_sufficiency(
        self,
        item: str,
        depot: str,
    ) -> Decimal:
        """Get recommended sufficiency from GenericDetail."""
        item_obj = await self.item_repo.get(item)
        generic_detail = await self.generic_detail_repo.find(
            generic=item_obj.generic,
            depot=depot,
        )
        return generic_detail.recommended_sufficiency_days if generic_detail else 60

3. PRGO Generator

class PRGOGenerator:
    """Generate PRGO from MSTC data."""

    async def generate(
        self,
        prgo_date: date,
        review_factor: int,
        blocking_factor: int,
        prgo_type: str,
        filters: dict,
    ) -> PRGO:
        """Generate PRGO for given parameters."""

        prgo = PRGO(
            prgo_date=prgo_date,
            review_factor_months=review_factor,
            blocking_factor_months=blocking_factor,
            prgo_type=prgo_type,
            **filters,
        )
        await self.prgo_repo.save(prgo)

        # Get items based on filters
        items = await self.get_filtered_items(prgo_type, filters)
        depots = await self.get_depots_for_type(prgo_type)

        for item in items:
            for depot in depots:
                detail = await self.calculate_prgo_qty(
                    prgo, item, depot, review_factor, blocking_factor
                )
                await self.detail_repo.save(detail)

        prgo.status = "Generated"
        await self.prgo_repo.save(prgo)
        return prgo

    async def calculate_prgo_qty(
        self,
        prgo: PRGO,
        item: Item,
        depot: Depot,
        review_factor: int,
        blocking_factor: int,
    ) -> PRGOItemDetail:
        """Calculate PRGO quantity for item+depot."""

        # Get MSTC data
        mstc = await self.mstc_repo.get_latest(item.id, depot.id)

        # Calculate sufficiency
        suff = await self.sufficiency.calculate_sufficiency(
            item.id, depot.id, prgo.prgo_date
        )

        recommended_suff = await self.sufficiency.get_recommended_sufficiency(
            item.id, depot.id
        )

        # PRGO Qty = (Recommended Sufficiency - Actual Sufficiency) * Daily Consumption
        shortfall_days = max(0, recommended_suff - suff["actual_sufficiency_days"])
        daily_consumption = suff["avg_monthly_sales"] / 30
        prgo_qty = shortfall_days * daily_consumption

        # Adjust for review + blocking factor
        prgo_qty = prgo_qty * (review_factor + blocking_factor) / review_factor

        return PRGOItemDetail(
            prgo=prgo.id,
            item=item.id,
            depot=depot.id,
            opening_stock=mstc.stock_qty if mstc else 0,
            sales=mstc.sold_qty if mstc else 0,
            avg_monthly_sales=suff["avg_monthly_sales"],
            actual_sufficiency_days=suff["actual_sufficiency_days"],
            recommended_sufficiency_days=recommended_suff,
            prgo_qty=prgo_qty,
            recommended_qty=prgo_qty,  # Can be modified by user
        )

4. PRGO Modification UI

For manual adjustments:

class PRGOModificationService:
    """Bulk modification of PRGO quantities."""

    async def apply_region_shortcut(
        self,
        prgo: str,
        item: str,
        quantities: dict[str, Decimal],  # {region: qty}
    ):
        """Apply quantity by region."""
        for region, qty in quantities.items():
            depots = await self.depot_repo.find_by_region(region)
            await self.detail_repo.update_many(
                filters={"prgo": prgo, "item": item, "depot__in": [d.id for d in depots]},
                values={"recommended_qty": qty},
            )

    async def apply_size_shortcut(
        self,
        prgo: str,
        item: str,
        quantities: dict[str, Decimal],  # {size: qty}
    ):
        """Apply quantity by depot size."""
        ...

    async def copy_prgo_to_recommended(self, prgo: str, item: str):
        """Copy PRGO qty to recommended qty."""
        await self.detail_repo.update_many(
            filters={"prgo": prgo, "item": item},
            values={"recommended_qty": F("prgo_qty")},
        )

    async def copy_bg_to_recommended(self, prgo: str, item: str):
        """Copy BG qty to recommended (for I3 items)."""
        await self.detail_repo.update_many(
            filters={"prgo": prgo, "item": item},
            values={"recommended_qty": F("bg_qty")},
        )

5. PRGO to PO Conversion

class PRGOToPOConverter:
    """Convert finalized PRGO to Purchase Orders."""

    async def convert(self, prgo: PRGO) -> list[PurchaseOrder]:
        """Group PRGO items by supplier and create POs."""

        details = await self.detail_repo.find(prgo=prgo.id)

        # Group by supplier + depot
        grouped = defaultdict(list)
        for detail in details:
            item = await self.item_repo.get(detail.item)
            supplier = await self.get_default_supplier(item.id, detail.depot)
            grouped[(supplier, detail.depot)].append(detail)

        pos = []
        for (supplier, depot), items in grouped.items():
            po = PurchaseOrder(
                supplier=supplier,
                depot=depot,
                order_date=date.today(),
                prgo=prgo.id,
            )
            for detail in items:
                po.items.append(PurchaseOrderItem(
                    item=detail.item,
                    qty=detail.recommended_qty,
                ))
            await self.po_repo.save(po)
            pos.append(po)

        prgo.status = "Converted"
        await self.prgo_repo.save(prgo)

        return pos

6. API Endpoints

Endpoint	Method	Description
`/api/v1/prgo`	GET, POST	List/Create PRGO
`/api/v1/prgo/{id}/generate`	POST	Run PRGO generation
`/api/v1/prgo/{id}/finalize`	POST	Finalize PRGO
`/api/v1/prgo/{id}/convert`	POST	Convert to POs
`/api/v1/prgo/{id}/items`	GET	Get PRGO details
`/api/v1/prgo/{id}/items/bulk-update`	PUT	Bulk update recommended qty

7. Verification Checklist