Testing Policy Thresholds
On Mandi Margins

Does the IMD declaring a region as "rain-deficient" drive agricultural market behavior, or is price movement merely a continuous function of physical precipitation?

We test this by placing rain departure on the horizontal axis and modeling the discontinuity at the official -20% deficit cutoff. If the policy declaration drives market sentiment, we should observe a sharp discontinuous jump (discontinuity) in prices at the boundary, holding all physical parameters constant.

Causal RDD Specification

We estimate local linear regressions on both sides of the -20% threshold:

Where Yit is the log market price, Xit is the actual rainfall departure percentage, c = -20% is the threshold, and Dit is the indicator binary equal to 1 if rainfall departure is below -20% (deficit territory).

Our findings suggest that while physical rainfall levels strongly affect yields and continuous pricing, the administrative -20% declaration itself produces statistically insignificant jumps in most crops, with the notable exception of Onions which show active hoarding signals.

MandiIQ decouples the heavy analytical warehouse from the customer-facing dashboard. The system runs an automated ELT pipeline using DuckDB, fits locally weighted regressions, tracks models in MLflow, and exposes predictions through FastAPI.

%%{init: {"theme": "dark", "themeVariables": { "primaryColor": "#1a1a2e", "primaryTextColor": "#fff", "lineColor": "#d7ff00", "secondaryColor": "#16213e", "tertiaryColor": "#0f3460"}}}%%
flowchart TB
    subgraph Sources["Data Sources"]
        A1["Agmarknet API
Daily Mandi Prices"]
        A6["Ashoka CEDA
Historical Price CSV"]
        A2["IMD Weather Grids
Rainfall Departure"]
    end
    subgraph Ingestion["Ingestion Pipeline"]
        B1["fetch_prices.py
Pagination + Retry"]
        B5["fetch_prices.py
Variety-Wise Archive"]
        B6["ingest_historical_csv.py
CSV Backfill"]
        B2["fetch_rainfall.py
Sub-division Parser"]
        B3["http_client.py
Shared Retry + SSL"]
        B4["scheduler.py
Orchestration"]
    end
    subgraph Storage["Analytical Store"]
        C1["DuckDB
mandi_iq.duckdb"]
        C2["🔍 data_lineage
Batch Provenance · Fingerprint
Source Type · Commodity List"]
        C3["📊 freshness_by_commodity
Per-Commodity: Latest Date
Row Count · District Coverage"]
    end
    subgraph Analysis["Analysis Engine"]
        D1["Causal RDD
Local-linear + FE"]
        D2["Forecast
Prophet vs LSTM"]
        D3["Classifier
XGBoost + SHAP"]
        D4["Prescriptive
Procurement Advice"]
    end
    subgraph AI["AI Orchestrator"]
        E1["Router
Multi-model + Circuit Breaker"]
        E2["Nightly Narratives
Commodity Reports"]
    end
    subgraph Serving["Serving Layer"]
        F1["FastAPI Gateway
Endpoints: /health /prices
/rdd-result /forecast
/freshness /lineage"]
        F2["Streamlit Dashboard
mandiiq.streamlit.app
Data Freshness Widget"]
    end
    A1 --> B1
    A6 --> B6
    A2 --> B2
    B1 --> B3
    B5 --> B3
    B2 --> B3
    B4 --> B1
    B4 --> B5
    B4 --> B6
    B4 --> B2
    B1 --> C1
    B5 --> C1
    B6 --> C1
    B2 --> C1
    C1 --> C2
    C2 --> C3
    C1 --> D1
    C1 --> D2
    C1 --> D3
    D1 --> D4
    D2 --> D4
    D3 --> D4
    D4 --> F1
    D1 --> F1
    D2 --> F1
    D3 --> F1
    F1 --> E1
    E1 --> E2
    F1 --> F2
    %% Source-labeled edges to data_lineage
    B1 -.->|API
9ef84268| C2
    B5 -.->|Variety
35985678| C2
    B6 -.->|CSV| C2
    B2 -.->|Rainfall| C2
    C3 -.->|Dashboard Feed| F2

Scaling to 10 Million Orders per Day

To transition MandiIQ to support massive transactional data volumes (e.g. 10M transactions per day), the infrastructure expands along several dimensions:

• Ingestion: Replace batch CSV loads with a streaming broker (Apache Kafka or AWS Kinesis) feeding analytical tables in real-time.
• Analytical Storage: Migrate the single-node local DuckDB file to a highly parallel serverless cloud warehouse such as Google BigQuery or Snowflake, maintaining similar SQL aggregation syntax but scaling horizontally.
• Feature Store: Introduce Feast or Tecton to serve model features with sub-50ms latency and prevent training-serving skew.
• Serving Layer: Run the FastAPI service inside containerized Kubernetes pods with horizontal autoscaling (HPA) and a Redis caching layer for active market queries.

A robust intelligence platform must anticipate failure. MandiIQ implements built-in fallback states and active telemetry checks across the following failure boundaries:

1. Model Registry Outages

If the active MLflow registry is unreachable at API startup, the FastAPI endpoints fall back to reading a locally serialized model pickle (models/loss_classifier_fallback.pkl). The service generates an active alert while continuing to serve inferences.

2. Data Drift & Anomalous Weather Jumps

Climate variables are highly volatile. When the pipeline detects feature distribution shift exceeding 3σ (e.g. during an unprecedented extreme monsoon departure), it automatically:

1. Sends telemetry warnings to monitoring dashboards.
2. Triggers an automated retraining run on the latest 90-day window.
3. Switches the prediction endpoint to a robust, simple baseline estimator until the new model passes champion verification tests.