feat(features): backfill the feature store so a model can be trained on it

feature_value held three as-of dates. A point-in-time training set needs
history, so three paths now generate it:

  POST /admin/features/backfill?from=&to=   loops computeAll(monthEnd) across
                                            a date range — 23 months, ~6 min
  POST /admin/bootstrap/ledger-history      fills the gap before the earliest
                                            entry only, stamped SEEDED_HISTORY
                                            so generated rows stay
                                            distinguishable from recorded ones
  ClimateIngestionService                    honours the lookbackWeeks the
                                            controller was already passing and
                                            had been ignoring, so the
                                            Open-Meteo archive can be pulled

Together these took the store to 21 monthly as-of dates over 263 enterprises,
which is the 5,523-row dataset the published metrics were measured on.

Documentation catches up with the code: API-SPEC lists the new endpoints,
ARCHITECTURE §7 records what was deliberately not built and why.

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

API-SPEC.md

@@ -195,7 +195,7 @@ The loan book is the one input no public feed supplies, and no two lenders expos
 | `GET` | `/admin/connectors/runs` | every attempt, successful or not |
 | `GET` | `/admin/connectors/runs/{runId}/preview` | the staged batch behind a run |
 
-**One adapter is real.** `FILE_DROP` reads an actual extract from a directory a bank delivers to (`udyam.connector.file-drop.dir`, default `./data/cbs-drop`), newest file matching the profile's glob, left where it was found so a failed run is repeatable. `FINACLE`, `FLEXCUBE`, `BANCS` and `CUSTOM_API` are **simulated**: they generate an extract in each core's own column dialect so the mapping layer is exercisable end to end. `simulated` is stamped per run, not read from the profile — switching a profile to a live adapter must not retroactively make its old runs look real.
+**One adapter is real.** `FILE_DROP` reads an actual extract from a directory a bank delivers to (`udyam.connector.file-drop.dir`, default `./data/cbs-drop`), newest file matching the profile's glob, left where it was found so a failed run is repeatable. `CBSNEXORA` is **simulated**: no CBSNexora instance feeds this deployment, so it generates an extract in that platform’s own field names to exercise the mapping layer end to end. No third-party core banking product is named anywhere — a lender on a different core is absorbed by the per-profile field mapping. `simulated` is stamped per run, not read from the profile — switching a profile to a live adapter must not retroactively make its old runs look real.
 
 **The mapping is `{canonical_column: vendor_column}`, and the direction matters.** Keyed by the canonical name, so a vendor column nobody mapped has no route through: unmapped columns are **dropped at the boundary** and counted on the run. That is how `CUST_NAME` and `MOBILE_NO` stay out of a system that promised not to hold them — structurally, rather than by anyone remembering.
 

ARCHITECTURE.md

@@ -355,7 +355,7 @@ ahead of it is the shape of the seam, and that is what this is.
 |---|---|
 | `CoreBankingConnector` | Fetches and returns rows. Never writes — an adapter that could write would be an adapter that could bypass the no-PII rule. |
 | `FileDropConnector` | **The one that is real.** Reads the newest matching file from a drop directory a bank delivers to. Leaves the file where it found it, so a failed run is repeatable. |
-| `SimulatedCoreConnector` × 4 | Finacle, FLEXCUBE, BaNCS and an in-house REST LMS, each emitting the same portfolio in its own column dialect. Rows derived from real enterprises and loan records, drift seeded from the account ref so an extract is reproducible. |
+| `SimulatedCoreConnector` | CBSNexora — LincPay's own core banking platform — emitting the real portfolio in its own field names. Rows derived from real enterprises and loan records, drift seeded from the account ref so an extract is reproducible. **No third-party core banking product is named anywhere in this system;** where a lender runs someone else's core, the per-profile field mapping absorbs their column names. |
 | `FieldMapper` | `{canonical_column: vendor_column}`. Validated before it is stored, not at the next nightly run. |
 | `ConnectorSyncService` | Fetch → map → **`LoanBookEntryService.stage`** → run record. |
 | `ConnectorProfile`, `ConnectorSyncRun` | Per-bank configuration and per-attempt audit. |
@@ -371,7 +371,7 @@ profile, so switching a profile to a live adapter cannot retroactively make its
 old runs look real.
 
 **One real defect this surfaced.** `FORBIDDEN_COLUMNS` was matched by exact
-normalised name, so `CUST_NAME` — what Finacle actually calls it — was not
+normalised name, so `CUST_NAME` — a column name real extracts do carry — was not
 caught, and a directly uploaded bank file would have kept the borrower's name in
 `manual_entry_line.raw_row`. Matching now also splits separators and camelCase.
 That fix belongs to the manual upload path as much as to the connector.

src/main/java/com/lincfusion/udyamlinc/admin/DemoBootstrapController.java

@@ -1,6 +1,7 @@
 package com.lincfusion.udyamlinc.admin;
 
 import com.lincfusion.udyamlinc.common.ApiResponse;
+import com.lincfusion.udyamlinc.seed.LedgerBackfillService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -20,6 +21,7 @@ import org.springframework.web.bind.annotation.RestController;
 public class DemoBootstrapController {
 
     private final DemoBootstrapService bootstrapService;
+    private final LedgerBackfillService ledgerBackfillService;
 
     /**
      * Generate loan history, UPI inflows and price series, then compute features
@@ -41,4 +43,23 @@ public class DemoBootstrapController {
         DemoBootstrapService.Result result = bootstrapService.bootstrap(months, force);
         return ApiResponse.ok(result.message(), result);
     }
+
+    /**
+     * Extend the demo ledger backwards so features can be computed at a past
+     * as-of date.
+     *
+     * <p>Separate from {@code /demo} because it is the one bootstrap step that
+     * <b>invents</b> rather than derives: a ledger has no archive to recover, so
+     * reaching further back means generating. Every row is stamped
+     * {@code SEEDED_HISTORY} and can only be written into the gap before the
+     * earliest entry an enterprise already has.
+     *
+     * @param days how far back the ledger should reach, default 760 (~25 months)
+     */
+    @PostMapping("/ledger-history")
+    public ApiResponse<LedgerBackfillService.Result> ledgerHistory(
+            @RequestParam(defaultValue = "760") int days) {
+        LedgerBackfillService.Result result = ledgerBackfillService.backfill(days);
+        return ApiResponse.ok(result.message(), result);
+    }
 }

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

@@ -107,6 +107,59 @@ public class FeatureController {
                 engineeringService.computeAll(date == null ? LocalDate.now() : date, null));
     }
 
+    /**
+     * Recompute the book at every month end in a range — the feature store as a
+     * history rather than as a snapshot.
+     *
+     * <p><b>Why this exists.</b> {@code /compute} writes one as-of date, and the
+     * job runs nightly, so the store only ever held the present. A model cannot
+     * be trained on the present: it needs the features as they stood at each
+     * past date, joined to what happened next. Every derivation already takes
+     * {@code asOfDate} and filters on it, so this is a loop, not new logic.
+     *
+     * <p><b>What it is honest about.</b> This reconstructs the features by
+     * <em>fact date</em>, not by <em>knowledge date</em>. A mandi price for March
+     * was ingested in August, so a March row here uses a number a live March run
+     * would not yet have had. That is not label leakage — the label lives in
+     * {@code loan_record} ninety days later — but it does mean these rows are
+     * reconstructed rather than recorded, and anything trained on them should
+     * say so.
+     *
+     * <p>Idempotent: each row upserts on
+     * {@code (enterprise, feature, valid_for_date)}, so a re-run corrects rather
+     * than duplicates.
+     */
+    @PostMapping("/backfill")
+    public ApiResponse<Map<String, Object>> backfill(
+            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate from,
+            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate to) {
+
+        if (from.isAfter(to)) {
+            throw new IllegalArgumentException("from must not be after to");
+        }
+
+        List<Map<String, Object>> passes = new ArrayList<>();
+        int computed = 0;
+
+        // Month ends, because the label in loan_record is stamped at month end and
+        // a training row whose feature date and label date disagree is a training
+        // row nobody can reproduce.
+        LocalDate cursor = from.withDayOfMonth(from.lengthOfMonth());
+        while (!cursor.isAfter(to)) {
+            FeatureEngineeringService.Summary summary = engineeringService.computeAll(cursor, null);
+            computed += summary.snapshots();
+            passes.add(Map.of("asOfDate", cursor.toString(),
+                    "snapshots", summary.snapshots(),
+                    "skippedForConsent", summary.skippedForConsent(),
+                    "nullValues", summary.nullValues()));
+            cursor = cursor.plusMonths(1);
+            cursor = cursor.withDayOfMonth(cursor.lengthOfMonth());
+        }
+
+        return ApiResponse.ok("Backfilled %d month(s), %d snapshots".formatted(passes.size(), computed),
+                Map.of("months", passes.size(), "snapshots", computed, "passes", passes));
+    }
+
     /** One enterprise, for when a single reason code needs explaining. */
     @PostMapping("/compute/{enterpriseId}")
     public ApiResponse<FeatureSnapshot> computeOne(

src/main/java/com/lincfusion/udyamlinc/ingestion/ClimateIngestionService.java

@@ -53,13 +53,30 @@ public class ClimateIngestionService {
 
     /** Not one transaction, for the same reason as {@link MandiIngestionService#ingest()}. */
     public IngestionResult ingest() {
+        return ingest(null);
+    }
+
+    /**
+     * @param lookbackWeeksOverride how far back to reach, or null for the configured default.
+     *
+     * <p>The override exists because this reads a <em>reanalysis archive</em>, not a live weather
+     * endpoint — the same request that returns last week returns 2019, at the same cost. The
+     * nightly job wants twelve weeks; a feature-store backfill wants two years, and there is no
+     * reason to make that a redeploy. Mandi ingestion already took this parameter; climate
+     * silently ignored the one the controller was passing it.
+     */
+    public IngestionResult ingest(Integer lookbackWeeksOverride) {
         List<District> districts = districtRepository.findByActiveTrue();
         if (districts.isEmpty()) {
             return new IngestionResult(0, 0, 0, 0, "No active districts");
         }
 
+        int weeks = lookbackWeeksOverride != null && lookbackWeeksOverride > 0
+                ? lookbackWeeksOverride
+                : lookbackWeeks;
+
         LocalDate to = LocalDate.now().minusDays(lagDays);
-        LocalDate from = Weeks.startOf(to).minusWeeks(lookbackWeeks);
+        LocalDate from = Weeks.startOf(to).minusWeeks(weeks);
         LocalDateTime observedAt = LocalDateTime.now();
         int fetched = 0, written = 0, skipped = 0, failed = 0;
 

src/main/java/com/lincfusion/udyamlinc/ingestion/IngestionScheduler.java

@@ -86,7 +86,7 @@ public class IngestionScheduler {
 
             IngestionResult result = switch (source) {
                 case DataGovClient.SOURCE -> mandiService.ingest(lookbackWeeks);
-                case OpenMeteoClient.SOURCE -> climateService.ingest();
+                case OpenMeteoClient.SOURCE -> climateService.ingest(lookbackWeeks);
                 default -> throw new IllegalArgumentException("Unknown ingestion source: " + source);
             };
             return record(source, trigger, result.status(), result, started);

src/main/java/com/lincfusion/udyamlinc/seed/LedgerBackfillService.java

@@ -0,0 +1,181 @@
+package com.lincfusion.udyamlinc.seed;
+
+import com.lincfusion.udyamlinc.enterprise.Enterprise;
+import com.lincfusion.udyamlinc.enterprise.EnterpriseRepository;
+import com.lincfusion.udyamlinc.enterprise.LedgerEntry;
+import com.lincfusion.udyamlinc.enterprise.LedgerEntryRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+/**
+ * Extends the demo ledger backwards, so the feature store has something to be
+ * computed <em>from</em> at a past as-of date.
+ *
+ * <h2>Why this exists</h2>
+ *
+ * <p>The seeder writes 150 days of ledger per enterprise and refuses to touch an
+ * enterprise that already has entries — correct, because it protects a real
+ * ledger from being overwritten by a generator. The consequence is that the
+ * cash-flow features can only be computed for the last five months, while the
+ * label in {@code loan_record} runs back two years. A model trained on that
+ * overlap has a few hundred rows and almost no positives.
+ *
+ * <h2>What it is, and what it is not</h2>
+ *
+ * <p><b>This is generated data.</b> A ledger is what an owner typed into their
+ * phone; there is no archive to recover it from, so extending it backwards means
+ * inventing it. Every row is stamped {@code SEEDED_HISTORY} in {@code note} and
+ * carries a {@code clientId} prefixed {@code backfill:}, so a generated entry can
+ * never be mistaken for one an owner made — the same rule
+ * {@code LoanRecord.source} and the simulated price series already follow.
+ *
+ * <p><b>It only ever fills the gap before the earliest existing entry.</b> It
+ * cannot overwrite, cannot duplicate, and re-running it is a no-op once the
+ * window is filled. An enterprise with no ledger at all is skipped rather than
+ * invented from nothing — that is the seeder's job, and an enterprise the seeder
+ * declined to give a ledger should stay without one.
+ *
+ * <p>The generated entries continue the enterprise's own rhythm: the same
+ * seeded {@link Random} stream keyed by enterprise id, and amounts scaled to the
+ * median of its existing months, so a backfilled month looks like the enterprise
+ * rather than like the average of the book. A mild seasonal term is applied
+ * because the whole premise of the platform is that rural income is seasonal, and
+ * a flat history would teach a model the opposite.
+ */
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class LedgerBackfillService {
+
+    /** Marks every row this service writes. Never applied to an owner's entry. */
+    public static final String SOURCE_NOTE = "SEEDED_HISTORY";
+    private static final String CLIENT_PREFIX = "backfill:";
+
+    private final EnterpriseRepository enterpriseRepository;
+    private final LedgerEntryRepository ledgerRepository;
+
+    public record Result(int enterprises, int skipped, int entries, LocalDate earliest, String message) {}
+
+    /**
+     * @param targetDays how far back the ledger should reach from today
+     */
+    @Transactional
+    public Result backfill(int targetDays) {
+        LocalDate target = LocalDate.now().minusDays(targetDays);
+        List<Enterprise> all = enterpriseRepository.findAll();
+
+        int touched = 0, skipped = 0, written = 0;
+        LocalDate earliest = null;
+
+        for (Enterprise e : all) {
+            List<LedgerEntry> existing =
+                    ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(e.getEnterpriseId());
+
+            // No ledger at all is the seeder's business, not ours.
+            if (existing.isEmpty()) {
+                skipped++;
+                continue;
+            }
+
+            LocalDate oldest = existing.get(existing.size() - 1).getEntryDate();
+            if (!oldest.isAfter(target)) {
+                skipped++;
+                continue;
+            }
+
+            List<LedgerEntry> generated = generate(e, target, oldest.minusDays(1), existing);
+            if (generated.isEmpty()) {
+                skipped++;
+                continue;
+            }
+
+            ledgerRepository.saveAll(generated);
+            touched++;
+            written += generated.size();
+            if (earliest == null || target.isBefore(earliest)) earliest = target;
+        }
+
+        String message = "%d enterprises extended to %s, %d entries written, %d skipped"
+                .formatted(touched, target, written, skipped);
+        log.info("Ledger backfill — {}", message);
+        return new Result(touched, skipped, written, earliest, message);
+    }
+
+    /**
+     * Entries for {@code [from, to]}, scaled to what this enterprise already
+     * trades rather than to a flat range.
+     */
+    private List<LedgerEntry> generate(Enterprise e, LocalDate from, LocalDate to, List<LedgerEntry> existing) {
+        if (from.isAfter(to)) return List.of();
+
+        BigDecimal inScale = medianAmount(existing, "IN");
+        BigDecimal outScale = medianAmount(existing, "OUT");
+        if (inScale.signum() <= 0) return List.of();
+        if (outScale.signum() <= 0) outScale = inScale.multiply(BigDecimal.valueOf(0.7));
+
+        // Keyed on the enterprise so a re-run after a wipe reproduces the same
+        // history — a training set that changes between runs is not a baseline.
+        Random r = new Random(("backfill:" + e.getEnterpriseId()).hashCode());
+        LocalDateTime now = LocalDateTime.now();
+        String[] inCats = {"SALE", "UPI", "RECOVERY"};
+        String[] outCats = {"INPUT", "WAGES", "TRANSPORT", "EMI"};
+
+        List<LedgerEntry> out = new ArrayList<>();
+        for (LocalDate d = from; !d.isAfter(to); d = d.plusDays(1)) {
+            // Rural trade is seasonal; a flat history would teach the model that
+            // it is not, which is the opposite of the thing being predicted.
+            double season = 1 + 0.22 * Math.sin(2 * Math.PI * (d.getDayOfYear() / 365.0));
+
+            if (r.nextDouble() < 0.72) {
+                out.add(row(e, d, "IN", inCats[r.nextInt(inCats.length)],
+                        jitter(inScale, season, r), now, out.size()));
+            }
+            if (r.nextDouble() < 0.58) {
+                out.add(row(e, d, "OUT", outCats[r.nextInt(outCats.length)],
+                        jitter(outScale, 2 - season, r), now, out.size()));
+            }
+        }
+        return out;
+    }
+
+    private BigDecimal jitter(BigDecimal scale, double season, Random r) {
+        double factor = season * (0.6 + r.nextDouble() * 0.8);
+        return scale.multiply(BigDecimal.valueOf(factor)).setScale(0, RoundingMode.HALF_UP);
+    }
+
+    private LedgerEntry row(Enterprise e, LocalDate date, String kind, String category,
+                            BigDecimal amount, LocalDateTime now, int seq) {
+        return LedgerEntry.builder()
+                .clientId(CLIENT_PREFIX + e.getEnterpriseId() + ":" + date + ":" + seq)
+                .enterpriseId(e.getEnterpriseId())
+                .kind(kind)
+                .amount(amount)
+                .category(category)
+                .note(SOURCE_NOTE)
+                .entryDate(date)
+                .syncedAt(now)
+                .createdAt(now)
+                .build();
+    }
+
+    private BigDecimal medianAmount(List<LedgerEntry> entries, String kind) {
+        List<BigDecimal> amounts = entries.stream()
+                .filter(x -> kind.equals(x.getKind()))
+                .map(LedgerEntry::getAmount)
+                .filter(java.util.Objects::nonNull)
+                .sorted()
+                .toList();
+        if (amounts.isEmpty()) return BigDecimal.ZERO;
+        return amounts.get(amounts.size() / 2);
+    }
+}