chore: initial commit — UdyamLinc backend, phases A-E
Spring Boot 4 / Java 17 / Postgres. Masters and serving (A), Agmarknet and climate ingestion (B), feature store with 14 derivations (C), SpEL rule engine and risk scoring (D). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
LincPay
@@ -0,0 +1,123 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* The engineered indicators of ARCHITECTURE §4, as constants and as rows.
|
||||
*
|
||||
* <p>Raw prices are not features. An absolute maize price of ₹2,140 means
|
||||
* nothing; "feed cost 18 % above its eight-week mean while the egg rate is flat"
|
||||
* means something. Every derivation here turns a level into a change, a ratio or
|
||||
* an anomaly for exactly that reason.
|
||||
*
|
||||
* <p>Codes are snake_case because a trip-wire expression is written against them
|
||||
* directly: {@code emi_delay > 30 and cash_flow_change < -0.40}.
|
||||
*
|
||||
* <p>Three codes beyond §4's list are here because the seeded trip-wires already
|
||||
* name them — {@code dscr_p10}, {@code days_since_last_entry} and
|
||||
* {@code emi_to_inflow}. A rule referring to a feature nobody computes is a rule
|
||||
* that silently never fires, which is worse than no rule at all. They also give
|
||||
* the OBLIGATION_LOAD family the inputs it otherwise lacked.
|
||||
*/
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
public class FeatureCatalog {
|
||||
|
||||
// ── codes ───────────────────────────────────────────────
|
||||
|
||||
public static final String FEED_COST_CHANGE = "feed_cost_change";
|
||||
public static final String OUTPUT_PRICE_CHANGE = "output_price_change";
|
||||
public static final String MARGIN_SQUEEZE = "margin_squeeze";
|
||||
public static final String PRICE_Z_SCORE = "price_z_score";
|
||||
public static final String RAINFALL_ANOMALY = "rainfall_anomaly";
|
||||
public static final String HEAT_ANOMALY = "heat_anomaly";
|
||||
public static final String CASH_FLOW_CHANGE = "cash_flow_change";
|
||||
public static final String EMI_DELAY = "emi_delay";
|
||||
public static final String INVENTORY_DAYS = "inventory_days";
|
||||
public static final String UPI_INFLOW_CHANGE = "upi_inflow_change";
|
||||
public static final String FEED_STALENESS_DAYS = "feed_staleness_days";
|
||||
public static final String DSCR_P10 = "dscr_p10";
|
||||
public static final String DAYS_SINCE_LAST_ENTRY = "days_since_last_entry";
|
||||
public static final String EMI_TO_INFLOW = "emi_to_inflow";
|
||||
|
||||
public static final List<String> ALL = List.of(
|
||||
FEED_COST_CHANGE, OUTPUT_PRICE_CHANGE, MARGIN_SQUEEZE, PRICE_Z_SCORE,
|
||||
RAINFALL_ANOMALY, HEAT_ANOMALY, CASH_FLOW_CHANGE, EMI_DELAY,
|
||||
INVENTORY_DAYS, UPI_INFLOW_CHANGE, FEED_STALENESS_DAYS,
|
||||
DSCR_P10, DAYS_SINCE_LAST_ENTRY, EMI_TO_INFLOW);
|
||||
|
||||
// ── families ────────────────────────────────────────────
|
||||
|
||||
public static final String LIQUIDITY = "LIQUIDITY";
|
||||
public static final String BEHAVIOURAL = "BEHAVIOURAL";
|
||||
public static final String TREND_VOLATILITY = "TREND_VOLATILITY";
|
||||
public static final String OBLIGATION_LOAD = "OBLIGATION_LOAD";
|
||||
public static final String EXTERNAL_SHOCK = "EXTERNAL_SHOCK";
|
||||
public static final String QUALITY = "QUALITY";
|
||||
|
||||
// ── missing reasons ─────────────────────────────────────
|
||||
|
||||
public static final String NO_SIGNAL = "NO_SIGNAL";
|
||||
public static final String NO_LEDGER = "NO_LEDGER";
|
||||
public static final String NO_CBS = "NO_CBS";
|
||||
public static final String NO_CONSENT = "NO_CONSENT";
|
||||
public static final String NO_FORECAST = "NO_FORECAST";
|
||||
public static final String INSUFFICIENT_HISTORY = "INSUFFICIENT_HISTORY";
|
||||
|
||||
private final FeatureDefinitionRepository repository;
|
||||
|
||||
/** 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,
|
||||
"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,
|
||||
"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,
|
||||
"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"),
|
||||
def(HEAT_ANOMALY, "Heat anomaly", EXTERNAL_SHOCK, "INDEX", "CLIMATE", 14,
|
||||
"HIGHER_IS_WORSE", "Mean weekly max temperature minus the 30-year normal, over the trailing window"),
|
||||
def(CASH_FLOW_CHANGE, "Cash flow change", LIQUIDITY, "RATIO", "LEDGER", 7,
|
||||
"LOWER_IS_WORSE", "Ledger net position over the recent 4 weeks vs the prior 8-week mean"),
|
||||
def(EMI_DELAY, "EMI delay", BEHAVIOURAL, "DAYS", "CBS", 35,
|
||||
"HIGHER_IS_WORSE", "Worst DPD across open loans at the extract date"),
|
||||
def(INVENTORY_DAYS, "Inventory days", LIQUIDITY, "DAYS", "LEDGER", 14,
|
||||
"HIGHER_IS_WORSE", "Stock on hand divided by daily sales velocity over the recent window"),
|
||||
def(UPI_INFLOW_CHANGE, "UPI inflow change", LIQUIDITY, "RATIO", "UPI", 21,
|
||||
"LOWER_IS_WORSE", "Trailing 8-week consented inflow vs the 8 weeks before it"),
|
||||
def(FEED_STALENESS_DAYS, "Feed staleness", QUALITY, "DAYS", "AGMARKNET", null,
|
||||
"HIGHER_IS_WORSE", "Age of the newest market observation for the district — guards every price-derived feature above"),
|
||||
def(DSCR_P10, "DSCR at P10", OBLIGATION_LOAD, "RATIO", "FORECAST", 7,
|
||||
"LOWER_IS_WORSE", "Worst month of the horizon: min(p10 / obligation) over the latest scoring run — the downside case failing to cover what is due"),
|
||||
def(DAYS_SINCE_LAST_ENTRY, "Days since last ledger entry", BEHAVIOURAL, "DAYS", "LEDGER", 3,
|
||||
"HIGHER_IS_WORSE", "Days between the newest ledger entry date and the scoring date; silence is itself a signal"),
|
||||
def(EMI_TO_INFLOW, "Instalment share of inflow", OBLIGATION_LOAD, "RATIO", "CBS", 30,
|
||||
"HIGHER_IS_WORSE", "Monthly instalment across open loans divided by the monthly ledger inflow rate"));
|
||||
|
||||
int written = 0;
|
||||
for (FeatureDefinition candidate : catalogue) {
|
||||
if (repository.findByCode(candidate.getCode()).isEmpty()) {
|
||||
repository.save(candidate);
|
||||
written++;
|
||||
}
|
||||
}
|
||||
if (written > 0) log.info("Seeded {} feature definitions", written);
|
||||
}
|
||||
|
||||
private FeatureDefinition def(String code, String label, String family, String unit, String sourceKind,
|
||||
Integer staleness, String direction, String derivation) {
|
||||
return FeatureDefinition.builder()
|
||||
.code(code).label(label).family(family).unit(unit).sourceKind(sourceKind)
|
||||
.stalenessToleranceDays(staleness).direction(direction).derivation(derivation)
|
||||
.active(true).build();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import com.lincfusion.udyamlinc.common.ApiResponse;
|
||||
import com.lincfusion.udyamlinc.common.ResourceNotFoundException;
|
||||
import com.lincfusion.udyamlinc.enterprise.Enterprise;
|
||||
import com.lincfusion.udyamlinc.enterprise.EnterpriseRepository;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.springframework.format.annotation.DateTimeFormat;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* Phase C terminates in a table you can inspect; this is how you inspect it.
|
||||
*
|
||||
* <p>Under {@code /admin}, so reads are open to any officer and the recompute
|
||||
* trigger is supervisor-only under the existing rule.
|
||||
*/
|
||||
@RestController
|
||||
@RequestMapping("/admin/features")
|
||||
@RequiredArgsConstructor
|
||||
public class FeatureController {
|
||||
|
||||
private final FeatureDefinitionRepository definitionRepository;
|
||||
private final FeatureValueRepository valueRepository;
|
||||
private final FeatureSnapshotRepository snapshotRepository;
|
||||
private final FeatureEngineeringService engineeringService;
|
||||
private final EnterpriseRepository enterpriseRepository;
|
||||
|
||||
@GetMapping("/definitions")
|
||||
public ApiResponse<List<FeatureDefinition>> definitions() {
|
||||
return ApiResponse.ok(definitionRepository.findByActiveTrue());
|
||||
}
|
||||
|
||||
/** The wide row the engine reads — latest, or as it stood on a given date. */
|
||||
@GetMapping("/snapshot/{enterpriseId}")
|
||||
public ApiResponse<FeatureSnapshot> snapshot(
|
||||
@PathVariable String enterpriseId,
|
||||
@RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date) {
|
||||
|
||||
Optional<FeatureSnapshot> snapshot = date == null
|
||||
? snapshotRepository.findFirstByEnterpriseIdOrderByAsOfDateDesc(enterpriseId)
|
||||
: snapshotRepository.findByEnterpriseIdAndAsOfDate(enterpriseId, date);
|
||||
|
||||
return ApiResponse.ok(snapshot.orElseThrow(() ->
|
||||
new ResourceNotFoundException("No feature snapshot for " + enterpriseId
|
||||
+ (date == null ? "" : " at " + date))));
|
||||
}
|
||||
|
||||
/** The tall rows behind it, including the nulls and why they are null. */
|
||||
@GetMapping("/values/{enterpriseId}")
|
||||
public ApiResponse<List<FeatureValue>> values(
|
||||
@PathVariable String enterpriseId,
|
||||
@RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date) {
|
||||
|
||||
LocalDate asOf = date != null ? date
|
||||
: snapshotRepository.findFirstByEnterpriseIdOrderByAsOfDateDesc(enterpriseId)
|
||||
.map(FeatureSnapshot::getAsOfDate)
|
||||
.orElseThrow(() -> new ResourceNotFoundException("No features computed for " + enterpriseId));
|
||||
|
||||
return ApiResponse.ok(valueRepository.findByEnterpriseIdAndValidForDateOrderByFeatureCode(enterpriseId, asOf));
|
||||
}
|
||||
|
||||
/**
|
||||
* How much of the catalogue is actually populated across the book. A feature
|
||||
* at 4 % coverage is not a feature — it is a column, and the engine should be
|
||||
* told to stop weighting it.
|
||||
*/
|
||||
@GetMapping("/coverage")
|
||||
public ApiResponse<Map<String, Object>> coverage(
|
||||
@RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date) {
|
||||
|
||||
LocalDate asOf = date == null ? LocalDate.now() : date;
|
||||
List<FeatureSnapshot> snapshots = snapshotRepository.findByAsOfDate(asOf);
|
||||
|
||||
Map<String, Integer> present = new LinkedHashMap<>();
|
||||
FeatureCatalog.ALL.forEach(code -> present.put(code, 0));
|
||||
for (FeatureSnapshot snapshot : snapshots) {
|
||||
snapshot.asFeatureMap().forEach((code, value) -> {
|
||||
if (value != null) present.merge(code, 1, Integer::sum);
|
||||
});
|
||||
}
|
||||
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
body.put("asOfDate", asOf);
|
||||
body.put("snapshots", snapshots.size());
|
||||
Map<String, Object> byFeature = new LinkedHashMap<>();
|
||||
present.forEach((code, count) -> {
|
||||
Map<String, Object> row = new LinkedHashMap<>();
|
||||
row.put("populated", count);
|
||||
row.put("coveragePct", snapshots.isEmpty() ? 0
|
||||
: Math.round(count * 1000.0 / snapshots.size()) / 10.0);
|
||||
byFeature.put(code, row);
|
||||
});
|
||||
body.put("features", byFeature);
|
||||
return ApiResponse.ok(body);
|
||||
}
|
||||
|
||||
// ── trigger ─────────────────────────────────────────────
|
||||
|
||||
/** Recompute the whole book. Supervisor-only under the {@code /admin/**} write rule. */
|
||||
@PostMapping("/compute")
|
||||
public ApiResponse<FeatureEngineeringService.Summary> compute(
|
||||
@RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date) {
|
||||
return ApiResponse.ok("Features computed",
|
||||
engineeringService.computeAll(date == null ? LocalDate.now() : date, null));
|
||||
}
|
||||
|
||||
/** One enterprise, for when a single reason code needs explaining. */
|
||||
@PostMapping("/compute/{enterpriseId}")
|
||||
public ApiResponse<FeatureSnapshot> computeOne(
|
||||
@PathVariable String enterpriseId,
|
||||
@RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date) {
|
||||
|
||||
Enterprise enterprise = enterpriseRepository.findByEnterpriseId(enterpriseId)
|
||||
.orElseThrow(() -> new ResourceNotFoundException("No enterprise " + enterpriseId));
|
||||
return ApiResponse.ok("Features computed", engineeringService.computeFor(
|
||||
enterprise, date == null ? LocalDate.now() : date, java.time.LocalDateTime.now(), null));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import jakarta.persistence.*;
|
||||
import lombok.*;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* The feature catalogue. A row here is the contract for one engineered
|
||||
* indicator: what it means, what unit it carries, and how old it is allowed to
|
||||
* be before anything downstream may act on it.
|
||||
*
|
||||
* <p>{@code stalenessToleranceDays} is the column trip-wires read. A rule firing
|
||||
* on a three-week-old price is worse than no rule, and the tolerance is per
|
||||
* feature because a mandi price goes stale in days while a CBS extract is
|
||||
* monthly by design.
|
||||
*/
|
||||
@Entity
|
||||
@Table(name = "feature_definition", indexes = @Index(name = "ix_feature_def_code", columnList = "code", unique = true))
|
||||
@Getter
|
||||
@Setter
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
@Builder
|
||||
public class FeatureDefinition {
|
||||
|
||||
@Id
|
||||
@GeneratedValue(strategy = GenerationType.IDENTITY)
|
||||
private Long id;
|
||||
|
||||
/** snake_case, matching the name a trip-wire expression uses. */
|
||||
@Column(nullable = false, unique = true, length = 48)
|
||||
private String code;
|
||||
|
||||
@Column(nullable = false, length = 96)
|
||||
private String label;
|
||||
|
||||
/** LIQUIDITY | BEHAVIOURAL | TREND_VOLATILITY | OBLIGATION_LOAD | EXTERNAL_SHOCK | QUALITY */
|
||||
@Column(nullable = false, length = 32)
|
||||
private String family;
|
||||
|
||||
/** RATIO | PCT | DAYS | INDEX */
|
||||
@Column(length = 16)
|
||||
private String unit;
|
||||
|
||||
/** Which upstream table it derives from — AGMARKNET, CLIMATE, LEDGER, CBS, UPI. */
|
||||
@Column(length = 32)
|
||||
private String sourceKind;
|
||||
|
||||
/** Beyond this age the value must not be acted on. Null means it never goes stale. */
|
||||
private Integer stalenessToleranceDays;
|
||||
|
||||
/** HIGHER_IS_WORSE | LOWER_IS_WORSE */
|
||||
@Column(length = 24)
|
||||
private String direction;
|
||||
|
||||
@Column(length = 512)
|
||||
private String derivation;
|
||||
|
||||
private Boolean active;
|
||||
|
||||
private LocalDateTime updatedAt;
|
||||
|
||||
@PrePersist
|
||||
@PreUpdate
|
||||
void stamp() {
|
||||
if (active == null) active = Boolean.TRUE;
|
||||
updatedAt = LocalDateTime.now();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
public interface FeatureDefinitionRepository extends JpaRepository<FeatureDefinition, Long> {
|
||||
|
||||
Optional<FeatureDefinition> findByCode(String code);
|
||||
|
||||
List<FeatureDefinition> findByActiveTrue();
|
||||
|
||||
List<FeatureDefinition> findByFamilyAndActiveTrue(String family);
|
||||
}
|
||||
@@ -0,0 +1,535 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import com.lincfusion.udyamlinc.enterprise.*;
|
||||
import com.lincfusion.udyamlinc.ingestion.*;
|
||||
import com.lincfusion.udyamlinc.master.SectorCommodity;
|
||||
import com.lincfusion.udyamlinc.master.SectorCommodityRepository;
|
||||
import com.lincfusion.udyamlinc.risk.ForecastPoint;
|
||||
import com.lincfusion.udyamlinc.risk.ForecastPointRepository;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.math.RoundingMode;
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* Turns signals into indicators. The only writer of the feature store, and the
|
||||
* only path by which anything reaches the model — ingestion writes signals, this
|
||||
* writes features, the AI layer consumes features and nothing else. That is what
|
||||
* lets the model be retrained or replaced without touching ingestion or the API.
|
||||
*
|
||||
* <p>Every read here is point-in-time: each repository query filters on both
|
||||
* {@code validForDate <= asOfDate} and {@code observedAt <= observedAsOf}, so
|
||||
* recomputing last month's snapshot uses only what was actually knowable last
|
||||
* month. Without that, a backtest looks excellent and production does not
|
||||
* reproduce it.
|
||||
*
|
||||
* <p>A feature that cannot be computed is written as null with a reason, not
|
||||
* skipped. "We did not know" and "we never looked" are different facts, and the
|
||||
* forecast widens honestly only if the first one is recorded.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
public class FeatureEngineeringService {
|
||||
|
||||
private final EnterpriseRepository enterpriseRepository;
|
||||
private final SectorCommodityRepository sectorCommodityRepository;
|
||||
private final MarketSignalRepository marketSignalRepository;
|
||||
private final ClimateSignalRepository climateSignalRepository;
|
||||
private final LedgerEntryRepository ledgerRepository;
|
||||
private final LoanRecordRepository loanRepository;
|
||||
private final UpiSummaryRepository upiRepository;
|
||||
private final ForecastPointRepository forecastPointRepository;
|
||||
private final FeatureValueRepository valueRepository;
|
||||
private final FeatureSnapshotRepository snapshotRepository;
|
||||
private final FeatureCatalog catalog;
|
||||
|
||||
/** Weeks of price history averaged for a "vs trailing mean" change. */
|
||||
@Value("${udyam.features.price-window-weeks:8}")
|
||||
private int priceWindowWeeks;
|
||||
|
||||
/** Recent window for ledger and climate aggregates. */
|
||||
@Value("${udyam.features.recent-window-weeks:4}")
|
||||
private int recentWindowWeeks;
|
||||
|
||||
@Value("${udyam.features.climate-window-weeks:8}")
|
||||
private int climateWindowWeeks;
|
||||
|
||||
/** Minimum observations before a z-score is meaningful rather than noise. */
|
||||
@Value("${udyam.features.min-zscore-samples:6}")
|
||||
private int minZScoreSamples;
|
||||
|
||||
// ── entry points ────────────────────────────────────────
|
||||
|
||||
/** One pass over the whole book. Consent-revoked enterprises are skipped, not computed. */
|
||||
public Summary computeAll(LocalDate asOfDate, Long scoringRunId) {
|
||||
catalog.seed();
|
||||
LocalDateTime observedAsOf = LocalDateTime.now();
|
||||
int computed = 0, skipped = 0, missing = 0;
|
||||
|
||||
for (Enterprise enterprise : enterpriseRepository.findAll()) {
|
||||
if (Boolean.FALSE.equals(enterprise.getConsentActive())) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
FeatureSnapshot snapshot = computeFor(enterprise, asOfDate, observedAsOf, scoringRunId);
|
||||
computed++;
|
||||
missing += snapshot.getMissingCount() == null ? 0 : snapshot.getMissingCount();
|
||||
}
|
||||
|
||||
String message = "%d snapshots at %s · %d skipped for consent · %d null feature values"
|
||||
.formatted(computed, asOfDate, skipped, missing);
|
||||
log.info("Feature engineering — {}", message);
|
||||
return new Summary(computed, skipped, missing, message);
|
||||
}
|
||||
|
||||
/** The eleven catalogue features for one enterprise, tall rows and a wide snapshot. */
|
||||
public FeatureSnapshot computeFor(Enterprise enterprise, LocalDate asOfDate,
|
||||
LocalDateTime observedAsOf, Long scoringRunId) {
|
||||
|
||||
Map<String, Computed> features = new LinkedHashMap<>();
|
||||
|
||||
List<SectorCommodity> inputs = sectorCommodityRepository.findBySectorIdAndRole(enterprise.getSectorId(), "INPUT");
|
||||
List<SectorCommodity> outputs = sectorCommodityRepository.findBySectorIdAndRole(enterprise.getSectorId(), "OUTPUT");
|
||||
|
||||
Integer marketStaleness = marketStaleness(enterprise.getDistrictId(), asOfDate);
|
||||
|
||||
Computed feedCost = weightedPriceChange(enterprise.getDistrictId(), inputs, asOfDate, observedAsOf, marketStaleness);
|
||||
Computed outputPrice = weightedPriceChange(enterprise.getDistrictId(), outputs, asOfDate, observedAsOf, marketStaleness);
|
||||
|
||||
features.put(FeatureCatalog.FEED_COST_CHANGE, feedCost);
|
||||
features.put(FeatureCatalog.OUTPUT_PRICE_CHANGE, outputPrice);
|
||||
features.put(FeatureCatalog.MARGIN_SQUEEZE, marginSqueeze(feedCost, outputPrice, marketStaleness));
|
||||
features.put(FeatureCatalog.PRICE_Z_SCORE,
|
||||
priceZScore(enterprise.getDistrictId(), outputs, asOfDate, observedAsOf, marketStaleness));
|
||||
features.put(FeatureCatalog.RAINFALL_ANOMALY, rainfallAnomaly(enterprise.getDistrictId(), asOfDate, observedAsOf));
|
||||
features.put(FeatureCatalog.HEAT_ANOMALY, heatAnomaly(enterprise.getDistrictId(), asOfDate, observedAsOf));
|
||||
features.put(FeatureCatalog.CASH_FLOW_CHANGE, cashFlowChange(enterprise.getEnterpriseId(), asOfDate));
|
||||
features.put(FeatureCatalog.EMI_DELAY, emiDelay(enterprise.getEnterpriseId(), asOfDate));
|
||||
features.put(FeatureCatalog.INVENTORY_DAYS, inventoryDays(enterprise.getEnterpriseId(), asOfDate));
|
||||
features.put(FeatureCatalog.UPI_INFLOW_CHANGE, upiInflowChange(enterprise.getEnterpriseId(), asOfDate, observedAsOf));
|
||||
features.put(FeatureCatalog.FEED_STALENESS_DAYS, marketStaleness == null
|
||||
? Computed.missing(FeatureCatalog.NO_SIGNAL)
|
||||
: Computed.of(BigDecimal.valueOf(marketStaleness), marketStaleness));
|
||||
features.put(FeatureCatalog.DAYS_SINCE_LAST_ENTRY, daysSinceLastEntry(enterprise.getEnterpriseId(), asOfDate));
|
||||
features.put(FeatureCatalog.EMI_TO_INFLOW, emiToInflow(enterprise.getEnterpriseId(), asOfDate));
|
||||
features.put(FeatureCatalog.DSCR_P10, dscrP10(enterprise.getEnterpriseId(), asOfDate));
|
||||
|
||||
return persist(enterprise.getEnterpriseId(), asOfDate, observedAsOf, scoringRunId, features);
|
||||
}
|
||||
|
||||
// ── price derivations ───────────────────────────────────
|
||||
|
||||
/**
|
||||
* A level is not a feature. This is the modal price against its trailing mean,
|
||||
* averaged across the sector's commodities for that role and weighted by how
|
||||
* much of the trade each one actually is.
|
||||
*/
|
||||
private Computed weightedPriceChange(Long districtId, List<SectorCommodity> commodities,
|
||||
LocalDate asOfDate, LocalDateTime observedAsOf, Integer staleness) {
|
||||
if (commodities.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
|
||||
BigDecimal weightedSum = BigDecimal.ZERO;
|
||||
int totalWeight = 0;
|
||||
boolean anyHistory = false;
|
||||
|
||||
for (SectorCommodity link : commodities) {
|
||||
List<MarketSignal> series = marketSignalRepository.findTrailing(
|
||||
districtId, link.getCommodityId(), asOfDate, observedAsOf);
|
||||
BigDecimal change = changeVsTrailingMean(series);
|
||||
if (change == null) continue;
|
||||
anyHistory = true;
|
||||
int weight = link.getWeight() == null || link.getWeight() <= 0 ? 1 : link.getWeight();
|
||||
weightedSum = weightedSum.add(change.multiply(BigDecimal.valueOf(weight)));
|
||||
totalWeight += weight;
|
||||
}
|
||||
|
||||
if (!anyHistory || totalWeight == 0) {
|
||||
return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
}
|
||||
return Computed.of(weightedSum.divide(BigDecimal.valueOf(totalWeight), 4, RoundingMode.HALF_UP), staleness);
|
||||
}
|
||||
|
||||
/** (latest − mean of the prior window) / mean. Null until there is enough history to mean anything. */
|
||||
private BigDecimal changeVsTrailingMean(List<MarketSignal> descendingByWeek) {
|
||||
List<BigDecimal> modal = descendingByWeek.stream()
|
||||
.map(MarketSignal::getModalPrice)
|
||||
.filter(Objects::nonNull)
|
||||
.toList();
|
||||
if (modal.size() < 4) return null;
|
||||
|
||||
BigDecimal latest = modal.get(0);
|
||||
List<BigDecimal> prior = modal.subList(1, Math.min(modal.size(), priceWindowWeeks + 1));
|
||||
BigDecimal mean = mean(prior);
|
||||
if (mean == null || mean.signum() == 0) return null;
|
||||
return latest.subtract(mean).divide(mean, 4, RoundingMode.HALF_UP);
|
||||
}
|
||||
|
||||
/** A feed rise matched by an output rise is not a squeeze; the difference is. */
|
||||
private Computed marginSqueeze(Computed feedCost, Computed outputPrice, Integer staleness) {
|
||||
if (feedCost.value() == null || outputPrice.value() == null) {
|
||||
return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
}
|
||||
return Computed.of(feedCost.value().subtract(outputPrice.value()), staleness);
|
||||
}
|
||||
|
||||
/**
|
||||
* Seasonal z-score on the sale-side commodity: this week against the same week
|
||||
* of the year across all held history, ±2 weeks for sample size. Prices are
|
||||
* seasonal, so comparing March with the annual mean would call every harvest a
|
||||
* shock.
|
||||
*/
|
||||
private Computed priceZScore(Long districtId, List<SectorCommodity> outputs,
|
||||
LocalDate asOfDate, LocalDateTime observedAsOf, Integer staleness) {
|
||||
if (outputs.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
|
||||
SectorCommodity primary = outputs.stream()
|
||||
.max(Comparator.comparing(sc -> sc.getWeight() == null ? 0 : sc.getWeight()))
|
||||
.orElse(outputs.get(0));
|
||||
|
||||
List<MarketSignal> series = marketSignalRepository.findTrailing(
|
||||
districtId, primary.getCommodityId(), asOfDate, observedAsOf);
|
||||
if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
|
||||
MarketSignal latest = series.get(0);
|
||||
if (latest.getModalPrice() == null) return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
int week = Weeks.isoWeek(latest.getWeekStart());
|
||||
|
||||
List<BigDecimal> seasonal = series.stream()
|
||||
.skip(1)
|
||||
.filter(s -> s.getModalPrice() != null)
|
||||
.filter(s -> weekDistance(Weeks.isoWeek(s.getWeekStart()), week) <= 2)
|
||||
.map(MarketSignal::getModalPrice)
|
||||
.toList();
|
||||
|
||||
if (seasonal.size() < minZScoreSamples) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
BigDecimal mean = mean(seasonal);
|
||||
BigDecimal stdev = stdev(seasonal, mean);
|
||||
if (stdev == null || stdev.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
return Computed.of(latest.getModalPrice().subtract(mean).divide(stdev, 4, RoundingMode.HALF_UP), staleness);
|
||||
}
|
||||
|
||||
/** ISO weeks wrap, so week 52 and week 1 are two apart, not fifty-one. */
|
||||
private int weekDistance(int a, int b) {
|
||||
int raw = Math.abs(a - b);
|
||||
return Math.min(raw, 53 - raw);
|
||||
}
|
||||
|
||||
// ── climate derivations ─────────────────────────────────
|
||||
|
||||
private Computed rainfallAnomaly(Long districtId, LocalDate asOfDate, LocalDateTime observedAsOf) {
|
||||
List<ClimateSignal> series = climateSignalRepository.findTrailing(districtId, asOfDate, observedAsOf);
|
||||
if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
|
||||
List<ClimateSignal> window = series.subList(0, Math.min(series.size(), climateWindowWeeks));
|
||||
Integer staleness = staleness(window.get(0).getValidForDate(), asOfDate);
|
||||
|
||||
BigDecimal observed = BigDecimal.ZERO, normal = BigDecimal.ZERO;
|
||||
boolean anyNormal = false;
|
||||
for (ClimateSignal week : window) {
|
||||
if (week.getRainfallMm() != null) observed = observed.add(week.getRainfallMm());
|
||||
if (week.getRainfallNormalMm() != null) {
|
||||
normal = normal.add(week.getRainfallNormalMm());
|
||||
anyNormal = true;
|
||||
}
|
||||
}
|
||||
if (!anyNormal || normal.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
BigDecimal pct = observed.subtract(normal)
|
||||
.multiply(BigDecimal.valueOf(100))
|
||||
.divide(normal, 4, RoundingMode.HALF_UP);
|
||||
return Computed.of(pct, staleness);
|
||||
}
|
||||
|
||||
private Computed heatAnomaly(Long districtId, LocalDate asOfDate, LocalDateTime observedAsOf) {
|
||||
List<ClimateSignal> series = climateSignalRepository.findTrailing(districtId, asOfDate, observedAsOf);
|
||||
if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_SIGNAL);
|
||||
|
||||
List<ClimateSignal> window = series.subList(0, Math.min(series.size(), climateWindowWeeks));
|
||||
Integer staleness = staleness(window.get(0).getValidForDate(), asOfDate);
|
||||
|
||||
List<BigDecimal> anomalies = window.stream()
|
||||
.map(ClimateSignal::getHeatAnomalyC)
|
||||
.filter(Objects::nonNull)
|
||||
.toList();
|
||||
if (anomalies.isEmpty()) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
return Computed.of(mean(anomalies), staleness);
|
||||
}
|
||||
|
||||
// ── ledger derivations ──────────────────────────────────
|
||||
|
||||
/**
|
||||
* Net position over the recent weeks against the mean of the weeks before it.
|
||||
* Entries are filtered on {@code entryDate}, not sync time — an entry made
|
||||
* offline three weeks ago describes three weeks ago.
|
||||
*/
|
||||
private Computed cashFlowChange(String enterpriseId, LocalDate asOfDate) {
|
||||
List<LedgerEntry> entries = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
|
||||
.filter(e -> !e.getEntryDate().isAfter(asOfDate))
|
||||
.toList();
|
||||
if (entries.isEmpty()) return Computed.missing(FeatureCatalog.NO_LEDGER);
|
||||
|
||||
Integer staleness = staleness(entries.get(0).getEntryDate(), asOfDate);
|
||||
LocalDate recentFrom = asOfDate.minusWeeks(recentWindowWeeks);
|
||||
LocalDate priorFrom = recentFrom.minusWeeks(priceWindowWeeks);
|
||||
|
||||
BigDecimal recent = net(entries, recentFrom, asOfDate);
|
||||
BigDecimal prior = net(entries, priorFrom, recentFrom);
|
||||
if (prior.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
// Both windows normalised to a weekly rate before comparison.
|
||||
BigDecimal recentRate = recent.divide(BigDecimal.valueOf(recentWindowWeeks), 4, RoundingMode.HALF_UP);
|
||||
BigDecimal priorRate = prior.divide(BigDecimal.valueOf(priceWindowWeeks), 4, RoundingMode.HALF_UP);
|
||||
if (priorRate.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
BigDecimal change = recentRate.subtract(priorRate).divide(priorRate.abs(), 4, RoundingMode.HALF_UP);
|
||||
return Computed.of(change, staleness);
|
||||
}
|
||||
|
||||
private BigDecimal net(List<LedgerEntry> entries, LocalDate from, LocalDate to) {
|
||||
BigDecimal sum = BigDecimal.ZERO;
|
||||
for (LedgerEntry e : entries) {
|
||||
if (e.getEntryDate().isBefore(from) || !e.getEntryDate().isBefore(to)) continue;
|
||||
if ("IN".equals(e.getKind())) sum = sum.add(e.getAmount());
|
||||
else if ("OUT".equals(e.getKind())) sum = sum.subtract(e.getAmount());
|
||||
}
|
||||
return sum;
|
||||
}
|
||||
|
||||
/** Stock on hand against sales velocity. Rising inventory days is cash turning into unsold goods. */
|
||||
private Computed inventoryDays(String enterpriseId, LocalDate asOfDate) {
|
||||
LocalDate from = asOfDate.minusWeeks(recentWindowWeeks);
|
||||
List<LedgerEntry> entries = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
|
||||
.filter(e -> !e.getEntryDate().isAfter(asOfDate) && !e.getEntryDate().isBefore(from))
|
||||
.toList();
|
||||
if (entries.isEmpty()) return Computed.missing(FeatureCatalog.NO_LEDGER);
|
||||
|
||||
Integer staleness = staleness(entries.get(0).getEntryDate(), asOfDate);
|
||||
BigDecimal stock = BigDecimal.ZERO, sales = BigDecimal.ZERO;
|
||||
for (LedgerEntry e : entries) {
|
||||
if ("STOCK".equals(e.getKind())) stock = stock.add(e.getAmount());
|
||||
else if ("IN".equals(e.getKind())) sales = sales.add(e.getAmount());
|
||||
}
|
||||
if (sales.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
long days = ChronoUnit.DAYS.between(from, asOfDate);
|
||||
BigDecimal dailySales = sales.divide(BigDecimal.valueOf(Math.max(days, 1)), 4, RoundingMode.HALF_UP);
|
||||
if (dailySales.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
return Computed.of(stock.divide(dailySales, 4, RoundingMode.HALF_UP), staleness);
|
||||
}
|
||||
|
||||
/** Silence is a signal. An enterprise that stops recording has usually stopped trading normally. */
|
||||
private Computed daysSinceLastEntry(String enterpriseId, LocalDate asOfDate) {
|
||||
Optional<LocalDate> newest = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
|
||||
.map(LedgerEntry::getEntryDate)
|
||||
.filter(date -> !date.isAfter(asOfDate))
|
||||
.max(LocalDate::compareTo);
|
||||
return newest
|
||||
.map(date -> {
|
||||
int days = staleness(date, asOfDate);
|
||||
return Computed.of(BigDecimal.valueOf(days), days);
|
||||
})
|
||||
.orElseGet(() -> Computed.missing(FeatureCatalog.NO_LEDGER));
|
||||
}
|
||||
|
||||
// ── institutional derivations ───────────────────────────
|
||||
|
||||
private Computed emiDelay(String enterpriseId, LocalDate asOfDate) {
|
||||
List<LoanRecord> loans = loanRepository
|
||||
.findByEnterpriseIdAndAsOfDateLessThanEqual(enterpriseId, asOfDate).stream()
|
||||
.filter(l -> !Boolean.TRUE.equals(l.getClosed()))
|
||||
.toList();
|
||||
if (loans.isEmpty()) return Computed.missing(FeatureCatalog.NO_CBS);
|
||||
|
||||
LocalDate newest = loans.stream().map(LoanRecord::getAsOfDate).max(LocalDate::compareTo).orElse(asOfDate);
|
||||
int worstDpd = loans.stream().map(LoanRecord::getDpd).filter(Objects::nonNull)
|
||||
.mapToInt(Integer::intValue).max().orElse(0);
|
||||
return Computed.of(BigDecimal.valueOf(worstDpd), staleness(newest, asOfDate));
|
||||
}
|
||||
|
||||
private Computed upiInflowChange(String enterpriseId, LocalDate asOfDate, LocalDateTime observedAsOf) {
|
||||
List<UpiSummary> series = upiRepository.findTrailing(enterpriseId, asOfDate, observedAsOf);
|
||||
if (series.isEmpty()) return Computed.missing(FeatureCatalog.NO_CONSENT);
|
||||
|
||||
Integer staleness = staleness(series.get(0).getValidForDate(), asOfDate);
|
||||
int window = priceWindowWeeks;
|
||||
if (series.size() < window * 2) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
BigDecimal recent = sumInflow(series.subList(0, window));
|
||||
BigDecimal prior = sumInflow(series.subList(window, Math.min(series.size(), window * 2)));
|
||||
if (prior.signum() == 0) return Computed.missing(FeatureCatalog.INSUFFICIENT_HISTORY);
|
||||
|
||||
return Computed.of(recent.subtract(prior).divide(prior, 4, RoundingMode.HALF_UP), staleness);
|
||||
}
|
||||
|
||||
/**
|
||||
* What the instalment takes out of what comes in. Inflow is the ledger's own
|
||||
* money-in over the recent window, normalised to a month — the enterprise's
|
||||
* record of its trade, not an institutional estimate of it.
|
||||
*/
|
||||
private Computed emiToInflow(String enterpriseId, LocalDate asOfDate) {
|
||||
List<LoanRecord> loans = loanRepository
|
||||
.findByEnterpriseIdAndAsOfDateLessThanEqual(enterpriseId, asOfDate).stream()
|
||||
.filter(l -> !Boolean.TRUE.equals(l.getClosed()))
|
||||
.toList();
|
||||
if (loans.isEmpty()) return Computed.missing(FeatureCatalog.NO_CBS);
|
||||
|
||||
BigDecimal instalment = loans.stream()
|
||||
.map(LoanRecord::getInstalmentAmount)
|
||||
.filter(Objects::nonNull)
|
||||
.reduce(BigDecimal.ZERO, BigDecimal::add);
|
||||
if (instalment.signum() == 0) return Computed.missing(FeatureCatalog.NO_CBS);
|
||||
|
||||
LocalDate from = asOfDate.minusWeeks(recentWindowWeeks);
|
||||
BigDecimal inflow = ledgerRepository.findByEnterpriseIdOrderByEntryDateDesc(enterpriseId).stream()
|
||||
.filter(e -> "IN".equals(e.getKind()))
|
||||
.filter(e -> !e.getEntryDate().isAfter(asOfDate) && !e.getEntryDate().isBefore(from))
|
||||
.map(LedgerEntry::getAmount)
|
||||
.reduce(BigDecimal.ZERO, BigDecimal::add);
|
||||
if (inflow.signum() == 0) return Computed.missing(FeatureCatalog.NO_LEDGER);
|
||||
|
||||
long days = Math.max(ChronoUnit.DAYS.between(from, asOfDate), 1);
|
||||
BigDecimal monthlyInflow = inflow.divide(BigDecimal.valueOf(days), 4, RoundingMode.HALF_UP)
|
||||
.multiply(BigDecimal.valueOf(30));
|
||||
if (monthlyInflow.signum() == 0) return Computed.missing(FeatureCatalog.NO_LEDGER);
|
||||
|
||||
LocalDate newest = loans.stream().map(LoanRecord::getAsOfDate).max(LocalDate::compareTo).orElse(asOfDate);
|
||||
return Computed.of(instalment.divide(monthlyInflow, 4, RoundingMode.HALF_UP), staleness(newest, asOfDate));
|
||||
}
|
||||
|
||||
/**
|
||||
* The product's entire trigger, as a number: the worst month of the horizon
|
||||
* where the downside case has to cover what is due. Read from the latest
|
||||
* scoring run's stored quantile path, so it reflects the engine that actually
|
||||
* produced the forecast an officer saw.
|
||||
*/
|
||||
private Computed dscrP10(String enterpriseId, LocalDate asOfDate) {
|
||||
List<ForecastPoint> points = forecastPointRepository.findByEnterpriseIdOrderByHorizonAsc(enterpriseId);
|
||||
if (points.isEmpty()) return Computed.missing(FeatureCatalog.NO_FORECAST);
|
||||
|
||||
Long latestRun = points.stream().map(ForecastPoint::getRunId)
|
||||
.filter(Objects::nonNull).max(Long::compareTo).orElse(null);
|
||||
if (latestRun == null) return Computed.missing(FeatureCatalog.NO_FORECAST);
|
||||
|
||||
BigDecimal worst = null;
|
||||
for (ForecastPoint point : points) {
|
||||
if (!latestRun.equals(point.getRunId())) continue;
|
||||
if (point.getP10() == null || point.getObligation() == null || point.getObligation().signum() == 0) continue;
|
||||
BigDecimal ratio = point.getP10().divide(point.getObligation(), 4, RoundingMode.HALF_UP);
|
||||
if (worst == null || ratio.compareTo(worst) < 0) worst = ratio;
|
||||
}
|
||||
return worst == null ? Computed.missing(FeatureCatalog.NO_FORECAST) : Computed.of(worst, 0);
|
||||
}
|
||||
|
||||
private BigDecimal sumInflow(List<UpiSummary> weeks) {
|
||||
return weeks.stream()
|
||||
.map(UpiSummary::getInflowAmount)
|
||||
.filter(Objects::nonNull)
|
||||
.reduce(BigDecimal.ZERO, BigDecimal::add);
|
||||
}
|
||||
|
||||
// ── persistence ─────────────────────────────────────────
|
||||
|
||||
private FeatureSnapshot persist(String enterpriseId, LocalDate asOfDate, LocalDateTime observedAsOf,
|
||||
Long scoringRunId, Map<String, Computed> features) {
|
||||
|
||||
FeatureSnapshot snapshot = snapshotRepository.findByEnterpriseIdAndAsOfDate(enterpriseId, asOfDate)
|
||||
.orElseGet(() -> FeatureSnapshot.builder()
|
||||
.enterpriseId(enterpriseId)
|
||||
.asOfDate(asOfDate)
|
||||
.build());
|
||||
|
||||
int present = 0, missing = 0, maxStaleness = 0;
|
||||
|
||||
for (Map.Entry<String, Computed> entry : features.entrySet()) {
|
||||
String code = entry.getKey();
|
||||
Computed computed = entry.getValue();
|
||||
|
||||
FeatureValue row = valueRepository
|
||||
.findByEnterpriseIdAndFeatureCodeAndValidForDate(enterpriseId, code, asOfDate)
|
||||
.orElseGet(() -> FeatureValue.builder()
|
||||
.enterpriseId(enterpriseId)
|
||||
.featureCode(code)
|
||||
.validForDate(asOfDate)
|
||||
.build());
|
||||
row.setValue(computed.value());
|
||||
row.setStalenessDays(computed.stalenessDays());
|
||||
row.setMissingReason(computed.missingReason());
|
||||
row.setObservedAt(observedAsOf);
|
||||
valueRepository.save(row);
|
||||
|
||||
snapshot.put(code, computed.value());
|
||||
if (computed.value() == null) missing++;
|
||||
else present++;
|
||||
if (computed.stalenessDays() != null) maxStaleness = Math.max(maxStaleness, computed.stalenessDays());
|
||||
}
|
||||
|
||||
snapshot.setScoringRunId(scoringRunId);
|
||||
snapshot.setFeatureCount(present);
|
||||
snapshot.setMissingCount(missing);
|
||||
snapshot.setMaxStalenessDays(maxStaleness);
|
||||
snapshot.setComputedAt(observedAsOf);
|
||||
return snapshotRepository.save(snapshot);
|
||||
}
|
||||
|
||||
// ── helpers ─────────────────────────────────────────────
|
||||
|
||||
private Integer marketStaleness(Long districtId, LocalDate asOfDate) {
|
||||
return marketSignalRepository.findFirstByDistrictIdOrderByValidForDateDesc(districtId)
|
||||
.map(signal -> staleness(signal.getValidForDate(), asOfDate))
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private Integer staleness(LocalDate validFor, LocalDate asOfDate) {
|
||||
if (validFor == null) return null;
|
||||
return (int) Math.max(0, ChronoUnit.DAYS.between(validFor, asOfDate));
|
||||
}
|
||||
|
||||
private BigDecimal mean(List<BigDecimal> values) {
|
||||
if (values == null || values.isEmpty()) return null;
|
||||
return values.stream().reduce(BigDecimal.ZERO, BigDecimal::add)
|
||||
.divide(BigDecimal.valueOf(values.size()), 4, RoundingMode.HALF_UP);
|
||||
}
|
||||
|
||||
/** Sample standard deviation — n−1, because these are samples of a season, not the population. */
|
||||
private BigDecimal stdev(List<BigDecimal> values, BigDecimal mean) {
|
||||
if (values.size() < 2 || mean == null) return null;
|
||||
BigDecimal sumSquares = BigDecimal.ZERO;
|
||||
for (BigDecimal v : values) {
|
||||
BigDecimal diff = v.subtract(mean);
|
||||
sumSquares = sumSquares.add(diff.multiply(diff));
|
||||
}
|
||||
double variance = sumSquares.divide(BigDecimal.valueOf(values.size() - 1L), 6, RoundingMode.HALF_UP)
|
||||
.doubleValue();
|
||||
return BigDecimal.valueOf(Math.sqrt(variance)).setScale(4, RoundingMode.HALF_UP);
|
||||
}
|
||||
|
||||
/**
|
||||
* One feature's outcome. A null value always carries a reason — that is the
|
||||
* difference between a hole we can explain and one we cannot.
|
||||
*/
|
||||
public record Computed(BigDecimal value, Integer stalenessDays, String missingReason) {
|
||||
|
||||
static Computed of(BigDecimal value, Integer stalenessDays) {
|
||||
return new Computed(value, stalenessDays, null);
|
||||
}
|
||||
|
||||
static Computed missing(String reason) {
|
||||
return new Computed(null, null, reason);
|
||||
}
|
||||
}
|
||||
|
||||
/** What one pass over the book did. */
|
||||
public record Summary(int snapshots, int skippedForConsent, int nullValues, String message) {}
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import jakarta.persistence.*;
|
||||
import lombok.*;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* The wide store — one row per (enterprise, scoring run). What the risk engine
|
||||
* and, later, the trainer actually read.
|
||||
*
|
||||
* <p>The tall table alone would mean thousands of row reads per scoring pass and
|
||||
* far worse for training; this alone would mean a migration every time a feature
|
||||
* is added. Both exist on purpose, and {@link FeatureValue} remains the system of
|
||||
* record — this is a materialised read.
|
||||
*
|
||||
* <p>{@link #asFeatureMap()} exposes the same values under their snake_case
|
||||
* catalogue codes, which is what a trip-wire expression is written against.
|
||||
*/
|
||||
@Entity
|
||||
@Table(name = "feature_snapshot", indexes = {
|
||||
@Index(name = "ix_feature_snap_key", columnList = "enterpriseId,asOfDate", unique = true),
|
||||
@Index(name = "ix_feature_snap_run", columnList = "scoringRunId"),
|
||||
})
|
||||
@Getter
|
||||
@Setter
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
@Builder
|
||||
public class FeatureSnapshot {
|
||||
|
||||
@Id
|
||||
@GeneratedValue(strategy = GenerationType.IDENTITY)
|
||||
private Long id;
|
||||
|
||||
@Column(nullable = false, length = 24)
|
||||
private String enterpriseId;
|
||||
|
||||
@Column(nullable = false)
|
||||
private LocalDate asOfDate;
|
||||
|
||||
/** Set when the snapshot was built for a specific scoring pass. */
|
||||
private Long scoringRunId;
|
||||
|
||||
// ── external shock ──────────────────────────────────────
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal feedCostChange;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal outputPriceChange;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal marginSqueeze;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal priceZScore;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal rainfallAnomaly;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal heatAnomaly;
|
||||
|
||||
// ── liquidity & behaviour ───────────────────────────────
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal cashFlowChange;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal emiDelay;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal inventoryDays;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal upiInflowChange;
|
||||
|
||||
/** Silence is a signal: an enterprise that stops recording has usually stopped trading normally. */
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal daysSinceLastEntry;
|
||||
|
||||
// ── obligation load ─────────────────────────────────────
|
||||
|
||||
/** Worst month of the horizon — min(p10 / obligation). Below 1.0 the downside case does not cover what is due. */
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal dscrP10;
|
||||
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal emiToInflow;
|
||||
|
||||
// ── the guard ───────────────────────────────────────────
|
||||
|
||||
/** Age of the newest market observation behind this row. Guards all of the above. */
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal feedStalenessDays;
|
||||
|
||||
// ── provenance ──────────────────────────────────────────
|
||||
|
||||
/** How many catalogue features carry a value. */
|
||||
private Integer featureCount;
|
||||
|
||||
/** How many came back null — a snapshot mostly made of holes must not look confident. */
|
||||
private Integer missingCount;
|
||||
|
||||
/** The oldest input age across all features, in days. */
|
||||
private Integer maxStalenessDays;
|
||||
|
||||
@Column(nullable = false)
|
||||
private LocalDateTime computedAt;
|
||||
|
||||
@PrePersist
|
||||
void onCreate() {
|
||||
if (computedAt == null) computedAt = LocalDateTime.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* The catalogue-code view, for rule evaluation and for the model. Insertion
|
||||
* ordered so a debug dump reads the way the catalogue does.
|
||||
*/
|
||||
public Map<String, Object> asFeatureMap() {
|
||||
Map<String, Object> map = new LinkedHashMap<>();
|
||||
map.put(FeatureCatalog.FEED_COST_CHANGE, feedCostChange);
|
||||
map.put(FeatureCatalog.OUTPUT_PRICE_CHANGE, outputPriceChange);
|
||||
map.put(FeatureCatalog.MARGIN_SQUEEZE, marginSqueeze);
|
||||
map.put(FeatureCatalog.PRICE_Z_SCORE, priceZScore);
|
||||
map.put(FeatureCatalog.RAINFALL_ANOMALY, rainfallAnomaly);
|
||||
map.put(FeatureCatalog.HEAT_ANOMALY, heatAnomaly);
|
||||
map.put(FeatureCatalog.CASH_FLOW_CHANGE, cashFlowChange);
|
||||
map.put(FeatureCatalog.EMI_DELAY, emiDelay);
|
||||
map.put(FeatureCatalog.INVENTORY_DAYS, inventoryDays);
|
||||
map.put(FeatureCatalog.UPI_INFLOW_CHANGE, upiInflowChange);
|
||||
map.put(FeatureCatalog.FEED_STALENESS_DAYS, feedStalenessDays);
|
||||
map.put(FeatureCatalog.DSCR_P10, dscrP10);
|
||||
map.put(FeatureCatalog.DAYS_SINCE_LAST_ENTRY, daysSinceLastEntry);
|
||||
map.put(FeatureCatalog.EMI_TO_INFLOW, emiToInflow);
|
||||
return map;
|
||||
}
|
||||
|
||||
/** Setter counterpart, so the engineering service writes columns by code. */
|
||||
public void put(String code, BigDecimal value) {
|
||||
switch (code) {
|
||||
case FeatureCatalog.FEED_COST_CHANGE -> feedCostChange = value;
|
||||
case FeatureCatalog.OUTPUT_PRICE_CHANGE -> outputPriceChange = value;
|
||||
case FeatureCatalog.MARGIN_SQUEEZE -> marginSqueeze = value;
|
||||
case FeatureCatalog.PRICE_Z_SCORE -> priceZScore = value;
|
||||
case FeatureCatalog.RAINFALL_ANOMALY -> rainfallAnomaly = value;
|
||||
case FeatureCatalog.HEAT_ANOMALY -> heatAnomaly = value;
|
||||
case FeatureCatalog.CASH_FLOW_CHANGE -> cashFlowChange = value;
|
||||
case FeatureCatalog.EMI_DELAY -> emiDelay = value;
|
||||
case FeatureCatalog.INVENTORY_DAYS -> inventoryDays = value;
|
||||
case FeatureCatalog.UPI_INFLOW_CHANGE -> upiInflowChange = value;
|
||||
case FeatureCatalog.FEED_STALENESS_DAYS -> feedStalenessDays = value;
|
||||
case FeatureCatalog.DSCR_P10 -> dscrP10 = value;
|
||||
case FeatureCatalog.DAYS_SINCE_LAST_ENTRY -> daysSinceLastEntry = value;
|
||||
case FeatureCatalog.EMI_TO_INFLOW -> emiToInflow = value;
|
||||
default -> throw new IllegalArgumentException("Unknown feature code: " + code);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
public interface FeatureSnapshotRepository extends JpaRepository<FeatureSnapshot, Long> {
|
||||
|
||||
Optional<FeatureSnapshot> findByEnterpriseIdAndAsOfDate(String enterpriseId, LocalDate asOfDate);
|
||||
|
||||
Optional<FeatureSnapshot> findFirstByEnterpriseIdOrderByAsOfDateDesc(String enterpriseId);
|
||||
|
||||
List<FeatureSnapshot> findByAsOfDate(LocalDate asOfDate);
|
||||
|
||||
List<FeatureSnapshot> findByScoringRunId(Long scoringRunId);
|
||||
|
||||
long countByAsOfDate(LocalDate asOfDate);
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import jakarta.persistence.*;
|
||||
import lombok.*;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* The tall store — one row per (enterprise, feature, date). This is the system of
|
||||
* record: auditable, and schema-free as features are added.
|
||||
*
|
||||
* <p>Both timestamps are carried for the reason in ARCHITECTURE §2.3.
|
||||
* {@code stalenessDays} is computed at write time from the age of the newest
|
||||
* input the value was derived from, so a consumer never has to re-derive
|
||||
* freshness from the source tables.
|
||||
*/
|
||||
@Entity
|
||||
@Table(name = "feature_value", indexes = {
|
||||
@Index(name = "ix_feature_value_key", columnList = "enterpriseId,featureCode,validForDate", unique = true),
|
||||
@Index(name = "ix_feature_value_code", columnList = "featureCode,validForDate"),
|
||||
})
|
||||
@Getter
|
||||
@Setter
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
@Builder
|
||||
public class FeatureValue {
|
||||
|
||||
@Id
|
||||
@GeneratedValue(strategy = GenerationType.IDENTITY)
|
||||
private Long id;
|
||||
|
||||
@Column(nullable = false, length = 24)
|
||||
private String enterpriseId;
|
||||
|
||||
@Column(nullable = false, length = 48)
|
||||
private String featureCode;
|
||||
|
||||
/** Null is a value: it means the inputs were absent, not that the row is missing. */
|
||||
@Column(precision = 16, scale = 4)
|
||||
private BigDecimal value;
|
||||
|
||||
/** The date the fact describes. */
|
||||
@Column(nullable = false)
|
||||
private LocalDate validForDate;
|
||||
|
||||
/** When we computed it. */
|
||||
@Column(nullable = false)
|
||||
private LocalDateTime observedAt;
|
||||
|
||||
/** Age of the newest input behind this value, in days. */
|
||||
private Integer stalenessDays;
|
||||
|
||||
/** Why the value is null — NO_SIGNAL | NO_LEDGER | NO_CBS | NO_CONSENT | INSUFFICIENT_HISTORY */
|
||||
@Column(length = 32)
|
||||
private String missingReason;
|
||||
|
||||
@PrePersist
|
||||
void onCreate() {
|
||||
if (observedAt == null) observedAt = LocalDateTime.now();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package com.lincfusion.udyamlinc.features;
|
||||
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
import org.springframework.data.jpa.repository.Query;
|
||||
import org.springframework.data.repository.query.Param;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
public interface FeatureValueRepository extends JpaRepository<FeatureValue, Long> {
|
||||
|
||||
Optional<FeatureValue> findByEnterpriseIdAndFeatureCodeAndValidForDate(
|
||||
String enterpriseId, String featureCode, LocalDate validForDate);
|
||||
|
||||
List<FeatureValue> findByEnterpriseIdAndValidForDateOrderByFeatureCode(String enterpriseId, LocalDate validForDate);
|
||||
|
||||
List<FeatureValue> findByEnterpriseIdAndFeatureCodeOrderByValidForDateDesc(String enterpriseId, String featureCode);
|
||||
|
||||
/** The training read: one feature's history for one enterprise, point-in-time. */
|
||||
@Query("""
|
||||
select f from FeatureValue f
|
||||
where f.enterpriseId = :enterpriseId and f.featureCode = :featureCode
|
||||
and f.validForDate <= :asOfDate and f.observedAt <= :asOf
|
||||
order by f.validForDate desc
|
||||
""")
|
||||
List<FeatureValue> findTrailing(@Param("enterpriseId") String enterpriseId,
|
||||
@Param("featureCode") String featureCode,
|
||||
@Param("asOfDate") LocalDate asOfDate,
|
||||
@Param("asOf") LocalDateTime asOf);
|
||||
|
||||
long countByValidForDate(LocalDate validForDate);
|
||||
}
|
||||
Reference in New Issue
Block a user