Phase 10: Data Collection & Import

Duration: 2 days Goal: Centralized data import from 34 Depots and 3,630 URCs

---

Overview

Legacy ERP operates on a centralized procurement model:

• All purchases happen at HQ
• URCs and Depots only report demand and stock balances
• Data flows up via Excel uploads (current) or API (future)

This phase implements WMS-specific import controllers for:

• Demand collection (Material Requests)
• Stock balance reporting
• Delivery confirmations

---

1. Demand Submission

1.1 Demand Upload DocType

class DemandUpload(BaseDocType):
    """Depot uploads consolidated URC demand."""

    # Source
    depot: str  # FK to Depot
    period: str  # "2024-01" (month)

    # Upload
    demand_file: str  # FK to File (Excel)

    # Processing
    status: str = "Pending"  # Pending, Processing, Validated, Imported, Error

    # Statistics
    total_rows: int = 0
    urc_count: int = 0
    item_count: int = 0
    total_qty: Decimal = Decimal("0")

    # Errors
    error_count: int = 0
    error_log: str | None = None  # JSON of row errors

    class Meta:
        table_name = "demand_upload"
        indexes = ["depot", "period", "status"]

1.2 Demand Import Controller

class DemandImportController(BaseController[DemandUpload]):
    """WMS-specific demand import from Depot Excel."""

    # Expected columns in Excel
    REQUIRED_COLUMNS = ["urc_code", "item_code", "demand_qty"]
    OPTIONAL_COLUMNS = ["required_by_date", "remarks"]

    async def on_create(self, doc: DemandUpload) -> None:
        """Auto-validate on upload."""
        doc.status = "Processing"
        await self.validate_file(doc)

    async def validate_file(self, doc: DemandUpload) -> None:
        """Parse and validate Excel file."""
        df = await self.parse_excel(doc.demand_file)

        # Check required columns
        missing = set(self.REQUIRED_COLUMNS) - set(df.columns)
        if missing:
            doc.status = "Error"
            doc.error_log = f"Missing columns: {missing}"
            return

        errors = []
        valid_rows = []

        for idx, row in df.iterrows():
            row_errors = []

            # Validate URC exists under this depot
            if not await self.urc_repo.exists_under_depot(
                row["urc_code"], doc.depot
            ):
                row_errors.append(f"URC {row['urc_code']} not found")

            # Validate Item exists
            if not await self.item_repo.exists_by_code(row["item_code"]):
                row_errors.append(f"Item {row['item_code']} not found")

            # Validate qty
            if row["demand_qty"] <= 0:
                row_errors.append("Qty must be positive")

            if row_errors:
                errors.append({"row": idx + 2, "errors": row_errors})
            else:
                valid_rows.append(row)

        doc.total_rows = len(df)
        doc.error_count = len(errors)
        doc.urc_count = df["urc_code"].nunique()
        doc.item_count = df["item_code"].nunique()
        doc.total_qty = df["demand_qty"].sum()

        if errors:
            doc.status = "Error"
            doc.error_log = json.dumps(errors)
        else:
            doc.status = "Validated"

    async def on_submit(self, doc: DemandUpload) -> None:
        """Import validated demand into Material Requests."""
        df = await self.parse_excel(doc.demand_file)

        # Group by URC to create one MR per URC
        for urc_code, urc_df in df.groupby("urc_code"):
            mr = MaterialRequest(
                urc=urc_code,
                depot=doc.depot,
                request_date=date.today(),
                source="Import",
                source_upload=doc.id,
            )

            for _, row in urc_df.iterrows():
                mr.items.append(MaterialRequestItem(
                    item=row["item_code"],
                    qty=row["demand_qty"],
                ))

            await self.mr_repo.save(mr)

        doc.status = "Imported"

1.3 Tasks

• Create DemandUpload DocType
• Excel template with required columns
• Validation controller
• Create Material Requests on submit
• Error report download

---

2. Stock Balance Reporting

2.1 Balance Upload DocType

class BalanceUpload(BaseDocType):
    """Monthly stock balance report from Depot."""

    depot: str
    as_of_date: date  # Last day of month

    balance_file: str  # FK to File (Excel)

    status: str = "Pending"

    # Stats
    warehouse_count: int = 0
    item_count: int = 0
    total_qty: Decimal = Decimal("0")
    total_value: Decimal = Decimal("0")

    # Variance
    variance_count: int = 0  # Items with qty mismatch vs WMS
    variance_log: str | None = None

    class Meta:
        table_name = "balance_upload"

2.2 Balance Reconciliation

class BalanceImportController(BaseController[BalanceUpload]):
    """Compare uploaded balance with WMS stock ledger."""

    async def on_submit(self, doc: BalanceUpload) -> None:
        """Reconcile uploaded balance with system balance."""
        df = await self.parse_excel(doc.balance_file)

        variances = []

        for _, row in df.iterrows():
            # Get system balance
            system_qty = await self.stock_ledger.get_balance(
                item=row["item_code"],
                warehouse=row["warehouse_code"],
                as_of=doc.as_of_date,
            )

            reported_qty = row["balance_qty"]

            if system_qty != reported_qty:
                variances.append({
                    "item": row["item_code"],
                    "warehouse": row["warehouse_code"],
                    "system_qty": str(system_qty),
                    "reported_qty": str(reported_qty),
                    "variance": str(reported_qty - system_qty),
                })

        doc.variance_count = len(variances)
        doc.variance_log = json.dumps(variances) if variances else None

        # Create Stock Reconciliation for variances
        if variances:
            await self.create_reconciliation(doc, variances)

2.3 Tasks

• Create BalanceUpload DocType
• Reconciliation logic vs stock ledger
• Variance report
• Stock Reconciliation creation for adjustments

---

3. Delivery Confirmation

3.1 Delivery Confirmation DocType

class DeliveryConfirmation(BaseDocType):
    """Depot confirms receipt of goods from Supplier."""

    # Reference
    purchase_receipt: str | None = None  # If from GRN
    delivery_note: str | None = None  # If from internal transfer

    depot: str
    confirmation_date: date

    # Confirmer
    confirmed_by: str  # User

    # Quantities
    items: list["DeliveryConfirmationItem"] = []

    # Discrepancies
    has_discrepancy: bool = False
    discrepancy_notes: str | None = None

    class Meta:
        table_name = "delivery_confirmation"

3.2 Tasks

• Create DeliveryConfirmation DocType
• Link to GRN/Delivery Note
• Discrepancy reporting
• Auto-update receipt status

---

4. Excel Templates

4.1 Downloadable Templates

# GET /api/v1/templates/demand
# GET /api/v1/templates/balance

TEMPLATES = {
    "demand": {
        "columns": ["urc_code", "item_code", "demand_qty", "required_by_date", "remarks"],
        "sample_data": [
            ["URC-MUM-001", "ITM-00001", 100, "2024-02-15", "Urgent"],
        ],
    },
    "balance": {
        "columns": ["warehouse_code", "item_code", "balance_qty", "balance_value"],
        "sample_data": [
            ["WH-DEPOT-01", "ITM-00001", 500, 50000],
        ],
    },
}

@get("/api/v1/templates/{template_type}")
async def download_template(template_type: str) -> StreamingResponse:
    """Generate Excel template for download."""
    template = TEMPLATES.get(template_type)
    df = pd.DataFrame([template["sample_data"]], columns=template["columns"])

    buffer = BytesIO()
    df.to_excel(buffer, index=False)
    buffer.seek(0)

    return StreamingResponse(
        buffer,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename={template_type}_template.xlsx"},
    )

---

5. Upload Portal (Frontend)

// Depot user uploads demand Excel
export const DemandUploadPage = () => {
    const { mutate: upload } = useCreate({ resource: "demand-upload" });

    return (
        <Card title="Upload Monthly Demand">
            <Form onFinish={(values) => upload(values)}>
                <Form.Item name="period" label="Period">
                    <DatePicker picker="month" />
                </Form.Item>
                <Form.Item name="demand_file" label="Demand Excel">
                    <Upload accept=".xlsx,.xls">
                        <Button>Upload Excel</Button>
                    </Upload>
                </Form.Item>
                <Button type="primary" htmlType="submit">
                    Submit
                </Button>
            </Form>

            <Alert message="Download template" description={
                <a href="/api/v1/templates/demand">demand_template.xlsx</a>
            } />
        </App>
    );
};

---

6. Processing Queue

For large uploads (34 depots × 1000s of rows):

@broker.task
async def process_demand_upload(upload_id: str):
    """Background task for large file processing."""
    doc = await demand_upload_repo.get(upload_id)
    controller = DemandImportController()

    await controller.validate_file(doc)
    await demand_upload_repo.save(doc)

    # Notify user
    await notifications.send_in_app(
        user_id=doc.owner,
        title="Demand Upload Processed",
        body=f"{doc.total_rows} rows processed, {doc.error_count} errors",
        link=f"/app/demand-upload/{upload_id}",
    )

---

7. API Endpoints

Endpoint	Method	Description
/api/v1/demand-upload	POST	Upload demand Excel
/api/v1/demand-upload/{id}/submit	POST	Import validated demand
/api/v1/balance-upload	POST	Upload balance Excel
/api/v1/balance-upload/{id}/reconcile	POST	Run reconciliation
/api/v1/templates/{type}	GET	Download Excel template
/api/v1/delivery-confirmation	POST	Confirm delivery
/api/v1/variances	GET	List variances with filters
/api/v1/variances/anomalies	GET	Get flagged anomalies

---

8. Variance Detection & Anomaly Flagging

8.1 Variance Threshold Configuration

class VarianceThreshold(BaseDocType):
    """Configurable tolerance levels per item group."""

    item_group: str | None = None  # None = default for all
    category: str | None = None  # GS, LIF, AFD-I, AFD-II

    # Quantity variance tolerance
    qty_variance_percent: Decimal = Decimal("2.0")  # ±2% acceptable
    qty_variance_absolute: int = 10  # OR ±10 units (whichever is higher)

    # Value variance
    value_variance_percent: Decimal = Decimal("1.0")

    # Stricter for high-value/liquor
    # Liquor: 0.5%, High-value electronics: 0.5%

    class Meta:
        table_name = "variance_threshold"

8.2 Stock Variance DocType

class StockVariance(BaseDocType):
    """Individual variance record for analysis."""

    # Source
    balance_upload: str  # FK to BalanceUpload
    depot: str
    warehouse: str
    item: str

    # Quantities
    system_qty: Decimal
    reported_qty: Decimal
    variance_qty: Decimal  # reported - system
    variance_percent: Decimal

    # Value
    unit_value: Decimal
    variance_value: Decimal  # variance_qty * unit_value

    # Classification
    variance_type: str = "normal"  # normal, minor, major, suspicious

    # Anomaly flags
    anomaly_flags: list[str] = Field(default_factory=list)
    # ["CONSISTENT_SHORTFALL", "ROUND_NUMBER", "HIGH_VALUE"]

    # Resolution
    status: str = "Open"  # Open, Acknowledged, Investigated, Resolved
    resolution_notes: str | None = None
    resolved_by: str | None = None

    class Meta:
        table_name = "stock_variance"
        indexes = ["depot", "item", "variance_type", "status"]

8.3 Anomaly Detection Service

class AnomalyDetector:
    """Detect suspicious patterns in variances."""

    PATTERNS = [
        ("CONSISTENT_SHORTFALL", "Same item short 3+ consecutive months"),
        ("ROUND_NUMBER", "Variance is suspiciously round (100, 500, 1000)"),
        ("HIGH_VALUE", "Single variance exceeds ₹50,000"),
        ("DEPOT_CONCENTRATION", "80%+ of item's variance from one depot"),
        ("PRE_AUDIT_SPIKE", "Unusual variance before scheduled audit"),
        ("LIQUOR_CATEGORY", "Any variance in AFD category"),
    ]

    async def analyze(self, variance: StockVariance) -> list[str]:
        """Return list of anomaly flags for a variance."""
        flags = []

        # Pattern 1: Consistent shortfall (3+ months)
        history = await self.get_history(variance.item, variance.depot, months=6)
        shortfall_streak = sum(1 for v in history[-3:] if v.variance_qty < 0)
        if shortfall_streak >= 3:
            flags.append("CONSISTENT_SHORTFALL")

        # Pattern 2: Round number
        abs_qty = abs(variance.variance_qty)
        if abs_qty >= 100 and abs_qty % 100 == 0:
            flags.append("ROUND_NUMBER")

        # Pattern 3: High value
        if abs(variance.variance_value) >= 50000:
            flags.append("HIGH_VALUE")

        # Pattern 4: Depot concentration
        total_variance = await self.get_total_variance(variance.item)
        depot_variance = await self.get_depot_variance(variance.item, variance.depot)
        if total_variance > 0 and (depot_variance / total_variance) > 0.8:
            flags.append("DEPOT_CONCENTRATION")

        # Pattern 5: Liquor category (zero tolerance)
        item = await self.item_repo.get(variance.item)
        if item.category in ["AFD-I", "AFD-II"]:
            flags.append("LIQUOR_CATEGORY")

        return flags

    async def classify(self, variance: StockVariance, flags: list[str]) -> str:
        """Classify variance severity."""
        if flags:
            return "suspicious"

        threshold = await self.get_threshold(variance.item)
        abs_percent = abs(variance.variance_percent)

        if abs_percent <= threshold.qty_variance_percent:
            return "normal"
        elif abs_percent <= threshold.qty_variance_percent * 2:
            return "minor"
        else:
            return "major"

8.4 Enhanced Balance Controller

class BalanceImportController(BaseController[BalanceUpload]):

    async def on_submit(self, doc: BalanceUpload) -> None:
        """Reconcile with variance detection."""
        df = await self.parse_excel(doc.balance_file)
        detector = AnomalyDetector()

        for _, row in df.iterrows():
            system_qty = await self.stock_ledger.get_balance(
                item=row["item_code"],
                warehouse=row["warehouse_code"],
                as_of=doc.as_of_date,
            )
            reported_qty = row["balance_qty"]
            variance_qty = reported_qty - system_qty

            if variance_qty != 0:
                # Create variance record
                variance = StockVariance(
                    balance_upload=doc.id,
                    depot=doc.depot,
                    warehouse=row["warehouse_code"],
                    item=row["item_code"],
                    system_qty=system_qty,
                    reported_qty=reported_qty,
                    variance_qty=variance_qty,
                    variance_percent=(variance_qty / system_qty * 100) if system_qty else Decimal("100"),
                    unit_value=row.get("unit_value", 0),
                    variance_value=variance_qty * row.get("unit_value", 0),
                )

                # Detect anomalies
                flags = await detector.analyze(variance)
                variance.anomaly_flags = flags
                variance.variance_type = await detector.classify(variance, flags)

                await self.variance_repo.save(variance)

                # Auto-escalate suspicious variances
                if variance.variance_type == "suspicious":
                    await self.escalate_to_audit(variance)

8.5 Demand Variance (for Prediction)

class DemandVariance(BaseDocType):
    """Track demand vs actual consumption for prediction."""

    period: str  # "2024-01"
    depot: str
    item: str

    # Demand
    demanded_qty: Decimal

    # Actual (from deliveries/consumption)
    actual_qty: Decimal

    # Variance
    variance_qty: Decimal  # actual - demanded
    variance_percent: Decimal

    # Trend
    is_over_demand: bool  # Demanded more than consumed
    is_under_demand: bool  # Demanded less than consumed

    class Meta:
        table_name = "demand_variance"

---

9. Variance Dashboard

9.1 Dashboard Components

┌────────────────────────────────────────────────────────────────┐
│  📊 Variance Dashboard - January 2024                          │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│  Summary:  ✅ Normal: 89%  ⚠️ Minor: 8%  🔴 Major: 2%  🚨 Suspicious: 1% │
│                                                                │
│  ┌──────────────────────────────────────────────────────────┐ │
│  │  🚨 REQUIRING INVESTIGATION                              │ │
│  ├──────────────────────────────────────────────────────────┤ │
│  │  Item              Depot        Variance   Flags         │ │
│  │  ─────────────────────────────────────────────────────── │ │
│  │  Whisky Royal      Mumbai       -₹2.3L     CONSISTENT,   │ │
│  │                                            LIQUOR        │ │
│  │  Cooking Oil 5L    Delhi        -500 units ROUND_NUMBER  │ │
│  │  Samsung TV        Chennai      -₹1.8L     HIGH_VALUE    │ │
│  └──────────────────────────────────────────────────────────┘ │
│                                                                │
│  ┌──────────────────────────────────────────────────────────┐ │
│  │  📈 DEPOT PERFORMANCE                                    │ │
│  ├──────────────────────────────────────────────────────────┤ │
│  │  Pune         ████████████████████ 0.3%   ✅              │ │
│  │  Bangalore    ███████████████████  0.5%   ✅              │ │
│  │  Kolkata      ██████████████       1.8%   ⚠️              │ │
│  │  Lucknow      ████████             4.2%   🚨 ← Audit      │ │
│  └──────────────────────────────────────────────────────────┘ │
│                                                                │
│  ┌──────────────────────────────────────────────────────────┐ │
│  │  📉 DEMAND PREDICTION (Based on Variance Trends)         │ │
│  ├──────────────────────────────────────────────────────────┤ │
│  │  ▼ Over-demanded (reduce next PO): Rice 10kg, Dal 5kg   │ │
│  │  ▲ Under-demanded (increase PO): Cooking Oil, Sugar     │ │
│  └──────────────────────────────────────────────────────────┘ │
└────────────────────────────────────────────────────────────────┘

9.2 Reports

• Suspicious Variance Report: All items flagged for investigation
• Depot Variance Trend: 6-month trend per depot
• Item Shrinkage Report: Items with consistent negative variance
• Demand Accuracy Report: Predicted vs actual for planning

---

10. Verification Checklist

• Depot can download demand template
• Depot can upload filled Excel
• Validation catches bad URC/Item codes
• Error log downloadable
• Material Requests created on submit
• Balance reconciliation finds variances
• Variance thresholds configurable per item group
• Anomaly detection flags suspicious patterns
• Suspicious variances auto-escalate to audit
• Variance dashboard shows depot performance
• Stock Reconciliation created for adjustments
• Delivery confirmation updates GRN status
• Large files processed in background

---

11. Time Estimate

Task	Duration
DemandUpload + Controller	0.5 day
BalanceUpload + Reconciliation	0.5 day
Variance Detection + Anomaly	0.5 day
DeliveryConfirmation	0.25 day
Excel templates	0.25 day
Frontend (upload + dashboard)	0.75 day
Total	2.75 days