diff --git a/src/main/java/com/lincfusion/udyamlinc/ingestion/BasketProbeService.java b/src/main/java/com/lincfusion/udyamlinc/ingestion/BasketProbeService.java new file mode 100644 index 0000000..58f634b --- /dev/null +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/BasketProbeService.java @@ -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. + * + *

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. + * + *

The probe separates them using two facts the ingestion loop does not have: + * whether this commodity has ever 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. + * + *

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 suspectCommodities, List suspectDistricts, + int newlyVerified, List 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 commodities = new HashMap<>(); + commodityRepository.findAll().forEach(c -> commodities.put(c.getId(), c)); + List districts = districtRepository.findByActiveTrue(); + + List 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 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 workingDistricts = new HashSet<>(); + Set workingCommodities = new HashSet<>(); + for (PairResult r : results) { + if (r.rows() > 0) { + workingDistricts.add(r.districtCode()); + workingCommodities.add(r.commodityCode()); + } + } + List 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 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 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); + } +} diff --git a/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionController.java b/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionController.java index db8d755..b9bb527 100644 --- a/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionController.java +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionController.java @@ -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 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. + * + *

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 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 runOne(@PathVariable String source) { - return ApiResponse.ok("Ingestion triggered", scheduler.run(source.toUpperCase(), "MANUAL")); + public ApiResponse runOne(@PathVariable String source, + @RequestParam(required = false) Integer lookbackWeeks) { + return ApiResponse.ok("Ingestion triggered", + scheduler.run(source.toUpperCase(), "MANUAL", lookbackWeeks)); } } diff --git a/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionScheduler.java b/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionScheduler.java index 0cd0c84..5df94ed 100644 --- a/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionScheduler.java +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionScheduler.java @@ -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); }; diff --git a/src/main/java/com/lincfusion/udyamlinc/ingestion/MandiIngestionService.java b/src/main/java/com/lincfusion/udyamlinc/ingestion/MandiIngestionService.java index 9b29733..cdc4d48 100644 --- a/src/main/java/com/lincfusion/udyamlinc/ingestion/MandiIngestionService.java +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/MandiIngestionService.java @@ -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 districts = districtRepository.findByActiveTrue(); List 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 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 records = client.fetchPrices(district.getState(), district.getName(), sourceCode); + List 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}. + * + *

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 basketFor(District district, Map byId, List all) { + List 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 rows, LocalDateTime observedAt) { diff --git a/src/main/java/com/lincfusion/udyamlinc/ingestion/client/DataGovClient.java b/src/main/java/com/lincfusion/udyamlinc/ingestion/client/DataGovClient.java index 6a72c3d..2cde9b2 100644 --- a/src/main/java/com/lincfusion/udyamlinc/ingestion/client/DataGovClient.java +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/client/DataGovClient.java @@ -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 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. + * + *

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. + * + *

{@code lookbackWeeks} of zero or less means "one page", preserving the + * previous behaviour for callers that do not care. + */ + public List 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 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 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. + * + *

The limit is a quota per window, not per call, so it is reached by the + * run 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. + * + *

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>) body.getOrDefault("records", List.of()); List parsed = new ArrayList<>(records.size()); + LocalDate oldest = null; for (Map 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; } } diff --git a/src/main/java/com/lincfusion/udyamlinc/master/Commodity.java b/src/main/java/com/lincfusion/udyamlinc/master/Commodity.java index b67add4..cfa5beb 100644 --- a/src/main/java/com/lincfusion/udyamlinc/master/Commodity.java +++ b/src/main/java/com/lincfusion/udyamlinc/master/Commodity.java @@ -42,10 +42,30 @@ public class Commodity { @Column(length = 16) private String unit; + /** + * CEREAL | PULSE | OILSEED | VEGETABLE | FRUIT | SPICE | LIVESTOCK. + * + *

{@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; } } diff --git a/src/main/java/com/lincfusion/udyamlinc/master/District.java b/src/main/java/com/lincfusion/udyamlinc/master/District.java index 722f688..1dfc9ab 100644 --- a/src/main/java/com/lincfusion/udyamlinc/master/District.java +++ b/src/main/java/com/lincfusion/udyamlinc/master/District.java @@ -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; diff --git a/src/main/java/com/lincfusion/udyamlinc/master/DistrictCommodity.java b/src/main/java/com/lincfusion/udyamlinc/master/DistrictCommodity.java new file mode 100644 index 0000000..bf99cb9 --- /dev/null +++ b/src/main/java/com/lincfusion/udyamlinc/master/DistrictCommodity.java @@ -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. + * + *

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. + * + *

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. + * + *

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; + } +} diff --git a/src/main/java/com/lincfusion/udyamlinc/master/DistrictCommodityRepository.java b/src/main/java/com/lincfusion/udyamlinc/master/DistrictCommodityRepository.java new file mode 100644 index 0000000..c48f70c --- /dev/null +++ b/src/main/java/com/lincfusion/udyamlinc/master/DistrictCommodityRepository.java @@ -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 { + + List findByDistrictIdAndActiveTrue(Long districtId); + + Optional findByDistrictIdAndCommodityId(Long districtId, Long commodityId); + + List findByActiveTrue(); +} diff --git a/src/main/java/com/lincfusion/udyamlinc/master/SectorCommodityRepository.java b/src/main/java/com/lincfusion/udyamlinc/master/SectorCommodityRepository.java index f17c68f..be5394c 100644 --- a/src/main/java/com/lincfusion/udyamlinc/master/SectorCommodityRepository.java +++ b/src/main/java/com/lincfusion/udyamlinc/master/SectorCommodityRepository.java @@ -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 { List findBySectorId(Long sectorId); List findBySectorIdAndRole(Long sectorId, String role); + + Optional findBySectorIdAndCommodityId(Long sectorId, Long commodityId); } diff --git a/src/main/java/com/lincfusion/udyamlinc/seed/CommodityBaskets.java b/src/main/java/com/lincfusion/udyamlinc/seed/CommodityBaskets.java new file mode 100644 index 0000000..8cf769d --- /dev/null +++ b/src/main/java/com/lincfusion/udyamlinc/seed/CommodityBaskets.java @@ -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. + * + *

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. + * + *

On {@code sourceCode} and {@code verified}

+ * + *

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. + * + *

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. + * + *

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 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. + * + *

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> 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. + * + *

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 every + * commodity nightly, forever, all of it years stale. + * + *

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 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() { + } +}