feat(ingestion): per-district commodity baskets, and a probe that verifies them

Commodities 7 -> 21 and a district_commodity table holding a per-district
basket of 60 pairs, replacing the district x commodity cross product that would
have been 252 mostly-impossible calls a night. District, not zone: Kolar and
Belagavi are both SOUTH and trade almost disjoint baskets.

Only the four original cereals had verified Agmarknet spellings, so the rest
carry sourceVerified = false until POST /admin/ingestion/probe confirms them.
The probe separates a wrong spelling from an absent local market by
cross-referencing which districts and commodities returned rows anywhere — a
distinction that matters because both fail silently and identically.

Three district names were costing us every price in their district: Nasik ->
Nashik, Ahmadnagar -> Ahilyanagar, Belgaum -> Belagavi. The last is the
nastiest: Belgaum still answers with 1000 rows, but they stop in May 2026 while
Belagavi is current, so a retired name serving a frozen archive reads as a
healthy feed. Only the arrival dates give it away.

data.gov.in meters by quota per window, so the limit is reached by the run and
not by any one call: a 62-pair run straight after a 62-pair probe returned 429
on all 62 and recorded the whole ingestion FAILED with zero rows. DataGovClient
now retries with a 45s backoff, long because the window is minutes not seconds.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
LincPay
2026-08-15 15:28:05 +05:30
co-authored by Claude Opus 5
parent 4a1fe69ff6
commit ec057a8124
11 changed files with 802 additions and 17 deletions
@@ -0,0 +1,153 @@
package com.lincfusion.udyamlinc.ingestion;
import com.lincfusion.udyamlinc.ingestion.client.DataGovClient;
import com.lincfusion.udyamlinc.ingestion.client.MandiRecord;
import com.lincfusion.udyamlinc.master.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.util.*;
/**
* Asks the live feed one question per basket pair: does this spelling return
* anything, and if not, whose fault is it likely to be.
*
* <p>This exists because of how Agmarknet fails. A misspelt commodity, a
* district renamed on our side but not theirs, and a market that genuinely had
* no arrivals all produce the same response — HTTP 200, empty record list. A
* nightly run folds all three into one number and the console shows a feed that
* looks healthy while a commodity quietly contributes nothing.
*
* <p>The probe separates them using two facts the ingestion loop does not have:
* whether this commodity has <em>ever</em> returned rows anywhere (so the
* spelling is known good), and whether this district has returned rows for
* anything else (so the district filter is known good). A pair that is empty
* while both of its axes work elsewhere is a real market fact. A commodity
* empty in every district it is basketed in is almost certainly our spelling.
*
* <p>It writes nothing but the verified flag. Deliberately: a probe that
* ingested would make "check the spelling" and "load a year of prices" the same
* button, and the first needs to be cheap enough to run whenever a basket
* changes.
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class BasketProbeService {
/** One week back is enough to prove a filter matches; the goal is not history. */
private static final int PROBE_WEEKS = 8;
private final DataGovClient client;
private final DistrictRepository districtRepository;
private final CommodityRepository commodityRepository;
private final DistrictCommodityRepository districtCommodityRepository;
public record PairResult(String districtCode, String district, String commodityCode,
String sourceCode, boolean benchmark, boolean wasVerified,
int rows, LocalDate newestArrival, String verdict) {
}
public record ProbeReport(boolean configured, int pairs, int returningRows, int empty, int failed,
List<String> suspectCommodities, List<String> suspectDistricts,
int newlyVerified, List<PairResult> results) {
}
/**
* @param commodityCode probe only this commodity, or null for the whole basket set
* @param persist write {@code sourceVerified = true} for commodities that
* returned rows. Off by default — a probe should be safe to
* run against production without changing what it measures.
*/
public ProbeReport probe(String commodityCode, boolean persist) {
if (!client.configured()) {
return new ProbeReport(false, 0, 0, 0, 0, List.of(), List.of(), 0, List.of());
}
Map<Long, Commodity> commodities = new HashMap<>();
commodityRepository.findAll().forEach(c -> commodities.put(c.getId(), c));
List<District> districts = districtRepository.findByActiveTrue();
List<PairResult> results = new ArrayList<>();
int failed = 0;
for (District district : districts) {
for (DistrictCommodity dc : districtCommodityRepository.findByDistrictIdAndActiveTrue(district.getId())) {
Commodity commodity = commodities.get(dc.getCommodityId());
if (commodity == null || !DataGovClient.SOURCE.equals(commodity.getSource())) continue;
if (commodityCode != null && !commodityCode.equalsIgnoreCase(commodity.getCode())) continue;
String sourceCode = commodity.getSourceCode() == null
? commodity.getName() : commodity.getSourceCode();
List<MandiRecord> records = client.fetchPrices(
district.feedState(), district.feedDistrict(), sourceCode, PROBE_WEEKS);
if (records == null) {
failed++;
results.add(new PairResult(district.getCode(), district.getName(), commodity.getCode(),
sourceCode, Boolean.TRUE.equals(dc.getBenchmark()),
Boolean.TRUE.equals(commodity.getSourceVerified()), 0, null, "CALL_FAILED"));
continue;
}
LocalDate newest = records.stream().map(MandiRecord::arrivalDate)
.filter(Objects::nonNull).max(LocalDate::compareTo).orElse(null);
results.add(new PairResult(district.getCode(), district.getName(), commodity.getCode(),
sourceCode, Boolean.TRUE.equals(dc.getBenchmark()),
Boolean.TRUE.equals(commodity.getSourceVerified()),
records.size(), newest, records.isEmpty() ? "EMPTY" : "ROWS"));
}
}
// A commodity that came back empty in every district that trades it, on a
// run where other commodities in those same districts returned rows. The
// district axis is therefore working, which leaves the spelling.
Set<String> workingDistricts = new HashSet<>();
Set<String> workingCommodities = new HashSet<>();
for (PairResult r : results) {
if (r.rows() > 0) {
workingDistricts.add(r.districtCode());
workingCommodities.add(r.commodityCode());
}
}
List<String> suspectCommodities = results.stream()
.map(PairResult::commodityCode).distinct()
.filter(c -> !workingCommodities.contains(c))
.filter(c -> results.stream()
.filter(r -> r.commodityCode().equals(c))
.anyMatch(r -> workingDistricts.contains(r.districtCode())))
.sorted().toList();
// The mirror image: a district empty on everything while its commodities
// work elsewhere — the district or state spelling, not the crop.
List<String> suspectDistricts = results.stream()
.map(PairResult::districtCode).distinct()
.filter(d -> !workingDistricts.contains(d))
.filter(d -> results.stream()
.filter(r -> r.districtCode().equals(d))
.anyMatch(r -> workingCommodities.contains(r.commodityCode())))
.sorted().toList();
int newlyVerified = 0;
if (persist) {
for (String code : workingCommodities) {
Optional<Commodity> c = commodityRepository.findByCode(code);
if (c.isPresent() && !Boolean.TRUE.equals(c.get().getSourceVerified())) {
c.get().setSourceVerified(true);
commodityRepository.save(c.get());
newlyVerified++;
}
}
}
int returning = (int) results.stream().filter(r -> r.rows() > 0).count();
int empty = (int) results.stream().filter(r -> "EMPTY".equals(r.verdict())).count();
log.info("Basket probe — {} pairs, {} returned rows, {} empty, {} failed. Suspect spellings: {}",
results.size(), returning, empty, failed,
suspectCommodities.isEmpty() ? "none" : suspectCommodities);
return new ProbeReport(true, results.size(), returning, empty, failed,
suspectCommodities, suspectDistricts, newlyVerified, results);
}
}
@@ -33,6 +33,41 @@ public class IngestionController {
private final ApiLogRepository apiLogRepository;
private final DistrictRepository districtRepository;
private final CommodityRepository commodityRepository;
private final BasketProbeService basketProbeService;
private final NoFeedPriceSimulationService noFeedPriceSimulationService;
/**
* Three years of weekly prices for milk, egg, fodder and sugarcane — the
* commodities no API publishes. Rows carry {@code source = SIMULATED} and
* never collide with a real MANUAL or AGMARKNET row, because the upsert key
* includes the source.
*/
@PostMapping("/simulate/no-feed")
public ApiResponse<NoFeedPriceSimulationService.SimulationResult> simulateNoFeed() {
NoFeedPriceSimulationService.SimulationResult result = noFeedPriceSimulationService.simulate();
return ApiResponse.ok(result.message(), result);
}
/**
* One call per basket pair, reporting which spellings actually match the
* feed. A write only in the {@code persist=true} sense — it sets the
* verified flag and ingests nothing.
*
* <p>Run this after changing a basket or adding a commodity, and before
* reading anything into a night of empty results. Under {@code /admin} it is
* supervisor-only by the existing write rule.
*/
@PostMapping("/probe")
public ApiResponse<BasketProbeService.ProbeReport> probe(
@RequestParam(required = false) String commodity,
@RequestParam(defaultValue = "false") boolean persist) {
BasketProbeService.ProbeReport report = basketProbeService.probe(commodity, persist);
if (!report.configured()) {
return ApiResponse.ok("No DATAGOV_API_KEY configured — nothing probed", report);
}
return ApiResponse.ok("%d pairs probed · %d returned rows · %d suspect spellings"
.formatted(report.pairs(), report.returningRows(), report.suspectCommodities().size()), report);
}
// ── what ran ────────────────────────────────────────────
@@ -154,8 +189,18 @@ public class IngestionController {
return ApiResponse.ok("Ingestion triggered", scheduler.runAll("MANUAL"));
}
/**
* @param lookbackWeeks a one-off deep window, for the backfill that gives
* {@code price_z_score} a multi-year seasonal mean.
* {@code POST /admin/ingestion/run/AGMARKNET?lookbackWeeks=156}
* is the three-year pull. Omitted, the configured nightly
* window applies — the deep pull is many times the calls,
* so it is asked for explicitly and never scheduled.
*/
@PostMapping("/run/{source}")
public ApiResponse<IngestionRun> runOne(@PathVariable String source) {
return ApiResponse.ok("Ingestion triggered", scheduler.run(source.toUpperCase(), "MANUAL"));
public ApiResponse<IngestionRun> runOne(@PathVariable String source,
@RequestParam(required = false) Integer lookbackWeeks) {
return ApiResponse.ok("Ingestion triggered",
scheduler.run(source.toUpperCase(), "MANUAL", lookbackWeeks));
}
}
@@ -60,6 +60,15 @@ public class IngestionScheduler {
* nothing" must not look the same in the console.
*/
public IngestionRun run(String source, String trigger) {
return run(source, trigger, null);
}
/**
* @param lookbackWeeks a one-off window for this run, or {@code null} for the
* configured one. Only Agmarknet honours it; the climate
* archive is already pulled against a fixed baseline.
*/
public IngestionRun run(String source, String trigger, Integer lookbackWeeks) {
if (!running.compareAndSet(false, true)) {
return record(source, trigger, IngestionRun.SKIPPED,
new IngestionResult(0, 0, 0, 0, "An ingestion pass is already running"),
@@ -76,7 +85,7 @@ public class IngestionScheduler {
}
IngestionResult result = switch (source) {
case DataGovClient.SOURCE -> mandiService.ingest();
case DataGovClient.SOURCE -> mandiService.ingest(lookbackWeeks);
case OpenMeteoClient.SOURCE -> climateService.ingest();
default -> throw new IllegalArgumentException("Unknown ingestion source: " + source);
};
@@ -39,6 +39,7 @@ public class MandiIngestionService {
private final CommodityRepository commodityRepository;
private final MarketRepository marketRepository;
private final MarketSignalRepository signalRepository;
private final DistrictCommodityRepository districtCommodityRepository;
@Value("${udyam.ingestion.mandi.lookback-weeks:12}")
private int lookbackWeeks;
@@ -52,6 +53,20 @@ public class MandiIngestionService {
* how a nightly job takes the pool down. Each upsert commits on its own.
*/
public IngestionResult ingest() {
return ingest(null);
}
/**
* @param lookbackWeeksOverride how far back to reach for this run only, or
* {@code null} for the configured nightly window.
* A deep value is the one-off backfill that gives
* {@code price_z_score} the multi-year seasonal
* mean it needs; it is deliberately not the
* default, because paging three years of arrivals
* for every district on every nightly run would
* be several thousand calls to earn nothing.
*/
public IngestionResult ingest(Integer lookbackWeeksOverride) {
List<District> districts = districtRepository.findByActiveTrue();
List<Commodity> commodities = commodityRepository.findBySourceAndActiveTrue(DataGovClient.SOURCE);
@@ -59,18 +74,25 @@ public class MandiIngestionService {
return new IngestionResult(0, 0, 0, 0, "Nothing to pull — " + districts.size()
+ " active districts, " + commodities.size() + " Agmarknet commodities");
}
Map<Long, Commodity> byId = commodities.stream()
.collect(Collectors.toMap(Commodity::getId, c -> c));
LocalDate horizon = Weeks.startOf(LocalDate.now()).minusWeeks(lookbackWeeks);
int weeks = lookbackWeeksOverride == null || lookbackWeeksOverride <= 0
? lookbackWeeks
: lookbackWeeksOverride;
LocalDate horizon = Weeks.startOf(LocalDate.now()).minusWeeks(weeks);
LocalDateTime observedAt = LocalDateTime.now();
int fetched = 0, written = 0, skipped = 0, failed = 0;
int fetched = 0, written = 0, skipped = 0, failed = 0, pairs = 0;
for (District district : districts) {
Long marketId = marketRepository.findFirstByDistrictIdAndPrimaryMarketTrue(district.getId())
.map(Market::getId).orElse(null);
for (Commodity commodity : commodities) {
for (Commodity commodity : basketFor(district, byId, commodities)) {
pairs++;
String sourceCode = commodity.getSourceCode() == null ? commodity.getName() : commodity.getSourceCode();
List<MandiRecord> records = client.fetchPrices(district.getState(), district.getName(), sourceCode);
List<MandiRecord> records = client.fetchPrices(
district.feedState(), district.feedDistrict(), sourceCode, weeks);
if (records == null) {
failed++;
@@ -94,12 +116,38 @@ public class MandiIngestionService {
}
}
String message = "%d districts × %d commodities · %d rows fetched · %d weeks written · %d calls failed"
.formatted(districts.size(), commodities.size(), fetched, written, failed);
String message = "%d districts · %d basket pairs · %d week lookback · %d rows fetched · %d weeks written · %d calls failed"
.formatted(districts.size(), pairs, weeks, fetched, written, failed);
log.info("Agmarknet ingestion — {}", message);
return new IngestionResult(fetched, written, skipped, failed, message);
}
/**
* What this district trades, from {@code district_commodity}.
*
* <p>Falls back to every Agmarknet commodity when a district has no basket
* rows, and says so. The fallback is the old cross-product behaviour, kept
* because an unseeded basket should degrade to pulling too much rather than
* to pulling nothing — a district silently ingesting no prices at all would
* show up only as features going stale a week later, whereas over-pulling
* shows up immediately in the run's failed-call count.
*/
private List<Commodity> basketFor(District district, Map<Long, Commodity> byId, List<Commodity> all) {
List<DistrictCommodity> basket = districtCommodityRepository
.findByDistrictIdAndActiveTrue(district.getId());
if (basket.isEmpty()) {
log.warn("District {} has no basket — falling back to all {} Agmarknet commodities",
district.getCode(), all.size());
return all;
}
// A basket may name a commodity that is inactive or not Agmarknet-sourced
// (milk, fodder); byId holds only the pullable ones, so those drop out.
return basket.stream()
.map(dc -> byId.get(dc.getCommodityId()))
.filter(Objects::nonNull)
.toList();
}
private void upsert(District district, Commodity commodity, Long marketId,
LocalDate weekStart, List<MandiRecord> rows, LocalDateTime observedAt) {
@@ -51,6 +51,14 @@ public class DataGovClient {
@Value("${udyam.ingestion.datagov.page-limit:1000}")
private int pageLimit;
/**
* Hard ceiling on pages per (district, commodity), whatever the lookback asks
* for. A backstop, not a tuning knob: without it a mis-set lookback would page
* an 81-million-row resource until the key's quota ran out.
*/
@Value("${udyam.ingestion.datagov.max-pages:40}")
private int maxPages;
// Filter field names, because they must match the resource's own field names.
@Value("${udyam.ingestion.datagov.field.state:state}")
private String fieldState;
@@ -69,6 +77,12 @@ public class DataGovClient {
@Value("${udyam.ingestion.datagov.sort-field:Arrival_Date}")
private String sortField;
@Value("${udyam.ingestion.datagov.retry-attempts:3}")
private int retryAttempts;
@Value("${udyam.ingestion.datagov.retry-backoff-ms:45000}")
private long retryBackoffMs;
public DataGovClient(@Qualifier("externalFeedClient") RestClient http, ApiLogRepository apiLogRepository) {
this.http = http;
this.apiLogRepository = apiLogRepository;
@@ -88,10 +102,103 @@ public class DataGovClient {
* the run continues — a feed being down degrades tomorrow's freshness, it
* does not abort ingestion.
*/
@SuppressWarnings("unchecked")
public List<MandiRecord> fetchPrices(String state, String district, String commoditySourceCode) {
return fetchPrices(state, district, commoditySourceCode, 0);
}
/**
* Pages until the feed runs out, the requested window is covered, or the page
* ceiling is hit.
*
* <p>One page was enough while the lookback was twelve weeks. It is not enough
* for the multi-year window {@code price_z_score} needs: a district reports
* several varieties across several markets every arrival day, so a thousand
* rows sorted newest-first can be a few months rather than a few years — and
* the shortfall is invisible, because a truncated page and a district that
* genuinely stopped reporting look identical downstream.
*
* <p>{@code lookbackWeeks} of zero or less means "one page", preserving the
* previous behaviour for callers that do not care.
*/
public List<MandiRecord> fetchPrices(String state, String district, String commoditySourceCode,
int lookbackWeeks) {
if (!configured()) return null;
LocalDate horizon = lookbackWeeks <= 0 ? null : LocalDate.now().minusWeeks(lookbackWeeks);
boolean sorted = sortField != null && !sortField.isBlank();
int pageCeiling = horizon == null ? 1 : Math.max(1, maxPages);
List<MandiRecord> all = new ArrayList<>();
boolean anyCallSucceeded = false;
for (int page = 0; page < pageCeiling; page++) {
Page result = fetchPage(state, district, commoditySourceCode, page * pageLimit);
if (result == null) {
// An outage mid-way through a deep backfill: keep what earlier pages
// returned rather than discarding it, but never report a total
// failure as an empty district.
return anyCallSucceeded ? all : null;
}
anyCallSucceeded = true;
all.addAll(result.records());
// A short page is the end of the result set.
if (result.rawCount() < pageLimit) break;
if (horizon == null) break;
// Sorted newest-first, so once a page's oldest usable row predates the
// window, every later page does too. Without the sort there is no such
// guarantee and the ceiling is the only stop.
if (sorted && result.oldest() != null && result.oldest().isBefore(horizon)) break;
}
if (all.isEmpty()) {
log.warn("data.gov.in returned no usable records for {}/{}/{} — check the filter field names and the district's source spelling",
state, district, commoditySourceCode);
}
return all;
}
/** One page of records, or {@code null} when the call itself failed. */
private record Page(List<MandiRecord> records, int rawCount, LocalDate oldest) {
}
/** Set by {@link #attemptPage} so the retry loop can tell a 429 from a real failure. */
private volatile boolean lastWasRateLimited;
/**
* A page, retried when data.gov.in rate-limits us.
*
* <p>The limit is a quota per window, not per call, so it is reached by the
* <em>run</em> rather than by any one request — and once reached, every
* remaining call in that run fails. A 62-pair basket run immediately after a
* 62-pair probe took all 62 pairs down together and recorded the whole
* ingestion as FAILED with zero rows.
*
* <p>Without this the failure is also silently misattributed: a 429 arrives
* as a failed call on the district that happened to be next in the loop, so
* the freshness screen blames whichever district the quota ran out on rather
* than the run. Open-Meteo already retried for exactly this reason; data.gov
* did not, and it is the noisier of the two.
*/
private Page fetchPage(String state, String district, String commoditySourceCode, int offset) {
for (int attempt = 1; ; attempt++) {
Page result = attemptPage(state, district, commoditySourceCode, offset);
if (result != null || attempt >= Math.max(1, retryAttempts)) return result;
if (!lastWasRateLimited) return null;
log.info("data.gov.in rate-limited ({}/{}) — waiting {}ms before retrying {}/{}",
attempt, retryAttempts, retryBackoffMs, district, commoditySourceCode);
try {
Thread.sleep(retryBackoffMs);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
return null;
}
}
}
@SuppressWarnings("unchecked")
private Page attemptPage(String state, String district, String commoditySourceCode, int offset) {
UriComponentsBuilder builder = UriComponentsBuilder.fromUriString(baseUrl + "/" + resourceId)
.queryParam("api-key", apiKey)
.queryParam("format", "json")
@@ -100,6 +207,10 @@ public class DataGovClient {
.queryParam("filters[" + fieldDistrict + "]", district)
.queryParam("filters[" + fieldCommodity + "]", commoditySourceCode);
if (offset > 0) {
builder.queryParam("offset", offset);
}
// Newest first, so the page the caller keeps is the page it asked for.
if (sortField != null && !sortField.isBlank()) {
builder.queryParam("sort[" + sortField + "]", "desc");
@@ -115,20 +226,27 @@ public class DataGovClient {
: (List<Map<String, Object>>) body.getOrDefault("records", List.of());
List<MandiRecord> parsed = new ArrayList<>(records.size());
LocalDate oldest = null;
for (Map<String, Object> row : records) {
MandiRecord rec = parse(row);
if (rec.usable()) parsed.add(rec);
if (!rec.usable()) continue;
parsed.add(rec);
if (rec.arrivalDate() != null && (oldest == null || rec.arrivalDate().isBefore(oldest))) {
oldest = rec.arrivalDate();
}
}
lastWasRateLimited = false;
logCall(uri, 200, started, records.size(), true, null);
if (records.isEmpty()) {
log.warn("data.gov.in returned no records for {}/{} — check the filter field names", district, commoditySourceCode);
}
return parsed;
return new Page(parsed, records.size(), oldest);
} catch (Exception e) {
logCall(uri, null, started, null, false, e.getMessage());
log.warn("data.gov.in call failed for {}/{}: {}", district, commoditySourceCode, e.toString());
String message = e.getMessage() == null ? e.toString() : e.getMessage();
lastWasRateLimited = message.contains("429") || message.contains("Too Many Requests")
|| message.contains("Rate limit exceeded");
logCall(uri, lastWasRateLimited ? 429 : null, started, null, false, message);
log.warn("data.gov.in call failed for {}/{} at offset {}: {}",
district, commoditySourceCode, offset, e.toString());
return null;
}
}
@@ -42,10 +42,30 @@ public class Commodity {
@Column(length = 16)
private String unit;
/**
* CEREAL | PULSE | OILSEED | VEGETABLE | FRUIT | SPICE | LIVESTOCK.
*
* <p>{@code group} is reserved in SQL, hence the column name. Perishability
* is the reason this exists rather than a display nicety: a vegetable price
* moves on a weather event within days and a cereal price does not, so the
* two cannot share one staleness tolerance.
*/
@Column(name = "commodity_group", length = 16)
private String commodityGroup;
/**
* Whether {@link #sourceCode} has actually been seen to return rows from the
* live feed. False is not a defect — it is the honest state of a spelling
* nobody has checked yet, and it is what lets the probe separate "our name
* for this is wrong" from "this market had no arrivals".
*/
private Boolean sourceVerified;
private Boolean active;
@PrePersist
void onCreate() {
if (active == null) active = Boolean.TRUE;
if (sourceVerified == null) sourceVerified = Boolean.FALSE;
}
}
@@ -31,9 +31,27 @@ public class District {
@Column(length = 64)
private String nameHi;
/**
* The name Agmarknet itself uses, which is often the pre-rename spelling —
* {@code Nasik} for Nashik, {@code Belgaum} for Belagavi. Held separately
* because the feed answers HTTP 200 with an empty record list when a filter
* matches nothing, so a district renamed on our side and not on theirs looks
* exactly like a district that simply had no arrivals. Null means the display
* name is also the feed name.
*/
@Column(length = 64)
private String sourceName;
@Column(nullable = false, length = 64)
private String state;
/**
* The state name as the feed spells it, for the same reason as
* {@link #sourceName}. Null means {@link #state} is also the feed name.
*/
@Column(length = 64)
private String sourceState;
/**
* NORTH | SOUTH | EAST | WEST | CENTRAL | NORTH_EAST — the zone NABARD reads
* a national portfolio in, above the state.
@@ -52,6 +70,16 @@ public class District {
private Boolean active;
/** The district name to send to an upstream feed. */
public String feedDistrict() {
return sourceName == null || sourceName.isBlank() ? name : sourceName;
}
/** The state name to send to an upstream feed. */
public String feedState() {
return sourceState == null || sourceState.isBlank() ? state : sourceState;
}
@PrePersist
void onCreate() {
if (active == null) active = Boolean.TRUE;
@@ -0,0 +1,64 @@
package com.lincfusion.udyamlinc.master;
import jakarta.persistence.*;
import lombok.*;
/**
* What a district actually trades — the basket ingestion pulls for it.
*
* <p>Without this table ingestion is a cross product: every active district
* against every active commodity. That is affordable at four cereals and four
* districts and stops being affordable the moment fruit and vegetables are
* tracked across six states, because most of the pairs it generates cannot
* exist. Kolar has no soyabean arrivals, Kamrup no grapes, Nashik no chana.
*
* <p>The cost is not really the wasted calls, it is what they do to the
* freshness screen. Agmarknet answers a filter that matches nothing with
* HTTP 200 and an empty record list, so a structurally impossible pair and a
* genuinely broken filter produce byte-identical responses. Under a cross
* product the majority of pairs return empty on a healthy night, and a
* commodity whose {@code sourceCode} is misspelt disappears into that crowd.
* Pulling only what a district trades makes an empty result mean something
* again: either arrivals genuinely stopped, or something is wrong.
*
* <p>Zone is deliberately not the grain. Kolar and Belagavi are both SOUTH and
* trade almost disjoint baskets — tomato and mango against onion and maize.
*/
@Entity
@Table(name = "district_commodity", indexes = {
@Index(name = "ix_dc_district", columnList = "districtId"),
@Index(name = "ix_dc_pair", columnList = "districtId,commodityId", unique = true),
})
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class DistrictCommodity {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(nullable = false)
private Long districtId;
@Column(nullable = false)
private Long commodityId;
/**
* True where this district is a reference market for the commodity — Nashik
* for onion, Kolar for tomato. Not used to decide what to pull; it marks the
* rows whose absence is worth a louder complaint, because a national onion
* signal that has lost Nashik has lost the market that sets the price.
*/
private Boolean benchmark;
private Boolean active;
@PrePersist
void onCreate() {
if (active == null) active = Boolean.TRUE;
if (benchmark == null) benchmark = Boolean.FALSE;
}
}
@@ -0,0 +1,15 @@
package com.lincfusion.udyamlinc.master;
import org.springframework.data.jpa.repository.JpaRepository;
import java.util.List;
import java.util.Optional;
public interface DistrictCommodityRepository extends JpaRepository<DistrictCommodity, Long> {
List<DistrictCommodity> findByDistrictIdAndActiveTrue(Long districtId);
Optional<DistrictCommodity> findByDistrictIdAndCommodityId(Long districtId, Long commodityId);
List<DistrictCommodity> findByActiveTrue();
}
@@ -3,10 +3,13 @@ package com.lincfusion.udyamlinc.master;
import org.springframework.data.jpa.repository.JpaRepository;
import java.util.List;
import java.util.Optional;
public interface SectorCommodityRepository extends JpaRepository<SectorCommodity, Long> {
List<SectorCommodity> findBySectorId(Long sectorId);
List<SectorCommodity> findBySectorIdAndRole(Long sectorId, String role);
Optional<SectorCommodity> findBySectorIdAndCommodityId(Long sectorId, Long commodityId);
}
@@ -0,0 +1,282 @@
package com.lincfusion.udyamlinc.seed;
import java.util.List;
import java.util.Map;
/**
* The commodities the pilot tracks, and which district trades each of them.
*
* <p>Pure data, like {@link PortfolioPlan} and for the same reason: the seeder
* treats it as a target to converge on rather than a one-shot insert, so adding
* a commodity here reaches a database that was seeded months ago.
*
* <h2>On {@code sourceCode} and {@code verified}</h2>
*
* <p>Agmarknet's commodity vocabulary is its own. It writes {@code Bengal
* Gram(Gram)(Whole)}, not "Gram", and punctuation is part of the match. A wrong
* value here does not raise anything — the API returns HTTP 200 with an empty
* record list, indistinguishable from a week with no arrivals.
*
* <p>So each spec carries whether its source code has actually been seen to
* return rows. Only the four cereals and pulses have; they were checked against
* the live API on 2026-08-06. Every fruit and vegetable below is spelt as
* Agmarknet is understood to spell it and is marked {@code verified = false}
* until {@code POST /admin/ingestion/probe} says otherwise. The flag is not
* decoration — the probe reports unverified-and-empty separately from
* verified-and-empty, because the first is probably our spelling and the second
* is probably the market.
*
* <p>The probe's suspect-commodity test finds a name that is empty in every
* district that baskets it while those districts return rows for other
* commodities. That narrows the cause to two and cannot separate them: the name
* is wrong, or the name is right and the feed has no coverage here. Sugarcane
* was the second — spelt correctly, 2,926 rows nationally, and none of them in
* the two states that grow the most of it, because cane is sold to mills at a
* set price and never reaches a mandi. Settling which cause it is takes one
* unfiltered national query per suspect, and the probe cannot do it for you.
*/
public final class CommodityBaskets {
/** CEREAL | PULSE | OILSEED | VEGETABLE | FRUIT | SPICE | CASH | LIVESTOCK */
public record CommoditySpec(String code, String name, String nameHi, String group,
String source, String sourceCode, String unit, boolean verified) {
}
private static CommoditySpec agri(String code, String name, String nameHi, String group,
String sourceCode, boolean verified) {
return new CommoditySpec(code, name, nameHi, group, "AGMARKNET", sourceCode, "QUINTAL", verified);
}
/**
* Everything with a price feed. The four verified entries keep the exact
* source codes the original seeder used — changing one would silently orphan
* the market_signal rows already ingested under it.
*/
public static final List<CommoditySpec> COMMODITIES = List.of(
// ── verified against the live API, 2026-08-06 ────────────────
agri("MAIZE", "Maize", "मक्का", "CEREAL", "Maize", true),
agri("SOYA", "Soyabean", "सोयाबीन", "OILSEED", "Soyabean", true),
agri("WHEAT", "Wheat", "गेहूँ", "CEREAL", "Wheat", true),
agri("GRAM", "Gram", "चना", "PULSE", "Bengal Gram(Gram)(Whole)", true),
// ── staples ─────────────────────────────────────────────────
// "Rice" and "Lentil(Masur)(Whole)" were read straight off live
// Kamrup rows dated 10/08/2026, so they are verified on sight rather
// than on trust.
agri("RICE", "Rice", "चावल", "CEREAL", "Rice", true),
agri("MASOOR", "Lentil (masur)", "मसूर", "PULSE", "Lentil(Masur)(Whole)", true),
// Paddy is the grower's commodity and rice the miller's; a bank
// lending against a kharif crop wants the first. Agmarknet splits it
// by variety, and Common is the one smallholders sell.
agri("PADDY", "Paddy", "धान", "CEREAL", "Paddy(Dhan)(Common)", false),
// Whole grain, not the dal. Kamrup publishes "Green Gram Dal(Moong
// Dal)" — that is the processed form a retailer buys, and it moves
// with milling margins rather than with the farm gate. If the probe
// rejects this spelling, that observed one is the fallback.
agri("MOONG", "Green gram (moong)", "मूँग", "PULSE", "Green Gram (Moong)(Whole)", false),
agri("ARHAR", "Arhar (tur)", "अरहर", "PULSE", "Arhar (Tur/Red Gram)(Whole)", false),
agri("MUSTARD", "Mustard", "सरसों", "OILSEED", "Mustard", false),
agri("BAJRA", "Bajra", "बाजरा", "CEREAL", "Bajra(Pearl Millet/Cumbu)", false),
agri("JOWAR", "Jowar", "ज्वार", "CEREAL", "Jowar(Sorghum)", false),
// ── vegetables — spelling unverified ─────────────────────────
agri("ONION", "Onion", "प्याज़", "VEGETABLE", "Onion", false),
agri("TOMATO", "Tomato", "टमाटर", "VEGETABLE", "Tomato", false),
agri("POTATO", "Potato", "आलू", "VEGETABLE", "Potato", false),
agri("BRINJAL", "Brinjal", "बैंगन", "VEGETABLE", "Brinjal", false),
agri("CAULIFLOWER", "Cauliflower", "फूलगोभी", "VEGETABLE", "Cauliflower", false),
agri("CABBAGE", "Cabbage", "पत्तागोभी", "VEGETABLE", "Cabbage", false),
agri("GREEN_CHILLI", "Green chilli", "हरी मिर्च", "VEGETABLE", "Green Chilli", false),
// Agmarknet's own name, parenthesis and all. The freshest line in both
// Bihar districts on 2026-08-11.
agri("OKRA", "Okra", "भिंडी", "VEGETABLE", "Bhindi(Ladies Finger)", false),
// Read off live rows on 2026-08-11 — Kamrup publishes capsicum,
// cucumber and green peas daily, Nagaon pumpkin, beetroot and bitter
// gourd, Begusarai bitter gourd and capsicum. Spellings taken from
// the feed rather than guessed, parentheses and all.
agri("CAPSICUM", "Capsicum", "शिमला मिर्च", "VEGETABLE", "Capsicum", true),
agri("CUCUMBER", "Cucumber", "खीरा", "VEGETABLE", "Cucumbar(Kheera)", true),
agri("GREEN_PEAS", "Green peas", "हरी मटर", "VEGETABLE", "Green Peas", true),
agri("PUMPKIN", "Pumpkin", "कद्दू", "VEGETABLE", "Pumpkin", true),
agri("BEETROOT", "Beetroot", "चुकंदर", "VEGETABLE", "Beetroot", true),
agri("BITTER_GOURD", "Bitter gourd", "करेला", "VEGETABLE", "Bitter gourd", true),
agri("BOTTLE_GOURD", "Bottle gourd", "लौकी", "VEGETABLE", "Bottle gourd", false),
agri("CARROT", "Carrot", "गाजर", "VEGETABLE", "Carrot", false),
agri("PAPAYA", "Papaya", "पपीता", "FRUIT", "Papaya", true),
agri("GUAVA", "Guava", "अमरूद", "FRUIT", "Guava", false),
// ── cash crops ──────────────────────────────────────────────
// Cotton and sugarcane are what a Maharashtra or north-Karnataka
// borrower is actually financed against, and neither is a vegetable
// or a staple — a portfolio that prices onions but not cane is
// reading the wrong half of the Belagavi book.
agri("COTTON", "Cotton", "कपास", "CASH", "Cotton", true),
// Correctly spelt and all but useless here. "Sugarcane" returns 2,926
// rows nationally against cotton's 958,962, and nearly all of them are
// Chhattisgarh — Maharashtra and Karnataka, the two states that grow
// the most of it, have none. Cane does not pass through a mandi: it
// goes to the mill at a government-set FRP, so there is no market
// price to fetch. Same class as milk, and manual entry is the answer.
new CommoditySpec("SUGARCANE", "Sugarcane", "गन्ना", "CASH", "MANUAL", null, "QUINTAL", false),
agri("TURMERIC", "Turmeric", "हल्दी", "SPICE", "Turmeric", false),
// ── fruit — spelling unverified ──────────────────────────────
agri("BANANA", "Banana", "केला", "FRUIT", "Banana", false),
agri("MANGO", "Mango", "आम", "FRUIT", "Mango", false),
agri("GRAPES", "Grapes", "अंगूर", "FRUIT", "Grapes", false),
agri("POMEGRANATE", "Pomegranate", "अनार", "FRUIT", "Pomegranate", false),
agri("LITCHI", "Litchi", "लीची", "FRUIT", "Litchi", false),
// ── spice / bulb — spelling unverified ───────────────────────
agri("GARLIC", "Garlic", "लहसुन", "SPICE", "Garlic", false),
agri("GINGER", "Ginger", "अदरक", "SPICE", "Ginger(Green)", false),
// ── no usable API; manual entry exists because of these ──────
new CommoditySpec("EGG", "Egg", "अंडा", "LIVESTOCK", "NECC", "Bhopal", "HUNDRED", false),
new CommoditySpec("MILK", "Milk", "दूध", "LIVESTOCK", "DAIRY_FEDERATION", null, "LITRE", false),
new CommoditySpec("FODDER", "Green fodder", "हरा चारा", "LIVESTOCK", "MANUAL", null, "QUINTAL", false),
// ── simulated output indices ────────────────────────────────
// These are not commodities. Nothing publishes them, no mandi trades
// them, and unlike milk or egg there is no real number they stand in
// for — a finished sari and a kirana basket have no market price at
// all. They exist so that margin_squeeze, output_price_change and
// price_z_score are computable for the three packs that otherwise
// carry none, which was an explicit decision to prefer a complete
// console over an honest gap.
//
// Everything about them is therefore labelled twice over: source
// SIMULATED, group INDEX, "(simulated)" in the display name, and a
// base of 100 so the series reads as an index rather than a rupee
// price anyone might quote. If these ever need to become real, the
// replacement is a published WPI series, not a better generator.
new CommoditySpec("PROCESSED_FOOD_INDEX", "Processed food output (simulated)",
"प्रसंस्कृत खाद्य सूचकांक (सिम्युलेटेड)", "INDEX", "SIMULATED", null, "INDEX", false),
new CommoditySpec("RETAIL_BASKET_INDEX", "Retail basket output (simulated)",
"खुदरा बास्केट सूचकांक (सिम्युलेटेड)", "INDEX", "SIMULATED", null, "INDEX", false),
new CommoditySpec("HANDLOOM_INDEX", "Handloom output (simulated)",
"हथकरघा उत्पाद सूचकांक (सिम्युलेटेड)", "INDEX", "SIMULATED", null, "INDEX", false),
new CommoditySpec("YARN_INDEX", "Yarn input (simulated)",
"सूत सूचकांक (सिम्युलेटेड)", "INDEX", "SIMULATED", null, "INDEX", false));
/**
* What each district trades, by district code. Chosen from what the district
* is actually known for rather than from what would make the map look full —
* a basket entry is a nightly API call, and one that can never return rows
* costs the same as one that can.
*
* <p>A leading {@code *} marks a benchmark market: the one whose price the
* rest of the country reads. Nashik sets the onion rate and Kolar the tomato
* rate, so losing either is worth more than losing an ordinary row.
*/
public static final Map<String, List<String>> BASKETS = Map.ofEntries(
// CENTRAL — wheat and paddy are the state's two main crops and carry
// every district here; soya is the commercial third; and Sehore,
// Vidisha and Raisen are India's summer-moong belt, so a moong price
// is a first-order input to a Madhya Pradesh loan book rather than a
// nice-to-have. Paddy sits in all four districts, not just the two
// canal ones — a rabi wheat crop and a kharif paddy crop are the two
// halves of the same borrower's year, and pricing only one of them
// reads the year at the wrong end.
Map.entry("SEHORE", List.of("WHEAT", "PADDY", "SOYA", "GRAM", "MAIZE", "MOONG", "TOMATO")),
Map.entry("VIDISHA", List.of("WHEAT", "PADDY", "SOYA", "GRAM", "MOONG", "MASOOR")),
Map.entry("RAISEN", List.of("WHEAT", "PADDY", "SOYA", "GRAM", "MAIZE", "MOONG")),
// Bhopal is the terminal market, so it carries milled Rice alongside
// the paddy the surrounding districts sell.
Map.entry("BHOPAL", List.of("WHEAT", "PADDY", "RICE", "MAIZE", "MOONG", "MASOOR",
"TOMATO", "POTATO", "ONION", "GREEN_CHILLI", "CAPSICUM", "CUCUMBER")),
// WEST — Nashik is the national onion benchmark and a table-grape
// belt; both districts are also rainfed bajra and jowar country,
// which is the crop a marginal Maharashtra borrower actually grows,
// and sugarcane and cotton are what the bigger loans are against.
Map.entry("NASHIK", List.of("*ONION", "GRAPES", "TOMATO", "POMEGRANATE",
"SOYA", "BAJRA")),
Map.entry("AHMEDNAGAR", List.of("ONION", "POMEGRANATE", "WHEAT", "TOMATO",
"BAJRA", "JOWAR", "COTTON")),
// NORTH — Rajasthan is the country's mustard bowl and its bajra belt.
// Kota sits in the Chambal command, so unlike the rest of the state
// it grows paddy.
Map.entry("JAIPUR", List.of("WHEAT", "GRAM", "MUSTARD", "BAJRA", "MOONG",
"ONION", "GARLIC", "CAULIFLOWER", "GREEN_PEAS", "CUCUMBER")),
Map.entry("KOTA", List.of("SOYA", "WHEAT", "GRAM", "MUSTARD", "PADDY", "GARLIC")),
// EAST — Muzaffarpur really is India's litchi belt, and Agmarknet
// does not carry it: the newest litchi row is from 2015 and the
// newest banana row from 2004. What the district actually reports
// daily is vegetables, so that is what the basket asks for. The
// reputation of a district is not evidence that the feed covers it.
// Bihar is a rice-wheat-maize state before it is anything else, and
// Begusarai sits in the summer-moong belt of the Kosi plain.
// Paddy is dropped from both: Agmarknet carries none for Begusarai
// and eighteen rows ending in 2022 for Muzaffarpur. Bihar is a rice
// state with no usable paddy price in the feed — a coverage gap, not
// a spelling we can fix, and worth stating rather than papering over.
Map.entry("MUZAFFARPUR", List.of("WHEAT", "MAIZE",
"POTATO", "ONION", "OKRA", "BRINJAL", "CAULIFLOWER", "PUMPKIN", "PAPAYA")),
Map.entry("BEGUSARAI", List.of("WHEAT", "MAIZE",
"POTATO", "ONION", "OKRA", "BRINJAL", "TOMATO", "BITTER_GOURD", "CAPSICUM")),
// SOUTH — Kolar is Asia's largest tomato market and a mango belt;
// Belagavi is jowar and sugarcane country in the same breath as
// onion. Belagavi paddy returned nothing at all, so it is dropped.
Map.entry("KOLAR", List.of("*TOMATO", "POTATO", "MANGO", "RICE", "ARHAR",
"CAPSICUM", "BEETROOT")),
Map.entry("BELAGAVI", List.of("ONION", "POTATO", "MAIZE", "CABBAGE",
"JOWAR", "COTTON")),
// NORTH_EAST — assumed to be the thinnest coverage of the six and it
// is not: Kamrup reports 110k rows and Nagaon 50k, both current to
// yesterday across a wide vegetable basket. Banana was the mistake,
// not the zone — 13 rows in Kamrup and a newest arrival of 2011 in
// Nagaon. Ginger is real in both and is what these districts are
// actually known for.
// Assam is a rice state: Kamrup publishes Rice, moong dal and masur
// daily, all current to 10/08/2026 — those rows are where the RICE
// and MASOOR spellings above were read from.
// Kamrup's paddy stops in 2017 and its moong in 2019; Nagaon's rice
// and paddy both stop in 2019. Assam's rice reaches this feed as
// milled Rice through Kamrup and nowhere else, so that is the only
// rice line either district keeps.
Map.entry("KAMRUP", List.of("RICE", "MASOOR",
"POTATO", "ONION", "GINGER", "TOMATO", "BRINJAL",
"CAPSICUM", "CUCUMBER", "GREEN_PEAS")),
Map.entry("NAGAON", List.of("POTATO", "GREEN_CHILLI", "GINGER", "BRINJAL", "TOMATO",
"PUMPKIN", "BEETROOT", "BITTER_GOURD")));
/**
* Districts dropped from the plan whose row must stay in the database.
*
* <p>Patna stopped publishing to Agmarknet in March 2025 — onion and potato
* both end 05/03/2025, everything else in 2023 — and no spelling change
* brings it back. It cannot simply vanish from {@link PortfolioPlan}: it
* still holds seeded enterprises and a bank branch, and a district with no
* basket entry would hit the ingestion fallback and pull <em>every</em>
* commodity nightly, forever, all of it years stale.
*
* <p>So the district is deactivated instead. Ingestion reads
* {@code findByActiveTrue} and skips it; the enterprises stay where they are
* and keep their history. What they lose is a live price signal, which is
* the honest consequence of the market they sit in having gone quiet.
*/
public static final Map<String, String> RETIRED = Map.of(
"PATNA", "Stopped reporting to Agmarknet in March 2025; replaced by Begusarai");
/** Strips the benchmark marker from a basket entry. */
public static String commodityCode(String entry) {
return entry.startsWith("*") ? entry.substring(1) : entry;
}
public static boolean isBenchmark(String entry) {
return entry.startsWith("*");
}
/** Every (district, commodity) pair the pilot intends to pull. */
public static int pairCount() {
return BASKETS.values().stream().mapToInt(List::size).sum();
}
private CommodityBaskets() {
}
}