src/server/simulation/economic.py

# src/server/simulation/economic.py
"""

Economic model for AquaGuard-RL.



Simulates farmer income, poverty dynamics, MSP price evolution,

and market demand fluctuations.



References:

    - NSSO Situation Assessment Survey 2021: https://mospi.gov.in/web/mospi

    - CACP MSP schedule: https://cacp.dacnet.nic.in

    - Log-normal income distribution: calibrated from NSSO household survey data

"""

from __future__ import annotations

import math
import logging
import random
from typing import Dict, Optional, Tuple

logger = logging.getLogger(__name__)

# Average holding size per farm household (operational holding, Census 2011)
AVG_FARM_SIZE_HA = 1.8

# Groundwater pumping energy cost (INR per m³ equivalent)
PUMP_COST_INR_PER_M3 = 3.5

# Log-normal sigma for income distribution (calibrated from NSSO 2021)
INCOME_DISTRIBUTION_SIGMA = 0.45

# Rural poverty line (NSSO 2021, inflation-adjusted)
POVERTY_LINE_INR = 125000

# MSP random walk parameters
MSP_ANNUAL_DRIFT = 0.04       # ~4% annual increase (historical trend)
MSP_VOLATILITY = 0.05         # ±5% standard deviation

# Market demand parameters
MARKET_DEMAND_MEAN_REVERSION = 0.15   # Speed of reversion to 1.0
MARKET_DEMAND_VOLATILITY = 0.08       # Noise amplitude


class EconomicModel:
    """

    Simulates the agricultural economic system including:

    - Farm household gross income from crop sales

    - Input cost deductions (seeds, fertilizers, labor, irrigation)

    - MSP price evolution (random walk with drift)

    - Market demand fluctuations (mean-reverting process)

    - Poverty fraction estimation via log-normal income distribution

    """

    def __init__(self) -> None:
        """Initialize economic model with baseline MSP prices and market state."""
        self.current_msp: Dict[str, float] = {}
        self._market_demand: Dict[str, float] = {}
        self.last_income_result: Tuple[float, float] = (0.0, 0.0)  # (avg_income, poverty_frac)

    def initialize(self, crop_data: Dict[str, Dict]) -> None:
        """

        Initialize MSP prices and market demand from crop constants.



        Args:

            crop_data: Crop data dictionary from constants.py.

        """
        for crop_id, crop_def in crop_data.items():
            self.current_msp[crop_id] = crop_def["msp_inr_per_ton"]
            self._market_demand[crop_id] = 1.0  # Start at balanced demand

        logger.debug(f"Economic model initialized with {len(self.current_msp)} crops")

    def update_msp_prices(self, year: int) -> None:
        """

        Update MSP prices with random walk + annual drift.



        MSP prices follow a log-normal random walk with:

        - Annual drift of ~4% (historical GoI trend)

        - Seasonal volatility of ±5%



        Args:

            year: Current simulation year (for annual adjustment).

        """
        for crop_id in self.current_msp:
            # Log-normal increment: drift + noise
            log_change = MSP_ANNUAL_DRIFT / 3 + random.gauss(0, MSP_VOLATILITY / 3)
            multiplier = math.exp(log_change)
            self.current_msp[crop_id] = self.current_msp[crop_id] * multiplier

    def update_market_demand(self, current_demand: float) -> float:
        """

        Update market demand with mean-reverting random walk (Ornstein-Uhlenbeck).



        The process reverts toward 1.0 (balanced supply-demand) with noise.



        Args:

            current_demand: Current market demand index for a crop.



        Returns:

            Updated market demand index, clamped to [0.3, 2.0].

        """
        # Mean-reverting: ΔD = κ(μ - D) + σε
        kappa = MARKET_DEMAND_MEAN_REVERSION
        sigma = MARKET_DEMAND_VOLATILITY
        mean = 1.0

        delta = kappa * (mean - current_demand) + sigma * random.gauss(0, 1)
        new_demand = current_demand + delta
        return max(0.3, min(2.0, new_demand))

    def compute_farmer_income(

        self,

        crop_states: Dict[str, Dict],

        zone_states: Dict[str, Dict],

        zone_data: Dict[str, Dict],

        crop_data: Dict[str, Dict],

    ) -> Tuple[float, float]:
        """

        Compute average farm household income and poverty fraction.



        Income Model (per average 1.8-ha farm household):

            gross_income = Σ crops (allocated_fraction × farm_size × yield × msp × subsidy)

            water_cost = extraction_volume × pump_cost

            net_income = gross_income - input_costs - water_cost



        Poverty fraction estimated via log-normal CDF:

            P(income < poverty_line) = Φ((ln(poverty_line) - ln(net_income)) / σ)



        Args:

            crop_states: Current crop state dictionaries.

            zone_states: Current zone state dictionaries (for water extraction).

            zone_data: Static zone parameter dictionaries.

            crop_data: Static crop parameter dictionaries.



        Returns:

            Tuple of (average_income_inr, poverty_fraction).

        """
        if not self.current_msp:
            # Model not initialized — return reasonable defaults
            return (150000.0, 0.30)

        total_land_ha = sum(z["arable_land_ha"] for z in zone_states.values())
        total_farmers = sum(zd["farmer_households"] for zd in zone_data.values())

        if total_farmers <= 0:
            return (0.0, 1.0)

        avg_farm_size = total_land_ha / total_farmers

        # Gross income from crop sales
        gross_income = 0.0
        total_input_costs = 0.0

        for crop_id, crop_state in crop_states.items():
            alloc = crop_state.get("allocated_fraction", 0.0)
            if alloc < 0.001:
                continue

            crop_def = crop_data.get(crop_id, {})
            yield_t_per_ha = crop_state.get("yield_t_per_ha", crop_def.get("base_yield_t_per_ha", 0))
            msp = self.current_msp.get(crop_id, crop_def.get("msp_inr_per_ton", 0))
            subsidy_mult = crop_state.get("subsidy_multiplier", 1.0)
            market_demand = crop_state.get("market_demand_index", 1.0)
            input_cost_per_ha = crop_def.get("input_cost_inr_per_ha", 30000)

            # Effective price: MSP adjusted by subsidy and market demand
            effective_price = msp * subsidy_mult * min(1.5, max(0.8, market_demand))

            crop_land = alloc * avg_farm_size
            gross_income += crop_land * yield_t_per_ha * effective_price
            total_input_costs += crop_land * input_cost_per_ha

        # Water pumping cost (proportional to average actual extraction applied)
        avg_water_extracted_m = sum(z.get("actual_extracted_m", 0) for z in zone_states.values()) / max(len(zone_states), 1)
        # avg_water_extracted_m is volume per unit area. total volume = depth (m) * farm size (m²)
        water_volume_m3 = avg_water_extracted_m * avg_farm_size * 10000
        water_cost = water_volume_m3 * PUMP_COST_INR_PER_M3

        # Net income (per season)
        net_income = max(10000.0, gross_income - total_input_costs - water_cost)

        # Poverty fraction via log-normal CDF
        # Compare annualized net income against the annual poverty line
        annualized_income = net_income * 3.0
        poverty_fraction = self._lognormal_poverty_fraction(annualized_income)

        self.last_income_result = (annualized_income, poverty_fraction)

        logger.debug(
            f"Farmer income: gross={gross_income:.0f} - costs={total_input_costs:.0f} "
            f"- water={water_cost:.0f} = net={net_income:.0f} INR | "
            f"poverty_frac={poverty_fraction:.3f}"
        )
        return net_income, poverty_fraction

    def _lognormal_poverty_fraction(self, mean_income: float) -> float:
        """

        Estimate fraction of farmers below poverty line using log-normal distribution.



        The income distribution is modeled as log-normal with:

        - Location parameter: ln(mean_income) - σ²/2 (to maintain mean)

        - Scale parameter: σ = 0.45 (NSSO 2021 calibration)



        Args:

            mean_income: Average farm household income in INR.



        Returns:

            Fraction of farmers with income below poverty line [0, 1].

        """
        if mean_income <= 0:
            return 1.0

        sigma = INCOME_DISTRIBUTION_SIGMA
        poverty_line = POVERTY_LINE_INR

        # Log-normal: if X ~ LN(μ, σ²), then E[X] = exp(μ + σ²/2)
        # So μ = ln(mean_income) - σ²/2
        mu = math.log(max(mean_income, 1.0)) - (sigma ** 2) / 2

        # P(X < poverty_line) = Φ((ln(poverty_line) - μ) / σ)
        z = (math.log(poverty_line) - mu) / sigma

        # Standard normal CDF approximation (Abramowitz & Stegun)
        return self._standard_normal_cdf(z)

    @staticmethod
    def _standard_normal_cdf(z: float) -> float:
        """

        Compute standard normal CDF using math.erfc.



        Args:

            z: Z-score.



        Returns:

            CDF value in [0, 1].

        """
        return 0.5 * math.erfc(-z / math.sqrt(2))