fix(features): attribute staleness to the data that built the feature

Staleness was district-wide: the newest market row for the district,
whatever commodity it belonged to, stamped onto every price feature. That
was a fair approximation while every commodity came from the same daily
Agmarknet pull. Manual entry ended it — milk and egg arrive monthly, so a
dairy enterprise's output_price_change, built entirely from a month-old
milk rate, inherited yesterday's maize freshness and sailed through the
staleness gate that exists to catch exactly that.

Each price feature now carries the age of its own oldest contributing
commodity, and margin_squeeze takes the staler of its two halves — a
squeeze is only as current as the older side of it.

Measured: output_price_change and margin_squeeze now report 3–24 days
where they previously reported 1–8, which is the truth rather than maize's
answer to a question about milk.

Catalogue tolerances for the price features go to 35 days, the slowest
source that legitimately feeds them. Ten days would gate every dairy and
poultry price feature out of every rule permanently — uploaded data
landing in the table and changing nothing. RuleEvaluator takes the
stricter of rule and catalogue, so this is the permissive floor and the
trip-wire remains where tightening belongs.

FeatureCatalog.seed now reconciles existing rows instead of only inserting
absent ones. Insert-if-absent left every database seeded before a
tolerance changed running on the old value, silently, in the direction
that decides whether a rule may fire at all.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/main/java/com/lincfusion/udyamlinc/features/FeatureCatalog.java

@@ -5,6 +5,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
+import java.util.Objects;
 
 /**
  * The engineered indicators of ARCHITECTURE §4, as constants and as rows.
@@ -74,13 +75,21 @@ public class FeatureCatalog {
     /** Idempotent: writes any catalogue row that is not already present. */
     public void seed() {
         List<FeatureDefinition> catalogue = List.of(
-                def(FEED_COST_CHANGE, "Feed cost change", EXTERNAL_SHOCK, "RATIO", "AGMARKNET", 10,
+                // Tolerances below are set to the slowest source that legitimately
+                // feeds each feature, because a single number has to cover them all.
+                // Milk and egg have no API and arrive by hand monthly, so a
+                // ten-day tolerance would gate every dairy and poultry price
+                // feature out of every rule, permanently — the data would land and
+                // change nothing. A rule that needs it stricter says so on itself:
+                // RuleEvaluator takes the stricter of the two, so this is the
+                // permissive floor and the trip-wire is where tightening belongs.
+                def(FEED_COST_CHANGE, "Feed cost change", EXTERNAL_SHOCK, "RATIO", "AGMARKNET", 35,
                         "HIGHER_IS_WORSE", "Input-side modal price vs its 8-week trailing mean, weighted across the sector's INPUT commodities"),
-                def(OUTPUT_PRICE_CHANGE, "Output price change", EXTERNAL_SHOCK, "RATIO", "AGMARKNET", 10,
-                        "LOWER_IS_WORSE", "Same derivation on the sector's OUTPUT commodities"),
-                def(MARGIN_SQUEEZE, "Margin squeeze", EXTERNAL_SHOCK, "RATIO", "AGMARKNET", 10,
+                def(OUTPUT_PRICE_CHANGE, "Output price change", EXTERNAL_SHOCK, "RATIO", "MIXED", 35,
+                        "LOWER_IS_WORSE", "Same derivation on the sector's OUTPUT commodities — milk and egg arrive by manual entry, monthly"),
+                def(MARGIN_SQUEEZE, "Margin squeeze", EXTERNAL_SHOCK, "RATIO", "MIXED", 35,
                         "HIGHER_IS_WORSE", "feed_cost_change − output_price_change; a feed rise matched by an output rise is not a squeeze"),
-                def(PRICE_Z_SCORE, "Seasonal price z-score", TREND_VOLATILITY, "INDEX", "AGMARKNET", 10,
+                def(PRICE_Z_SCORE, "Seasonal price z-score", TREND_VOLATILITY, "INDEX", "MIXED", 35,
                         "HIGHER_IS_WORSE", "(current − seasonal mean for this ISO week ±2) / stdev, over all held history"),
                 def(RAINFALL_ANOMALY, "Rainfall anomaly", EXTERNAL_SHOCK, "PCT", "CLIMATE", 14,
                         "LOWER_IS_WORSE", "Season-to-date rainfall over the trailing window vs the 30-year normal for the same weeks"),
@@ -104,13 +113,33 @@ public class FeatureCatalog {
                         "HIGHER_IS_WORSE", "Monthly instalment across open loans divided by the monthly ledger inflow rate"));
 
         int written = 0;
+        int reconciled = 0;
         for (FeatureDefinition candidate : catalogue) {
-            if (repository.findByCode(candidate.getCode()).isEmpty()) {
+            FeatureDefinition existing = repository.findByCode(candidate.getCode()).orElse(null);
+            if (existing == null) {
                 repository.save(candidate);
                 written++;
+                continue;
+            }
+            // Insert-if-absent alone would leave every database seeded before a
+            // tolerance changed running on the old one — silently, and in the
+            // direction that decides whether a trip-wire may fire at all. The
+            // catalogue is this code, not whatever was written the first time.
+            if (!Objects.equals(existing.getStalenessToleranceDays(), candidate.getStalenessToleranceDays())
+                    || !Objects.equals(existing.getSourceKind(), candidate.getSourceKind())
+                    || !Objects.equals(existing.getDerivation(), candidate.getDerivation())) {
+                log.info("Feature {} tolerance {} → {} days",
+                        existing.getCode(), existing.getStalenessToleranceDays(),
+                        candidate.getStalenessToleranceDays());
+                existing.setStalenessToleranceDays(candidate.getStalenessToleranceDays());
+                existing.setSourceKind(candidate.getSourceKind());
+                existing.setDerivation(candidate.getDerivation());
+                repository.save(existing);
+                reconciled++;
             }
         }
         if (written > 0) log.info("Seeded {} feature definitions", written);
+        if (reconciled > 0) log.info("Reconciled {} feature definitions with the catalogue", reconciled);
     }
 
     private FeatureDefinition def(String code, String label, String family, String unit, String sourceKind,

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

@@ -18,6 +18,7 @@ import java.time.LocalDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * Turns signals into indicators. The only writer of the feature store, and the
@@ -145,6 +146,7 @@ public class FeatureEngineeringService {
         int totalWeight = 0;
         boolean anyHistory = false;
         boolean anyRows = false;
+        Integer ownStaleness = null;
 
         for (SectorCommodity link : commodities) {
             List<MarketSignal> series = marketSignalRepository.findTrailing(
@@ -156,6 +158,14 @@ public class FeatureEngineeringService {
             int weight = link.getWeight() == null || link.getWeight() <= 0 ? 1 : link.getWeight();
             weightedSum = weightedSum.add(change.multiply(BigDecimal.valueOf(weight)));
             totalWeight += weight;
+
+            // The age of the data that actually built this feature — the oldest
+            // contributing commodity, not the district's newest row. Once sources
+            // run at different cadences, a monthly milk rate would otherwise
+            // inherit the freshness of yesterday's maize and sail through the very
+            // staleness gate that exists to catch it.
+            Integer age = staleness(series.get(0).getValidForDate(), asOfDate);
+            if (age != null && (ownStaleness == null || age > ownStaleness)) ownStaleness = age;
         }
 
         if (!anyHistory || totalWeight == 0) {
@@ -168,7 +178,10 @@ public class FeatureEngineeringService {
                     ? FeatureCatalog.INSUFFICIENT_HISTORY
                     : FeatureCatalog.NO_SIGNAL);
         }
-        return Computed.of(weightedSum.divide(BigDecimal.valueOf(totalWeight), 4, RoundingMode.HALF_UP), staleness);
+        // Falls back to the district figure only when no contributing row carried a
+        // date of its own.
+        return Computed.of(weightedSum.divide(BigDecimal.valueOf(totalWeight), 4, RoundingMode.HALF_UP),
+                ownStaleness == null ? staleness : ownStaleness);
     }
 
     /** (latest − mean of the prior window) / mean. Null until there is enough history to mean anything. */
@@ -191,7 +204,13 @@ public class FeatureEngineeringService {
         if (feedCost.value() == null || outputPrice.value() == null) {
             return Computed.missing(FeatureCatalog.NO_SIGNAL);
         }
-        return Computed.of(feedCost.value().subtract(outputPrice.value()), staleness);
+        // A squeeze is only as current as the staler of its two halves. Taking the
+        // fresher one would let a month-old milk rate hide behind yesterday's maize.
+        Integer worst = Stream.of(feedCost.stalenessDays(), outputPrice.stalenessDays(), staleness)
+                .filter(Objects::nonNull)
+                .max(Integer::compareTo)
+                .orElse(null);
+        return Computed.of(feedCost.value().subtract(outputPrice.value()), worst);
     }
 
     /**