udyamlinc-api/src/main/java/com/lincfusion/udyamlinc/features/FeatureEngineeringService.java

package com.lincfusion.udyamlinc.features;

import com.lincfusion.udyamlinc.enterprise.*;
import com.lincfusion.udyamlinc.ingestion.*;
import com.lincfusion.udyamlinc.master.SectorCommodity;
import com.lincfusion.udyamlinc.master.SectorCommodityRepository;
import com.lincfusion.udyamlinc.risk.ForecastPoint;
import com.lincfusion.udyamlinc.risk.ForecastPointRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Turns signals into indicators. The only writer of the feature store, and the
 * only path by which anything reaches the model — ingestion writes signals, this
 * writes features, the AI layer consumes features and nothing else. That is what
 * lets the model be retrained or replaced without touching ingestion or the API.
 *
 * <p>Every read here is point-in-time: each repository query filters on both
 * {@code validForDate <= asOfDate} and {@code observedAt <= observedAsOf}, so
 * recomputing last month's snapshot uses only what was actually knowable last
 * month. Without that, a backtest looks excellent and production does not
 * reproduce it.
 *
 * <p>A feature that cannot be computed is written as null with a reason, not
 * skipped. "We did not know" and "we never looked" are different facts, and the
 * forecast widens honestly only if the first one is recorded.
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class FeatureEngineeringService {

    private final EnterpriseRepository enterpriseRepository;
    private final SectorCommodityRepository sectorCommodityRepository;
    private final MarketSignalRepository marketSignalRepository;
    private final ClimateSignalRepository climateSignalRepository;
    private final LedgerEntryRepository ledgerRepository;
    private final LoanRecordRepository loanRepository;
    private final UpiSummaryRepository upiRepository;
    private final ForecastPointRepository forecastPointRepository;
    private final FeatureValueRepository valueRepository;
    private final FeatureSnapshotRepository snapshotRepository;
    private final FeatureCatalog catalog;

    /** Weeks of price history averaged for a "vs trailing mean" change. */
    @Value("${udyam.features.price-window-weeks:8}")
    private int priceWindowWeeks;

    /** Recent window for ledger and climate aggregates. */
    @Value("${udyam.features.recent-window-weeks:4}")
    private int recentWindowWeeks;

    @Value("${udyam.features.climate-window-weeks:8}")
    private int climateWindowWeeks;

    /** Minimum observations before a z-score is meaningful rather than noise. */
    @Value("${udyam.features.min-zscore-samples:6}")
    private int minZScoreSamples;

    /** Whether {@code dscr_p10} may read forecast rows the seeder wrote. See {@code dscrP10}. */
    @Value("${udyam.forecast.trust-seeded:true}")
    private boolean trustSeededForecasts;

    // ── entry points ────────────────────────────────────────

    /** One pass over the whole book. Consent-revoked enterprises are skipped, not computed. */
    public Summary computeAll(LocalDate asOfDate, Long scoringRunId) {
        catalog.seed();
        LocalDateTime observedAsOf = LocalDateTime.now();
        int computed = 0, skipped = 0, missing = 0;

        for (Enterprise enterprise : enterpriseRepository.findAll()) {
            if (Boolean.FALSE.equals(enterprise.getConsentActive())) {
                skipped++;
                continue;
            }
            FeatureSnapshot snapshot = computeFor(enterprise, asOfDate, observedAsOf, scoringRunId);
            computed++;
            missing += snapshot.getMissingCount() == null ? 0 : snapshot.getMissingCount();
        }

        String message = "%d snapshots at %s · %d skipped for consent · %d null feature values"
                .formatted(computed, asOfDate, skipped, missing);
        log.info("Feature engineering — {}", message);
        return new Summary(computed, skipped, missing, message);
    }

    /** The eleven catalogue features for one enterprise, tall rows and a wide snapshot. */
    public FeatureSnapshot computeFor(Enterprise enterprise, LocalDate asOfDate,
                                      LocalDateTime observedAsOf, Long scoringRunId) {

        Map<String, Computed> features = new LinkedHashMap<>();

        List<SectorCommodity> inputs = sectorCommodityRepository.findBySectorIdAndRole(enterprise.getSectorId(), "INPUT");
        List<SectorCommodity> outputs = sectorCommodityRepository.findBySectorIdAndRole(enterprise.getSectorId(), "OUTPUT");

        Integer marketStaleness = marketStaleness(enterprise.getDistrictId(), asOfDate);

        Computed feedCost = weightedPriceChange(enterprise.getDistrictId(), inputs, asOfDate, observedAsOf, marketStaleness);
        Computed outputPrice = weightedPriceChange(enterprise.getDistrictId(), outputs, asOfDate, observedAsOf, marketStaleness);

        features.put(FeatureCatalog.FEED_COST_CHANGE, feedCost);
        features.put(FeatureCatalog.OUTPUT_PRICE_CHANGE, outputPrice);
        features.put(FeatureCatalog.MARGIN_SQUEEZE, marginSqueeze(feedCost, outputPrice, marketStaleness));
        features.put(FeatureCatalog.PRICE_Z_SCORE,
                priceZScore(enterprise.getDistrictId(), outputs, asOfDate, observedAsOf, marketStaleness));
        features.put(FeatureCatalog.RAINFALL_ANOMALY, rainfallAnomaly(enterprise.getDistrictId(), asOfDate, observedAsOf));
        features.put(FeatureCatalog.HEAT_ANOMALY, heatAnomaly(enterprise.getDistrictId(), asOfDate, observedAsOf));
        features.put(FeatureCatalog.CASH_FLOW_CHANGE, cashFlowChange(enterprise.getEnterpriseId(), asOfDate));
        features.put(FeatureCatalog.EMI_DELAY, emiDelay(enterprise.getEnterpriseId(), asOfDate));
        features.put(FeatureCatalog.INVENTORY_DAYS, inventoryDays(enterprise.getEnterpriseId(), asOfDate));
        features.put(FeatureCatalog.UPI_INFLOW_CHANGE, upiInflowChange(enterprise.getEnterpriseId(), asOfDate, observedAsOf));
        features.put(FeatureCatalog.FEED_STALENESS_DAYS, marketStaleness == null
                ? Computed.missing(FeatureCatalog.NO_SIGNAL)
                : Computed.of(BigDecimal.valueOf(marketStaleness), marketStaleness));
        features.put(FeatureCatalog.DAYS_SINCE_LAST_ENTRY, daysSinceLastEntry(enterprise.getEnterpriseId(), asOfDate));
        features.put(FeatureCatalog.EMI_TO_INFLOW, emiToInflow(enterprise.getEnterpriseId(), asOfDate));
        features.put(FeatureCatalog.DSCR_P10, dscrP10(enterprise.getEnterpriseId(), asOfDate));

        return persist(enterprise.getEnterpriseId(), asOfDate, observedAsOf, scoringRunId, features);
    }

    // ── price derivations ───────────────────────────────────

    /**
     * A level is not a feature. This is the modal price against its trailing mean,
     * averaged across the sector's commodities for that role and weighted by how
     * much of the trade each one actually is.
     */
    private Computed weightedPriceChange(Long districtId, List<SectorCommodity> commodities,
                                         LocalDate asOfDate, LocalDateTime observedAsOf, Integer staleness) {
        if (commodities.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);

        BigDecimal weightedSum = BigDecimal.ZERO;
        int totalWeight = 0;
        boolean anyHistory = false;
        boolean anyRows = false;

        for (SectorCommodity link : commodities) {
            List<MarketSignal> series = marketSignalRepository.findTrailing(
                    districtId, link.getCommodityId(), asOfDate, observedAsOf);
            if (!series.isEmpty()) anyRows = true;
            BigDecimal change = changeVsTrailingMean(series);
            if (change == null) continue;
            anyHistory = true;
            int weight = link.getWeight() == null || link.getWeight() <= 0 ? 1 : link.getWeight();
            weightedSum = weightedSum.add(change.multiply(BigDecimal.valueOf(weight)));
            totalWeight += weight;
        }

        if (!anyHistory || totalWeight == 0) {
            // "Nobody has ever sent us a price" and "the series has started but is
            // still too short to mean anything" are different facts, and only the
            // second one gets better on its own. A newly onboarded manual feed
            // reporting NO_SIGNAL reads as a broken integration for the weeks it
            // spends filling up.
            return Computed.missing(anyRows
                    ? FeatureCatalog.INSUFFICIENT_HISTORY
                    : FeatureCatalog.NO_SIGNAL);
        }
        return Computed.of(weightedSum.divide(BigDecimal.valueOf(totalWeight), 4, RoundingMode.HALF_UP), staleness);
    }

    /** (latest − mean of the prior window) / mean. Null until there is enough history to mean anything. */
    private BigDecimal changeVsTrailingMean(List<MarketSignal> descendingByWeek) {
        List<BigDecimal> modal = descendingByWeek.stream()
                .map(MarketSignal::getModalPrice)
                .filter(Objects::nonNull)
                .toList();
        if (modal.size() < 4) return null;

        BigDecimal latest = modal.get(0);
        List<BigDecimal> prior = modal.subList(1, Math.min(modal.size(), priceWindowWeeks + 1));
        BigDecimal mean = mean(prior);
        if (mean == null || mean.signum() == 0) return null;
        return latest.subtract(mean).divide(mean, 4, RoundingMode.HALF_UP);
    }

    /** A feed rise matched by an output rise is not a squeeze; the difference is. */
    private Computed marginSqueeze(Computed feedCost, Computed outputPrice, Integer staleness) {
        if (feedCost.value() == null || outputPrice.value() == null) {
            return Computed.missing(FeatureCatalog.NO_SIGNAL);
        }
        return Computed.of(feedCost.value().subtract(outputPrice.value()), staleness);
    }

    /**
     * Seasonal z-score on the sale-side commodity: this week against the same week
     * of the year across all held history, ±2 weeks for sample size. Prices are
     * seasonal, so comparing March with the annual mean would call every harvest a
     * shock.
     */
    private Computed priceZScore(Long districtId, List<SectorCommodity> outputs,
                                 LocalDate asOfDate, LocalDateTime observedAsOf, Integer staleness) {
        if (outputs.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);

        SectorCommodity primary = outputs.stream()
                .max(Comparator.comparing(sc -> sc.getWeight() == null ? 0 : sc.getWeight()))
                .orElse(outputs.get(0));

        List<MarketSignal> series = marketSignalRepository.findTrailing(
                districtId, primary.getCommodityId(), asOfDate, observedAsOf);
        if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);

        MarketSignal latest = series.get(0);
        if (latest.getModalPrice() == null) return Computed.missing(FeatureCatalog.NO_SIGNAL);
        int week = Weeks.isoWeek(latest.getWeekStart());

        List<BigDecimal> seasonal = series.stream()
                .skip(1)
                .filter(s -> s.getModalPrice() != null)
                .filter(s -> weekDistance(Weeks.isoWeek(s.getWeekStart()), week) <= 2)
                .map(MarketSignal::getModalPrice)
                .toList();

        if (seasonal.size() < minZScoreSamples) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        BigDecimal mean = mean(seasonal);
        BigDecimal stdev = stdev(seasonal, mean);
        if (stdev == null || stdev.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        return Computed.of(latest.getModalPrice().subtract(mean).divide(stdev, 4, RoundingMode.HALF_UP), staleness);
    }

    /** ISO weeks wrap, so week 52 and week 1 are two apart, not fifty-one. */
    private int weekDistance(int a, int b) {
        int raw = Math.abs(a - b);
        return Math.min(raw, 53 - raw);
    }

    // ── climate derivations ─────────────────────────────────

    private Computed rainfallAnomaly(Long districtId, LocalDate asOfDate, LocalDateTime observedAsOf) {
        List<ClimateSignal> series = climateSignalRepository.findTrailing(districtId, asOfDate, observedAsOf);
        if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);

        List<ClimateSignal> window = series.subList(0, Math.min(series.size(), climateWindowWeeks));
        Integer staleness = staleness(window.get(0).getValidForDate(), asOfDate);

        BigDecimal observed = BigDecimal.ZERO, normal = BigDecimal.ZERO;
        boolean anyNormal = false;
        for (ClimateSignal week : window) {
            if (week.getRainfallMm() != null) observed = observed.add(week.getRainfallMm());
            if (week.getRainfallNormalMm() != null) {
                normal = normal.add(week.getRainfallNormalMm());
                anyNormal = true;
            }
        }
        if (!anyNormal || normal.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        BigDecimal pct = observed.subtract(normal)
                .multiply(BigDecimal.valueOf(100))
                .divide(normal, 4, RoundingMode.HALF_UP);
        return Computed.of(pct, staleness);
    }

    private Computed heatAnomaly(Long districtId, LocalDate asOfDate, LocalDateTime observedAsOf) {
        List<ClimateSignal> series = climateSignalRepository.findTrailing(districtId, asOfDate, observedAsOf);
        if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);

        List<ClimateSignal> window = series.subList(0, Math.min(series.size(), climateWindowWeeks));
        Integer staleness = staleness(window.get(0).getValidForDate(), asOfDate);

        List<BigDecimal> anomalies = window.stream()
                .map(ClimateSignal::getHeatAnomalyC)
                .filter(Objects::nonNull)
                .toList();
        if (anomalies.isEmpty()) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        return Computed.of(mean(anomalies), staleness);
    }

    // ── ledger derivations ──────────────────────────────────

    /**
     * Net position over the recent weeks against the mean of the weeks before it.
     * Entries are filtered on {@code entryDate}, not sync time — an entry made
     * offline three weeks ago describes three weeks ago.
     */
    private Computed cashFlowChange(String enterpriseId, LocalDate asOfDate) {
        List<LedgerEntry> entries = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
                .filter(e -> !e.getEntryDate().isAfter(asOfDate))
                .toList();
        if (entries.isEmpty()) return Computed.missing(FeatureCatalog.NO_LEDGER);

        Integer staleness = staleness(entries.get(0).getEntryDate(), asOfDate);
        LocalDate recentFrom = asOfDate.minusWeeks(recentWindowWeeks);
        LocalDate priorFrom = recentFrom.minusWeeks(priceWindowWeeks);

        BigDecimal recent = net(entries, recentFrom, asOfDate);
        BigDecimal prior = net(entries, priorFrom, recentFrom);
        if (prior.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        // Both windows normalised to a weekly rate before comparison.
        BigDecimal recentRate = recent.divide(BigDecimal.valueOf(recentWindowWeeks), 4, RoundingMode.HALF_UP);
        BigDecimal priorRate = prior.divide(BigDecimal.valueOf(priceWindowWeeks), 4, RoundingMode.HALF_UP);
        if (priorRate.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        BigDecimal change = recentRate.subtract(priorRate).divide(priorRate.abs(), 4, RoundingMode.HALF_UP);
        return Computed.of(change, staleness);
    }

    private BigDecimal net(List<LedgerEntry> entries, LocalDate from, LocalDate to) {
        BigDecimal sum = BigDecimal.ZERO;
        for (LedgerEntry e : entries) {
            if (e.getEntryDate().isBefore(from) || !e.getEntryDate().isBefore(to)) continue;
            if ("IN".equals(e.getKind())) sum = sum.add(e.getAmount());
            else if ("OUT".equals(e.getKind())) sum = sum.subtract(e.getAmount());
        }
        return sum;
    }

    /** Stock on hand against sales velocity. Rising inventory days is cash turning into unsold goods. */
    private Computed inventoryDays(String enterpriseId, LocalDate asOfDate) {
        LocalDate from = asOfDate.minusWeeks(recentWindowWeeks);
        List<LedgerEntry> entries = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
                .filter(e -> !e.getEntryDate().isAfter(asOfDate) && !e.getEntryDate().isBefore(from))
                .toList();
        if (entries.isEmpty()) return Computed.missing(FeatureCatalog.NO_LEDGER);

        Integer staleness = staleness(entries.get(0).getEntryDate(), asOfDate);
        BigDecimal stock = BigDecimal.ZERO, sales = BigDecimal.ZERO;
        for (LedgerEntry e : entries) {
            if ("STOCK".equals(e.getKind())) stock = stock.add(e.getAmount());
            else if ("IN".equals(e.getKind())) sales = sales.add(e.getAmount());
        }
        if (sales.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        long days = ChronoUnit.DAYS.between(from, asOfDate);
        BigDecimal dailySales = sales.divide(BigDecimal.valueOf(Math.max(days, 1)), 4, RoundingMode.HALF_UP);
        if (dailySales.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        return Computed.of(stock.divide(dailySales, 4, RoundingMode.HALF_UP), staleness);
    }

    /** Silence is a signal. An enterprise that stops recording has usually stopped trading normally. */
    private Computed daysSinceLastEntry(String enterpriseId, LocalDate asOfDate) {
        Optional<LocalDate> newest = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
                .map(LedgerEntry::getEntryDate)
                .filter(date -> !date.isAfter(asOfDate))
                .max(LocalDate::compareTo);
        return newest
                .map(date -> {
                    int days = staleness(date, asOfDate);
                    return Computed.of(BigDecimal.valueOf(days), days);
                })
                .orElseGet(() -> Computed.missing(FeatureCatalog.NO_LEDGER));
    }

    // ── institutional derivations ───────────────────────────

    /**
     * The state of each open loan as of the scoring date — the newest extract per
     * {@code accountRef}, never the union of every extract.
     *
     * <p>Once the book holds monthly snapshots, taking the worst DPD across all
     * rows would report a loan that was sixty days down in March as sixty days
     * down today. The history is there to train on, not to be summarised as the
     * present.
     */
    private List<LoanRecord> currentLoans(String enterpriseId, LocalDate asOfDate) {
        return loanRepository.findByEnterpriseIdAndAsOfDateLessThanEqual(enterpriseId, asOfDate).stream()
                .collect(Collectors.toMap(
                        LoanRecord::getAccountRef,
                        loan -> loan,
                        (a, b) -> a.getAsOfDate().isAfter(b.getAsOfDate()) ? a : b))
                .values().stream()
                .filter(loan -> !Boolean.TRUE.equals(loan.getClosed()))
                .toList();
    }

    private Computed emiDelay(String enterpriseId, LocalDate asOfDate) {
        List<LoanRecord> loans = currentLoans(enterpriseId, asOfDate);
        if (loans.isEmpty()) return Computed.missing(FeatureCatalog.NO_CBS);

        LocalDate newest = loans.stream().map(LoanRecord::getAsOfDate).max(LocalDate::compareTo).orElse(asOfDate);
        int worstDpd = loans.stream().map(LoanRecord::getDpd).filter(Objects::nonNull)
                .mapToInt(Integer::intValue).max().orElse(0);
        return Computed.of(BigDecimal.valueOf(worstDpd), staleness(newest, asOfDate));
    }

    private Computed upiInflowChange(String enterpriseId, LocalDate asOfDate, LocalDateTime observedAsOf) {
        List<UpiSummary> series = upiRepository.findTrailing(enterpriseId, asOfDate, observedAsOf);
        if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_CONSENT);

        Integer staleness = staleness(series.get(0).getValidForDate(), asOfDate);
        int window = priceWindowWeeks;
        if (series.size() < window * 2) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        BigDecimal recent = sumInflow(series.subList(0, window));
        BigDecimal prior = sumInflow(series.subList(window, Math.min(series.size(), window * 2)));
        if (prior.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);

        return Computed.of(recent.subtract(prior).divide(prior, 4, RoundingMode.HALF_UP), staleness);
    }

    /**
     * What the instalment takes out of what comes in. Inflow is the ledger's own
     * money-in over the recent window, normalised to a month — the enterprise's
     * record of its trade, not an institutional estimate of it.
     */
    private Computed emiToInflow(String enterpriseId, LocalDate asOfDate) {
        List<LoanRecord> loans = currentLoans(enterpriseId, asOfDate);
        if (loans.isEmpty()) return Computed.missing(FeatureCatalog.NO_CBS);

        BigDecimal instalment = loans.stream()
                .map(LoanRecord::getInstalmentAmount)
                .filter(Objects::nonNull)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        if (instalment.signum() == 0) return Computed.missing(FeatureCatalog.NO_CBS);

        LocalDate from = asOfDate.minusWeeks(recentWindowWeeks);
        BigDecimal inflow = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
                .filter(e -> "IN".equals(e.getKind()))
                .filter(e -> !e.getEntryDate().isAfter(asOfDate) && !e.getEntryDate().isBefore(from))
                .map(LedgerEntry::getAmount)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        if (inflow.signum() == 0) return Computed.missing(FeatureCatalog.NO_LEDGER);

        long days = Math.max(ChronoUnit.DAYS.between(from, asOfDate), 1);
        BigDecimal monthlyInflow = inflow.divide(BigDecimal.valueOf(days), 4, RoundingMode.HALF_UP)
                .multiply(BigDecimal.valueOf(30));
        if (monthlyInflow.signum() == 0) return Computed.missing(FeatureCatalog.NO_LEDGER);

        LocalDate newest = loans.stream().map(LoanRecord::getAsOfDate).max(LocalDate::compareTo).orElse(asOfDate);
        return Computed.of(instalment.divide(monthlyInflow, 4, RoundingMode.HALF_UP), staleness(newest, asOfDate));
    }

    /**
     * The product's entire trigger, as a number: the worst month of the horizon
     * where the downside case has to cover what is due. Read from the latest
     * scoring run's stored quantile path, so it reflects the engine that actually
     * produced the forecast an officer saw.
     */
    private Computed dscrP10(String enterpriseId, LocalDate asOfDate) {
        List<ForecastPoint> points = forecastPointRepository.findByEnterpriseIdOrderByHorizonAsc(enterpriseId);

        // Seeded rows carry no engine. Once a real forecast engine exists, a
        // demo path and a computed one are indistinguishable downstream — the
        // score, the band and the alert all look equally earned. Setting this
        // false makes an enterprise nobody can forecast report NO_FORECAST
        // instead of borrowing the seeder's optimism.
        if (!trustSeededForecasts) {
            points = points.stream().filter(point -> point.getEngine() != null).toList();
        }
        if (points.isEmpty()) return Computed.missing(FeatureCatalog.NO_FORECAST);

        Long latestRun = points.stream().map(ForecastPoint::getRunId)
                .filter(Objects::nonNull).max(Long::compareTo).orElse(null);
        if (latestRun == null) return Computed.missing(FeatureCatalog.NO_FORECAST);

        BigDecimal worst = null;
        for (ForecastPoint point : points) {
            if (!latestRun.equals(point.getRunId())) continue;
            if (point.getP10() == null || point.getObligation() == null || point.getObligation().signum() == 0) continue;
            BigDecimal ratio = point.getP10().divide(point.getObligation(), 4, RoundingMode.HALF_UP);
            if (worst == null || ratio.compareTo(worst) < 0) worst = ratio;
        }
        return worst == null ? Computed.missing(FeatureCatalog.NO_FORECAST) : Computed.of(worst, 0);
    }

    private BigDecimal sumInflow(List<UpiSummary> weeks) {
        return weeks.stream()
                .map(UpiSummary::getInflowAmount)
                .filter(Objects::nonNull)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
    }

    // ── persistence ─────────────────────────────────────────

    private FeatureSnapshot persist(String enterpriseId, LocalDate asOfDate, LocalDateTime observedAsOf,
                                    Long scoringRunId, Map<String, Computed> features) {

        FeatureSnapshot snapshot = snapshotRepository.findByEnterpriseIdAndAsOfDate(enterpriseId, asOfDate)
                .orElseGet(() -> FeatureSnapshot.builder()
                        .enterpriseId(enterpriseId)
                        .asOfDate(asOfDate)
                        .build());

        int present = 0, missing = 0, maxStaleness = 0;

        for (Map.Entry<String, Computed> entry : features.entrySet()) {
            String code = entry.getKey();
            Computed computed = entry.getValue();

            FeatureValue row = valueRepository
                    .findByEnterpriseIdAndFeatureCodeAndValidForDate(enterpriseId, code, asOfDate)
                    .orElseGet(() -> FeatureValue.builder()
                            .enterpriseId(enterpriseId)
                            .featureCode(code)
                            .validForDate(asOfDate)
                            .build());
            row.setValue(computed.value());
            row.setStalenessDays(computed.stalenessDays());
            row.setMissingReason(computed.missingReason());
            row.setObservedAt(observedAsOf);
            valueRepository.save(row);

            snapshot.put(code, computed.value());
            if (computed.value() == null) missing++;
            else present++;
            if (computed.stalenessDays() != null) maxStaleness = Math.max(maxStaleness, computed.stalenessDays());
        }

        snapshot.setScoringRunId(scoringRunId);
        snapshot.setFeatureCount(present);
        snapshot.setMissingCount(missing);
        snapshot.setMaxStalenessDays(maxStaleness);
        snapshot.setComputedAt(observedAsOf);
        return snapshotRepository.save(snapshot);
    }

    // ── helpers ─────────────────────────────────────────────

    private Integer marketStaleness(Long districtId, LocalDate asOfDate) {
        return marketSignalRepository.findFirstByDistrictIdOrderByValidForDateDesc(districtId)
                .map(signal -> staleness(signal.getValidForDate(), asOfDate))
                .orElse(null);
    }

    private Integer staleness(LocalDate validFor, LocalDate asOfDate) {
        if (validFor == null) return null;
        return (int) Math.max(0, ChronoUnit.DAYS.between(validFor, asOfDate));
    }

    private BigDecimal mean(List<BigDecimal> values) {
        if (values == null || values.isEmpty()) return null;
        return values.stream().reduce(BigDecimal.ZERO, BigDecimal::add)
                .divide(BigDecimal.valueOf(values.size()), 4, RoundingMode.HALF_UP);
    }

    /** Sample standard deviation — n−1, because these are samples of a season, not the population. */
    private BigDecimal stdev(List<BigDecimal> values, BigDecimal mean) {
        if (values.size() < 2 || mean == null) return null;
        BigDecimal sumSquares = BigDecimal.ZERO;
        for (BigDecimal v : values) {
            BigDecimal diff = v.subtract(mean);
            sumSquares = sumSquares.add(diff.multiply(diff));
        }
        double variance = sumSquares.divide(BigDecimal.valueOf(values.size() - 1L), 6, RoundingMode.HALF_UP)
                .doubleValue();
        return BigDecimal.valueOf(Math.sqrt(variance)).setScale(4, RoundingMode.HALF_UP);
    }

    /**
     * One feature's outcome. A null value always carries a reason — that is the
     * difference between a hole we can explain and one we cannot.
     */
    public record Computed(BigDecimal value, Integer stalenessDays, String missingReason) {

        static Computed of(BigDecimal value, Integer stalenessDays) {
            return new Computed(value, stalenessDays, null);
        }

        static Computed missing(String reason) {
            return new Computed(null, null, reason);
        }
    }

    /** What one pass over the book did. */
    public record Summary(int snapshots, int skippedForConsent, int nullValues, String message) {}
}