learning_loop/runtime/model_core.py

"""Extended research inputs and scoring conventions (learning-loop release runtime).

Identical to the 24 Oct release except that provenance counts and labels are read from
metadata.json, so a release retrained on observed outcomes reports them honestly."""
from pathlib import Path
import hashlib
import json
import math
import numpy as np
import pandas as pd
import xgboost as xgb

ROOT = Path(__file__).resolve().parent
FUSION_FEATURES = ['sentinel_ndvi_score', 'cv_health_score_0_100',
    'soil_moisture_vwc_pct', 'soil_temperature_c', 'soil_n_mg_kg', 'soil_p_mg_kg',
    'soil_k_mg_kg', 'soil_ph', 'soil_ec_ms_cm', 'days_after_planting']
CREDIT_FEATURES = FUSION_FEATURES + ['projected_cash_coverage', 'days_until_due', 'prior_loans_count',
    'prior_late_rate', 'prior_default_rate', 'history_available', 'crop_maize', 'crop_potato', 'crop_tomato']
HEALTHY = {'maize':'maize_healthy', 'potato':'Potato___healthy', 'tomato':'Tomato_healthy'}
CV_HASH = '153e9abf180f3823de24b082a3a1f188463dbc650cf9cc15792d582110d14598'
SCORE_POLICY = dict(minimum=300, maximum=850, reference_score=600,
    reference_repayment_odds=9, points_per_odds_doubling=50,
    origin='Chosen research display convention; not a fitted or lender-approved KCS policy')
SIMULATED_MATURITY_DAYS = dict(maize=130, potato=120, tomato=150)


def sigmoid(x):
    return 1/(1+np.exp(-np.clip(x,-50,50)))


def research_score(probability):
    p = np.clip(probability, 1e-6, 1-1e-6)
    factor = SCORE_POLICY['points_per_odds_doubling']/math.log(2)
    score = SCORE_POLICY['reference_score'] + factor*(np.log(p/(1-p))-math.log(SCORE_POLICY['reference_repayment_odds']))
    return np.clip(score, SCORE_POLICY['minimum'], SCORE_POLICY['maximum'])


def canonical_crop(value):
    result = str(value).strip().lower()
    return {'potatoes':'potato','tomatoes':'tomato','corn':'maize'}.get(result,result)


def checked_number(value, name, low, high, optional=False):
    if value is None and optional:
        return None
    if value is None or isinstance(value,bool):
        raise ValueError('Enter '+name)
    value = float(value)
    if not math.isfinite(value) or not low <= value <= high:
        raise ValueError(f'{name} must be finite and between {low} and {high}.')
    return value


def normalise_record(record):
    if 'feature_snapshot' in record:
        record = record['feature_snapshot']
    ctx = record.get('context', {})
    crop = canonical_crop(ctx.get('crop',record.get('crop','')))
    if crop not in HEALTHY:
        raise ValueError('This research release supports maize, potatoes and tomatoes only.')
    source = record.get('features',record)
    values = {key:source.get(key) for key in FUSION_FEATURES}
    if values['cv_health_score_0_100'] is None:
        values['cv_health_score_0_100'] = source.get('cv_health_score_daily_mean_14d_0_100')
    if source.get('cv_available') is False:
        values['cv_health_score_0_100'] = None
    if values['days_after_planting'] is None:
        values['days_after_planting'] = record.get('management',{}).get('days_after_planting')
    if record.get('cv_checkpoint_sha256') not in [None, CV_HASH]:
        raise ValueError('Unexpected CV checkpoint. Its scores require a new model version.')
    if record.get('cv_score_version') not in [None,'cv-healthy-class-score-v1']:
        raise ValueError('Unexpected CV score definition.')
    bounds = dict(sentinel_ndvi_score=(-1,1),cv_health_score_0_100=(0,100),
        soil_moisture_vwc_pct=(0,100),soil_temperature_c=(-30,80),
        soil_n_mg_kg=(0,10000),soil_p_mg_kg=(0,10000),soil_k_mg_kg=(0,10000),
        soil_ph=(0,14),soil_ec_ms_cm=(0,100),days_after_planting=(0,365))
    for key,(low,high) in bounds.items():
        values[key] = checked_number(values[key],key,low,high,optional=key=='cv_health_score_0_100')
    return record,crop,values


def normalise_finance(finance):
    currency = str(finance.get('currency','')).strip().upper()
    if not currency or len(currency)>12:
        raise ValueError('Specify the common currency for every monetary input.')
    limits = dict(area_ha=(0.001,100000),farmgate_price_per_tonne=(0.000001,1e15),
        remaining_costs_excluding_loan_financed=(0,1e15),other_cash_available_by_due=(0,1e15),
        household_cash_needed_by_due=(0,1e15),amount_due=(0.000001,1e15),days_until_due=(1,3650))
    result = {key:checked_number(finance.get(key),key,*limit) for key,limit in limits.items()}
    result['currency'] = currency
    keys = ['prior_loans_count','prior_late_repayments','prior_defaults']
    counts = [finance.get(key) for key in keys]
    if all(v is None for v in counts):
        result.update(history_available=False,prior_loans_count=None,prior_late_repayments=None,prior_defaults=None)
    elif any(v is None for v in counts):
        raise ValueError('Supply all three repayment-history counts, or leave all three blank if unknown.')
    else:
        for key in keys:
            value = checked_number(finance[key],key,0,10000)
            if value != int(value):
                raise ValueError('History counts must be whole numbers.')
            result[key] = int(value)
        if result['prior_late_repayments']+result['prior_defaults'] > result['prior_loans_count']:
            raise ValueError('Late non-default loans plus defaults cannot exceed the number of prior loans.')
        result['history_available'] = True
    return result


def build_credit_features(frame):
    """Currency cancels in coverage; no credit label or actual harvest is an input."""
    maturity = frame.crop.map(SIMULATED_MATURITY_DAYS).to_numpy()
    days_to_harvest = np.maximum(0,maturity-frame.days_after_planting.to_numpy())
    # Explicit simulator timing assumption, not a crop-specific harvest schedule.
    fraction = sigmoid((frame.days_until_due.to_numpy()-days_to_harvest)/15)
    revenue = frame.predicted_yield_t_ha.to_numpy()*frame.area_ha.to_numpy()*frame.farmgate_price_per_tonne.to_numpy()
    cash = revenue*fraction + frame.other_cash_available_by_due.to_numpy() - frame.remaining_costs_excluding_loan_financed.to_numpy() - frame.household_cash_needed_by_due.to_numpy()
    coverage = np.maximum(0,cash)/frame.amount_due.to_numpy()
    known = frame.history_available.astype(bool).to_numpy()
    count = frame.prior_loans_count.to_numpy(dtype=float,copy=True)
    # A known history of zero previous loans has known zero rates; unknown stays missing.
    late = np.divide(frame.prior_late_repayments.to_numpy(dtype=float), count,
        out=np.zeros(len(frame),dtype=float),where=count>0)
    default = np.divide(frame.prior_defaults.to_numpy(dtype=float),count,
        out=np.zeros(len(frame),dtype=float),where=count>0)
    late[~known],default[~known],count[~known] = np.nan,np.nan,np.nan
    features = pd.DataFrame(dict(projected_cash_coverage=coverage,
        days_until_due=frame.days_until_due.to_numpy(),prior_loans_count=count,
        prior_late_rate=late,prior_default_rate=default,history_available=known.astype(int),
        crop_maize=frame.crop.eq('maize').astype(int),crop_potato=frame.crop.eq('potato').astype(int),crop_tomato=frame.crop.eq('tomato').astype(int)))
    for key in FUSION_FEATURES:
        features[key]=frame[key].to_numpy()
    return features[CREDIT_FEATURES], dict(projected_gross_revenue=revenue,
        assumed_harvest_fraction_available_by_due=fraction,projected_cash_available_by_due=cash,
        projected_cash_coverage=coverage)


def verified_model(path, expected_hash):
    if hashlib.sha256(path.read_bytes()).hexdigest()!=expected_hash:
        raise ValueError('Model checksum mismatch: '+path.name)


def assess(record, finance=None, *, enable_synthetic_credit=False, root=ROOT):
    root = Path(root)
    metadata = json.loads((root/'metadata.json').read_text())
    record,crop,values = normalise_record(record)
    spec = metadata['yield_models'][crop]
    path = root/'models'/spec['file']
    verified_model(path,spec['sha256'])
    yield_model = xgb.XGBRegressor()
    yield_model.load_model(path)
    X = pd.DataFrame([values])[FUSION_FEATURES].astype(np.float32)
    predicted_yield = max(0,float(np.expm1(yield_model.predict(X)[0]+spec.get('log_offset',0.0))))
    yield_splits=yield_model.get_booster().get_score(importance_type='weight')
    output = dict(status='EXTENDED_RESEARCH_ONLY_NOT_VALIDATED_FARM_OR_CREDIT_MODEL',
        model_version=metadata['version'],crop=crop,inputs=values,
        experimental_yield_t_ha=predicted_yield,
        validated_kcs_score=None,validated_probability_of_repayment=None,lending_recommendation=None,
        research_credit_estimate=None,
        real_full_fusion_training_rows=metadata.get('real_full_fusion_rows',0),real_repayment_training_outcomes=metadata.get('real_repayment_outcomes',0),
        model_release=metadata.get('learning_loop',{}).get('release','base-synthetic-prior'),
        cv_input_supplied=values['cv_health_score_0_100'] is not None,
        cv_used_for_yield=values['cv_health_score_0_100'] is not None and 'cv_health_score_0_100' in yield_splits,
        features_selected_by_yield_trees=list(yield_splits),
        nutrients_ph_ec_used_as_model_inputs=True,
        calibration_verified=False,real_farm_performance_validated=False,
        yield_training_origin=spec['training_origin'],
        outside_training_ranges=[key for key,value in values.items() if value is not None and
            not spec['bounds'][key][0]<=value<=spec['bounds'][key][1]],
        warnings=metadata['limitations'])
    if finance is None:
        return output
    if not enable_synthetic_credit:
        output['credit_status']='Synthetic credit output requires explicitly enabled research mode.'
        return output
    finance = normalise_finance(finance)
    frame = pd.DataFrame([{**values,**finance,'crop':crop,'predicted_yield_t_ha':predicted_yield}])
    credit_features,cash = build_credit_features(frame)
    credit_spec = metadata['credit_model']
    path = root/'models'/credit_spec['file']
    verified_model(path,credit_spec['sha256'])
    credit_model = xgb.XGBClassifier()
    credit_model.load_model(path)
    raw = float(credit_model.predict_proba(credit_features.astype(np.float32))[0,1])
    calibration = metadata['synthetic_probability_calibration']
    logit = math.log(max(1e-8,raw)/max(1e-8,1-raw))
    probability = float(sigmoid(calibration['coefficient']*logit+calibration['intercept']))
    score = int(round(float(research_score(probability))))
    real_loans = int(metadata.get('real_repayment_outcomes',0))
    output['research_credit_estimate'] = dict(
        label=('RESEARCH KCS: SYNTHETIC PRIOR UPDATED WITH %d OBSERVED REPAYMENTS' % real_loans) if real_loans else 'SIMULATED RESEARCH KCS AND REPAYMENT ESTIMATE',
        kcs_score=score,score_range=[300,850],probability_of_repayment=probability,
        probability_target='Synthetic probability of paying the full stated amount by its due date',
        training_labels=('Synthetic prior blended with %d observed repayment outcomes' % real_loans) if real_loans else 'Entirely simulated loan outcomes; zero observed farmer repayments',
        calibration_domain=calibration.get('domain','Separate synthetic calibration set only'),
        features_selected_by_repayment_trees=list(credit_model.get_booster().get_score(importance_type='weight')),
        financial_inputs=finance,currency=finance['currency'],
        cash_flow={key:float(value[0]) for key,value in cash.items()},
        credit_features={key:None if pd.isna(value) else float(value) for key,value in credit_features.iloc[0].items()},
        scoring_scale=SCORE_POLICY,
        eligible_as_observed_outcome_training_label=False)
    return output