Use the Notebook (Corridor Python Package)

Ask AI

TL;DR

The Notebook is a hosted Jupyter environment inside Corridor (Model → Notebook). It comes with the corridor Python package, which lets you import corridor and work with registered objects (Data Elements, Features, Models, Policies, and more) from code.

What is it?

The Notebook is a hosted JupyterLab environment built into Corridor, under Model → Notebook. It is where you build and experiment in Python without leaving the platform, and it ships with the corridor Python package already installed and authenticated.

The corridor package lets you work with your platform objects (Data Elements, Features, Models, and more) from Python instead of a form: query registered objects and their metadata, read the results of jobs run on the platform, and declare and run new objects on data before registering them back.

Setup

Inside the Corridor Notebook, the package is pre-configured, so you can import it right away:

# import only what you need
import corridor
from corridor import Feature, DataElement, Model, Policy, create_data

# or pull in everything at once
from corridor import *

To switch the workspace you are working in, call set_workspace:

from corridor import set_workspace

set_workspace('Fair Lending Team')

What you can access

Every registered object type has a class in the package. Import the ones you need from corridor:

Object type	Fetch by	What it represents
`DataTable`	`alias` / `name`	A registered raw data table.
`DataElement`	`alias`	A column-backed or aggregated Data Element.
`Feature`	`alias`	A derived Feature.
`Model`	`alias`	A registered Model and its output.
`ModelTransform`	`model` + `alias`	A transform attached to a Model.
`Policy`	`name`	A decision Policy and its workflow blocks.
`GlobalFunction`	`alias`	A reusable function, callable from Python.
`GlobalVariable`	`alias`	A registered global variable.
`RuntimeParameter`	`alias`	A runtime parameter.
`ReasonCode`	`alias` / `name`	A reason code attached to a Model or rule.
`Report`	`name`	A report template and its outputs.
`Dataset`	`alias`	A registered dataset.
`Framework` / `Product`	`name`	Valuation frameworks and products.
`QualityCheck`	`name`	A data quality check.
`MonitoringDashboard`	filters	A monitoring dashboard.
`Entity` / `TableKey`	`name`	Platform entities and their keys.
`User`	`username`	A registered platform user.

Alongside the classes, a few helper functions let you move data in and out of the package:

create_data(...) - run one or more registered objects on data and get back a DataFrame.
get_object_by_alias(...) - fetch an object when you do not know its type.
sync(...) - push a locally declared object to the platform.

Find registered objects

Every registry object type is searched and fetched the same way, so the patterns below apply to Feature, DataElement, Model, Policy, Dataset, and the rest.

List objects with `.all()`

.all() returns a list of objects, optionally filtered:

from corridor import Feature

# every Feature
all_features = Feature.all()

# filter by status, type, keyword, group, permissible purpose, entity, ...
approved  = Feature.all(status='Approved')
numerical = Feature.all(type='Numerical')
income    = Feature.all(contains='income')

Any filter takes a single value or a list, for example Feature.all(type=['Numerical', 'String']). The filters accepted are:

Every class accepts id, alias, name, contains (free-text search), and group.
Feature, DataElement, and Model also accept status, type, keyword, permissible_purpose, and platform_entity.
Model additionally accepts algorithm_type.

For a Model, type and algorithm_type are different filters. type is the model’s output data type - 'Numerical', 'String', 'DateTime', or 'Boolean'. algorithm_type is the kind of model - 'Binary Classification', 'Regression', 'Time Series', or 'Survival'. They are independent: a Binary Classification model usually has a 'Numerical' output (a probability).

from corridor import Model

# probability-output models trained as binary classifiers
Model.all(type='Numerical', algorithm_type='Binary Classification')

Policy.all() is the one exception: it returns a dictionary keyed by policy type, so you index into it by type.

from corridor import Policy

policies = Policy.all()
underwriting = policies['UnderWriting']   # list of UnderWriting policies

Fetch one object

Pass the alias (or name, for Policy and Report) to the class. Without a version, you get the latest approved (current) version; pass version= for a specific one.

from corridor import Feature, Model, Policy

dti      = Feature('debt_to_income')
model    = Model('default_prob')            # Model is fetched by alias
model_v1 = Model('default_prob', version=1)
policy   = Policy('Optimized Return Policy') # Policy is fetched by name

If you do not know an object’s type, use get_object_by_alias:

from corridor import get_object_by_alias

obj = get_object_by_alias('annual_income')  # returns a Feature, DataElement, Model, ...

Read metadata

Once you have an object, read its attributes directly. The set below is common to DataElement, Feature, and Model:

feature = Feature('debt_to_income')

# not sure what an object exposes? ask it:
help(feature)   # or feature? in Jupyter - lists every attribute with a description
dir(feature)    # the attribute and method names

feature.name                 # 'Debt to Income'
feature.alias                # 'debt_to_income'
feature.version              # 1
feature.type                 # 'Numerical'
feature.is_aggregated        # False
feature.current_status       # 'Approved'
feature.platform_entity      # <Entity name="Applicant">
feature.permissible_purpose  # ['Underwriting']
feature.group                # 'Financial'
feature.description          # the registered description
feature.definition           # the definition code (or None)
feature.inputs               # the objects that flow into it
feature.created_by           # 'analyst'
feature.created_date         # datetime

Model adds model-specific attributes:

model = Model('default_prob')

model.alias             # 'default_prob'
model.algorithm_type    # 'Binary Classification'
model.input_type        # how the model was defined - 'python function', 'PMML', ...
model.type              # the output type - 'Numerical', ...
model.dependent_variable
model.reason_codes      # reason codes by input, when configured

Approval history

Objects that go through approval expose their review state:

model = Model('default_prob')

model.current_status       # 'Approved'
model.approval_statuses    # per-responsibility review records (None if no approval workflow)
model.approval_histories   # the full audit trail

for status in model.approval_statuses:
    print(status.responsibility, status.status, status.reviewers)

Access raw data

Read the data behind a registered DataTable with to_spark():

from corridor import DataTable

loans = DataTable('application')

loans.columns          # the DataColumn objects
loans.dtypes           # {column_alias: column_type, ...}
loans.location         # where the data lives
loans.is_primary       # is this a primary table

df = loans.to_spark()  # a PySpark DataFrame
df.select('int_rate', 'loan_amnt').limit(5).show()

For external data, read it with native Spark - point a Spark session at the location:

import pyspark
spark = pyspark.sql.SparkSession.builder.getOrCreate()

application_table = spark.read.parquet('s3a://corridor.dev/master/sampleAppData.parquet')
application_table.limit(5).toPandas()

Work with Policies

A Policy exposes its decision workflow as blocks. Each block may contain segments, and each segment contains rules:

from corridor import Policy

policy = Policy('UW Policy with PD Model and Framework')

policy.type               # 'UnderWriting'
policy.strategies         # the decision blocks in the policy
policy.derived_variables  # derived-variable blocks
policy.decision_workflow  # every block (strategies + derived variables), in order
policy.product            # the Product used

A block prints with its type, for example:

policy.strategies
# [<ApplicationDecisionBlock name="Eligibility and Risk Gate">,
#  <ApplicationDecisionBlock name="Credit Capacity Gate">,
#  <ApplicationDecisionBlock name="Final Approval Gate">]

Walk down from blocks to segments to rules. Not every block is split into segments: when a block has no segmentation, block.segments comes back as None rather than an empty list. The simplest way to reach every rule in a block is block.rules, which returns them all whether the block is segmented or not:

for block in policy.strategies:
    print(f'block: {block.name}')
    for rule in block.rules:          # all rules in the block, segmented or not
        print(f'  rule: {rule.name}')
        print(f'    condition: {rule.definition}')   # the rule's condition tree

If you do want to walk segment by segment, guard for the None case:

for block in policy.strategies:
    if block.segments:                # None when the block is not segmented
        for segment in block.segments:
            print(segment.name, [rule.name for rule in segment.rules])

Each rule’s condition is on rule.definition - a parsed condition tree of operators and operands, not a plain string. Other useful rule attributes are rule.name, rule.colname (its column in job output), rule.actions, and rule.reason_codes.

Use Global Functions

A GlobalFunction is callable straight from Python once you have fetched it:

from corridor import GlobalFunction

# search for one
GlobalFunction.all(contains='exponent')

gf = GlobalFunction('exponential_function_with_rounding')

gf.type         # the return type
gf.arguments    # [InputArgument(alias='base', ...), InputArgument(alias='power', ...)]
gf.definition   # the function body

# call it like any Python function
gf(9, 0.5)      # 3.0

To apply a Global Function across a Spark DataFrame, turn it into a UDF with get_python_function():

import pyspark.sql.functions as F
from pyspark.sql.types import FloatType
from corridor import create_data

# start with a DataFrame that has the 'annual_income' column
data = create_data('annual_income')
# add the second argument the function needs, as a constant column
data = data.withColumn('scaling_factor', F.lit(0.95))

# wrap the Global Function as a Spark UDF, declaring its return type
gf_udf = F.udf(gf.get_python_function(), FloatType())
# apply it row-by-row, passing the two columns as arguments
data = data.withColumn('scaled_income', gf_udf('annual_income', 'scaling_factor'))
data.show(5)

Read job results

Jobs (simulations and what-if analyses) are created and run on the platform, not from the package - there is no method to trigger one from Python. What the package gives you is read access to jobs that have already run, so you can pull their results, logs, and reports into the Notebook.

Objects that support jobs - Model, Feature, DataElement, Policy, Dataset, QualityCheck - expose the same job surface:

model = Model('default_prob')

model.jobs                 # names of every job run on this Model
job = model.get_job('My Simulation')  # fetch one job by name
model.default_simulation   # the simulation marked as default for this Model (else None)

default_simulation returns the one simulation the platform has flagged as this object’s default - the same one the UI treats as its primary simulation - or None if none is set.

Job status and timing

job.status      # SCHEDULED | QUEUED | COMPILING | RUNNING | COMPLETED | FAILED
job.job_type    # Simulation | What-If Analysis | ...
job.runtime     # a timedelta - how long the job took (None if it has not run)
job.logs        # the execution logs as text (None if none attached)
job.is_old      # True if a newer job supersedes this one

Job configuration

Read back exactly how the job was set up:

job.sample_size           # the sample size, or ratio, the job ran on
job.date_filter           # the date field used to filter, if any
job.date_filter_from_date # start of the date filter
job.date_filter_to_date   # end of the date filter
job.runtime_parameters    # {rtp_alias: {proxy_type: value}} supplied to the job
job.dependencies          # global-variable / product-config values supplied to the job
job.report_versions       # the Report objects attached to the job
job.parameters            # the report parameters the job ran with

Load a job’s output data

The quickest way to pull a job’s output into the Notebook is to let the platform write the snippet for you. Open the job on any object that has one, go to the Job Results tab, and in the Output Data panel click </> Code. Choose:

Pandas or Spark - which kind of DataFrame you want.
All, Inputs Only, or Outputs Only - which columns to load.

The panel generates a ready-to-run snippet that reads the job’s result file, for example:

import pandas as pd
result_current = pd.read_parquet("/opt/corridor/data/results/sim_SIM_MODEL_235_job_1440_ent_1_simplified_data.parquet")

Copy it straight into a Notebook cell and run it - result_current is your job output as a DataFrame.

Job results and reports from code

You can also reach a job’s output programmatically. A job exposes its computed results and its rendered reports.

job.results          # the computed result data, keyed by entity label then output alias
job.report_dashboard # the rendered reports (see below)

job.results only works for a completed job that actually produced result data, such as a simulation. On a job that has no results to parse - one that has not finished, or a report-only job - it raises an error instead of returning empty. If you hit an error here, use the </> Code snippet above, or check that the job is COMPLETED and is the kind of job that produces results.

job.report_dashboard returns the job’s reports grouped by the report group they belong to, as a nested dictionary of {group_name: {report_name: JobReport}} (or None if the job has no reports). Printed, it looks like this:

job.report_dashboard
# OrderedDict([
#   ('Corridor Model Output Summary Statistics',
#     OrderedDict([('Summary Statistics', <JobReport #134 name="Summary Statistics">)])),
#   ('Corridor Model Output Histogram',
#     OrderedDict([('Histogram', <JobReport #137 name="Histogram">)])),
#   ...
# ])

Index in with the group name then the report name to get a single JobReport, which you can render inline in the Notebook:

report = job.report_dashboard['Corridor Model Output Histogram']['Histogram']

report.name         # the report's name
report.status       # COMPLETED | ...
report.report_type  # Figure | Downloadable File | Metric
report.raw_data     # renders inline: a Plotly figure, DataFrame, PDF, image, or Markdown
report.logs         # logs attached to this report

What-if jobs

A what-if job references the objects and baseline it was run against:

job.reused_job_object   # the object used for Existing Job Results
job.reused_job          # the specific job that was reused
job.baseline_job        # the baseline job this what-if compares against (else None)

For a Policy what-if job, you can also reach the what-if policy version and resolve a block, segment, or rule from its column name in the job output:

policy_job = Policy('My Policy').get_job('What-If Run')
policy_job.whatif_policy   # the what-if Policy version being tested

# resolve a column like 'block-1', 'segment-1.2', or 'rule-1.1.3' to its registered object
rule = policy_job.get_registered_object_by_colname('rule-1.1.2', 'current')
rule.name

Run objects on data

create_data(...) runs one or more registered objects and returns a DataFrame with the results. Use it to test a definition before or after you register it.

Most of the time you just name the object. create_data looks up which tables it reads, finds those tables on the platform, and loads them for you:

from corridor import create_data

output = create_data('debt_to_income')

Pass data only when you want to supply the tables yourself, such as test data. When you do, give it every table the object reads. The key for each table is the table’s registered alias (the alias it has on the platform), and the value is your DataFrame:

output = create_data(
    'debt_to_income',
    data={'application': application_df},     # key = the table's registered alias
    runtime_params={'default_months': 12},    # values for any runtime parameters
)

If you pass data but leave out a table the object needs, create_data stops and tells you which table is missing. The data you pass is used instead of the platform’s, so it does not read the registered tables in this case.

You can pass Feature, Data Element, Model, or Policy aliases (or the objects themselves). A Policy runs on its own: you cannot mix it with other objects, and only one Policy is allowed per call.

Register/Sync your work back to the platform

Building an object is two steps: declare it in Python, then sync it to the platform. sync creates the object if it is new, or updates your Draft version if it already exists.

Declare an object

Use the .declare(...) decorator (or classmethod) to define a new object. Feature, Model, GlobalFunction, and Report wrap a function whose body becomes the definition; DataElement and DataTable wrap a Spark DataFrame, since they describe existing columns rather than computed logic.

The function-based forms need one thing the DataFrame-based forms do not: an anchor comment.

Feature, Model, Global Function, Report (function-based): the body must contain the line # -- BEGIN DEFINITION --. Everything above it (the signature and docstring) is discarded; everything below it becomes the definition registered on the platform.
Leave it out and sync fails with Expected anchor comment # -- BEGIN DEFINITION --.
Data Element, Data Table (DataFrame-based): no anchor needed.

Declare a Feature from a function. Each argument is resolved to a registered input by its alias, exactly like the Input section of the form:

from corridor import Feature

@Feature.declare(entity='Applicant', name='Debt to Income', group='Financial')
def debt_to_income(annual_income, monthly_debt) -> float:
    """Ratio of monthly debt to annual income."""
    # -- BEGIN DEFINITION --
    return (monthly_debt * 12) / annual_income

Declare a Data Element from a column of a Spark DataFrame (the DataFrame must itself come from DataTable.declare()):

from corridor import DataElement

annual_income = DataElement.declare(
    table=applicant_df,        # a DataFrame from DataTable.declare()
    column='annual_income',    # the column to register
    group='Financial',
)

Declare a Data Table to register a raw table at a location:

from corridor import DataTable

application = DataTable.declare(
    location='s3a://corridor.dev/master/application.parquet',
    alias='application',
    table=application_df,      # a Spark DataFrame whose schema defines the columns
)

Declare a Model with .declare(...), then attach its scoring function. Note that type takes a Python type (like float), not a string - sync maps it to the platform data type:

from corridor import Model

model = Model.declare(
    alias='default_prob',
    entity='Applicant',
    algorithm_type='Binary Classification',  # the kind of model
    type=float,                              # a Python type -> 'Numerical'
    input_type='python function',
)

@model.model_function()
def score(annual_income, debt_to_income) -> float:
    # -- BEGIN DEFINITION --
    ...

Sync the declared object

from corridor import sync

sync(debt_to_income)                          # prompts for confirmation
sync(debt_to_income, skip_confirmation=True)  # push without prompting

sync supports these declared types: Global Function, Global Variable, Runtime Parameter, Report, Data Table, Data Element, Feature, and Model (a Model also syncs its transforms). Datasets, Frameworks, and Policies cannot be synced.

What sync reads from your declaration. What you write in Python becomes the registered definition:

The function name becomes the alias (and the name, unless you passed one to .declare).
The docstring becomes the description.
The return type annotation becomes the data type.
The function body below the anchor becomes the definition.
Any registered object referenced inside the body is picked up as an input - and if that input was also declared locally, it is synced first, automatically.

Values the declare parameters accept:

Parameter	On	Accepts
`algorithm_type`	Model	`'Binary Classification'`, `'Regression'`, `'Time Series'`, or `'Survival'`
`type`	Model, Global Variable, Runtime Parameter	a Python type (see the mapping below)
`input_type`	Model	`'python function'` (attach a `@model.model_function()`) or `'custom'` (attach `@model.init` and `@model.run`)
`entity`	Feature, Model	the name of a registered platform Entity (e.g. `'Applicant'`)
`permissible_purpose`	Feature, DataElement, Model	a list of permissible-purpose tag names; omit to use all
`group`	all	any group name; omit to use `'Default'`

sync maps the Python type you pass as type to a platform data type:

Python type	Data type
`float`, `int`	Numerical
`str`	String
`bool`	Boolean
`datetime.datetime`	DateTime
`list[...]`	the matching Array type

The confirmation flow. sync compares your declaration against the platform, prints a git-style diff of what will change, then asks before writing:

Creating new Feature 'debt_to_income' with values:
[Alias]
  + debt_to_income
[Definition]
  + return (monthly_debt * 12) / annual_income
...
Confirm pushing changes to platform? [y/N]:

Answer y to push, anything else to cancel. If your declaration already matches the platform, it prints Feature 'debt_to_income' up to date with platform and does nothing. Pass skip_confirmation=True to push without the prompt - useful in scripted runs.

Share a draft object with other users or roles before it is approved, without leaving the Notebook:

de = DataElement('annual_income')

de.grant_share_for_user('analyst', access_type='Write')
de.grant_share_for_role('Underwriting', access_type='Read')

de.user_accesses   # who the object is shared with
de.role_accesses

de.revoke_share_for_user('analyst')

What’s Next

Register the objects you build here through the UI: a Data Element, Feature, or Model.
Run a Simulation on a registered object to validate it on a full sample.
Send your work for approval so it can be used outside your draft workspace.

Was this page helpful?

Thanks for the feedback.