learning_loop/learning_loop.py

"""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).')]))

    @property
    def registry(self):
        return json.loads(self.registry_path.read_text())

    @property
    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))