KCS learning loop: v0 synthetic-prior release, outcome capture, retraining, season replay
48dbbe2 verified | """Maminda KCS learning loop: observed outcomes -> prior-blended retraining -> champion/challenger release. | |
| How the model "gets better as farmers use it" | |
| -------------------------------------------- | |
| 1. Farmers' photos, NDVI and soil readings are saved and scored. These are model INPUTS. | |
| 2. At season end the measured harvest is recorded (outcomes.py). When a scored loan falls | |
| due, its repayment result is recorded against the exact estimate that scored it. These | |
| are the only LABELS the model learns from. | |
| 3. Training rows use the inputs exactly as they were at prediction time (no look-ahead). | |
| 4. The synthetic data is a PRIOR worth S real examples. With n real examples the synthetic | |
| share of training weight is S / (S + n), so real farms progressively take over. | |
| 5. The data decide how much to trust the simulator. Each run builds candidate challengers: | |
| * recalibrate - keep the champion model, correct its level (yield) or its probability | |
| mapping (credit) from real outcomes. Works with very few outcomes. | |
| * retrain_S - refit on real outcomes + synthetic prior at strength S, for S in | |
| PRIOR_GRID (S = 0 means real data only). | |
| Every candidate is scored by grouped cross-validation over farmers on outcomes the | |
| champion has never seen. The best candidate is promoted only if it beats the champion. | |
| 6. Every release is versioned, checksummed, logged in registry.json and reversible. | |
| """ | |
| from __future__ import annotations | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import argparse | |
| import hashlib | |
| import json | |
| import shutil | |
| import sqlite3 | |
| import sys | |
| import numpy as np | |
| import pandas as pd | |
| import xgboost as xgb | |
| from sklearn.linear_model import LogisticRegression | |
| from sklearn.metrics import brier_score_loss, log_loss, roc_auc_score | |
| from sklearn.model_selection import GroupKFold | |
| HERE = Path(__file__).resolve().parent | |
| sys.path.insert(0, str(HERE / 'runtime')) | |
| from model_core import CREDIT_FEATURES, FUSION_FEATURES, build_credit_features, sigmoid # noqa: E402 | |
| CROPS = ('maize', 'potato', 'tomato') | |
| SEED = 20261025 | |
| PRIOR_GRID_YIELD = (0.0, 5.0, 20.0, 80.0, 320.0) # synthetic prior worth S real field-seasons | |
| PRIOR_GRID_CREDIT = (0.0, 25.0, 100.0, 400.0) # synthetic prior worth S real loans | |
| OFFSET_SHRINK = 3.0 # yield level correction shrinks toward 0 like 3 pseudo field-seasons | |
| PLATT_PRIOR_WEIGHT = 10.0 # recalibration keeps the champion mapping worth 10 loans | |
| MIN_NEW_FIELD_SEASONS = 5 # per crop, before a yield challenger is evaluated | |
| MIN_NEW_LOANS = 20 # before a credit challenger is evaluated | |
| YIELD_PARAMS = dict(objective='reg:squarederror', n_estimators=220, max_depth=3, min_child_weight=.2, | |
| learning_rate=.035, reg_lambda=10, subsample=1, colsample_bytree=1, tree_method='hist', | |
| n_jobs=4, random_state=SEED) | |
| CREDIT_PARAMS = dict(objective='binary:logistic', n_estimators=240, max_depth=3, min_child_weight=10, | |
| learning_rate=.04, reg_lambda=10, subsample=1, colsample_bytree=1, tree_method='hist', | |
| n_jobs=4, random_state=SEED, eval_metric='logloss', | |
| monotone_constraints=(1, 1, 0, 0, 1, 1, 1, 0, -1, 1, 1, 1, 0, -1, -1, 0, 0, 0, 0)) | |
| RUNTIME_FILES = ('model_core.py', 'predict_extended.py') | |
| CARRIED_FILES = ('generator_assumptions.json', 'evaluation.json') | |
| def now(): | |
| return datetime.now(timezone.utc) | |
| def sha(path): | |
| return hashlib.sha256(Path(path).read_bytes()).hexdigest() | |
| def write_json(path, value): | |
| tmp = Path(str(path) + '.tmp') | |
| tmp.write_text(json.dumps(value, indent=2, allow_nan=False, default=float), encoding='utf-8') | |
| tmp.replace(path) | |
| def clean(result, drop=('model', 'keys', 'bounds', 'calibration', 'log_offset')): | |
| return {k: v for k, v in result.items() if k not in drop} | |
| # ---------------------------------------------------------------- weights | |
| def blend_weights(prior_w, local_w, strength): | |
| """prior_w sums to 1 (shape of the prior); local_w sums to n real examples. | |
| Returned weights have mean 1, so XGBoost regularisation behaves as for unweighted data.""" | |
| w = np.r_[np.asarray(prior_w, float) * strength, np.asarray(local_w, float)] | |
| return w * len(w) / w.sum() | |
| def prior_share(strength, n_real): | |
| return strength / (strength + n_real) if strength + n_real else 1.0 | |
| # ---------------------------------------------------------------- prior data | |
| def prior_yield(prior_dir, crop): | |
| d = pd.read_csv(Path(prior_dir) / 'yield_training_WITH_ORIGIN.csv', low_memory=False) | |
| d = d[d.crop.eq(crop)] | |
| real = d[d.data_origin.eq('real_partial')] | |
| syn = d[d.data_origin.eq('synthetic') & d.partition.eq('yield_fit')] | |
| # Same internal proportions as the 24 Oct release (synthetic mass = 25% of real partial mass). | |
| w = np.r_[np.ones(len(real)), np.full(len(syn), (len(real) * .25 / len(syn)) if len(real) else 1.0)] | |
| frame = pd.concat([real, syn], ignore_index=True) | |
| return frame[FUSION_FEATURES].astype(float), frame.yield_t_ha.to_numpy(float), w / w.sum() | |
| def prior_credit(prior_dir): | |
| loans = pd.read_csv(Path(prior_dir) / 'loan_outcomes_ENTIRELY_SYNTHETIC.csv', low_memory=False) | |
| loans['history_available'] = loans.history_available.astype(str).str.lower().isin(['true', '1']) | |
| X, _ = build_credit_features(loans) | |
| y = loans.synthetic_paid_in_full_by_due.to_numpy(int) | |
| fit, cal = loans.partition.eq('credit_fit').to_numpy(), loans.partition.eq('credit_calibration').to_numpy() | |
| return (X[fit].astype(float).reset_index(drop=True), y[fit], np.full(fit.sum(), 1 / fit.sum()), | |
| X[cal].astype(float).reset_index(drop=True), y[cal]) | |
| # ---------------------------------------------------------------- observed (local) data | |
| def connect(db_path): | |
| db = sqlite3.connect(db_path) | |
| db.row_factory = sqlite3.Row | |
| return db | |
| def tables(db): | |
| return {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} | |
| def local_yield_rows(db_path): | |
| """One row per saved estimate made before harvest; each field-season carries total weight 1.""" | |
| with connect(db_path) as db: | |
| if not {'harvest_outcomes', 'research_predictions'} <= tables(db): | |
| return pd.DataFrame() | |
| rows = db.execute('''SELECT rp.id prediction_id, rp.payload, h.context_key, h.yield_t_ha, | |
| h.harvest_date, fs.farmer_id | |
| FROM research_predictions rp | |
| JOIN harvest_outcomes h ON h.context_key = rp.context_key | |
| JOIN field_seasons fs ON fs.context_key = rp.context_key | |
| WHERE h.label_grade = 1 AND rp.created_at < h.harvest_date''').fetchall() | |
| records = [] | |
| for r in rows: | |
| payload = json.loads(r['payload']) | |
| inputs = payload.get('inputs') or {} | |
| if payload.get('crop') not in CROPS or any(inputs.get(k) is None for k in FUSION_FEATURES if k != 'cv_health_score_0_100'): | |
| continue | |
| records.append(dict({k: inputs.get(k) for k in FUSION_FEATURES}, crop=payload['crop'], | |
| yield_t_ha=r['yield_t_ha'], key=r['context_key'], farmer_id=r['farmer_id'], | |
| prediction_id=r['prediction_id'], outcome_time=r['harvest_date'])) | |
| frame = pd.DataFrame(records) | |
| if len(frame): | |
| frame[FUSION_FEATURES] = frame[FUSION_FEATURES].astype(float) | |
| frame['weight'] = 1.0 / frame.groupby('key').key.transform('size') | |
| return frame | |
| def local_credit_rows(db_path): | |
| """One row per repayment outcome, using the credit features saved with the scoring estimate.""" | |
| with connect(db_path) as db: | |
| if 'repayment_outcomes' not in tables(db): | |
| return pd.DataFrame() | |
| rows = db.execute('''SELECT r.prediction_id, r.paid_in_full_by_due, r.due_date, rp.payload, fs.farmer_id | |
| FROM repayment_outcomes r | |
| JOIN research_predictions rp ON rp.id = r.prediction_id | |
| JOIN field_seasons fs ON fs.context_key = r.context_key''').fetchall() | |
| records = [] | |
| for r in rows: | |
| features = (json.loads(r['payload']).get('research_credit_estimate') or {}).get('credit_features') or {} | |
| records.append(dict({k: features.get(k) for k in CREDIT_FEATURES}, paid=int(r['paid_in_full_by_due']), | |
| key=r['prediction_id'], farmer_id=r['farmer_id'], outcome_time=r['due_date'])) | |
| frame = pd.DataFrame(records) | |
| if len(frame): | |
| frame[CREDIT_FEATURES] = frame[CREDIT_FEATURES].astype(float) | |
| frame['weight'] = 1.0 | |
| return frame | |
| # ---------------------------------------------------------------- learners | |
| def fit_yield(X, y, w): | |
| model = xgb.XGBRegressor(**YIELD_PARAMS) | |
| model.fit(X[FUSION_FEATURES].astype(np.float32), np.log1p(y), sample_weight=w) | |
| return model | |
| def yield_margin(model, X): | |
| return model.predict(X[FUSION_FEATURES].astype(np.float32)) | |
| def predict_yield(model, X, log_offset=0.0): | |
| return np.maximum(0, np.expm1(yield_margin(model, X) + log_offset)) | |
| def fit_offset(margin, y, w): | |
| """Shrunken weighted mean residual in log1p space: corrects the level of a biased model.""" | |
| residual = np.log1p(np.asarray(y)) - margin | |
| return float(np.sum(w * residual) / (np.sum(w) + OFFSET_SHRINK)) | |
| def fit_credit(X, y, w): | |
| model = xgb.XGBClassifier(**CREDIT_PARAMS) | |
| model.fit(X[CREDIT_FEATURES].astype(np.float32), y, sample_weight=w) | |
| return model | |
| def raw_credit(model, X): | |
| return model.predict_proba(X[CREDIT_FEATURES].astype(np.float32))[:, 1] | |
| def logit(p): | |
| p = np.clip(p, 1e-8, 1 - 1e-8) | |
| return np.log(p / (1 - p)) | |
| def fit_platt(raw, y, w): | |
| lr = LogisticRegression(C=1e6, max_iter=1000).fit(logit(raw).reshape(-1, 1), y, sample_weight=w) | |
| return dict(coefficient=float(lr.coef_[0, 0]), intercept=float(lr.intercept_[0])) | |
| def apply_platt(raw, cal): | |
| return sigmoid(cal['coefficient'] * logit(raw) + cal['intercept']) | |
| def recalibrate_platt(raw, y, champion_cal): | |
| """Refit the probability mapping on real loans, anchored by pseudo-loans drawn from the champion mapping.""" | |
| grid = np.linspace(.02, .98, 25) | |
| anchor_y = np.r_[np.ones(len(grid)), np.zeros(len(grid))] | |
| anchor_raw = np.r_[grid, grid] | |
| p = apply_platt(grid, champion_cal) | |
| anchor_w = np.r_[p, 1 - p] * PLATT_PRIOR_WEIGHT / len(grid) | |
| return fit_platt(np.r_[raw, anchor_raw], np.r_[y, anchor_y], np.r_[np.ones(len(raw)), anchor_w]) | |
| def groups_for(frame): | |
| return frame.farmer_id if frame.farmer_id.nunique() >= 3 else frame.key | |
| def folds(frame): | |
| groups = groups_for(frame) | |
| return list(GroupKFold(int(min(5, pd.Series(groups).nunique()))).split(frame, groups=groups)) | |
| def wmae(y, p, w): | |
| return float(np.average(np.abs(np.asarray(y) - np.asarray(p)), weights=w)) | |
| def credit_metrics(y, p): | |
| out = dict(log_loss=float(log_loss(y, p, labels=[0, 1])), brier=float(brier_score_loss(y, p)), | |
| mean_predicted=float(np.mean(p)), observed_rate=float(np.mean(y))) | |
| out['roc_auc'] = float(roc_auc_score(y, p)) if len(set(y)) == 2 else None | |
| return out | |
| # ---------------------------------------------------------------- releases | |
| class Releases: | |
| def __init__(self, directory, base_dir): | |
| self.dir = Path(directory) | |
| self.base = Path(base_dir) | |
| self.registry_path = self.dir / 'registry.json' | |
| self.dir.mkdir(parents=True, exist_ok=True) | |
| if not self.registry_path.exists(): | |
| self._bootstrap() | |
| def _bootstrap(self): | |
| """Release v0 = the 24 Oct synthetic-prior models, unchanged, so the loop starts from them.""" | |
| version = 'v0-synthetic-prior' | |
| target = self.dir / version | |
| (target / 'models').mkdir(parents=True, exist_ok=True) | |
| for name in RUNTIME_FILES: | |
| shutil.copy2(HERE / 'runtime' / name, target / name) | |
| for name in CARRIED_FILES: | |
| if (self.base / name).exists(): | |
| shutil.copy2(self.base / name, target / name) | |
| meta = json.loads((self.base / 'metadata.json').read_text()) | |
| for spec in list(meta['yield_models'].values()) + [meta['credit_model']]: | |
| shutil.copy2(self.base / 'models' / spec['file'], target / 'models' / spec['file']) | |
| assert sha(target / 'models' / spec['file']) == spec['sha256'], spec['file'] | |
| for spec in meta['yield_models'].values(): | |
| spec['log_offset'] = 0.0 | |
| meta['learning_loop'] = dict(release=version, parent=None, prior_grid_yield=PRIOR_GRID_YIELD, | |
| prior_grid_credit=PRIOR_GRID_CREDIT, observed_training_keys={}) | |
| write_json(target / 'metadata.json', meta) | |
| write_json(self.registry_path, dict(active=version, releases=[dict( | |
| version=version, created_at=now().isoformat(), parent=None, heads_changed=[], | |
| note='Bootstrap from the 24 Oct 2026 extended research release (synthetic prior).')])) | |
| def registry(self): | |
| return json.loads(self.registry_path.read_text()) | |
| def active_dir(self): | |
| return self.dir / self.registry['active'] | |
| def active_meta(self): | |
| return json.loads((self.active_dir / 'metadata.json').read_text()) | |
| def champion_keys(self, head): | |
| return set(self.active_meta()['learning_loop']['observed_training_keys'].get(head, [])) | |
| def load_yield(self, crop): | |
| spec = self.active_meta()['yield_models'][crop] | |
| model = xgb.XGBRegressor() | |
| model.load_model(self.active_dir / 'models' / spec['file']) | |
| return model, float(spec.get('log_offset', 0.0)) | |
| def load_credit(self): | |
| meta = self.active_meta() | |
| model = xgb.XGBClassifier() | |
| model.load_model(self.active_dir / 'models' / meta['credit_model']['file']) | |
| return model, meta['synthetic_probability_calibration'] | |
| def _new_version(self): | |
| version = 'v' + now().strftime('%Y%m%d-%H%M%S') | |
| suffix = 1 | |
| while (self.dir / version).exists(): | |
| suffix += 1 | |
| version = 'v' + now().strftime('%Y%m%d-%H%M%S') + f'-{suffix}' | |
| return version | |
| def publish(self, promoted, decisions): | |
| parent = self.registry['active'] | |
| version = self._new_version() | |
| target = self.dir / version | |
| shutil.copytree(self.dir / parent, target) | |
| meta = json.loads((target / 'metadata.json').read_text()) | |
| loop = meta['learning_loop'] | |
| for head, item in promoted.items(): | |
| if 'model' in item: | |
| name = f'{head}_{version}.json' | |
| item['model'].save_model(target / 'models' / name) | |
| if head.startswith('yield_'): | |
| spec = meta['yield_models'][head[6:]] | |
| if 'model' in item: | |
| spec.update(file=name, sha256=sha(target / 'models' / name)) | |
| spec.update(log_offset=item['log_offset'], learning_method=item['chosen'], | |
| training_origin='observed_local_outcomes_plus_synthetic_prior', | |
| local_observed_field_seasons=item['n_real'], prior_strength=item.get('prior_strength'), | |
| prior_share_of_training_weight=item.get('prior_share')) | |
| for k, (low, high) in item['bounds'].items(): | |
| spec['bounds'][k] = [float(min(spec['bounds'][k][0], low)), float(max(spec['bounds'][k][1], high))] | |
| else: | |
| if 'model' in item: | |
| meta['credit_model'].update(file=name, sha256=sha(target / 'models' / name)) | |
| meta['credit_model'].update(learning_method=item['chosen'], | |
| training_origin='observed_repayments_plus_synthetic_prior', | |
| observed_repayment_outcomes=item['n_real'], | |
| prior_strength=item.get('prior_strength'), | |
| prior_share_of_training_weight=item.get('prior_share')) | |
| meta['synthetic_probability_calibration'] = dict( | |
| item['calibration'], method='sigmoid_on_logit_output', | |
| domain=f"Fitted with {item['n_real']} observed repayment outcomes ({item['chosen']})") | |
| loop['observed_training_keys'][head] = sorted(item['keys']) | |
| keys = loop['observed_training_keys'] | |
| meta['real_full_fusion_rows'] = int(sum(len(v) for k, v in keys.items() if k.startswith('yield_'))) | |
| meta['real_repayment_outcomes'] = int(len(keys.get('credit', []))) | |
| meta['version'] = version | |
| meta['status'] = 'LEARNING_LOOP_RESEARCH_RELEASE_NOT_VALIDATED_FOR_LENDING' | |
| loop.update(release=version, parent=parent, last_decisions=decisions) | |
| write_json(target / 'metadata.json', meta) | |
| registry = self.registry | |
| registry['releases'].append(dict(version=version, created_at=now().isoformat(), parent=parent, | |
| heads_changed=sorted(promoted), decisions=decisions)) | |
| registry['active'] = version | |
| write_json(self.registry_path, registry) | |
| return version | |
| def rollback(self, version): | |
| registry = self.registry | |
| if not (self.dir / version / 'metadata.json').exists(): | |
| raise ValueError('Unknown release ' + version) | |
| registry['releases'].append(dict(version=version, created_at=now().isoformat(), | |
| parent=registry['active'], heads_changed=[], note='Manual rollback')) | |
| registry['active'] = version | |
| write_json(self.registry_path, registry) | |
| # ---------------------------------------------------------------- challengers | |
| def evaluate_yield(crop, local, releases, prior_dir): | |
| head = 'yield_' + crop | |
| rows = local[local.crop.eq(crop)].reset_index(drop=True) if len(local) else pd.DataFrame() | |
| if not len(rows): | |
| return dict(head=head, decision='waiting', new_field_seasons=0, | |
| reason='No measured harvests linked to pre-harvest estimates yet.') | |
| new_mask = ~rows.key.isin(releases.champion_keys(head)) | |
| new, old = rows[new_mask].reset_index(drop=True), rows[~new_mask] | |
| n_new, n_all = int(new.key.nunique()), int(rows.key.nunique()) | |
| if n_new < MIN_NEW_FIELD_SEASONS: | |
| return dict(head=head, decision='waiting', new_field_seasons=n_new, | |
| reason=f'{n_new} new field-season(s); need {MIN_NEW_FIELD_SEASONS} before testing a challenger.') | |
| champion, offset0 = releases.load_yield(crop) | |
| champ_margin_new = yield_margin(champion, new) + offset0 | |
| pX, py, pw = prior_yield(prior_dir, crop) | |
| split = folds(new) | |
| oof = {'recalibrate': np.zeros(len(new)), **{f'retrain_S{S:g}': np.zeros(len(new)) for S in PRIOR_GRID_YIELD}} | |
| old_margin = yield_margin(champion, old) + offset0 if len(old) else np.zeros(0) | |
| for train_idx, test_idx in split: | |
| fit_local = pd.concat([old, new.iloc[train_idx]]) | |
| # Candidate 1: champion + level correction. | |
| margin = np.r_[old_margin, champ_margin_new[train_idx]] | |
| delta = fit_offset(margin, fit_local.yield_t_ha, fit_local.weight.to_numpy()) | |
| oof['recalibrate'][test_idx] = np.maximum(0, np.expm1(champ_margin_new[test_idx] + delta)) | |
| # Candidates 2..: refit with the synthetic prior at each strength. | |
| X = pd.concat([pX, fit_local[FUSION_FEATURES]], ignore_index=True) | |
| y = np.r_[py, fit_local.yield_t_ha] | |
| for S in PRIOR_GRID_YIELD: | |
| model = fit_yield(X, y, blend_weights(pw, fit_local.weight, S)) | |
| oof[f'retrain_S{S:g}'][test_idx] = predict_yield(model, new.iloc[test_idx]) | |
| w = new.weight.to_numpy() | |
| champ_mae = wmae(new.yield_t_ha, np.maximum(0, np.expm1(champ_margin_new)), w) | |
| scores = {name: wmae(new.yield_t_ha, pred, w) for name, pred in oof.items()} | |
| chosen = min(scores, key=scores.get) | |
| result = dict(head=head, new_field_seasons=n_new, total_observed_field_seasons=n_all, | |
| champion_mae_t_ha=champ_mae, candidate_mae_t_ha=scores, chosen=chosen, | |
| challenger_mae_t_ha=scores[chosen], | |
| improvement_pct=100 * (champ_mae - scores[chosen]) / champ_mae if champ_mae else 0.0, | |
| test='Grouped cross-validation over farmers on outcomes the champion never saw', | |
| selection_note='Best of several candidates on the same folds, so its score is slightly optimistic.') | |
| if scores[chosen] >= champ_mae: | |
| result.update(decision='keep_champion', reason='No candidate was more accurate on the new outcomes.') | |
| return result | |
| bounds = {k: [float(rows[k].min()), float(rows[k].max())] for k in FUSION_FEATURES if rows[k].notna().any()} | |
| if chosen == 'recalibrate': | |
| margin = np.r_[old_margin, champ_margin_new] | |
| delta = fit_offset(margin, np.r_[old.yield_t_ha, new.yield_t_ha], np.r_[old.weight, new.weight]) | |
| result.update(log_offset=offset0 + delta, level_correction_factor=float(np.exp(delta))) | |
| else: | |
| S = float(chosen.split('_S')[1]) | |
| X = pd.concat([pX, rows[FUSION_FEATURES]], ignore_index=True) | |
| result.update(model=fit_yield(X, np.r_[py, rows.yield_t_ha], blend_weights(pw, rows.weight, S)), | |
| log_offset=0.0, prior_strength=S, prior_share=prior_share(S, n_all)) | |
| result.update(decision='promote', n_real=n_all, keys=set(rows.key), bounds=bounds) | |
| return result | |
| def evaluate_credit(local, releases, prior_dir): | |
| head = 'credit' | |
| if not len(local): | |
| return dict(head=head, decision='waiting', new_loans=0, reason='No repayment outcomes recorded yet.') | |
| new_mask = ~local.key.isin(releases.champion_keys(head)) | |
| new, old = local[new_mask].reset_index(drop=True), local[~new_mask] | |
| if len(new) < MIN_NEW_LOANS: | |
| return dict(head=head, decision='waiting', new_loans=int(len(new)), | |
| reason=f'{len(new)} new repayment outcome(s); need {MIN_NEW_LOANS} before testing a challenger.') | |
| champion, cal0 = releases.load_credit() | |
| raw_new, raw_old = raw_credit(champion, new), (raw_credit(champion, old) if len(old) else np.zeros(0)) | |
| pX, py, pw, cX, cy = prior_credit(prior_dir) | |
| y_new = new.paid.to_numpy() | |
| names = ['recalibrate'] + [f'retrain_S{S:g}' for S in PRIOR_GRID_CREDIT] | |
| oof = {n: np.zeros(len(new)) for n in names} | |
| oof_raw = {n: np.zeros(len(new)) for n in names[1:]} | |
| for train_idx, test_idx in folds(new): | |
| fit_local = pd.concat([old, new.iloc[train_idx]]) | |
| cal = recalibrate_platt(np.r_[raw_old, raw_new[train_idx]], fit_local.paid.to_numpy(), cal0) | |
| oof['recalibrate'][test_idx] = apply_platt(raw_new[test_idx], cal) | |
| X = pd.concat([pX, fit_local[CREDIT_FEATURES]], ignore_index=True) | |
| for S in PRIOR_GRID_CREDIT: | |
| name = f'retrain_S{S:g}' | |
| model = fit_credit(X, np.r_[py, fit_local.paid], blend_weights(pw, fit_local.weight, S)) | |
| # Scored on its own logistic output: refitting a calibrator on the loans the model was | |
| # trained on would make it over-confident. | |
| oof_raw[name][test_idx] = raw_credit(model, new.iloc[test_idx]) | |
| oof[name][test_idx] = oof_raw[name][test_idx] | |
| champ_m = credit_metrics(y_new, apply_platt(raw_new, cal0)) | |
| scores = {n: credit_metrics(y_new, p) for n, p in oof.items()} | |
| chosen = min(scores, key=lambda n: scores[n]['log_loss']) | |
| result = dict(head=head, new_loans=int(len(new)), total_observed_loans=int(len(local)), | |
| champion=champ_m, candidates=scores, chosen=chosen, challenger=scores[chosen], | |
| test='Grouped cross-validation over farmers on loans the champion never saw', | |
| selection_note='Best of several candidates on the same folds, so its score is slightly optimistic.') | |
| if scores[chosen]['log_loss'] >= champ_m['log_loss']: | |
| result.update(decision='keep_champion', reason='No candidate lowered log loss on the new outcomes.') | |
| return result | |
| y_all = np.r_[old.paid, y_new] | |
| if chosen == 'recalibrate': | |
| cal = recalibrate_platt(np.r_[raw_old, raw_new], y_all, cal0) | |
| else: | |
| S = float(chosen.split('_S')[1]) | |
| X = pd.concat([pX, local[CREDIT_FEATURES]], ignore_index=True) | |
| model = fit_credit(X, np.r_[py, local.paid], blend_weights(pw, local.weight, S)) | |
| # Same mapping that was tested: the model's own logistic output (identity calibration). | |
| cal = dict(coefficient=1.0, intercept=0.0) | |
| result.update(model=model, prior_strength=S, prior_share=prior_share(S, len(local))) | |
| if cal['coefficient'] <= 0: | |
| result.update(decision='keep_champion', reason='Calibration would reverse the score direction.') | |
| return result | |
| result.update(decision='promote', calibration=cal, n_real=int(len(local)), keys=set(local.key)) | |
| return result | |
| def retrain(db_path, releases_dir, base_dir, prior_dir=None, report_dir=None): | |
| prior_dir = Path(prior_dir or Path(base_dir) / 'data') | |
| releases = Releases(releases_dir, base_dir) | |
| yield_rows, credit_rows = local_yield_rows(db_path), local_credit_rows(db_path) | |
| decisions, promoted = {}, {} | |
| for crop in CROPS: | |
| result = evaluate_yield(crop, yield_rows, releases, prior_dir) | |
| decisions[result['head']] = clean(result) | |
| if result['decision'] == 'promote': | |
| promoted[result['head']] = result | |
| result = evaluate_credit(credit_rows, releases, prior_dir) | |
| decisions['credit'] = clean(result) | |
| if result['decision'] == 'promote': | |
| promoted['credit'] = result | |
| version = releases.publish(promoted, decisions) if promoted else releases.registry['active'] | |
| report = dict(run_at=now().isoformat(), active_release=version, new_release=bool(promoted), | |
| observed_harvest_rows=int(len(yield_rows)), observed_repayments=int(len(credit_rows)), | |
| decisions=decisions) | |
| report_dir = Path(report_dir or Path(releases_dir) / 'reports') | |
| report_dir.mkdir(exist_ok=True) | |
| write_json(report_dir / f"retrain_{now().strftime('%Y%m%d-%H%M%S-%f')}.json", report) | |
| return report | |
| if __name__ == '__main__': | |
| parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| parser.add_argument('--db', type=Path, required=True, help='maminda_crop_history.sqlite3') | |
| parser.add_argument('--releases', type=Path, required=True, help='Folder holding versioned model releases') | |
| parser.add_argument('--base', type=Path, required=True, help='24 Oct extended research package (bootstrap source)') | |
| parser.add_argument('--prior', type=Path, help='Folder with the synthetic prior CSVs (default: <base>/data)') | |
| parser.add_argument('--rollback', help='Make an earlier release active again') | |
| args = parser.parse_args() | |
| if args.rollback: | |
| Releases(args.releases, args.base).rollback(args.rollback) | |
| print(json.dumps(dict(active_release=args.rollback))) | |
| else: | |
| print(json.dumps(retrain(args.db, args.releases, args.base, args.prior), indent=2, default=float)) | |