Output Contracts
This page documents the return types and output shapes for key public functions. Use it alongside the Input Data Contracts to understand the full data flow through each workflow.
Shared example data
Run this setup once, then run the examples below in order. The data is synthetic; replace it with the matching shapes from Input Data Contracts for your own portfolio. Forecast dates are explicit so the results are reproducible.
import numpy as np
import pandas as pd
from pathlib import Path
portfolio_df = pd.DataFrame({
"loan_id": ["A", "B"], "principal": [1000.0, 2000.0],
"annual_rate": [0.10, 0.12], "term_months": [24, 36],
"start_date": pd.to_datetime(["2024-01-01", "2024-01-01"]),
"status": ["Current", "Current"], "fico_score": [700, 650],
})
states = ["Current", "Charged Off"]
transition_input = pd.DataFrame(
[[0.98, 0.02], [0.0, 1.0]], index=states, columns=states,
)
matrix = transition_input
initial_states = pd.Series([2.0, 0.0], index=states)
as_of_date = pd.Timestamp("2024-06-01")
prior_portfolio = portfolio_df
current_portfolio = portfolio_df.assign(principal=[900.0, 1800.0])
prior_assumptions = {"migration_matrix": matrix, "as_of_date": as_of_date}
current_assumptions = dict(prior_assumptions)
loan_df = portfolio_df.assign(collateral_value=[500.0, 0.0])
mob_array = np.arange(1, 25)
loss_rate_array = 0.08 * (1.0 - np.exp(-mob_array / 12.0))
future_mob_array = np.arange(1, 37)
vintage_df = pd.DataFrame({
"vintage_date": "2022-Q1", "months_on_book": mob_array,
"cumulative_loss_rate": loss_rate_array,
})
flow_df = pd.DataFrame({
"segment_id": ["prime"] * 24, "month_on_book": mob_array,
"payments": 50.0, "chargeoffs": 5.0,
"outstanding_balance": 2000.0 - 55.0 * mob_array,
"amtloan": 2000.0,
})
known_actuals = flow_df.iloc[:12].copy()
df = pd.DataFrame({
"origination_month": np.repeat(np.arange(1, 7), 4),
"fico_score": np.tile([600.0, 650.0, 750.0, 800.0], 6),
"fpf30_flag": np.tile([1.0, 0.0, 1.0, 0.0], 6),
})
feature_cols = ["fico_score"]
target_col = "fpf30_flag"
split_col = "origination_month"
expected_scores = pd.Series(np.linspace(0.01, 0.99, 100))
actual_scores = pd.Series(np.linspace(0.05, 0.95, 100))
Loss Forecasting
forecast_lifetime_loss
from cranalytics import forecast_lifetime_loss
loss = forecast_lifetime_loss(portfolio_df, transition_input, as_of_date=as_of_date)
Returns: float — total expected lifetime loss across the portfolio in its currency units.
summarize_lifetime_loss
from cranalytics import summarize_lifetime_loss
summary = summarize_lifetime_loss(portfolio_df, transition_input, as_of_date=as_of_date)
Returns: dict[str, float]
| Key | Type | Description |
|---|---|---|
total_portfolio_balance |
float | Current outstanding balance: sum of principal for loans not already Charged Off/Default or Paid Off |
estimated_lifetime_loss |
float | Total expected loss |
reserve_ratio |
float | estimated_lifetime_loss / total_portfolio_balance (0 if nothing is outstanding) |
lgd_assumption |
float | The LGD value used in the calculation |
reserve_walk
from cranalytics.reserve import reserve_walk
walk = reserve_walk(prior_portfolio, current_portfolio, prior_assumptions, current_assumptions)
Returns: pd.DataFrame with columns step, amount, cumulative and ten rows in this order:
Prior reserve, the eight steps (Runoff, Seasoning, New originations, Balance change,
Credit migration, Transition matrix, LGD, Prepayment and amortization), Current reserve.
- Step rows:
amountis the change in forecast lifetime loss from that input;cumulativeis the running reserve after it. - First and last rows:
amountequalscumulative(the prior and current reserve). - Reconciliation: step amounts sum to current minus prior; the last row equals
forecast_lifetime_losson the current portfolio. - Every step is always present, with
0.0when its input did not change.
reserve_walk_by_segment(..., by=...) returns the same rows per segment, with columns segment, step, amount, cumulative.
forecast_portfolio_states
from cranalytics import forecast_portfolio_states
states_df = forecast_portfolio_states(matrix, initial_states, n_periods=24)
Returns: pd.DataFrame
- Index:
period— integer from 0 (current) ton_periodsinclusive - Columns: One column per state in the transition matrix
| Column | Type | Description |
|---|---|---|
(state names, e.g. Current, Delinquent, Charged Off) |
float | Count or weight of loans in that state at each period |
Shape: (n_periods + 1, n_states)
Portfolio & Segmentation
segment_fico
from cranalytics import segment_fico
segmented_df = segment_fico(portfolio_df)
Returns: pd.DataFrame — input DataFrame with two columns added.
| New Column | Type | Values |
|---|---|---|
fico_band |
str | <600, 600-649, 650-699, 700-749, 750-799, 800+ |
risk_grade |
int | 1 (lowest risk, 800+) → 6 (highest risk, <600) |
All original columns are preserved.
calculate_lgd
from cranalytics import calculate_lgd
lgd_series = calculate_lgd(loan_df)
Returns: pd.Series of float — one LGD value per loan, same index as input.
- Range: 0.0 – 1.0 (decimal, not percent)
- Formula:
LGD = 1 − (collateral_value × (1 − haircut)) / principal - Unsecured loans: LGD = 1.0
- Single loan: pass a
pd.Seriesto get onefloat. It is validated exactly like a one-row DataFrame. - Rejected input:
principalmust be present and positive for both input types (LGD on a zero balance is undefined); invalid rows raisepandera.errors.SchemaErrors.
Vintage Curve Fitting
vintage.run
from cranalytics import vintage
session = vintage.run(vintage_df, min_maturity_months=12)
Returns: VintageAnalysisSessionResult — dataclass with attribute access
| Attribute | Type | Description |
|---|---|---|
triangle |
pd.DataFrame | Vintage triangle used to detect maturity coverage |
incomplete_vintages |
list[str] | Vintage names flagged as incomplete |
selected_vintage |
pd.DataFrame | Long-format curve chosen for comparison and validation |
validation_issues |
pd.DataFrame | Issue table surfaced from the contract boundary |
comparison_metrics |
pd.DataFrame | Per-method metric table from smoothing comparison |
rankings |
pd.DataFrame | Ranked method scores, best first |
validation_summary |
pd.DataFrame | Temporal validation summary by smoothing method |
summary_table |
str | Human-readable validation summary table |
tail_projections |
pd.DataFrame | Projected rows for incomplete vintages when requested |
Access fields as attributes, e.g. session.rankings. Dict-style access
(session["rankings"]) was removed in 0.3.0.
CurveFitter.predict / CurveFitter.forecast
from cranalytics import CurveFitter
fitter = CurveFitter(method="weibull")
fitter.fit(mob_array, loss_rate_array)
predictions = fitter.predict(future_mob_array)
Returns: np.ndarray — 1-D array of predicted cumulative loss rates.
- Shape:
(n_samples,)matching inputfuture_mob_array - Range:
[0, fitter.ultimate_]for non-negative MOB, from the fitted cumulative distribution
Key fitted attributes after fit():
| Attribute | Type | Description |
|---|---|---|
ultimate_ |
float | Fitted ultimate (terminal) loss rate |
params_ |
np.ndarray | Raw curve parameters |
smooth_vintage
from cranalytics.vintage import smooth_vintage
curve_df = vintage_df.rename(columns={
"vintage_date": "vintage_name", "months_on_book": "mob",
})
result = smooth_vintage(curve_df, method="moving_average", window=3)
Returns: SmoothedCurve dataclass
| Attribute | Type | Description |
|---|---|---|
mob |
np.ndarray | Months on book (same as input) |
smoothed_values |
np.ndarray | Smoothed cumulative loss rates |
original_values |
np.ndarray | Original (unsmoothed) cumulative loss rates |
method_name |
str | Human-readable method name and parameters |
parameters |
dict | Fitted parameters (method-specific) |
fitted_model |
Any | None | Underlying model object (spline, isotonic, etc.) |
Key method: result.forecast(future_mob) → np.ndarray — extrapolate to future months.
Rollforward
fit_flow_hazard_curves
from cranalytics.survival_flows import fit_flow_hazard_curves
curves_df = fit_flow_hazard_curves(flow_df)
Returns: pd.DataFrame
| Column | Type | Description |
|---|---|---|
segment_id |
str | Segment identifier |
month_on_book |
int | Month on book |
payment_hazard_rate |
float | Monthly payment hazard (0.0–1.0) |
chargeoff_hazard_rate |
float | Monthly charge-off hazard (0.0–1.0) |
Constraint: payment_hazard_rate + chargeoff_hazard_rate ≤ 1.0 in every row.
forecast_balance_flows
from cranalytics.survival_flows import forecast_balance_flows
forecast_df = forecast_balance_flows(known_actuals, curves_df, max_month=36, amtloan=2000.0)
Returns: pd.DataFrame
| Column | Type | Description |
|---|---|---|
segment_id |
str | Segment identifier |
month_on_book |
int | Month on book |
outstanding_balance |
float | Balance at start of period ($) |
payments |
float | Payments received in period ($) |
chargeoffs |
float | Charge-offs in period ($) |
payment_hazard_rate |
float | Hazard rate used |
chargeoff_hazard_rate |
float | Hazard rate used |
forecast_flag |
str | "Actual" or "Forecast" |
outstanding_balance_ratio |
float | outstanding_balance / amtloan |
payments_ratio |
float | payments / amtloan |
chargeoffs_ratio |
float | chargeoffs / amtloan |
Shape: Observed rows plus forecast months after the latest observation, through max_month inclusive; select one segment per call.
Rollforward.run(mode="workflow")
from cranalytics.rollforward import Rollforward
result = Rollforward(flow_df).run(mode="workflow", output_dir=Path("./out"))
Returns: RollforwardWorkflowResult dataclass (frozen)
| Attribute | Type | Description |
|---|---|---|
status |
str | "ok", "data_issues", or "insufficient_data" |
output_dir |
Path | Directory where artifacts were written |
champion |
str | Name of selected champion variant |
challengers |
list[str] | Names of challenger variants |
promotion_reason |
str | Human-readable explanation for champion selection |
run_metadata |
dict | Portfolio KPIs: total_outstanding_balance, total_chargeoffs_last_month, n_segments, max_mob, champion_variant, data_issue_count |
run_summary |
RollforwardRunSummary |
Shared rollforward summary with surface, status, operational_status, issue_count, max_mob, and selection context |
operational_status |
str | Shared status vocabulary: "pass", "attention", or "blocker" |
Side effects — files written to output_dir: See CLI Reference — rollforward output files.
Compatibility note: status keeps the workflow-specific values
"ok", "data_issues", and "insufficient_data". Use result.run_summary
or result.operational_status when you want one shared status vocabulary across
rollforward facade, workflow, and readiness results.
The workflow output directory also writes rollforward_run_summary.json with
the same shared summary payload used by result.run_summary, plus
workflow_run_metadata.json with governance-friendly JSON metadata derived from
that shared summary.
Rollforward.readiness
import pandas as pd
from pathlib import Path
from cranalytics.rollforward import Rollforward
result = Rollforward(flow_df).readiness(output_dir=Path("./readiness_out"))
Returns: ReadinessRunResult dataclass (frozen)
| Attribute | Type | Description |
|---|---|---|
status |
str | "Ready", "Needs Review", or "Not Ready" |
output_dir |
Path | Directory where readiness artifacts were written |
summary |
dict | Readiness scoring summary including overall score, warning/fail counts, and max_mob |
run_summary |
RollforwardRunSummary |
Shared rollforward summary with surface, status, operational_status, issue_count, and output_dir |
operational_status |
str | Shared status vocabulary: "pass", "attention", or "blocker" |
Compatibility note: readiness keeps its onboarding-friendly status labels.
Use result.run_summary or result.operational_status when you want the same
pass/attention/blocker vocabulary used across rollforward surfaces.
The readiness output directory also writes rollforward_run_summary.json with
the same shared summary payload used by result.run_summary, plus
workflow_run_metadata.json with governance-friendly JSON metadata derived from
that shared summary.
Predictive Modeling
predictive.run
from cranalytics import predictive
session = predictive.run(
df,
feature_cols=feature_cols,
target_col="fpf30_flag",
split_col="origination_month",
model_family="logistic",
)
Returns: PredictiveModelingSessionResult — dataclass with attribute access
| Attribute | Type | Description |
|---|---|---|
estimator |
sklearn-compatible | Fitted estimator trained on the validated session frame |
training_metadata |
dict | Training configuration and target metadata |
training_diagnostics |
pd.DataFrame | In-sample diagnostics for the trained estimator |
validation_issues |
pd.DataFrame | Issue table surfaced from the predictive contract |
scored_data |
pd.DataFrame | Scored DataFrame with the requested output column |
backtest |
pd.DataFrame | Fold-level temporal backtest output |
backtest_summary |
pd.DataFrame | Aggregate backtest summary |
Access fields as attributes, e.g. session.backtest. Dict-style access
(session["backtest"]) was removed in 0.3.0.
train_binary_model
from cranalytics.predictive import train_binary_model
estimator, metadata, diagnostics = train_binary_model(
df, feature_cols, target_col="fpf30_flag", model_family="logistic",
)
Returns: tuple[estimator, dict, pd.DataFrame]
| Element | Type | Description |
|---|---|---|
estimator |
sklearn-compatible | Fitted model object with .predict_proba() |
metadata |
dict | target_name, target_type, model_family, feature_cols, train_period, split_strategy, random_state, model_params, calibration_method, label_definition |
diagnostics |
pd.DataFrame | One in-sample row with split="train", auc, gini, ks; these are not holdout metrics or feature importances |
score_model
from cranalytics.predictive import score_model
scored_df = score_model(
df, estimator, feature_cols, output_col="pd_score", prediction_type="probability",
)
Returns: pd.DataFrame — input DataFrame with one column appended.
| New Column | Type | Description |
|---|---|---|
(user-specified output_col) |
float | Predicted probability (0.0–1.0) or class label depending on prediction_type |
run_predictive_backtest / summarize_predictive_backtest
from cranalytics.predictive import (
run_predictive_backtest,
summarize_predictive_backtest,
)
results_df = run_predictive_backtest(
df, feature_cols, target_col, split_col, model_family="logistic",
)
summary_df = summarize_predictive_backtest(results_df)
run_predictive_backtest returns: pd.DataFrame — one row per temporal fold
| Column | Type | Description |
|---|---|---|
split |
int | Requested fold number, starting at 1 |
train_periods, val_periods |
list | Training and validation periods |
n_train, n_val |
int | Labeled observations in the respective windows |
auc, ks, gini |
float | Holdout metrics; NaN for skipped folds |
status |
str | "ok" or "skipped" |
reason |
str | Empty for evaluated folds; reason code for skipped folds |
Possible reasons are no_labeled_training_rows, no_labeled_validation_rows,
single_class_training, and single_class_validation. strict=True raises on
any skipped fold. Temporal ordering alone does not guarantee point-in-time label
availability; prepare training labels as they were known at each decision date.
summarize_predictive_backtest returns: one row overall, or per requested group.
| Column | Type | Description |
|---|---|---|
mean_auc, mean_ks, mean_gini |
float | Unweighted means across evaluated folds |
n_folds |
int | Number of evaluated folds |
n_requested_folds, n_skipped_folds |
int | Requested and skipped fold counts |
status |
str | "ok", "partial", or "insufficient_data" |
All-skipped evaluations return status="insufficient_data" and NaN metric means.
Score Monitoring
compute_psi
from cranalytics.feature_analytics import compute_psi
psi_table, total_psi = compute_psi(expected_scores, actual_scores)
Returns: tuple[pd.DataFrame, float]
DataFrame (PSI table) columns:
| Column | Type | Description |
|---|---|---|
bin |
str | Bin interval, e.g. "(0.1, 0.2]" |
expected_pct |
float | Fraction of expected population in this bin |
actual_pct |
float | Fraction of actual population in this bin |
psi_contribution |
float | Per-bin PSI: (actual − expected) × ln(actual / expected) |
Scalar (total PSI): Sum of all psi_contribution values.
| PSI range | Interpretation |
|---|---|
| < 0.10 | Distribution is stable |
| 0.10 – 0.25 | Moderate shift — investigate |
| > 0.25 | Significant shift — model may need recalibration |
score_performance_monitoring_report
from cranalytics.feature_analytics import score_performance_monitoring_report
report = score_performance_monitoring_report(
scored_df, score_col="pd_score", flag_col="fpf30_flag",
)
Returns: dict with keys:
| Key | Type | Description |
|---|---|---|
discrimination |
dict | auc, gini, ks, spearman_rho |
calibration_table |
pd.DataFrame | Score-to-event calibration table |
calibration_stats |
dict | Calibration method, fit statistics, and rank association |
actual_vs_expected |
pd.DataFrame or None | Breakdown when group_by is supplied |
psi |
dict or None | table and total_psi when baseline_df is supplied |
See also
- Input Data Contracts — what each function expects as input
- API Reference — full function signatures and docstrings