diff --git a/src/main/java/com/lincfusion/udyamlinc/config/ExternalFeedConfig.java b/src/main/java/com/lincfusion/udyamlinc/config/ExternalFeedConfig.java index 6699229..c1ad022 100644 --- a/src/main/java/com/lincfusion/udyamlinc/config/ExternalFeedConfig.java +++ b/src/main/java/com/lincfusion/udyamlinc/config/ExternalFeedConfig.java @@ -7,6 +7,8 @@ import org.springframework.http.client.SimpleClientHttpRequestFactory; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.web.client.RestClient; +import java.time.Duration; + /** * 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}") 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") - public RestClient externalFeedClient(RestClient.Builder builder) { + public RestClient externalFeedClient() { SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory(); - factory.setConnectTimeout(connectTimeoutMs); - factory.setReadTimeout(readTimeoutMs); - return builder.requestFactory(factory).build(); + factory.setConnectTimeout(Duration.ofMillis(connectTimeoutMs)); + factory.setReadTimeout(Duration.ofMillis(readTimeoutMs)); + return RestClient.builder().requestFactory(factory).build(); } } 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 28dcffa..6a72c3d 100644 --- a/src/main/java/com/lincfusion/udyamlinc/ingestion/client/DataGovClient.java +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/client/DataGovClient.java @@ -61,6 +61,14 @@ public class DataGovClient { @Value("${udyam.ingestion.datagov.field.commodity:commodity}") 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) { this.http = http; this.apiLogRepository = apiLogRepository; @@ -84,16 +92,20 @@ public class DataGovClient { public List fetchPrices(String state, String district, String commoditySourceCode) { if (!configured()) return null; - URI uri = UriComponentsBuilder.fromUriString(baseUrl + "/" + resourceId) + UriComponentsBuilder builder = UriComponentsBuilder.fromUriString(baseUrl + "/" + resourceId) .queryParam("api-key", apiKey) .queryParam("format", "json") .queryParam("limit", pageLimit) .queryParam("filters[" + fieldState + "]", state) .queryParam("filters[" + fieldDistrict + "]", district) - .queryParam("filters[" + fieldCommodity + "]", commoditySourceCode) - .build() - .encode() - .toUri(); + .queryParam("filters[" + fieldCommodity + "]", commoditySourceCode); + + // Newest first, so the page the caller keeps is the page it asked for. + if (sortField != null && !sortField.isBlank()) { + builder.queryParam("sort[" + sortField + "]", "desc"); + } + + URI uri = builder.build().encode().toUri(); long started = System.currentTimeMillis(); try { diff --git a/src/main/java/com/lincfusion/udyamlinc/ingestion/client/OpenMeteoClient.java b/src/main/java/com/lincfusion/udyamlinc/ingestion/client/OpenMeteoClient.java index b8b7207..11b1b23 100644 --- a/src/main/java/com/lincfusion/udyamlinc/ingestion/client/OpenMeteoClient.java +++ b/src/main/java/com/lincfusion/udyamlinc/ingestion/client/OpenMeteoClient.java @@ -40,6 +40,19 @@ public class OpenMeteoClient { @Value("${udyam.ingestion.climate.timezone:Asia/Kolkata}") 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) { this.http = http; this.apiLogRepository = apiLogRepository; @@ -49,7 +62,6 @@ public class OpenMeteoClient { * The daily series for one grid cell. {@code null} when the call failed, * empty when the archive held nothing for the window. */ - @SuppressWarnings("unchecked") public List fetchDaily(double latitude, double longitude, LocalDate from, LocalDate to) { URI uri = UriComponentsBuilder.fromUriString(baseUrl) .queryParam("latitude", latitude) @@ -62,7 +74,29 @@ public class OpenMeteoClient { .encode() .toUri(); + for (int attempt = 1; ; attempt++) { + List 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 attempt(URI uri, double latitude, double longitude, + LocalDate from, LocalDate to, int attemptNo) { long started = System.currentTimeMillis(); + lastWasRateLimited = false; try { Map body = http.get().uri(uri).retrieve().body(Map.class); Map daily = body == null ? null : (Map) body.get("daily"); @@ -87,8 +121,12 @@ public class OpenMeteoClient { return out; } catch (Exception e) { - logCall(uri, null, started, null, false, e.getMessage()); - log.warn("Open-Meteo call failed for {},{} {}..{}: {}", latitude, longitude, from, to, e.toString()); + String message = String.valueOf(e.getMessage()); + 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; } } diff --git a/src/main/java/com/lincfusion/udyamlinc/seed/UdyamDataSeeder.java b/src/main/java/com/lincfusion/udyamlinc/seed/UdyamDataSeeder.java index b80b7c5..4ca78a3 100644 --- a/src/main/java/com/lincfusion/udyamlinc/seed/UdyamDataSeeder.java +++ b/src/main/java/com/lincfusion/udyamlinc/seed/UdyamDataSeeder.java @@ -147,7 +147,11 @@ public class UdyamDataSeeder implements CommandLineRunner { commodity("GRAM", "Gram", "चना", "AGMARKNET", "Bengal Gram(Gram)(Whole)", "QUINTAL"), commodity("EGG", "Egg", "अंडा", "NECC", "Bhopal", "HUNDRED"), 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()); } @@ -283,7 +287,10 @@ public class UdyamDataSeeder implements CommandLineRunner { private static final BigDecimal PINNED_SHORTFALL = new BigDecimal("12400"); private void seedPortfolio() { - if (enterpriseRepository.count() > 0) return; + if (enterpriseRepository.count() > 0) { + backfillPortfolio(); + return; + } Map sectors = new HashMap<>(); 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 * carry. DPD tracks the band, because in the demo the band is the ground diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 669d02c..82ddd79 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -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. udyam.ingestion.datagov.api-key=${DATAGOV_API_KEY:} udyam.ingestion.datagov.base-url=https://api.data.gov.in/resource -# "Variety-wise Daily Market Prices Data of Commodity" (Agmarknet) -udyam.ingestion.datagov.resource-id=${DATAGOV_RESOURCE_ID:9ef84268-d588-465a-a308-a864a43d0070} +# "Variety-wise Daily Market Prices Data of Commodity" — the historical Agmarknet +# 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 -# Filter names must match the resource's own field names; they differ by resource version. -udyam.ingestion.datagov.field.state=state -udyam.ingestion.datagov.field.district=district -udyam.ingestion.datagov.field.commodity=commodity +# Filter names must match the resource's own field names; they differ by resource +# version. This resource capitalises them, and — unlike the current-price +# resource — matches exactly on the bare name. Appending ".keyword" here returns +# 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 # 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.baseline-from=1995-01-01 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) ──────────────────── # Window sizes for the derivations. A level is not a feature; every one of these