fix(ingestion): make the pipeline actually run against live feeds

First end-to-end run against Postgres and the real government APIs, which
surfaced four defects that static checks could not.

- Agmarknet pointed at 9ef84268-…, "Current Daily Price": a today-only
  snapshot holding one arrival_date, in which none of the four pilot
  districts appear. Switched to 35985678-…, the historical resource.
- Filter fields were lowercase. This resource matches exactly on the
  capitalised name; the lowercase form is tokenised, so "Madhya Pradesh"
  also matched Andhra, Himachal and Uttar Pradesh — wrong-state prices
  written under a pilot district, with freshness looking healthy.
- Rows arrive unordered and the lookback filter is client side, so a page
  was mostly years-old data and every commodity wrote zero weeks. Added
  sort[Arrival_Date]=desc.
- ExternalFeedConfig injected a RestClient.Builder bean Boot does not
  expose, so the context never built and the app could not start.

Also: Open-Meteo meters by request cost, so the 30-year baseline call
exhausted the minutely allowance and 429'd two districts out of their
climate data. Rate-limited calls now retry with backoff, while genuine
failures still fail once and are reported.

FODDER moves to MANUAL — Agmarknet carries Green Fodder, but every recent
row is a single market in Ambala, Haryana.

Result: 16,000 rows fetched, 202 market weeks and 52 climate weeks written,
0 failed calls; 142 scored, 33 trip-wires fired.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
LincPay
2026-08-06 14:01:25 +05:30
co-authored by Claude Opus 5
parent 1b9b02f2c7
commit 20879b5904
5 changed files with 121 additions and 20 deletions
@@ -7,6 +7,8 @@ import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.web.client.RestClient; import org.springframework.web.client.RestClient;
import java.time.Duration;
/** /**
* The HTTP client used for external feeds, and the scheduler that drives them. * The HTTP client used for external feeds, and the scheduler that drives them.
* *
@@ -24,11 +26,18 @@ public class ExternalFeedConfig {
@Value("${udyam.ingestion.http.read-timeout-ms:30000}") @Value("${udyam.ingestion.http.read-timeout-ms:30000}")
private int readTimeoutMs; private int readTimeoutMs;
/**
* Built from {@link RestClient#builder()} rather than an injected
* auto-configured {@code RestClient.Builder}, which Boot does not always
* expose as a bean. It also keeps this client free of any global customiser
* meant for the request path — the feed client's timeouts are deliberately
* its own.
*/
@Bean("externalFeedClient") @Bean("externalFeedClient")
public RestClient externalFeedClient(RestClient.Builder builder) { public RestClient externalFeedClient() {
SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory(); SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
factory.setConnectTimeout(connectTimeoutMs); factory.setConnectTimeout(Duration.ofMillis(connectTimeoutMs));
factory.setReadTimeout(readTimeoutMs); factory.setReadTimeout(Duration.ofMillis(readTimeoutMs));
return builder.requestFactory(factory).build(); return RestClient.builder().requestFactory(factory).build();
} }
} }
@@ -61,6 +61,14 @@ public class DataGovClient {
@Value("${udyam.ingestion.datagov.field.commodity:commodity}") @Value("${udyam.ingestion.datagov.field.commodity:commodity}")
private String fieldCommodity; private String fieldCommodity;
/**
* The resource holds years of history and answers in arbitrary order, so a
* page without this is mostly rows the caller's lookback window will discard.
* Blank disables the sort, for a resource that does not support it.
*/
@Value("${udyam.ingestion.datagov.sort-field:Arrival_Date}")
private String sortField;
public DataGovClient(@Qualifier("externalFeedClient") RestClient http, ApiLogRepository apiLogRepository) { public DataGovClient(@Qualifier("externalFeedClient") RestClient http, ApiLogRepository apiLogRepository) {
this.http = http; this.http = http;
this.apiLogRepository = apiLogRepository; this.apiLogRepository = apiLogRepository;
@@ -84,16 +92,20 @@ public class DataGovClient {
public List<MandiRecord> fetchPrices(String state, String district, String commoditySourceCode) { public List<MandiRecord> fetchPrices(String state, String district, String commoditySourceCode) {
if (!configured()) return null; if (!configured()) return null;
URI uri = UriComponentsBuilder.fromUriString(baseUrl + "/" + resourceId) UriComponentsBuilder builder = UriComponentsBuilder.fromUriString(baseUrl + "/" + resourceId)
.queryParam("api-key", apiKey) .queryParam("api-key", apiKey)
.queryParam("format", "json") .queryParam("format", "json")
.queryParam("limit", pageLimit) .queryParam("limit", pageLimit)
.queryParam("filters[" + fieldState + "]", state) .queryParam("filters[" + fieldState + "]", state)
.queryParam("filters[" + fieldDistrict + "]", district) .queryParam("filters[" + fieldDistrict + "]", district)
.queryParam("filters[" + fieldCommodity + "]", commoditySourceCode) .queryParam("filters[" + fieldCommodity + "]", commoditySourceCode);
.build()
.encode() // Newest first, so the page the caller keeps is the page it asked for.
.toUri(); if (sortField != null && !sortField.isBlank()) {
builder.queryParam("sort[" + sortField + "]", "desc");
}
URI uri = builder.build().encode().toUri();
long started = System.currentTimeMillis(); long started = System.currentTimeMillis();
try { try {
@@ -40,6 +40,19 @@ public class OpenMeteoClient {
@Value("${udyam.ingestion.climate.timezone:Asia/Kolkata}") @Value("${udyam.ingestion.climate.timezone:Asia/Kolkata}")
private String timezone; private String timezone;
/**
* Open-Meteo bills by request cost, not request count, so the single 30-year
* baseline call can exhaust the minutely allowance on its own and take the
* district calls behind it down with it. Losing a district's climate to a
* transient 429 is not the same as that district having no data, so a
* rate-limited call waits and tries again.
*/
@Value("${udyam.ingestion.climate.retry-attempts:3}")
private int retryAttempts;
@Value("${udyam.ingestion.climate.retry-backoff-ms:20000}")
private long retryBackoffMs;
public OpenMeteoClient(@Qualifier("externalFeedClient") RestClient http, ApiLogRepository apiLogRepository) { public OpenMeteoClient(@Qualifier("externalFeedClient") RestClient http, ApiLogRepository apiLogRepository) {
this.http = http; this.http = http;
this.apiLogRepository = apiLogRepository; this.apiLogRepository = apiLogRepository;
@@ -49,7 +62,6 @@ public class OpenMeteoClient {
* The daily series for one grid cell. {@code null} when the call failed, * The daily series for one grid cell. {@code null} when the call failed,
* empty when the archive held nothing for the window. * empty when the archive held nothing for the window.
*/ */
@SuppressWarnings("unchecked")
public List<DailyWeather> fetchDaily(double latitude, double longitude, LocalDate from, LocalDate to) { public List<DailyWeather> fetchDaily(double latitude, double longitude, LocalDate from, LocalDate to) {
URI uri = UriComponentsBuilder.fromUriString(baseUrl) URI uri = UriComponentsBuilder.fromUriString(baseUrl)
.queryParam("latitude", latitude) .queryParam("latitude", latitude)
@@ -62,7 +74,29 @@ public class OpenMeteoClient {
.encode() .encode()
.toUri(); .toUri();
for (int attempt = 1; ; attempt++) {
List<DailyWeather> result = attempt(uri, latitude, longitude, from, to, attempt);
if (result != null || attempt >= Math.max(1, retryAttempts)) return result;
// Only a rate-limited attempt asks to be retried; anything else has
// already returned null and logged its own reason.
if (!lastWasRateLimited) return null;
try {
Thread.sleep(retryBackoffMs);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
return null;
}
}
}
/** Set by {@link #attempt} so the retry loop can tell a 429 from a real failure. */
private volatile boolean lastWasRateLimited;
@SuppressWarnings("unchecked")
private List<DailyWeather> attempt(URI uri, double latitude, double longitude,
LocalDate from, LocalDate to, int attemptNo) {
long started = System.currentTimeMillis(); long started = System.currentTimeMillis();
lastWasRateLimited = false;
try { try {
Map<String, Object> body = http.get().uri(uri).retrieve().body(Map.class); Map<String, Object> body = http.get().uri(uri).retrieve().body(Map.class);
Map<String, Object> daily = body == null ? null : (Map<String, Object>) body.get("daily"); Map<String, Object> daily = body == null ? null : (Map<String, Object>) body.get("daily");
@@ -87,8 +121,12 @@ public class OpenMeteoClient {
return out; return out;
} catch (Exception e) { } catch (Exception e) {
logCall(uri, null, started, null, false, e.getMessage()); String message = String.valueOf(e.getMessage());
log.warn("Open-Meteo call failed for {},{} {}..{}: {}", latitude, longitude, from, to, e.toString()); lastWasRateLimited = message.contains("429") || message.contains("Too Many Requests");
logCall(uri, lastWasRateLimited ? 429 : null, started, null, false, message);
log.warn("Open-Meteo call failed for {},{} {}..{} (attempt {}{}): {}",
latitude, longitude, from, to, attemptNo,
lastWasRateLimited ? ", rate limited — will retry" : "", e.toString());
return null; return null;
} }
} }
@@ -147,7 +147,11 @@ public class UdyamDataSeeder implements CommandLineRunner {
commodity("GRAM", "Gram", "चना", "AGMARKNET", "Bengal Gram(Gram)(Whole)", "QUINTAL"), commodity("GRAM", "Gram", "चना", "AGMARKNET", "Bengal Gram(Gram)(Whole)", "QUINTAL"),
commodity("EGG", "Egg", "अंडा", "NECC", "Bhopal", "HUNDRED"), commodity("EGG", "Egg", "अंडा", "NECC", "Bhopal", "HUNDRED"),
commodity("MILK", "Milk", "दूध", "DAIRY_FEDERATION", null, "LITRE"), commodity("MILK", "Milk", "दूध", "DAIRY_FEDERATION", null, "LITRE"),
commodity("FODDER", "Green fodder", "हरा चारा", "AGMARKNET", "Green Fodder", "QUINTAL"))); // Agmarknet carries Green Fodder, but every recent row is one market
// in Ambala, Haryana — nothing for the pilot districts. Left as a
// manual-entry commodity rather than a nightly call that can only
// ever return an empty page.
commodity("FODDER", "Green fodder", "हरा चारा", "MANUAL", null, "QUINTAL")));
log.info("Seeded {} commodities", commodityRepository.count()); log.info("Seeded {} commodities", commodityRepository.count());
} }
@@ -283,7 +287,10 @@ public class UdyamDataSeeder implements CommandLineRunner {
private static final BigDecimal PINNED_SHORTFALL = new BigDecimal("12400"); private static final BigDecimal PINNED_SHORTFALL = new BigDecimal("12400");
private void seedPortfolio() { private void seedPortfolio() {
if (enterpriseRepository.count() > 0) return; if (enterpriseRepository.count() > 0) {
backfillPortfolio();
return;
}
Map<String, Sector> sectors = new HashMap<>(); Map<String, Sector> sectors = new HashMap<>();
sectorRepository.findAll().forEach(s -> sectors.put(s.getCode(), s)); sectorRepository.findAll().forEach(s -> sectors.put(s.getCode(), s));
@@ -403,6 +410,24 @@ public class UdyamDataSeeder implements CommandLineRunner {
} }
} }
/**
* A step added to the seeder after a database was first populated never
* reaches it — the portfolio guard short-circuits every per-enterprise step
* along with the enterprises themselves. Loan records arrived with Phase C,
* so a database seeded before then holds none and {@code emi_delay} resolves
* {@code NO_CBS} forever, which reads as "the bank sent us nothing" rather
* than "this seeder never ran". Each step below carries its own
* per-enterprise guard, so this is safe on every boot.
*/
private void backfillPortfolio() {
long before = loanRecordRepository.count();
enterpriseRepository.findAll().forEach(this::seedLoans);
long added = loanRecordRepository.count() - before;
if (added > 0) {
log.info("Backfilled {} loan records onto an already-seeded portfolio", added);
}
}
/** /**
* The institutional side of the record — what a nightly CBS extract would * The institutional side of the record — what a nightly CBS extract would
* carry. DPD tracks the band, because in the demo the band is the ground * carry. DPD tracks the band, because in the demo the band is the ground
+23 -6
View File
@@ -70,13 +70,25 @@ udyam.ingestion.http.read-timeout-ms=30000
# Register at https://data.gov.in for a key; without one the run records SKIPPED. # Register at https://data.gov.in for a key; without one the run records SKIPPED.
udyam.ingestion.datagov.api-key=${DATAGOV_API_KEY:} udyam.ingestion.datagov.api-key=${DATAGOV_API_KEY:}
udyam.ingestion.datagov.base-url=https://api.data.gov.in/resource udyam.ingestion.datagov.base-url=https://api.data.gov.in/resource
# "Variety-wise Daily Market Prices Data of Commodity" (Agmarknet) # "Variety-wise Daily Market Prices Data of Commodity" — the historical Agmarknet
udyam.ingestion.datagov.resource-id=${DATAGOV_RESOURCE_ID:9ef84268-d588-465a-a308-a864a43d0070} # resource, ~81M rows, current to yesterday. NOT 9ef84268-…, which is "Current
# Daily Price": a today-only snapshot carrying a single arrival_date, from which
# no week-over-week history can be built and in which none of the four pilot
# districts appear.
udyam.ingestion.datagov.resource-id=${DATAGOV_RESOURCE_ID:35985678-0d79-46b4-9ed6-6f13308a1d24}
udyam.ingestion.datagov.page-limit=1000 udyam.ingestion.datagov.page-limit=1000
# Filter names must match the resource's own field names; they differ by resource version. # Filter names must match the resource's own field names; they differ by resource
udyam.ingestion.datagov.field.state=state # version. This resource capitalises them, and — unlike the current-price
udyam.ingestion.datagov.field.district=district # resource — matches exactly on the bare name. Appending ".keyword" here returns
udyam.ingestion.datagov.field.commodity=commodity # zero rows; omitting the capital returns rows from every state whose name shares
# a token, so "Madhya Pradesh" would silently pull Andhra, Himachal and Uttar.
udyam.ingestion.datagov.field.state=State
udyam.ingestion.datagov.field.district=District
udyam.ingestion.datagov.field.commodity=Commodity
# Rows come back in arbitrary order and the lookback filter is applied client
# side, so without a descending sort a 1000-row page is mostly years-old data and
# every commodity reports zero weeks written.
udyam.ingestion.datagov.sort-field=Arrival_Date
udyam.ingestion.mandi.lookback-weeks=12 udyam.ingestion.mandi.lookback-weeks=12
# Climate — Open-Meteo reanalysis archive. No key. The archive trails real time, # Climate — Open-Meteo reanalysis archive. No key. The archive trails real time,
@@ -88,6 +100,11 @@ udyam.ingestion.climate.lag-days=5
udyam.ingestion.climate.heat-threshold-c=40 udyam.ingestion.climate.heat-threshold-c=40
udyam.ingestion.climate.baseline-from=1995-01-01 udyam.ingestion.climate.baseline-from=1995-01-01
udyam.ingestion.climate.baseline-to=2024-12-31 udyam.ingestion.climate.baseline-to=2024-12-31
# Open-Meteo meters by request cost, so the 30-year baseline call alone can trip
# the minutely limit and 429 the district calls queued behind it. A rate-limited
# call waits and retries; anything else fails once and is reported.
udyam.ingestion.climate.retry-attempts=3
udyam.ingestion.climate.retry-backoff-ms=20000
# ── Feature store (ARCHITECTURE Phase C) ──────────────────── # ── Feature store (ARCHITECTURE Phase C) ────────────────────
# Window sizes for the derivations. A level is not a feature; every one of these # Window sizes for the derivations. A level is not a feature; every one of these